On-failure alerting

Leoflow can notify you when a run fails — Slack or a generic webhook — with no extra task and no Python. You declare the rules in leoflow.yaml; the scheduler fires them in Go the moment a DagRun reaches the terminal failed state.

# leoflow.yaml
alerts:
  on_failure:
    - type: slack
      conn: slack_prod
      message: "🚨 {{dag}} run {{run_id}} failed on {{task}}"
    - type: webhook
      conn: pagerduty

That’s it. Compile and deploy as usual — when a run of this DAG fails, both the Slack message and the PagerDuty webhook fire.

Native, not a task. This runs in the control plane on the terminal failure — no task pod, no Python in the hot path, and it can’t be skipped by an upstream failure the way a notify task can. For the task-based pattern (and its trade-offs) see When to use a task instead.

How it works

flowchart TB
  A["leoflow.yaml<br/>alerts:"] -->|"compile<br/>(validate + overlay)"| B["dag.json<br/>(alerts baked in)"]
  B --> C["scheduler sees the run reach <code>failed</code>"]
  C -->|"off the tick, in a goroutine"| D["for each rule:<br/>resolve connection → endpoint URL,<br/>render the message, POST it"]
  D --> E["Slack / PagerDuty / Opsgenie / Teams …"]

Three guarantees hold by design:

• 🔒 The secret stays in the connection. The webhook URL (which is a secret) lives encrypted in a managed connection — never in leoflow.yaml and never in the compiled dag.json.
• 🛟 Best-effort. A delivery that fails (a 500, a bad URL, a missing connection) is logged and dropped — it can never fail the run, and one bad rule never stops the others.
• ⚡ Off the tick path. Alerts dispatch in a detached goroutine, so a slow endpoint can’t stall the scheduler.

Setting up the connection

The endpoint URL is stored as the connection’s secret (so it’s encrypted at rest). Create the connection once, then reference it by id from any DAG.

• Slack
• Generic webhook

$ curl -X POST .../api/v2/connections -d '{
    "connection_id":"slack_prod","conn_type":"slack",
    "password":"https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXX"}'

$ curl -X POST .../api/v2/connections -d '{
    "connection_id":"pagerduty","conn_type":"http",
    "password":"https://events.pagerduty.com/v2/enqueue"}'

The webhook URL goes in password on purpose: Leoflow encrypts it at rest and hands it straight to the sender at failure time. host/login/schema are ignored for alert connections.

The message

message is optional. When set, these placeholders are substituted at failure time; when omitted, Leoflow sends a default one-line summary.

A placeholder with no value for this run renders (none) rather than an empty string — {{logical_date}} on a manually triggered run, for example. An alert reading failed for logical date is indistinguishable from a truncated one, so the marker is deliberate.

Anything else in {{…}} is a compile error, naming the offending placeholder and listing what is available. A typo used to survive compile and reach the alert verbatim, which meant discovering it in the message that was supposed to explain an outage.

What each channel receives:

• slack → Slack’s incoming-webhook shape: {"text": "<your message>"}.

• webhook → a structured JSON body, so your service can route on it:

  {
    "dag_id": "sales",
    "run_id": "manual__2026-07-17T09:00:00Z",
    "logical_date": "2026-07-17T09:00:00Z",
    "failed_tasks": ["transform"],
    "message": "🚨 sales run … failed on transform"
  }
  

When to use a task instead

The alerts: block is the right default. But if you need the notification to run your own code (enrich the payload, hit an internal API, page via a provider operator), add a notify task guarded by a failure trigger rule — it runs in a pod with full operator fidelity:

alert = SlackWebhookOperator(
    task_id="alert", slack_webhook_conn_id="slack",
    message="pipeline failed 🚨", trigger_rule="one_failed",
)
[extract, transform] >> alert

Trade-off: a task runs a pod and is itself a task in the graph (it can be skipped if the whole run is torn down), whereas the native alerts: block always fires on the terminal failure, in the control plane, for free.

The Airflow on_failure_callback

If you already write Airflow, you can also use its native per-task on_failure_callback — a Python callable set on the operator or @task. Leoflow runs it in-process, inside the task’s own pod, on the task’s terminal failure:

def alert(context):
    # your code: enrich, page, post — runs in the task pod on final failure
    print("failed:", context["task_instance"].task_id)

@task(on_failure_callback=alert, retries=2)
def transform(): ...

How it differs from the alerts: block:

Only the terminal attempt fires it. The callback runs only once retries are exhausted (try_number >= max_tries), matching Airflow. With retries: 2 a fail → fail → success run fires it zero times; a fail × 3 run fires it once, on the last attempt. It’s best-effort and anti-loop: a callback that raises is logged and swallowed — it never changes the task’s own failed outcome and can’t re-trigger itself.

Where it works — and where the compiler stops you. It runs on any Python-executed task: a provider operator or a @task. It cannot run on the two task types that leave no Python to run it, and the compiler rejects those loudly rather than dropping the callback silently:

• bash — the runtime execs bash in place, so no Python is left.
• http_api — removed (ADR 0047/0048). HttpOperator now runs in a pod like any provider operator (declare connectors: [http]); the old inline path ran in the control-plane process and is being removed.

on_success_callback and on_retry_callback aren’t wired yet — they’re a loud compile error everywhere, never a silent drop. Reach for the alerts: block or a downstream @task with a trigger_rule instead.

Reference