feat(integrations): Add tracing to DramatiqIntegration #4571

Igreh · 2025-07-10T16:52:16Z

Adds tracing support to DramatiqIntegration #3454

Adds tracing support to DramatiqIntegration getsentry#3454

szokeasaurusrex

Hi @Igreh, I took a quick glance at your PR – at a high level this seems reasonable; however, before I do a full review, I would appreciate if you could please add some tests for these changes, since we need tests before we can consider merging this.

szokeasaurusrex · 2025-07-14T08:35:45Z

sentry_sdk/integrations/dramatiq.py

    """

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


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

To isolate variable between threads:

Dramatiq can be started in multithreading mode:
dramatiq --processes 1 --threads 5 ...

And I was inspired by official middleware:
CurrentMessageMiddleware

I will try recheck if it's actually needed

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

codecov · 2025-07-14T08:40:43Z

Codecov Report

❌ Patch coverage is 84.61538% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 84.55%. Comparing base (5a122b5) to head (39e3a50).
⚠️ Report is 10 commits behind head on master.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
sentry_sdk/integrations/dramatiq.py	84.21%	4 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #4571      +/-   ##
==========================================
+ Coverage   84.47%   84.55%   +0.07%     
==========================================
  Files         158      158              
  Lines       16506    16535      +29     
  Branches     2865     2868       +3     
==========================================
+ Hits        13944    13981      +37     
+ Misses       1712     1706       -6     
+ Partials      850      848       -2

Files with missing lines	Coverage Δ
sentry_sdk/consts.py	`99.33% <100.00%> (+<0.01%)`	⬆️
sentry_sdk/integrations/dramatiq.py	`83.16% <84.21%> (-0.59%)`	⬇️

... and 5 files with indirect coverage changes

Igreh · 2025-07-15T06:26:13Z

Hi @Igreh, I took a quick glance at your PR – at a high level this seems reasonable; however, before I do a full review, I would appreciate if you could please add some tests for these changes, since we need tests before we can consider merging this.

Hi!
Thanks for reply. I'm not very familiar about what should be tested exactly for now (transaction was successfully started? or what objects/props must be created? and etc ) but I will take a look and try to do my best =)
If you have any advice - would be very helpful

Edit:
Or I just need to cover one missed line?

sentrivana · 2025-07-15T07:18:46Z

Hey @Igreh, it'd be good to have "end-to-end" (i.e., not unit tests) for the different tracing code paths. Stuff like:

the happy path (if everything works fine, you get the transaction and it has the data you expect it to have)
how it behaves if there's an error (a transaction still gets sent, and it has the correct status, etc.; maybe also check if the user's program still behaves as expected -- the exception bubbles up if it should bubble up, etc.)
trace propagation (i.e., the transaction picks up propagation headers/vars if present) -- I think this might still be missing in your PR

...and/or whatever you consider important so that when the test suite is run and ends up green, you'd have a reasonable amount of confidence that tracing in the Dramatiq integration is working.

You can have a look at some of these similar integrations and their test suites (e.g. arq, huey) for inspiration.

Thanks for working on this! Let us know if you need any help.

- add trace propagation - set dramatiq_task_id as tag instead of extra - new tests - fix mypy issues Issue: getsentry#3454

Igreh · 2025-07-15T19:32:59Z

Hi @sentrivana !
Thanks for your reply. It was very helpful.
I believe I was able to solve propagation issue (see last commit)
Also, I've changed the way dramatiq_task_id is stored - now as tag (like in celery). It's actually very convenient to find transaction by ID.
New tests were added (huey integration I used for inspiration).
One thing I can't understand for now: mechanism.type of error event is "logging" instead of "dramatiq". https://github.com/getsentry/sentry-python/pull/4571/files#diff-ed170393871d90a02937ffecacd1b736805e6ea3c43e5549fd0df9bab883709eR82
Disabling LoggingIntegration fixes it (and actually it's the way we use sentry) but I suppose the general solution must work on defaults)
Probably, I have to dig deeper to dramatiq source code to understand what's happening... Not sure but I will keep investigating

Igreh · 2025-07-21T07:04:54Z

Hi @sentrivana !
Things are more complicated than I expected... =)

I'm using (and testing) my current integration on real project and
we're experiencing different side effects, but I'm curious about one thing for now:
Retried transactions must be connected by the same trace? Is any sentry agreement for such issue?
Retrying in terms of Dramatiq means requeuing message and it means baggage is propagated also. Bug or feature ? )

Failed task with 3 retries might look like this:

sentrivana · 2025-07-22T10:10:44Z

Hey again 👋🏻 Thanks for your time on this!

One thing I can't understand for now: mechanism.type of error event is "logging" instead of "dramatiq". https://github.com/getsentry/sentry-python/pull/4571/files#diff-ed170393871d90a02937ffecacd1b736805e6ea3c43e5549fd0df9bab883709eR82
Disabling LoggingIntegration fixes it (and actually it's the way we use sentry) but I suppose the general solution must work on defaults)

What you're seeing is likely dramatiq logging the error when it happens, and our LoggingIntegration picking it up. In case that happens, our deduplication logic (in DedupeIntegration) drops the second (in this case better) event coming directly from the integration.

What you can do is add an ignore_logger("<logger-name>") to the integration, see for example here where we were solving the same issue for strawberry-graphql. This will tell the LoggingIntegration to skip capturing events from that specific logger, so the actual events coming from the specialized integration should get through.

(I know you're not working on the error part of the integration in this PR but it's still a good improvement so feel free to include it.)

I'm using (and testing) my current integration on real project and
we're experiencing different side effects, but I'm curious about one thing for now:
Retried transactions must be connected by the same trace? Is any sentry agreement for such issue?
Retrying in terms of Dramatiq means requeuing message and it means baggage is propagated also. Bug or feature ? )

Not sure off the top of my head how this works in our other task queue integrations, but my hunch is that retries should be under one trace.

) - import ContextVar from utils for python3.6 support - do not import type "R" from dramatiq package because it was introduced only in newer version of dramatiq - fix mypy settings to ignore missing imports - fix tox.ini to include aiocontextvars for py3.6

- Start new trace in case of retrying task. Such decision makes life easier to investigate failed transactions in complicated traces. - also fix logging issue in tests Issue: getsentry#3454

Igreh · 2025-09-16T07:53:44Z

Good day everyone!
Sorry for long delay but now I'm back and going to complete this task)

Last patches are aimed to solve (I hope) all current issues:

py36 support (ContextVar). aiocontextvars as new deps for tox.
typing issues: incorrect import (type R was absent in earlier versions of dramatiq)
fix test (ignore logger)

Not sure off the top of my head how this works in our other task queue integrations, but my hunch is that retries should be under one trace.

I decided to not include retries under one trace. As mentioned in commit: "Start new trace in case of retrying task. Such decision makes life easier to investigate failed transactions in complicated traces.".
As example: one transaction may include 2 or more execution of the same task. If one or more failed and dramatiq starts retrying them then trace will include huge amount of task calls. Understanding what should have happened and what actually happened becomes an extremely difficult task. The sequence of calls (at least visually) is disrupted. It's really easier to find all retries by dramatiq_message_id (which is now became tag)

Let me know if anything needs to be improved or reworked. Thanks in advance!

sentrivana · 2025-09-17T10:21:40Z

Hey @Igreh, thanks for coming back to the PR and addressing the feedback! CI looks great, I'll give the PR a look this week.

sentrivana

Hey @Igreh, thanks for bearing with me. Left a few comments. :)

sentrivana · 2025-09-19T13:22:20Z

sentry_sdk/integrations/dramatiq.py

+            name=message.actor_name,
+            op=OP.QUEUE_TASK_DRAMATIQ,
+            source=TransactionSource.TASK,
+            # origin=DramatiqIntegration.origin,


The origin looks good, any reason it's commented out?

No reasons. Will fix it. Thanks

sentrivana · 2025-09-19T13:23:24Z

sentry_sdk/integrations/dramatiq.py

        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)


The SDK shouldn't set tags on its own (we still do this in a couple places, but are avoiding it for new code)

I'm a little bit confused here. What do you mean exactly by "SDK shouldn't set tags"?
Am I do it wrong now or you wanna say setting tags is only allowed for client's code and no tags allowed in SDK?
I definitely want to set dramatiq_message_id for each task like it happens in celery integration:

message_id is extremely helpful in investigating task failures

setting tags is only allowed for client's code and no tags allowed in SDK?

This. Going forward, we're not setting tags in the SDK. You can still set dramatiq_message_id with e.g. span.set_data() and it should be searchable.

sentrivana · 2025-09-19T13:24:39Z

sentry_sdk/integrations/dramatiq.py

    DramatiqIntegration.
    """

+    _transaction = ContextVar("_transaction")


I think I already may have asked this in an earlier iteration, but do we need the ContextVar here? The scopes themselves are stored in ContextVars and ideally they should govern the reference to the current transaction (this should already be working as long as the transaction/span is correctly set on the scope, which it is as long as you use start_span/transaction context manager or just __enter__ directly)

It makes sense.
Looks like I really needn't store transaction separately and I can get it directly from scope:

transaction = sentry_sdk.get_current_scope().transaction

Right? =)

Exactly, that'd be the idea.

Igreh · 2025-09-21T03:57:43Z

Hey @Igreh, thanks for bearing with me. Left a few comments. :)

Hi! Thanks for review - super helpful as usual)
Left replies with some more questions.
Waiting for you replies to prepare final patch.

sentrivana · 2025-10-01T11:31:26Z

@Igreh Left some feedback, but we're pretty much almost there :) Thanks for your continued work on this!

Looks like the config in scripts/populate_tox/config.py has a merge conflict now because I've added a new option to dramatiq there -- should be enough to just accept both changes from your branch and from master. In case CI complains about anything populate_tox related I'll take care of it.

- No need to store transaction separately. All contextvar mentions removed - dramatiq_mesage_id tag removed Issue: getsentry#3454

Igreh · 2025-10-02T05:00:06Z

@Igreh Left some feedback, but we're pretty much almost there :) Thanks for your continued work on this!

Looks like the config in scripts/populate_tox/config.py has a merge conflict now because I've added a new option to dramatiq there -- should be enough to just accept both changes from your branch and from master. In case CI complains about anything populate_tox related I'll take care of it.

@sentrivana
I've pushed corrections according to your replies. Getting rid of contextvar has significantly simplified the code and no tox related issues left) Thanks!
The only thing I'm upset for is removal of "dramatiq_message_id" tag from integration.
Yes, I can do it in my project code, but I will have to do it manually for each dramatiq task. So, I'll have to find a workaround. Some sort of middleware should help - another dramatiq middleware or just a func decorator. Will figure it out later =)

sentrivana

Heyo @Igreh, I'm back with the last round.

I've tried it out and it works great! The code looks so much nicer without the custom ContextVar, too. 🙏🏻 I've noticed a couple things in my last round of review, most of them minor, and one that we need to fix (but should be easy since the logic was mostly there before, just needs adapting a little to use a different kind of scope).

LMK if you still have time to make the changes, if not, I'm also happy to take this over the finish line. As always, if anything is unclear, just let me know.

sentrivana · 2025-10-02T09:57:07Z

sentry_sdk/integrations/dramatiq.py


+    def before_enqueue(self, broker, message, delay):
+        # type: (Broker, Message[R], int) -> None
+        message.options["sentry_headers"] = {


Nit: I don't think we're really in risk of collisions here, but let's maybe call this _sentry_headers just to denote it's something internal

Suggested change

message.options["sentry_headers"] = {

message.options["_sentry_headers"] = {

sentrivana · 2025-10-02T09:57:27Z

sentry_sdk/integrations/dramatiq.py

+        scope.clear_breadcrumbs()
        scope.add_event_processor(_make_message_event_processor(message, integration))

+        sentry_headers = message.options.get("sentry_headers") or {}


If you go with my renaming suggestion from above:

Suggested change

sentry_headers = message.options.get("sentry_headers") or {}

sentry_headers = message.options.get("_sentry_headers") or {}

sentrivana · 2025-10-02T10:00:35Z

sentry_sdk/integrations/dramatiq.py

-        integration = sentry_sdk.get_client().get_integration(DramatiqIntegration)
-        if integration is None:
-            return


Please keep this -- it's a fallback in case the integration/the SDK/the client has been deactivated in the meantime. In that case we shouldn't do anything.

We should have this also in the new before_enqueue.

Done, but I believe it's unnecessary due to this: https://github.com/getsentry/sentry-python/pull/4571/files#diff-2722e3fe31f13ffc24072313765f1fc89f0f0721154b7ca072bb46b1f9573f5bR84
Middleware won't be in use if integration is not enabled

sentrivana · 2025-10-02T10:08:43Z

sentry_sdk/integrations/dramatiq.py

-        message._scope_manager = sentry_sdk.new_scope()
-        message._scope_manager.__enter__()


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).

Done.
but please recheck it)

sentry_sdk/integrations/dramatiq.py

- new isolation scope for each dramatiq task - slightly changed sentry headers name to prevent collisions - revert: set dramatiq_message_id as additional data Issue: getsentry#3454

Issue: getsentry#3454

Igreh · 2025-10-03T06:30:20Z

Heyo @Igreh, I'm back with the last round.

I've tried it out and it works great! The code looks so much nicer without the custom ContextVar, too. 🙏🏻 I've noticed a couple things in my last round of review, most of them minor, and one that we need to fix (but should be easy since the logic was mostly there before, just needs adapting a little to use a different kind of scope).

LMK if you still have time to make the changes, if not, I'm also happy to take this over the finish line. As always, if anything is unclear, just let me know.

Hi @sentrivana!
one more final of final patch)
Fixed everything you mentioned. Your turn to find something more to fix)

sentrivana

Ty @Igreh! Let's get this merged! :)

feat(integrations): Add tracing to DramatiqIntegration

edee89d

Adds tracing support to DramatiqIntegration getsentry#3454

Igreh requested a review from a team as a code owner July 10, 2025 16:52

szokeasaurusrex reviewed Jul 14, 2025

View reviewed changes

feat(integrations): Add trace propagation to DramatiqIntegration

5562a9c

- add trace propagation - set dramatiq_task_id as tag instead of extra - new tests - fix mypy issues Issue: getsentry#3454

Merge branch 'getsentry:master' into feat-dramatiq-transaction-support

863d16b

antonpirker added the New Integration Integrating with a new framework or library label Sep 11, 2025

Merge branch 'master' into feat-dramatiq-transaction-support

5674a62

This comment was marked as outdated.

Sign in to view

Igreh and others added 2 commits September 15, 2025 11:39

Merge branch 'getsentry:master' into feat-dramatiq-transaction-support

9764f19

This comment was marked as outdated.

Sign in to view

feat(integrations): Start new trace on retrying

39e3a50

- Start new trace in case of retrying task. Such decision makes life easier to investigate failed transactions in complicated traces. - also fix logging issue in tests Issue: getsentry#3454

szokeasaurusrex requested a review from sentrivana September 16, 2025 12:14

sentrivana reviewed Sep 19, 2025

View reviewed changes

Merge branch 'master' into feat-dramatiq-transaction-support

a41b477

feat(integrations): Remove ContextVar usage.

4746db6

- No need to store transaction separately. All contextvar mentions removed - dramatiq_mesage_id tag removed Issue: getsentry#3454

This comment was marked as outdated.

Sign in to view

Merge branch 'master' into feat-dramatiq-transaction-support

a5e9ad0

sentrivana reviewed Oct 2, 2025

View reviewed changes

sentry_sdk/integrations/dramatiq.py Show resolved Hide resolved

feat(integrations): Isolation scope

1357bea

- new isolation scope for each dramatiq task - slightly changed sentry headers name to prevent collisions - revert: set dramatiq_message_id as additional data Issue: getsentry#3454

This comment was marked as outdated.

Sign in to view

feat(integrations): Fix scope usage

39d3532

Issue: getsentry#3454

sentrivana approved these changes Oct 3, 2025

View reviewed changes

Merge branch 'master' into feat-dramatiq-transaction-support

db239dc

sentrivana enabled auto-merge (squash) October 3, 2025 08:51

sentrivana merged commit bbd2a5d into getsentry:master Oct 3, 2025
113 checks passed

	message.options["sentry_headers"] = {
	message.options["_sentry_headers"] = {

	sentry_headers = message.options.get("sentry_headers") or {}
	sentry_headers = message.options.get("_sentry_headers") or {}

		message._scope_manager = sentry_sdk.new_scope()
		message._scope_manager.__enter__()

feat(integrations): Add tracing to DramatiqIntegration #4571

feat(integrations): Add tracing to DramatiqIntegration #4571

Uh oh!

Conversation

Igreh commented Jul 10, 2025

Uh oh!

szokeasaurusrex left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov bot commented Jul 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Igreh commented Jul 15, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sentrivana commented Jul 15, 2025

Uh oh!

Igreh commented Jul 15, 2025

Uh oh!

Igreh commented Jul 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

sentrivana commented Jul 22, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

This comment was marked as outdated.

Uh oh!

This comment was marked as outdated.

Uh oh!

Igreh commented Sep 16, 2025

Uh oh!

sentrivana commented Sep 17, 2025

Uh oh!

sentrivana left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Igreh commented Sep 21, 2025

Uh oh!

sentrivana commented Oct 1, 2025

Uh oh!

This comment was marked as outdated.

Uh oh!

Igreh commented Oct 2, 2025

Uh oh!

sentrivana left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov bot commented Jul 14, 2025 •

edited

Loading

Igreh commented Jul 15, 2025 •

edited

Loading

Igreh commented Jul 21, 2025 •

edited

Loading

sentrivana commented Jul 22, 2025 •

edited

Loading

sentrivana left a comment •

edited

Loading