Send Slack message based on job success retry sensor

[dagsir[bot]]

Is there a way to create a sensor that sends a Slack message if the retry of a job was successful? I've looked at the standard <https://docs.dagster.io/_apidocs/libraries/dagster-slack|dagster-slack sensor >and <https://docs.dagster.io/concepts/partitions-schedules-sensors/sensors#run-status-sensors|run status sensor> but didn't see how to check for the tag of retry_number.

---

The question was originally asked in Dagster Slack.

[jamiedemaria]

Inside of a run status sensor, you have access to a context. The context has an attribute called dagster_run that in turn contains the tags for a run. You could check those tags and then send a slack message if it was a retry

@run_status_sensor(run_status=DagsterRunStatus.SUCCESS)
def send_slack_on_successful_retry(context):
    tags = context.dagster_run.tags 
    # if retry tag in tags, then use SlackResource to send a message 

[jzamanxometry]

Thanks @jamiedemaria for the answer. I eventually figured it out. Here's the full code that worked in my case for future folks.

# this sensor only sends message if the retry is successful
@run_status_sensor(
    run_status=DagsterRunStatus.SUCCESS,
)
def slack_success_on_retry_sensor(context: SensorEvaluationContext) -> SensorDefinition:
    slack_client = WebClient(token=os.environ["DAGSTER_SLACK_TOKEN"])
   
    if (
        "dagster/retry_number" in context.dagster_run.tags
        and int(context.dagster_run.tags["dagster/retry_number"]) > 0
    ):
        # Slack Message Builder: copied from https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-slack/dagster_slack/sensors.py#L47-L94
        url = f"{webserver_base_url}/runs/{context.dagster_run.run_id}"
        blocks: List[Dict[Any, Any]] = []
        text = f'Job "{context.dagster_run.job_name}" succeeded on retry #{context.dagster_run.tags["dagster/retry_number"]}'
        main_body_text = "*SUCCESS ON RETRY!* 🥂"
        # building blocks
        blocks.extend(
            [
                {
                    "type": "section",
                    "text": {"type": "mrkdwn", "text": main_body_text},
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": text,
                    },
                },
            ]
        )
        blocks.append(
            {
                "type": "actions",
                "elements": [
                    {
                        "type": "button",
                        "text": {
                            "type": "plain_text",
                            "text": "View Successful Job in Dagster UI",
                        },
                        "url": url,
                    }
                ],
            }
        )
        slack_client.chat_postMessage(
            channel=slack_error_channel, text=text, blocks=blocks
        )
    else:
        SkipReason("Not success job or retry job attempt")