feat(integrations): Add tracing to DramatiqIntegration

pyproject.toml

@@ -195,6 +195,10 @@ ignore_missing_imports = true
 module = "agents.*"
 ignore_missing_imports = true
 
+[[tool.mypy.overrides]]
+module = "dramatiq.*"
+ignore_missing_imports = true
+
 #
 # Tool: Flake8
 #

scripts/populate_tox/config.py

@@ -100,6 +100,9 @@
     },
     "dramatiq": {
         "package": "dramatiq",
+        "deps": {
+            "py3.6": ["aiocontextvars"],
+        },
     },
     "falcon": {
         "package": "falcon",

sentry_sdk/consts.py

@@ -836,6 +836,7 @@ class OP:
     QUEUE_TASK_HUEY = "queue.task.huey"
     QUEUE_SUBMIT_RAY = "queue.submit.ray"
     QUEUE_TASK_RAY = "queue.task.ray"
+    QUEUE_TASK_DRAMATIQ = "queue.task.dramatiq"
     SUBPROCESS = "subprocess"
     SUBPROCESS_WAIT = "subprocess.wait"
     SUBPROCESS_COMMUNICATE = "subprocess.communicate"

sentry_sdk/integrations/dramatiq.py

@@ -1,18 +1,34 @@
 import json
 
 import sentry_sdk
-from sentry_sdk.integrations import Integration
+from sentry_sdk.consts import OP, SPANSTATUS
+from sentry_sdk.api import continue_trace, get_baggage, get_traceparent
+from sentry_sdk.integrations import Integration, DidNotEnable
 from sentry_sdk.integrations._wsgi_common import request_body_within_bounds
+from sentry_sdk.tracing import (
+    BAGGAGE_HEADER_NAME,
+    SENTRY_TRACE_HEADER_NAME,
+    TransactionSource,
+)
 from sentry_sdk.utils import (
     AnnotatedValue,
     capture_internal_exceptions,
     event_from_exception,
+    ContextVar,
+    HAS_REAL_CONTEXTVARS,
+    CONTEXTVARS_ERROR_MESSAGE,
 )
+from typing import TypeVar
+
+R = TypeVar("R")
 
-from dramatiq.broker import Broker  # type: ignore
-from dramatiq.message import Message  # type: ignore
-from dramatiq.middleware import Middleware, default_middleware  # type: ignore
-from dramatiq.errors import Retry  # type: ignore
+try:
+    from dramatiq.broker import Broker
+    from dramatiq.middleware import Middleware, default_middleware
+    from dramatiq.errors import Retry
+    from dramatiq.message import Message
+except ImportError:
+    raise DidNotEnable("Dramatiq is not installed")
 
 from typing import TYPE_CHECKING
 
@@ -38,6 +54,12 @@ class DramatiqIntegration(Integration):
     @staticmethod
     def setup_once():
         # type: () -> None
+        if not HAS_REAL_CONTEXTVARS:
+            raise DidNotEnable(
+                "The dramatiq integration for Sentry requires Python 3.7+ "
+                " or aiocontextvars package." + CONTEXTVARS_ERROR_MESSAGE
+            )
+
         _patch_dramatiq_broker()
 
 
@@ -85,50 +107,78 @@ class SentryMiddleware(Middleware):  # type: ignore[misc]
     DramatiqIntegration.
     """
 
+    _transaction = ContextVar("_transaction")
+
+    def before_enqueue(self, broker, message, delay):
+        # type: (Broker, Message[R], int) -> None
+        message.options["sentry_headers"] = {
+            BAGGAGE_HEADER_NAME: get_baggage(),
+            SENTRY_TRACE_HEADER_NAME: get_traceparent(),
+        }
+
     def before_process_message(self, broker, message):
-        # type: (Broker, Message) -> None
+        # type: (Broker, Message[R]) -> None
         integration = sentry_sdk.get_client().get_integration(DramatiqIntegration)
         if integration is None:
             return
 
-        message._scope_manager = sentry_sdk.new_scope()
-        message._scope_manager.__enter__()
-
         scope = sentry_sdk.get_current_scope()
-        scope.set_transaction_name(message.actor_name)
-        scope.set_extra("dramatiq_message_id", message.message_id)
+        scope.set_tag("dramatiq_message_id", message.message_id)
+        scope.clear_breadcrumbs()
         scope.add_event_processor(_make_message_event_processor(message, integration))
 
+        transaction = continue_trace(
+            message.options.get("sentry_headers") or {},
+            name=message.actor_name,
+            op=OP.QUEUE_TASK_DRAMATIQ,
+            source=TransactionSource.TASK,
+            # origin=DramatiqIntegration.origin,
+        )
+        transaction.set_status(SPANSTATUS.OK)
+        sentry_sdk.start_transaction(
+            transaction,
+            name=message.actor_name,
+            op=OP.QUEUE_TASK_DRAMATIQ,
+            source=TransactionSource.TASK,
+        )
+        transaction.__enter__()
+        self._transaction.set(transaction)
+
     def after_process_message(self, broker, message, *, result=None, exception=None):
-        # type: (Broker, Message, Any, Optional[Any], Optional[Exception]) -> None
-        integration = sentry_sdk.get_client().get_integration(DramatiqIntegration)
-        if integration is None:
-            return
+        # type: (Broker, Message[R], Optional[Any], Optional[Exception]) -> None
 
         actor = broker.get_actor(message.actor_name)
         throws = message.options.get("throws") or actor.options.get("throws")
 
-        try:
-            if (
-                exception is not None
-                and not (throws and isinstance(exception, throws))
-                and not isinstance(exception, Retry)
-            ):
-                event, hint = event_from_exception(
-                    exception,
-                    client_options=sentry_sdk.get_client().options,
-                    mechanism={
-                        "type": DramatiqIntegration.identifier,
-                        "handled": False,
-                    },
-                )
-                sentry_sdk.capture_event(event, hint=hint)
-        finally:
-            message._scope_manager.__exit__(None, None, None)
+        transaction = self._transaction.get(None)
+        if not transaction:
+            return None
+
+        is_event_capture_required = (
+            exception is not None
+            and not (throws and isinstance(exception, throws))
+            and not isinstance(exception, Retry)
+        )
+        if not is_event_capture_required:
+            # normal transaction finish
+            transaction.__exit__(None, None, None)
+            return
+
+        event, hint = event_from_exception(
+            exception,  # type: ignore[arg-type]
+            client_options=sentry_sdk.get_client().options,
+            mechanism={
+                "type": DramatiqIntegration.identifier,
+                "handled": False,
+            },
+        )
+        sentry_sdk.capture_event(event, hint=hint)
+        # transaction error
+        transaction.__exit__(type(exception), exception, None)
 
 
 def _make_message_event_processor(message, integration):
-    # type: (Message, DramatiqIntegration) -> Callable[[Event, Hint], Optional[Event]]
+    # type: (Message[R], DramatiqIntegration) -> Callable[[Event, Hint], Optional[Event]]
 
     def inner(event, hint):
         # type: (Event, Hint) -> Optional[Event]
@@ -142,7 +192,7 @@ def inner(event, hint):
 
 class DramatiqMessageExtractor:
     def __init__(self, message):
-        # type: (Message) -> None
+        # type: (Message[R]) -> None
         self.message_data = dict(message.asdict())
 
     def content_length(self):

tests/integrations/dramatiq/test_dramatiq.py

@@ -5,12 +5,21 @@
 from dramatiq.brokers.stub import StubBroker
 
 import sentry_sdk
+from sentry_sdk.tracing import TransactionSource
+from sentry_sdk import start_transaction
+from sentry_sdk.consts import SPANSTATUS
 from sentry_sdk.integrations.dramatiq import DramatiqIntegration
 
+# from sentry_sdk.integrations.logging import LoggingIntegration
 
-@pytest.fixture
-def broker(sentry_init):
-    sentry_init(integrations=[DramatiqIntegration()])
+
+@pytest.fixture(scope="function")
+def broker(request, sentry_init):
+    sentry_init(
+        integrations=[DramatiqIntegration()],
+        traces_sample_rate=getattr(request, "param", None),
+        # disabled_integrations=[LoggingIntegration()],
+    )
     broker = StubBroker()
     broker.emit_after("process_boot")
     dramatiq.set_broker(broker)
@@ -44,22 +53,61 @@ def dummy_actor(x, y):
     assert exception["type"] == "ZeroDivisionError"
 
 
-def test_that_actor_name_is_set_as_transaction(broker, worker, capture_events):
+@pytest.mark.parametrize(
+    "broker,expected_span_status",
+    [
+        (1.0, SPANSTATUS.INTERNAL_ERROR),
+        (1.0, SPANSTATUS.OK),
+    ],
+    ids=["error", "success"],
+    indirect=["broker"],
+)
+def test_task_transaction(broker, worker, capture_events, expected_span_status):
     events = capture_events()
+    task_fails = expected_span_status == SPANSTATUS.INTERNAL_ERROR
 
     @dramatiq.actor(max_retries=0)
     def dummy_actor(x, y):
         return x / y
 
-    dummy_actor.send(1, 0)
+    dummy_actor.send(1, int(not task_fails))
     broker.join(dummy_actor.queue_name)
     worker.join()
 
+    if task_fails:
+        error_event = events.pop(0)
+        exception = error_event["exception"]["values"][0]
+        assert exception["type"] == "ZeroDivisionError"
+        # todo: failed assert. Logging instead of dramatiq
+        # assert exception["mechanism"]["type"] == DramatiqIntegration.identifier
+
     (event,) = events
+    assert event["type"] == "transaction"
     assert event["transaction"] == "dummy_actor"
+    assert event["transaction_info"] == {"source": TransactionSource.TASK}
+    assert event["contexts"]["trace"]["status"] == expected_span_status
 
 
-def test_that_dramatiq_message_id_is_set_as_extra(broker, worker, capture_events):
+@pytest.mark.parametrize("broker", [1.0], indirect=True)
+def test_dramatiq_propagate_trace(broker, worker, capture_events):
+    events = capture_events()
+
+    @dramatiq.actor(max_retries=0)
+    def propagated_trace_task():
+        pass
+
+    with start_transaction() as outer_transaction:
+        propagated_trace_task.send()
+        broker.join(propagated_trace_task.queue_name)
+        worker.join()
+
+    assert (
+        events[0]["transaction"] == "propagated_trace_task"
+    )  # the "inner" transaction
+    assert events[0]["contexts"]["trace"]["trace_id"] == outer_transaction.trace_id
+
+
+def test_that_dramatiq_message_id_is_set_as_tag(broker, worker, capture_events):
     events = capture_events()
 
     @dramatiq.actor(max_retries=0)
@@ -72,13 +120,14 @@ def dummy_actor(x, y):
     worker.join()
 
     event_message, event_error = events
-    assert "dramatiq_message_id" in event_message["extra"]
-    assert "dramatiq_message_id" in event_error["extra"]
+
+    assert "dramatiq_message_id" in event_message["tags"]
+    assert "dramatiq_message_id" in event_error["tags"]
     assert (
-        event_message["extra"]["dramatiq_message_id"]
-        == event_error["extra"]["dramatiq_message_id"]
+        event_message["tags"]["dramatiq_message_id"]
+        == event_error["tags"]["dramatiq_message_id"]
     )
-    msg_ids = [e["extra"]["dramatiq_message_id"] for e in events]
+    msg_ids = [e["tags"]["dramatiq_message_id"] for e in events]
     assert all(uuid.UUID(msg_id) and isinstance(msg_id, str) for msg_id in msg_ids)
 
 

tox.ini

@@ -647,6 +647,7 @@ deps =
     dramatiq-v1.12.3: dramatiq==1.12.3
     dramatiq-v1.15.0: dramatiq==1.15.0
     dramatiq-v1.18.0: dramatiq==1.18.0
+    {py3.6}-dramatiq: aiocontextvars
 
     huey-v2.1.3: huey==2.1.3
     huey-v2.2.0: huey==2.2.0