Skip to content

feat(integrations): Add tracing to DramatiqIntegration #4571

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 13 commits into from
Oct 3, 2025
Merged

feat(integrations): Add tracing to DramatiqIntegration #4571

Changes from 1 commit
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
edee89d
feat(integrations): Add tracing to DramatiqIntegration
Jul 10, 2025
5562a9c
feat(integrations): Add trace propagation to DramatiqIntegration
Jul 15, 2025
863d16b
Merge branch 'getsentry:master' into feat-dramatiq-transaction-support
Igreh Jul 17, 2025
5674a62
Merge branch 'master' into feat-dramatiq-transaction-support
antonpirker Sep 11, 2025
9764f19
Merge branch 'getsentry:master' into feat-dramatiq-transaction-support
Igreh Sep 15, 2025
7156ac4
feat(integration): Fix dramatiq py3.6 and linters issues (#3454)
Sep 15, 2025
39e3a50
feat(integrations): Start new trace on retrying
Sep 16, 2025
a41b477
Merge branch 'master' into feat-dramatiq-transaction-support
Igreh Sep 20, 2025
4746db6
feat(integrations): Remove ContextVar usage.
Oct 2, 2025
a5e9ad0
Merge branch 'master' into feat-dramatiq-transaction-support
sentrivana Oct 2, 2025
1357bea
feat(integrations): Isolation scope
Oct 3, 2025
39d3532
feat(integrations): Fix scope usage
Oct 3, 2025
db239dc
Merge branch 'master' into feat-dramatiq-transaction-support
sentrivana Oct 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 38 additions & 21 deletions sentry_sdk/integrations/dramatiq.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
import json
import contextvars

import sentry_sdk
from sentry_sdk.consts import OP
from sentry_sdk.integrations import Integration
from sentry_sdk.integrations._wsgi_common import request_body_within_bounds
from sentry_sdk.tracing import TransactionSource
from sentry_sdk.utils import (
AnnotatedValue,
capture_internal_exceptions,
Expand All @@ -18,6 +21,7 @@

if TYPE_CHECKING:
from typing import Any, Callable, Dict, Optional, Union
from sentry_sdk.tracing import Transaction
from sentry_sdk._types import Event, Hint


Expand Down Expand Up @@ -85,20 +89,27 @@ class SentryMiddleware(Middleware): # type: ignore[misc]
DramatiqIntegration.
"""

# type: contextvars.ContextVar[Transaction]
_transaction = contextvars.ContextVar("_transaction", default=None)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are we using a ContextVar here, instead of just a simple instance variable on the SentryMiddleware?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To isolate variable between threads:

  1. Dramatiq can be started in multithreading mode:
    dramatiq --processes 1 --threads 5 ...
  2. And I was inspired by official middleware:
    CurrentMessageMiddleware

I will try recheck if it's actually needed

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Confirmed.
One SentryMiddleware instance will be shared between threads. So, isolating is necessary.
Previous (actually current) realisation adds _scope_manager to message, but I think ContextVar is much cleaner solution


def before_process_message(self, broker, message):
# type: (Broker, Message) -> 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__()
Comment on lines -94 to -95
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We still need some sort of scope management in order to make sure the data we collect about tasks is isolated.

The general rule of thumb is: if you start a transaction, you should start it in a new isolation scope. See for example huey.

So we should start an isolation scope right after the initial if integration is None: return check with

scope = sentry_sdk.isolation_scope()
message._scope_manager = scope
scope.__enter__()

Everything that we do on the scope later in the function can stay, but it should be done on the isolation scope, not current scope as before.

And finally, we need to __exit__ the saved scope in after_process_message with message._scope_manager.__exit__(None, None, None).

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.
but please recheck it)


scope = sentry_sdk.get_current_scope()
scope.set_transaction_name(message.actor_name)
scope.set_extra("dramatiq_message_id", message.message_id)
scope.add_event_processor(_make_message_event_processor(message, integration))

transaction = sentry_sdk.start_transaction(
name=message.actor_name,
op=OP.QUEUE_PROCESS,
source=TransactionSource.TASK,
)
transaction.__enter__()
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: Transaction Initialization Redundancy

The transaction initialization in before_process_message uses an incorrect pattern. It creates a transaction with continue_trace(), then passes this existing transaction object to sentry_sdk.start_transaction(), which is designed to create new transactions. This leads to redundant initialization, a manual transaction.__enter__() call, and causes the origin parameter set by continue_trace() to be lost.

Fix in Cursor Fix in Web

self._transaction.set(transaction)
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: Dramatiq Integration Fails Transaction Initialization

The Dramatiq integration incorrectly initializes transactions. It calls continue_trace() to create a transaction, then redundantly calls sentry_sdk.start_transaction() with that same transaction object, and finally manually calls transaction.__enter__(). This mixed lifecycle management can lead to improper transaction initialization, inconsistent state, or broken distributed tracing.

Fix in Cursor Fix in Web


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)
Expand All @@ -108,23 +119,29 @@ def after_process_message(self, broker, message, *, result=None, 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()

Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Bug: SentryMiddleware Scope Leak

The SentryMiddleware creates a scope leak. In after_process_message, if no transaction is found, the isolation scope entered in before_process_message is not properly exited, leading to a resource leak.

Fix in Cursor Fix in Web

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,
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):
Expand Down
Loading