Sending exception to Sentry

[abkfenris]

While Dagster's logging is really nice, it doesn't always give you quite as much information right when an exception happens as some other tools. Personally I use Sentry.io to capture exceptions in other things I build, so I tried to get it integrated into Dagster jobs.

The main way that data gets sent to Sentry is by attaching to sys.excepthook. Dagster however wraps all user functions with a broad exception handler, so exceptions never make it up to the system level. Sentry however will pick up that something bad has happened via log messages, but it won't have any of the context that usually makes it so useful.

It turns out that two can play the game of intercepting exceptions, so here's how I've gotten Sentry and Dagster to play nice.

First I made a helper to set up the Sentry client without a few of the usual integrations, and get it to ignore the dagster logger.

def setup_dagster_sentry():
    """
    Setup the sentry SDK for Dagster if SENTRY_DSN is defined for the environment.

    Additionally if IMAGE_TAG is defined it will be set as the release,
    and TRACES_SAMPLE_RATE can be set 0-1 otherwise will default to 0.

    Manually sets up a bunch of the default integrations and disables logging of dagster
    to quiet things down.
    """
    from sentry_sdk.integrations.argv import ArgvIntegration
    from sentry_sdk.integrations.atexit import AtexitIntegration
    from sentry_sdk.integrations.dedupe import DedupeIntegration
    from sentry_sdk.integrations.logging import LoggingIntegration, ignore_logger
    from sentry_sdk.integrations.modules import ModulesIntegration
    from sentry_sdk.integrations.stdlib import StdlibIntegration

    ignore_logger("dagster")

    SENTRY_DSN = os.environ.get("SENTRY_DSN")
    IMAGE_TAG = os.environ.get("IMAGE_TAG")
    TRACES_SAMPLE_RATE = float(os.environ.get("SENTRY_TRACES_SAMPLE_RATE", 0))

    if SENTRY_DSN:
        sentry_sdk.init(
            dsn=SENTRY_DSN,
            traces_sample_rate=TRACES_SAMPLE_RATE,
            release=IMAGE_TAG,
            default_integrations=False,
            integrations=[
                AtexitIntegration(),
                DedupeIntegration(),
                StdlibIntegration(),
                ModulesIntegration(),
                ArgvIntegration(),
                LoggingIntegration(),
            ],
        )

Then I created my own decorator to sit in between dagster.op send exceptions to Sentry before letting Dagster at them.

It also emits the unique Sentry event ID into the Dagster logs to help in finding the error.

def log_op_context(context: "OpExecutionContext"):
    """
    Capture Dagster OP context for Sentry Error handling
    """
    sentry_sdk.add_breadcrumb(
        category="dagster",
        message=f"{context.job_name} - {context.op_def.name}",
        level="info",
        data={
            "run_config": context.run_config,
            "job_name": context.job_name,
            "op_name": context.op_def.name,
            "run_id": context.run_id,
            "retry_number": context.retry_number,
        },
    )

    sentry_sdk.set_tag("job_name", context.job_name)
    sentry_sdk.set_tag("op_name", context.op_def.name)
    sentry_sdk.set_tag("run_id", context.run_id)

def capture_op_exceptions(func):
    """
    Captures exceptions thrown by Dagster Ops and forwards them to Sentry
    before re-throwing them for Dagster.

    Expects ops to receive Dagster context as the first argument,
    but it will continue if it doesn't (it just won't get as much context).

    It will log a unique ID that can be then entered into Sentry to find
    the exception.

    This should be used as a decorator between Dagster's `@op`,
    and the function to be handled.

    @op
    @sentry.capture_op_exceptions
    def op_with_error(context):
        raise Exception("Ahh!")
    """
    from dagster import get_dagster_logger

    @functools.wraps(func)
    def wrapped_fn(*args, **kwargs):
        logger = get_dagster_logger("sentry")

        try:
            log_op_context(args[0])
        except (AttributeError, IndexError):
            logger.warn("Sentry did not find execution context as the first arg")

        try:
            return func(*args, **kwargs)
        except Exception as e:
            event_id = sentry_sdk.capture_exception(e)
            logger.error(f"Sentry captured an exception. Event ID: {event_id}")
            raise e

    return wrapped_fn

Then it's used as a decorator on the ops.

@op
@capture_op_exception
def might_explode(context):
    raise Exception("Oops!")

Dagster now lets us know that a Sentry error has been made.

And Sentry has all sorts of context of what went wrong.

There are probably other ways that Sentry could be closer integrated (such as using dagster.logger with the various Sentry logging handlers), but now between Sentry and Dagster I have enough context to deal with many more errors without having to log every variable along the way.

[EtienneT]

Thank you @abkfenris for the amazing solution, works great. I was wondering if you figured out a way to avoid having to add @capture_op_exception to all your ops and assets manually?

For example I tried using capture_op_exception as a simple function wrapper instead of a decorator. This would allow us to wrap all assets with capture_op_exception before inserting them to the Definitions object. So we could do:

all_assets = [capture_asset_exceptions(x) for x in all_assets]

def = Definitions(assets = all_assets, ...)

Unfortunately we get an error related to type checking:

dagster._check.CheckError: Member of iterable mismatches type. Expected (<class 'dagster._core.definitions.assets.AssetsDefinition'>,
<class 'dagster._core.definitions.source_asset.SourceAsset'>, <class 'dagster._core.definitions.cacheable_assets.CacheableAssetsDefinition'>).
Got <function capture_asset_exceptions.<locals>.wrapped_fn at 0x000002604375FD80> of type <class 'function'>.

Looking at the code for @asset, I can't see an easy fix.

@capture_op_exception is also sensitive to where you insert it. You have to insert it after @asset to be able to add the asset in the Definitions object, otherwise you get the same error as above.

It would be great to have an official Sentry integration given Sentry's popularity.

Thanks again for the code!

[abkfenris]

Yes, it does have to between @op or @asset to get the context, and to make sure that Dagster can figure out when the right time to run it is.

If you don't want to manage remembering to sneak it in between, I'd probably make a decorator that combines @asset and @capture_op_exceptions, and takes all the kwargs of @asset and passes them along, then use that in place of @asset.

[whalesalad]

Want to bump this because this is something we would love to see. Our plan is to fork Dagster and submit a PR to integrate this, but we do not have rich understanding of how the system works currently. The investigation on this page though is priceless thank you for posting.

Would be great to do something like

import dagster

dagster.add_exception_handler('sentry', fn)

So that any number of these handlers can be added. The exception would get broadcast to all subscribed handlers, including the build-in one.

The caveat is that exceptions are heavy and in this case we don't want to handle them (in a blocking manner) as much as we just want to get notified of them and get them as data to pass it on elsewhere.

So perhaps dagster.on_exception is a better name.

Anyway, we've been huge fans of Dagster and have a lot of our workloads ported over to it, but atm we essentially need to manually visit the UI periodically to check in on failures. We'll embark on trying to do this ourselves unless the Dagster team has a better design idea or a quicker way to tackle.

[davidemerritt]

+1 - If the dagster team can help enable this, it would be a huge help to our organization. We'd love to have full exception tracking/tracing via sentry.

[ydennisy]

+1 on this, it would be a great integration to have!