Can @run_status_sensor ignore failure if job is being retried?

[stevenmurphy12]

I have a run status sensor to monitor for failures defined as follows:

@run_status_sensor(
    run_status=DagsterRunStatus.FAILURE,
    default_status=get_default_sensor_run_status(),
    description=f"Posts a message to {get_dagster_failure_channel()} when a job fails (excluding asset jobs)",
    monitor_all_code_locations=True
)
def job_run_failure_slack_sensor(context: RunStatusSensorContext, slack: SlackResource):
    text = ":x: Job failed"
    text += f"\nID: `{context.dagster_run.run_id}`"
    text += f"\nName: `{context.dagster_run.job_name}`"
    text += f"\n<{get_webserver_base_url()}/runs/{context.dagster_run.run_id}|View in Dagster UI>"

    post_to_slack(slack, context, get_dagster_failure_channel(), text)

I've recently reconfigured Dagster to retry runs in the event of a system failure, per the documentation here - https://docs.dagster.io/deployment/run-retries#combining-op-and-run-retries.

This sensor still fires even if the retry is successful, presumably because the job has still failed on the initial attempt. Is it possible to silence this sensor if Dagster is about to retry the run?

[gibsondan]

Hi @stevenmurphy12 - we actually just added a "dagster/will_retry" tag to the run that you can check in your sensor and tells you exactly this. You'll need to upgrade to the latest version (1.9.5) for it to be set.

[stevenmurphy12]

Wonderful timing, I just upgraded to 1.9.5 today. That tag works, thank you!

[stevenmurphy12]

For posterity, here's the code now that the dagster/will_retry tag is being checked:

@run_status_sensor(
    run_status=DagsterRunStatus.FAILURE,
    default_status=get_default_sensor_run_status(),
    description=f"Posts a message to {get_dagster_failure_channel()} when a job fails (excluding asset jobs)",
    monitor_all_code_locations=True
)
def job_run_failure_slack_sensor(context: RunStatusSensorContext, slack: SlackResource):
    logger.info(f"job_run_failure_slack_sensor run tags - {context.dagster_run.tags}")
    if context.dagster_run.tags.get('dagster/will_retry') == 'true':
        logger.info(f"run {context.dagster_run.run_id} will retry, not posting to slack")
        return

    text = ":x: Job failed"
    text += f"\nID: `{context.dagster_run.run_id}`"
    text += f"\nName: `{context.dagster_run.job_name}`"
    text += f"\n<{get_webserver_base_url()}/runs/{context.dagster_run.run_id}|View in Dagster UI>"

    post_to_slack(slack, context, get_dagster_failure_channel(), text)