Creating an Asset Check Sensor to get Slack notification whenever an Asset Check fails

[NicolasGuary]

Following my current needs and after looking at the following discussion, I started working on sensors that would send a Slack notification if a check fails (it seems to be a native feature of Dagster Cloud / Dagster+, but needs custom code for OSS deployments).

To avoid managing event logs, I used simple run_status_sensor for both SUCCESS and FAILURE states. It allows handling the job runs context only once in a quite simple manner.

Here's the function that handles the events :

def handle_asset_check_status(context: RunStatusSensorContext, slack: SlackResource):
    run_id = context.dagster_run.run_id
    event_records: Sequence[EventLogRecord] = context.instance.get_records_for_run(
        run_id,
        of_type=DagsterEventType.ASSET_CHECK_EVALUATION,
    ).records

    for event_record in event_records:
        build_and_post_slack_message_if_asset_check_failed(event_record, slack)

-> We get events from the context of the run and retrieve only ASSET_CHECK_EVALUATION

As it also contains some logic related to the event's context, here's my current implementation of the slack message formatting :

def build_and_post_slack_message_if_asset_check_failed(event_record: EventLogRecord, slack: SlackResource):
    event_log_entry: EventLogEntry = event_record.event_log_entry
    dagster_event: AssetCheckEvaluation = event_log_entry.dagster_event

    job_name = event_log_entry.job_name
    run_id = event_log_entry.run_id
    asset_key = dagster_event.event_specific_data.asset_key.to_user_string()
    check_name = dagster_event.event_specific_data.check_name
    asset_check_passed = dagster_event.event_specific_data.passed
    asset_check_severity = dagster_event.event_specific_data.metadata["status"].text

    if not asset_check_passed:
        severity_emoji = ":warning:" if asset_check_severity == "warn" else ":x:"
        instance_base_url = os.environ["DAGIT_BASE_URL"]
        slack_alerting_channel = os.environ["SLACK_CHANNEL"]

        slack_message_body = [
            {
                "color": "#E9D502" if asset_check_severity == "warn" else "#FF0000",
                "blocks": [
                    {"type": "header", "text": {"type": "plain_text", "text": f"{severity_emoji} Asset Check Failed"}},
                    {
                        "type": "section",
                        "fields": [
                            {"type": "mrkdwn", "text": f"*Asset Check:*\n `{check_name}`"},
                            {"type": "mrkdwn", "text": f"*Asset:*\n `{asset_key}`"},
                        ],
                    },
                    {"type": "section", "fields": [{"type": "mrkdwn", "text": f"*Job:*\n`{job_name}`"}]},
                    {
                        "type": "actions",
                        "elements": [
                            {
                                "type": "button",
                                "text": {"type": "plain_text", "text": "View in Dagster UI"},
                                "value": "dagster_button",
                                "url": instance_base_url + "/runs/" + run_id,
                            }
                        ],
                    },
                ],
            }
        ]

        slack.get_client().chat_postMessage(channel=slack_alerting_channel, attachments=slack_message_body)

Finally, here's how I defined two sensors (for both run failures and success) :

@run_status_sensor(run_status=DagsterRunStatus.FAILURE, default_status=DefaultSensorStatus.RUNNING)
def failed_asset_check_in_failed_job_sensor(context: RunStatusSensorContext, slack: SlackResource):
    handle_asset_check_status(context, slack)


@run_status_sensor(run_status=DagsterRunStatus.SUCCESS, default_status=DefaultSensorStatus.RUNNING)
def failed_asset_check_in_success_job_sensor(context: RunStatusSensorContext, slack: SlackResource):
    handle_asset_check_status(context, slack)

Hope this helps, feel free to use this code to fit your needs (for instance you can avoid posting if asset check is in warn state etc...)

[Daniel-Vetter-Coverwhale]

This is amazing! I had to make a few changes around the severity, but that's it. When I looked at the event I was getting I saw it as an attribute on event_specific_data, with the enum type. Thanks a ton for this!

changes:

    asset_check_severity = dagster_event.event_specific_data.severity

        severity_emoji = ":warning:" if asset_check_severity == dagster.AssetCheckSeverity.WARN else ":x:"