Add option to Dramatiq integration to ignore retried errors

[pencil]

Problem

When using the DramatiqIntegration that the Sentry SDK ships with, you will end up getting an event for every error in your task, even if the error is explicitly marked as retry-able and is handled as such by Dramatiq's built-in Retries middleware. This is usually not desired, as it adds noise for intermittent issues that are well known and handled accordingly.

Additionally, due to the Dramatiq integration monkey patching, it is challenging to work around this with a customized middleware.

Workaround

I am sharing my workaround here in hopes that it helps others running into this issue. I would of course much prefer official support for this behavior.

First up, you have to add the DramatiqIntegration to your Sentry integrations as you would normally, but you need to undo the aforementioned monkey patching afterwards, to remove the default Sentry middleware:

original_broker__init__ = Broker.__init__

sentry_sdk.init(
    dsn=SENTRY_DSN,
    integrations=[
        DjangoIntegration(),
        DramatiqIntegration(),
    ],
    environment=ENVIRONMENT,
)

# Undo Sentry monkey patching because we use our own Middleware
Broker.__init__ = original_broker__init__

Next, you have to implement a custom middleware that unifies the Retries and the official Sentry middleware:

from dramatiq import Message
from dramatiq.broker import Broker, MessageProxy
from dramatiq.middleware import Middleware, Retries
from sentry_sdk.integrations.dramatiq import SentryMiddleware


class LastRetrySentryMiddleware(Retries, SentryMiddleware, Middleware):
    """
    A middleware that combines the Retries and SentryMiddleware middlewares such that Sentry is notified only when the
    message has been retried the maximum number of times.
    """

    def __init__(self, *args, **kwargs):
        Retries.__init__(self, *args, **kwargs)
        SentryMiddleware.__init__(self)

    def after_process_message(self, broker: Broker, message: Message, *, result=None, exception=None):
        message_proxy: MessageProxy = message  # type: ignore

        Retries.after_process_message(self, broker, message, result=result, exception=exception)

        if exception is not None and not message_proxy.failed:
            # If the message is not marked as failed, it means it will be retried
            exception = None

        SentryMiddleware.after_process_message(self, broker, message, result=result, exception=exception)

Then, add this middleware instead of the Retries middelware to you Dramatiq broker via add_middleware or DRAMATIQ_BROKER["MIDDLEWARE"] if using django-dramatiq.

Proposed solution

Ideally the DramatiqIntegration would have an option to achieve this behavior. I'd be open to investigate how this could possibly be implemented in a PR if this is a feature the Sentry team is willing to support. At the very least, I thought it would be helpful to document the problem and the (somewhat ugly) workaround.