Guide · Orchestration de données

Surveiller Dagster avec un dead man's switch

L'interface de Dagster vous montre un run une fois qu'il a eu lieu. Elle ne peut pas vous dire qu'un schedule a cessé de se déclencher, qu'un sensor s'est bloqué, ou que le daemon est mort, puisque rien ne s'exécute pour signaler le problème. Un check-in envoyé par le job lui-même comble cette lacune.

1. Créer un check

Créez un check dans Mortemain avec la même période que le schedule de votre job (plus une marge de grâce raisonnable), et copiez son URL de ping unique.

2. Faire le check-in avec des hooks de succès et d'échec

Les décorateurs @success_hook et @failure_hook de Dagster enveloppent un callback recevant un HookContext, que vous attachez à un job. Dirigez-les vers l'URL de ping et /fail :

jobs.py

from dagster import success_hook, failure_hook, HookContext, job, op
import requests

PING_URL = "https://ping.mortemain.com/your-check-uuid"

@success_hook(required_resource_keys=set())
def ping_success(context: HookContext):
    requests.get(PING_URL, timeout=10)

@failure_hook(required_resource_keys=set())
def ping_failure(context: HookContext):
    requests.get(PING_URL + "/fail", timeout=10)

@op
def extract():
    ...

@job(hooks={ping_success, ping_failure})
def nightly_etl():
    extract()

Les hooks attachés au job s'exécutent une fois par op : sur un job à plusieurs op, ping_success fait donc plusieurs check-in lors d'un run réussi. Ce n'est pas gênant, Mortemain n'a besoin que d'un check-in par période, mais cela veut dire que la durée ou le payload d'un seul appel de hook ne représente pas l'ensemble du run.

3. Préférer un ping par run : un run status sensor

Pour un check-in unique reflétant le run dans son ensemble plutôt que chaque op, utilisez un run_status_sensor ciblé sur le job plutôt que des hooks :

sensors.py

from dagster import run_status_sensor, RunStatusSensorContext, DagsterRunStatus
import requests

PING_URL = "https://ping.mortemain.com/your-check-uuid"

@run_status_sensor(run_status=DagsterRunStatus.SUCCESS, monitored_jobs=[nightly_etl])
def nightly_etl_succeeded(context: RunStatusSensorContext):
    requests.get(PING_URL, timeout=10)

@run_status_sensor(run_status=DagsterRunStatus.FAILURE, monitored_jobs=[nightly_etl])
def nightly_etl_failed(context: RunStatusSensorContext):
    requests.get(PING_URL + "/fail", timeout=10)

Déclarez les deux sensors dans vos Definitions et activez-les depuis l'interface Dagster. Vous pouvez ajouter un troisième sensor sur DagsterRunStatus.STARTED qui appelle PING_URL + "/start", pour que Mortemain mesure la durée de chaque run et signale ceux qui traînent anormalement.

Tester

Matérialisez le job une fois depuis l'interface ou la CLI et observez le check passer à up. Désactivez ensuite le schedule (ou ciblez un job qui ne s'exécutera jamais) et vérifiez que l'alerte down arrive après la marge de grâce : c'est ce mode de panne que les logs de Dagster ne révèlent pas.