Merge pull request #757 from matysek/mzibrick-ccxdev15098-insights_core-workaround

[CCXDEV-16284] Implement memleak workaround from insights-core

Author: matysek

ccx_messaging/monitored_broker.py

@@ -1,4 +1,4 @@
-"""Utility broker to improve Ssentry exception handling."""
+"""Utility broker to improve Sentry exception handling."""
 
 import logging
 
@@ -22,4 +22,40 @@ def add_exception(self, component, ex, tb=None):
         # prevent MissingRequirements and ContentException from being sent to sentry
         if not isinstance(ex, (MissingRequirements, ContentException)):
             self.logger.debug("Sending exception to Sentry: %s", type(ex))
-            capture_exception(ex)
+            # Sentry failure must not prevent traceback cleanup below;
+            # otherwise the memory leak silently returns in production.
+            try:
+                capture_exception(ex)
+            except Exception:
+                self.logger.exception("Failed to send exception to Sentry")
+
+        # Break circular reference chain to prevent memory leak.
+        # TODO: remove when this is merged: https://github.com/RedHatInsights/insights-core/pull/4763
+        # Python attaches a live traceback to ex.__traceback__ which
+        # references the stack frame (including the broker), creating:
+        #   broker.exceptions -> ex -> __traceback__ -> frame -> broker
+        # The formatted traceback string is already stored in
+        # broker.tracebacks via super().add_exception(), and
+        # capture_exception() above receives the live traceback for
+        # error-tracking systems (Sentry/GlitchTip), so no debugging
+        # information is lost.
+        if isinstance(ex, BaseException):
+            # Defensive check: add_exception() is public API that could receive
+            # invalid input from plugin code (e.g., strings, None). Only process
+            # actual exceptions to prevent crashes downstream.
+            # Walk the full exception chain (explicit __cause__ from
+            # `raise X from Y` and implicit __context__) clearing
+            # __traceback__ at every level. Uses iterative depth-first
+            # traversal (not recursion) to handle arbitrarily deep chains
+            # without stack overflow. A `seen` set guards against cycles
+            # so the loop always terminates.
+            seen = set()
+            stack = [ex]
+            while stack:
+                cur = stack.pop()
+                if not isinstance(cur, BaseException) or id(cur) in seen:
+                    continue
+                seen.add(id(cur))
+                cur.__traceback__ = None
+                stack.append(getattr(cur, "__cause__", None))
+                stack.append(getattr(cur, "__context__", None))

pyproject.toml

@@ -28,7 +28,7 @@ dependencies = [
     "boto3>=1.42.91,<1.42.92",  # limited due to incompatibility between aiobotocore and newer versions of boto
     "confluent-kafka>=2.0.0",
     "insights-core>=3.1.2",
-    "insights-core-messaging>=1.2.15",
+    "insights-core-messaging @ git+https://github.com/RedHatInsights/insights-core-messaging@1.2.21",
     "jsonschema>=4.0.0",
     "prometheus-client>=0.16.0",
     "python-json-logger>=2.0.7,<5",

test/test_traceback_leak.py

@@ -0,0 +1,299 @@
+"""Tests for memory leak fix in SentryMonitoredBroker traceback handling."""
+
+import gc
+import traceback
+import weakref
+from unittest.mock import patch
+
+from insights.core.dr import MissingRequirements
+from insights.core.spec_factory import ContentException
+
+from ccx_messaging.monitored_broker import SentryMonitoredBroker
+
+
+def _make_exception_with_traceback(exc_type=Exception, msg="test exception"):
+    """Raise and catch an exception so it has a real __traceback__."""
+    try:
+        raise exc_type(msg)
+    except exc_type as ex:
+        tb_string = traceback.format_exc()
+        return ex, tb_string
+
+
+def test_traceback_cleared_keyboard_interrupt():
+    """Verify __traceback__ is cleared for BaseException subclasses.
+
+    The implementation uses `isinstance(ex, BaseException)` (not Exception),
+    which includes system exceptions like KeyboardInterrupt, SystemExit, and
+    GeneratorExit. These can occur during graceful shutdown, signal handling,
+    or async task cancellation. This test verifies the full BaseException
+    hierarchy is handled correctly, not just regular exceptions.
+
+    KeyboardInterrupt exception is in some frameworks mapped to signal SIGTERM
+    for graceful shutdown.
+    """
+    broker = SentryMonitoredBroker()
+    ex, tb = _make_exception_with_traceback(KeyboardInterrupt, "interrupt")
+    assert ex.__traceback__ is not None
+
+    with patch("ccx_messaging.monitored_broker.capture_exception"):
+        broker.add_exception("component", ex, tb)
+
+    assert ex.__traceback__ is None
+
+
+def test_formatted_traceback_preserved():
+    """The formatted traceback string must still be in broker.tracebacks."""
+    broker = SentryMonitoredBroker()
+    ex, tb = _make_exception_with_traceback(Exception, "preserved test")
+
+    with patch("ccx_messaging.monitored_broker.capture_exception"):
+        broker.add_exception("component", ex, tb)
+
+    assert broker.tracebacks[ex] == tb
+    assert "preserved test" in broker.tracebacks[ex]
+
+
+def test_sentry_called_with_traceback():
+    """capture_exception must be called while __traceback__ is still set."""
+    broker = SentryMonitoredBroker()
+    ex, tb = _make_exception_with_traceback(Exception, "sentry test")
+
+    traceback_was_set = []
+
+    def mock_capture(captured_ex):
+        traceback_was_set.append(captured_ex.__traceback__ is not None)
+
+    with patch(
+        "ccx_messaging.monitored_broker.capture_exception",
+        side_effect=mock_capture,
+    ):
+        broker.add_exception("component", ex, tb)
+
+    assert traceback_was_set == [True], (
+        "capture_exception must be called before __traceback__ is cleared"
+    )
+    assert ex.__traceback__ is None
+
+
+def test_sentry_not_called_for_excluded_types():
+    """MissingRequirements and ContentException skip Sentry but still clear traceback."""
+    broker = SentryMonitoredBroker()
+
+    # Don't use _make_exception_with_traceback helper here because these
+    # insights-core exception types have different constructor signatures:
+    # MissingRequirements expects a list, ContentException expects a string.
+    # The inline approach is clearer than abstracting a helper with complex
+    # parameter handling for just two test cases.
+    for exc_type, msg in [
+        (MissingRequirements, ["req"]),
+        (ContentException, "content"),
+    ]:
+        ex = exc_type(msg)
+        try:
+            raise ex
+        except type(ex):
+            tb = traceback.format_exc()
+
+        with patch("ccx_messaging.monitored_broker.capture_exception") as mock:
+            broker.add_exception("component", ex, tb)
+            assert not mock.called
+            assert ex.__traceback__ is None
+
+
+def test_traceback_cleared_even_when_sentry_fails():
+    """Traceback must be cleared even if capture_exception raises."""
+    broker = SentryMonitoredBroker()
+    ex, tb = _make_exception_with_traceback(Exception, "sentry failure")
+
+    with patch(
+        "ccx_messaging.monitored_broker.capture_exception",
+        side_effect=RuntimeError("sentry unavailable"),
+    ):
+        broker.add_exception("component", ex, tb)
+
+    assert ex.__traceback__ is None
+
+
+def test_chained_exception_tracebacks():
+    """Chained exception tracebacks (__cause__, __context__) should also be cleared."""
+    broker = SentryMonitoredBroker()
+    caught = None
+    try:
+        try:
+            raise ValueError("original")
+        except ValueError as orig:
+            raise RuntimeError("wrapper") from orig
+    except RuntimeError as ex:
+        tb = traceback.format_exc()
+        caught = ex
+
+    assert caught.__traceback__ is not None
+    assert caught.__cause__.__traceback__ is not None
+
+    with patch("ccx_messaging.monitored_broker.capture_exception"):
+        broker.add_exception("component", caught, tb)
+
+    assert caught.__traceback__ is None
+    assert caught.__cause__.__traceback__ is None
+
+
+def test_implicit_chained_exception_context():
+    """Implicit chaining (__context__) should also have traceback cleared."""
+    broker = SentryMonitoredBroker()
+    caught = None
+    try:
+        try:
+            raise ValueError("original")
+        except ValueError:
+            raise RuntimeError("during handling")  # noqa: B904
+    except RuntimeError as ex:
+        tb = traceback.format_exc()
+        caught = ex
+
+    assert caught.__traceback__ is not None
+    assert caught.__context__.__traceback__ is not None
+
+    with patch("ccx_messaging.monitored_broker.capture_exception"):
+        broker.add_exception("component", caught, tb)
+
+    assert caught.__traceback__ is None
+    assert caught.__context__.__traceback__ is None
+
+
+def test_deep_chained_exception_tracebacks():
+    """Multi-level exception chains must have all tracebacks cleared."""
+    broker = SentryMonitoredBroker()
+    caught = None
+    try:
+        try:
+            try:
+                raise ValueError("root cause")
+            except ValueError as e1:
+                raise TypeError("middle") from e1
+        except TypeError as e2:
+            raise RuntimeError("outer") from e2
+    except RuntimeError as ex:
+        tb = traceback.format_exc()
+        caught = ex
+
+    assert caught.__traceback__ is not None
+    assert caught.__cause__.__traceback__ is not None
+    assert caught.__cause__.__cause__.__traceback__ is not None
+
+    with patch("ccx_messaging.monitored_broker.capture_exception"):
+        broker.add_exception("component", caught, tb)
+
+    assert caught.__traceback__ is None
+    assert caught.__cause__.__traceback__ is None
+    assert caught.__cause__.__cause__.__traceback__ is None
+
+
+def test_multiple_exceptions_all_tracebacks_cleared():
+    """All tracebacks must be cleared when multiple exceptions are added to one broker."""
+    broker = SentryMonitoredBroker()
+    exceptions = []
+
+    for i in range(5):
+        ex, tb = _make_exception_with_traceback(Exception, f"error {i}")
+        with patch("ccx_messaging.monitored_broker.capture_exception"):
+            broker.add_exception(f"component_{i}", ex, tb)
+        exceptions.append(ex)
+
+    for ex in exceptions:
+        assert ex.__traceback__ is None, f"Traceback not cleared for: {ex}"
+
+
+def test_exception_without_traceback():
+    """Exceptions that were never raised (no __traceback__) should not crash."""
+    broker = SentryMonitoredBroker()
+    ex = Exception("never raised")
+    assert ex.__traceback__ is None
+
+    with patch("ccx_messaging.monitored_broker.capture_exception"):
+        broker.add_exception("component", ex, None)
+
+    assert ex.__traceback__ is None
+
+
+def test_broker_collected_after_add_exception():
+    """Verify that brokers are collected by reference counting alone.
+
+    The memory leak manifests as circular references that prevent garbage
+    collection: broker.exceptions -> ex -> __traceback__ -> frame -> broker.
+
+    This test disables Python's cyclic garbage collector to verify that
+    reference counting ALONE (immediate collection) successfully reclaims
+    brokers after exceptions are added. This proves that clearing
+    __traceback__ breaks the circular reference chain.
+
+    Without the fix, brokers would remain alive until the next GC cycle,
+    causing memory to accumulate over time in production (where thousands
+    of exceptions are processed before GC runs).
+    """
+    n_brokers = 5
+    refs = []
+
+    gc.collect()
+    was_enabled = gc.isenabled()
+    gc.disable()
+
+    try:
+        for _ in range(n_brokers):
+            broker = SentryMonitoredBroker()
+            ex, tb = _make_exception_with_traceback(Exception, "gc test")
+
+            with patch("ccx_messaging.monitored_broker.capture_exception"):
+                broker.add_exception("component", ex, tb)
+
+            refs.append(weakref.ref(broker))
+            del broker, ex, tb
+
+        collected = sum(1 for ref in refs if ref() is None)
+        assert collected == n_brokers, (
+            f"Only {collected}/{n_brokers} brokers were collected "
+            "by reference counting alone. Remaining brokers are held "
+            "by circular references from ex.__traceback__ -> frame -> broker."
+        )
+    finally:
+        if was_enabled:
+            gc.enable()
+        gc.collect()
+
+
+def test_broker_collected_with_chained_exceptions():
+    """Brokers must be collectable even when exceptions have __cause__ chains."""
+    n_brokers = 5
+    refs = []
+
+    gc.collect()
+    was_enabled = gc.isenabled()
+    gc.disable()
+
+    try:
+        for _ in range(n_brokers):
+            broker = SentryMonitoredBroker()
+            caught = None
+            try:
+                try:
+                    raise ValueError("cause")
+                except ValueError as orig:
+                    raise RuntimeError("effect") from orig
+            except RuntimeError as ex:
+                tb = traceback.format_exc()
+                caught = ex
+
+            with patch("ccx_messaging.monitored_broker.capture_exception"):
+                broker.add_exception("component", caught, tb)
+
+            refs.append(weakref.ref(broker))
+            del broker, caught, tb
+
+        collected = sum(1 for ref in refs if ref() is None)
+        assert collected == n_brokers, (
+            f"Only {collected}/{n_brokers} brokers collected with chained exceptions"
+        )
+    finally:
+        if was_enabled:
+            gc.enable()
+        gc.collect()