Get the original stack trace of the exception in run_failure_sensor

[dagsir[bot]]

In a @run_failure_sensor context, is there a way to get the original stack trace of the exception? This below just indicates the name of the step that failed.

context.failure_event.message

---

The question was originally asked in Dagster Slack.

[yuhan]

The original stack trace could be in multiple places. The code snippet should capture both run level exception and step level exception:

@run_failure_sensor(job_selection=[my_job], minimum_interval_seconds=5)
def my_failure_sensor(context: RunFailureSensorContext):
    errors = []
    # Run level failure
    if context.failure_event.event_specific_data.error:
        errors.append(context.failure_event.event_specific_data.error)

    # Step level failure - capture it separately because details of which won't be surfaced to the
    # run level context.failure_event
    events = context.instance.all_logs(
        run_id=context.dagster_run.run_id, of_type=DagsterEventType.STEP_FAILURE
    )

    for e in events:
        errors.append(e.dagster_event.event_specific_data.error)

    print(errors)