Merge pull request #539 from joselsegura/sentry_broker

New monitored broker in order to better handle the exceptions

Author: joselsegura

ccx_messaging/consumers/idp_kafka_consumer.py

@@ -5,10 +5,10 @@
 import re
 
 from confluent_kafka import Message
-from insights import dr
 
 from ccx_messaging.consumers.kafka_consumer import KafkaConsumer
 from ccx_messaging.error import CCXMessagingError
+from ccx_messaging.monitored_broker import SentryMonitoredBroker
 
 
 # Path example: <org_id>/<cluster_id>/<year><month><day><time>-<id>
@@ -80,7 +80,7 @@ def deserialize(self, msg):
     def create_broker(self, input_msg):
         """Create a suitable `Broker` to be pass arguments to the `Engine`."""
         path = input_msg.get("path")
-        broker = dr.Broker()
+        broker = SentryMonitoredBroker()
         broker["original_path"] = path
 
         if "cluster_id" in input_msg:

ccx_messaging/consumers/kafka_consumer.py

@@ -13,12 +13,12 @@
     Producer,
     TIMESTAMP_NOT_AVAILABLE,
 )
-from insights import dr
 from insights.core.exceptions import InvalidContentType
 from insights_messaging.consumers import Consumer
 
 from ccx_messaging.error import CCXMessagingError
 from ccx_messaging.ingress import parse_ingress_message
+from ccx_messaging.monitored_broker import SentryMonitoredBroker
 from ccx_messaging.utils.kafka_config import kafka_producer_config_cleanup
 
 
@@ -249,7 +249,7 @@ def check_last_message_received_time(self):
 
     def create_broker(self, input_msg):
         """Create a suitable `Broker` to be pass arguments to the `Engine`."""
-        broker = dr.Broker()
+        broker = SentryMonitoredBroker()
 
         # Some engines expect some data for its own usage, like the following:
         # (NOTE: The fields should be always present because of schema validation

ccx_messaging/consumers/rules_results_consumer.py

@@ -9,12 +9,12 @@
 
 
 from confluent_kafka import KafkaException
-from insights import dr
 from insights.core.exceptions import InvalidContentType
 
 from ccx_messaging.consumers.kafka_consumer import KafkaConsumer
 from ccx_messaging.error import CCXMessagingError
 from ccx_messaging.internal_pipeline import parse_rules_results_msg
+from ccx_messaging.monitored_broker import SentryMonitoredBroker
 
 
 LOG = logging.getLogger(__name__)
@@ -89,9 +89,9 @@ def deserialize(self, msg):
         LOG.debug("JSON message deserialized (%s): %s", self.log_pattern, deseralized_msg)
         return deseralized_msg
 
-    def create_broker(self, input_msg: dict[str, Any]) -> dr.Broker:
+    def create_broker(self, input_msg: dict[str, Any]) -> SentryMonitoredBroker:
         """Create a suitable `Broker`."""
-        broker = dr.Broker()
+        broker = SentryMonitoredBroker()
         broker["cluster_id"] = input_msg["metadata"]["cluster_id"]
         broker["report_path"] = self.create_report_path(input_msg)
         return broker

ccx_messaging/monitored_broker.py

@@ -0,0 +1,19 @@
+"""Utility broker to improve Ssentry exception handling."""
+
+from insights.core.dr import Broker
+from insights.core.dr import MissingRequirements
+from insights.core.spec_factory import ContentException
+
+from sentry_sdk import capture_exception
+
+
+class SentryMonitoredBroker(Broker):
+    """Implementation of Broker with custom Sentry capturing logic."""
+
+    def add_exception(self, component, ex, tb=None):
+        """Check added exception in order to use it with Sentry or not."""
+        super().add_exception(component, ex, tb)
+
+        # prevent MissingRequirements and ContentException from being sent to sentry
+        if not isinstance(ex, (MissingRequirements, ContentException)):
+            capture_exception(ex)

test/monitored_broker_test.py

@@ -0,0 +1,48 @@
+"""Tests for the utility class SentryMonitoredBroker."""
+
+import time
+from uuid import uuid4
+from unittest.mock import patch
+
+import pytest
+from insights.core.dr import MissingRequirements
+from insights.core.spec_factory import ContentException
+
+
+from ccx_messaging.monitored_broker import SentryMonitoredBroker
+
+
+def test_usage():
+    """Check that normal usage for the broker is kept across inheritance."""
+    broker = SentryMonitoredBroker()
+    cluster_id = uuid4()
+    timestamp = time.time()
+
+    broker["cluster_id"] = cluster_id
+    broker["timestamp"] = timestamp
+
+    assert broker["cluster_id"] == cluster_id
+    assert broker["timestamp"] == timestamp
+
+
+@patch("ccx_messaging.monitored_broker.capture_exception")
+def test_add_exception(sentry_capture_exception_mock):
+    """Check add_exception behavior."""
+    broker = SentryMonitoredBroker()
+
+    broker.add_exception(None, KeyboardInterrupt, None)
+    assert sentry_capture_exception_mock.called
+
+
+NO_CAPTURE_EXCEPTIONS = [MissingRequirements, ContentException]
+
+
+@pytest.mark.parametrize("exception", NO_CAPTURE_EXCEPTIONS)
+def test_add_exception_no_capture(exception):
+    """Check that selected exceptions are not captured by Sentry."""
+    broker = SentryMonitoredBroker()
+
+    with patch("ccx_messaging.monitored_broker.capture_exception") as capture_exception_mock:
+        # Adding an argument to constructor because it is required by some of the exceptions
+        broker.add_exception(None, exception("some message"), None)
+        assert not capture_exception_mock.called