How do I monitor an Airflow DAG with LastPing?

Pass on_success_callback and on_failure_callback functions to the DAG constructor. The success callback GETs https://ping.lastping.dev/ ; the failure callback POSTs a context summary to /fail. Both receive the Airflow context dict as their argument.

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

on_failure_callback fires on each failed attempt by default; on_success_callback fires only on overall DAG success. To fire only on final failure, check context['task_instance'].try_number > context['task'].retries inside the callback before posting to /fail. Set max_active_runs=1 and match your LastPing grace window to your retry total_time so intermediate failures don't open spurious incidents.

What if the Airflow scheduler itself is down?

A down scheduler never triggers DAG runs, so no callback fires and no ping is sent. Absence-detection monitoring catches this: the missing ping after the schedule+grace window opens an incident even if the scheduler is completely offline, with no callback code required.

Should I attach callbacks at the DAG level or task level?

DAG-level is simpler: one monitor per pipeline. DAG-level callbacks fire when the whole DAG succeeds or fails, which is usually the right scope for a single LastPing monitor, and most teams use one monitor per DAG run. Task-level callbacks give finer, per-task visibility if you want a separate monitor per task, but require more monitors to manage. Start with DAG-level and add task-level monitors only for tasks with their own SLAs.

Is Airflow DAG monitoring free?

Yes. LastPing is free for individuals with no monitor cap. See how it compares to Healthchecks.io and Cronitor.

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.

Your assistant can do this for you

Monitor my Airflow DAGs with LastPing

Start monitoring free All monitoring guides

On this page

The snippets
What this catches
Questions people ask

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.

How do I monitor an Airflow DAG with LastPing?

Pass on_success_callback and on_failure_callback functions to the DAG constructor. The success callback GETs https://ping.lastping.dev/<your-monitor-id>; the failure callback POSTs a context summary to /fail. Both receive the Airflow context dict as their argument.
Does on_failure_callback fire on each retry or only on final failure?

on_failure_callback fires on each failed attempt by default; on_success_callback fires only on overall DAG success. To fire only on final failure, check context['task_instance'].try_number > context['task'].retries inside the callback before posting to /fail. Set max_active_runs=1 and match your LastPing grace window to your retry total_time so intermediate failures don't open spurious incidents.
What if the Airflow scheduler itself is down?

A down scheduler never triggers DAG runs, so no callback fires and no ping is sent. Absence-detection monitoring catches this: the missing ping after the schedule+grace window opens an incident even if the scheduler is completely offline, with no callback code required.
Should I attach callbacks at the DAG level or task level?

DAG-level is simpler: one monitor per pipeline. DAG-level callbacks fire when the whole DAG succeeds or fails, which is usually the right scope for a single LastPing monitor, and most teams use one monitor per DAG run. Task-level callbacks give finer, per-task visibility if you want a separate monitor per task, but require more monitors to manage. Start with DAG-level and add task-level monitors only for tasks with their own SLAs.
Is Airflow DAG monitoring free?

Yes. LastPing is free for individuals with no monitor cap. See how it compares to Healthchecks.io and Cronitor.

FREE FOR INDIVIDUALS · FULLY HOSTED

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