GUIDE · AIRFLOW · FREE

Monitor Apache Airflow DAGs: on_success_callback and on_failure_callback with LastPing

Pass on_success_callback and on_failure_callback to the DAG() constructor. The success callback GETs the LastPing URL when all tasks complete; the failure callback POSTs the task-instance context to /fail. Absence detection handles what callbacks can't: a scheduler outage that never triggers the DAG at all. Free, no monitor cap.

Start monitoring — free All monitoring guides

The snippets

Create a monitor in the console, set the schedule to match your DAG schedule plus a grace window long enough to cover retries, then copy the ping URL.

DAG-level callbacks (recommended)

One monitor per DAG. The success callback fires when all tasks complete successfully; the failure callback fires when any task exhausts its retries:

import requests from airflow import DAG from airflow.operators.python import PythonOperator from datetime import datetime, timedelta PING = "https://ping.lastping.dev/<your-monitor-id>" def _on_success(context): requests.get(PING, timeout=10) def _on_failure(context): ti = context.get("task_instance") dag = context.get("dag") msg = ( f"DAG {dag.dag_id} / task {ti.task_id} " f"failed on {ti.execution_date} " f"(try {ti.try_number})" ) requests.post(f"{PING}/fail", data=msg[:10_000], timeout=10) with DAG( dag_id="daily_etl", start_date=datetime(2024, 1, 1), schedule="@daily", on_success_callback=_on_success, on_failure_callback=_on_failure, default_args={ "retries": 2, "retry_delay": timedelta(minutes=10), }, catchup=False, ) as dag: def extract(**kwargs): # your extraction logic pass def transform(**kwargs): # your transformation logic pass t1 = PythonOperator(task_id="extract", python_callable=extract) t2 = PythonOperator(task_id="transform", python_callable=transform) t1 >> t2

Task-level callbacks for granular monitoring

Attach callbacks directly to a task operator for per-task monitoring. Useful when one critical task in a large DAG should have its own monitor:

PING_LOAD = "https://ping.lastping.dev/<load-task-monitor-id>" def _load_success(context): requests.get(PING_LOAD, timeout=10) def _load_failure(context): exc = context.get("exception") msg = str(exc)[:10_000] if exc else "load task failed" requests.post(f"{PING_LOAD}/fail", data=msg, timeout=10) load = PythonOperator( task_id="load", python_callable=load_fn, on_success_callback=_load_success, on_failure_callback=_load_failure, )

Using Airflow Variables to store the ping URL

Store the ping URL in an Airflow Variable so it survives DAG code changes and is visible in the Airflow UI. Use Variable.get at module level (cached) or inside the callback (re-fetched each time):

from airflow.models import Variable PING = Variable.get("lastping_daily_etl_url", default_var="https://ping.lastping.dev/<your-monitor-id>") def _on_success(context): requests.get(PING, timeout=10)

Grace window and retry alignment

Set the LastPing grace window to at least expected_run_duration + (retries × retry_delay) so intermediate retries don't open a spurious incident. Only the final failure (when retries are exhausted) should fire /fail. See also the Celery monitoring guide for task-queue patterns, the Python guide for script-level monitoring, and the LastPing MCP server for AI-agent provisioning.

What this catches

The callbacks report what they can; the silence reports the rest.

  • A task failure after retries exhausted — on_failure_callback fires and POSTs the task-instance summary to /fail, opening an incident immediately.
  • DAG success — on_success_callback fires and GETs the success ping, resetting the absence timer.
  • The scheduler itself is down — no DAG run is triggered, no callback fires; the missing ping opens an incident after the grace window.
  • A hung task — Airflow's own execution_timeout kills the task, triggering the failure callback, which POSTs to /fail.
  • Recovery — the next successful DAG run fires the success callback, closing the incident automatically.

Questions people ask

Front-loaded answers — the most important fact first.

  • Does on_failure_callback fire on each retry or only on final failure?

    on_failure_callback fires on each failed attempt by default. To fire only on final failure, check context['task_instance'].try_number > context['task'].retries inside the callback before posting to /fail.

  • What if the Airflow scheduler is down?

    A down scheduler never triggers DAG runs. No callback fires, no ping is sent. Absence-detection monitoring catches this: the missing ping after the grace window opens an incident — no callback code required.

  • Should I use DAG-level or task-level callbacks?

    DAG-level is simpler: one monitor per pipeline. Task-level gives per-task visibility but requires more monitors to manage. Start with DAG-level and add task-level monitors only for tasks with their own SLAs.

  • Is this free?

    Yes. LastPing is free and fully hosted with no monitor cap and no paid tier. See how it compares to Healthchecks.io and Cronitor.

FREE · FULLY HOSTED · NO PAID TIER

A scheduler outage fires no callback. Absence detection does.

Two callback functions now beat a quiet ETL gap later. First alert wired up in about a minute — and LastPing monitors itself the same way.

Start monitoring — free All monitoring guides