Add on_skipped_intervals_callback for catchup=False DAGs

airflow-core/docs/administration-and-deployment/listeners.rst

@@ -64,6 +64,24 @@ Beginning with Airflow 3, listeners are also notified whenever a state change is
     :start-after: [START howto_listen_dagrun_failure_task]
     :end-before: [END howto_listen_dagrun_failure_task]
 
+Dag Skipped Interval Events
+---------------------------
+
+When a Dag with ``catchup=False`` skips one or more scheduled data intervals (for example after
+a scheduler restart or when a paused Dag is re-enabled), the scheduler invokes the
+``on_dag_skipped_intervals`` listener hook with a
+:class:`~airflow.timetables.base.SkippedIntervalsSummary`. This is the listener counterpart to the Dag-level
+``on_skipped_intervals_callback``; listeners run synchronously in the scheduler, while the callback
+is dispatched to the dag processor. Like the callback, it does not fire on a Dag's first
+automated run (no previous run to compare against).
+
+- ``on_dag_skipped_intervals``
+
+.. exampleinclude:: /../src/airflow/example_dags/plugins/event_listener.py
+    :language: python
+    :start-after: [START howto_listen_dag_skipped_intervals]
+    :end-before: [END howto_listen_dag_skipped_intervals]
+
 
 TaskInstance State Change Events
 --------------------------------
@@ -207,3 +225,5 @@ List of changes in the listener interfaces since 2.8.0 when they were introduced
 +-----------------+--------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------+
 | 3.2.0           | ``on_task_instance_skipped``               | New listener method added to the interface                                                                                    |
 +-----------------+--------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------+
+| 3.3.0           | ``on_dag_skipped_intervals``               | New listener method added; fires when a Dag with ``catchup=False`` skips scheduled intervals                                  |
++-----------------+--------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------+

airflow-core/docs/administration-and-deployment/logging-monitoring/callbacks.rst

@@ -67,6 +67,9 @@ Name                                        Description
                                             is not started to be executed because of a preceding branching
                                             decision in the Dag or a trigger rule which causes execution
                                             to skip so that the task execution is never scheduled.
+``on_skipped_intervals_callback``           Invoked when a Dag with ``catchup=False`` skips one or more             Dag
+                                            scheduled data intervals without creating Dag runs (for example
+                                            after a scheduler restart or when a paused Dag is re-enabled).
 =========================================== ======================================================================= =================
 
 
@@ -95,6 +98,57 @@ For example, a timeout may be caused by a number of stalling tasks, but only one
     Before Airflow 3.2.0, the rules above did not apply and the task instance passed to Dag callback was not related to Dag state, rather being selected as the latest task in the Dag
     lexicographically.
 
+Skipped Intervals Callback
+^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+When a Dag is defined with ``catchup=False``, the scheduler may advance past missed scheduled
+intervals without creating Dag runs for them. Set ``on_skipped_intervals_callback`` on the Dag
+to be notified when that happens.
+
+The callback is invoked once per scheduler decision that creates a new scheduled Dag run while
+intervals were skipped between the previous automated run and the new run. No Dag runs are
+created for the skipped intervals. It does not fire on a Dag's first automated run (no previous
+run to compare against), even when ``catchup=False`` leaves intervals after ``start_date``
+uncreated.
+
+The callback runs in the dag processor (like other Dag-level callbacks), not in the scheduler.
+The context mapping contains:
+
+* ``dag`` -- the Dag object
+* ``reason`` -- always ``"skipped_intervals"``
+* ``skipped_range`` -- a :class:`~airflow.timetables.base.DataInterval` from the
+  previous automated run's ``data_interval_end`` to the new run's ``data_interval_start``
+
+To enumerate every skipped interval, call
+:meth:`~airflow.serialization.definitions.dag.SerializedDAG.iter_dagrun_infos_between`
+on ``context["dag"]`` with ``skipped_range.start`` and ``skipped_range.end``.
+
+Example:
+
+.. code-block:: python
+
+    from airflow.sdk import DAG
+    from airflow.providers.standard.operators.empty import EmptyOperator
+
+
+    def log_skipped_intervals(context):
+        skipped_range = context["skipped_range"]
+        print(f"Gap: {skipped_range.start} -> {skipped_range.end}")
+        for info in context["dag"].iter_dagrun_infos_between(skipped_range.start, skipped_range.end):
+            if info.data_interval is None:
+                continue
+            interval = info.data_interval
+            print(f"Skipped interval: {interval.start} -> {interval.end}")
+
+
+    with DAG(
+        dag_id="example_skipped_intervals_callback",
+        schedule="@daily",
+        catchup=False,
+        on_skipped_intervals_callback=log_skipped_intervals,
+    ):
+        EmptyOperator(task_id="task")
+
 
 Examples
 --------

airflow-core/docs/core-concepts/dag-run.rst

@@ -128,6 +128,9 @@ If you set ``catchup=True`` in the Dag, the scheduler will kick off a Dag Run fo
 If your Dag is not written to handle its catchup (i.e., not limited to the interval, but instead to ``Now`` for instance.),
 then you will want to turn catchup off, which is the default setting or can be done explicitly by setting ``catchup=False`` in the Dag definition, if the default config has been changed for your Airflow environment.
 
+When intervals are skipped this way, Airflow does not create Dag runs for the missed periods.
+To observe those skips, set ``on_skipped_intervals_callback`` on the Dag (see :doc:`Callbacks </administration-and-deployment/logging-monitoring/callbacks>`) or implement the ``on_dag_skipped_intervals`` listener hook (see :doc:`Listeners </administration-and-deployment/listeners>`).
+
 .. code-block:: python
 
     """

airflow-core/docs/howto/listener-plugin.rst

@@ -44,6 +44,7 @@ Using this plugin, following events can be listened:
     * Dag run is in running state.
     * Dag run is in success state.
     * Dag run is in failure state.
+    * scheduler skips scheduled intervals for a Dag with ``catchup=False``
     * on start before event like Airflow job, scheduler
     * before stop for event like Airflow job, scheduler
 
@@ -91,5 +92,12 @@ This example listens when the Dag run is change to failed state
 
 Similarly, code to listen after dag_run success and during running state can be implemented.
 
+This example listens when the scheduler skips scheduled intervals for a Dag with ``catchup=False``
+
+.. exampleinclude:: /../src/airflow/example_dags/plugins/event_listener.py
+    :language: python
+    :start-after: [START howto_listen_dag_skipped_intervals]
+    :end-before: [END howto_listen_dag_skipped_intervals]
+
 The listener plugin files required to add the listener implementation is added as part of the
 Airflow plugin into ``$AIRFLOW_HOME/plugins/`` folder and loaded during Airflow startup.

airflow-core/newsfragments/68359.feature.rst

@@ -0,0 +1 @@
+Add ``on_skipped_intervals_callback`` for Dags with ``catchup=False`` and the ``on_dag_skipped_intervals`` listener hook, so operators are notified when the scheduler advances past missed scheduled intervals without creating Dag runs.

airflow-core/src/airflow/callbacks/callback_requests.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+from datetime import datetime
 from typing import TYPE_CHECKING, Annotated, Any, Literal, cast
 
 import structlog
@@ -26,7 +27,9 @@
 from sqlalchemy.orm.attributes import set_committed_value
 from sqlalchemy.orm.exc import DetachedInstanceError
 
+from airflow._shared.timezones import timezone
 from airflow.api_fastapi.execution_api.datamodels import taskinstance as ti_datamodel  # noqa: TC001
+from airflow.timetables.base import DataInterval, SkippedIntervalsSummary
 from airflow.utils.state import TaskInstanceState
 
 if TYPE_CHECKING:
@@ -174,8 +177,43 @@ class DagCallbackRequest(BaseCallbackRequest):
     type: Literal["DagCallbackRequest"] = "DagCallbackRequest"
 
 
+class DagSkippedIntervalsCallbackRequest(BaseCallbackRequest):
+    """Store skipped intervals callback data for execution by the Dag processor."""
+
+    dag_id: str
+    skipped_range: tuple[datetime, datetime]
+    """Envelope from the previous automated run's data_interval_end to the new run's start."""
+    type: Literal["DagSkippedIntervalsCallbackRequest"] = "DagSkippedIntervalsCallbackRequest"
+
+    @classmethod
+    def from_summary(
+        cls,
+        *,
+        filepath: str,
+        bundle_name: str,
+        bundle_version: str | None,
+        dag_id: str,
+        summary: SkippedIntervalsSummary,
+    ) -> Self:
+        return cls(
+            filepath=filepath,
+            bundle_name=bundle_name,
+            bundle_version=bundle_version,
+            dag_id=dag_id,
+            skipped_range=(summary.skipped_range.start, summary.skipped_range.end),
+        )
+
+    def to_summary(self) -> SkippedIntervalsSummary:
+        return SkippedIntervalsSummary(
+            skipped_range=DataInterval(
+                start=timezone.coerce_datetime(self.skipped_range[0]),
+                end=timezone.coerce_datetime(self.skipped_range[1]),
+            ),
+        )
+
+
 CallbackRequest = Annotated[
-    DagCallbackRequest | TaskCallbackRequest | EmailRequest,
+    DagCallbackRequest | DagSkippedIntervalsCallbackRequest | TaskCallbackRequest | EmailRequest,
     Field(discriminator="type"),
 ]
 

airflow-core/src/airflow/dag_processing/manager.py

@@ -48,7 +48,7 @@
 from airflow._shared.observability.metrics.stats import normalize_name_for_stats
 from airflow._shared.timezones import timezone
 from airflow.api_fastapi.execution_api.in_process import InProcessExecutionAPI
-from airflow.callbacks.callback_requests import DagCallbackRequest
+from airflow.callbacks.callback_requests import DagCallbackRequest, DagSkippedIntervalsCallbackRequest
 from airflow.configuration import conf
 from airflow.dag_processing.bundles.base import (
     BundleUsageTrackingManager,
@@ -821,6 +821,8 @@ def _add_callback_to_queue(self, request: CallbackRequest) -> None:
     def _log_dropped_lang_sdk_callback(self, request: CallbackRequest) -> None:
         if isinstance(request, DagCallbackRequest):
             target = f"dag_id={request.dag_id} run_id={request.run_id}"
+        elif isinstance(request, DagSkippedIntervalsCallbackRequest):
+            target = f"dag_id={request.dag_id}"
         else:
             ti = request.ti
             target = f"dag_id={ti.dag_id} run_id={ti.run_id} task_id={ti.task_id}"

airflow-core/src/airflow/dag_processing/processor.py

@@ -33,6 +33,7 @@
 from airflow.callbacks.callback_requests import (
     CallbackRequest,
     DagCallbackRequest,
+    DagSkippedIntervalsCallbackRequest,
     EmailRequest,
     TaskCallbackRequest,
 )
@@ -89,7 +90,7 @@
 
     from airflow.sdk.api.client import Client
     from airflow.sdk.bases.operator import BaseOperator
-    from airflow.sdk.definitions.context import Context
+    from airflow.sdk.definitions.context import Context, SkippedIntervalsCallbackContext
     from airflow.sdk.definitions.dag import DAG
     from airflow.sdk.definitions.mappedoperator import MappedOperator
     from airflow.sdk.execution_time.supervisor import RequestHandler, RequestResult, ResponseSent
@@ -342,8 +343,10 @@ def _execute_callbacks(
                 "run_id": request.ti.run_id,
                 "ti_id": str(request.ti.id),
             }
-        else:
+        elif isinstance(request, DagCallbackRequest):
             log_extra = {"dag_id": request.dag_id, "run_id": request.run_id}
+        else:
+            log_extra = {"dag_id": request.dag_id}
         # context_from_server can carry user-supplied run conf, and the masker cannot
         # redact inside an already-serialized string, so keep it out of log payloads.
         request_json = request.to_json(exclude={"context_from_server"})
@@ -360,6 +363,8 @@ def _execute_callbacks(
                     _execute_task_callbacks(dagbag, request, log)
                 elif isinstance(request, DagCallbackRequest):
                     _execute_dag_callbacks(dagbag, request, log)
+                elif isinstance(request, DagSkippedIntervalsCallbackRequest):
+                    _execute_dag_skipped_intervals_callback(dagbag, request, log)
                 elif isinstance(request, EmailRequest):
                     _execute_email_callbacks(dagbag, request, log)
         except Exception:
@@ -432,6 +437,44 @@ def _execute_dag_callbacks(dagbag: DagBag, request: DagCallbackRequest, log: Fil
             )
 
 
+def _execute_dag_skipped_intervals_callback(
+    dagbag: DagBag, request: DagSkippedIntervalsCallbackRequest, log: FilteringBoundLogger
+) -> None:
+    dag, _ = _get_dag_with_task(dagbag, request.dag_id)
+    callbacks = dag.on_skipped_intervals_callback
+    if not callbacks:
+        log.warning("Skipped intervals callback requested, but dag didn't have any", dag_id=request.dag_id)
+        return
+
+    callbacks = callbacks if isinstance(callbacks, list) else [callbacks]
+    summary = request.to_summary()
+    context: SkippedIntervalsCallbackContext = {
+        "dag": dag,
+        "reason": "skipped_intervals",
+        "skipped_range": summary.skipped_range,
+    }
+
+    for callback in callbacks:
+        log.info("Executing on_skipped_intervals_callback", dag_id=request.dag_id)
+        try:
+            callback(context)
+        except Exception:
+            log.exception("Callback failed", dag_id=request.dag_id)
+            stats.incr(
+                "dag.callback_exceptions",
+                tags=prune_dict(
+                    {
+                        "dag_id": request.dag_id,
+                        "team_name": (
+                            DagModel.get_team_name(request.dag_id)
+                            if conf.getboolean("core", "multi_team")
+                            else None
+                        ),
+                    }
+                ),
+            )
+
+
 def _execute_task_callbacks(dagbag: DagBag, request: TaskCallbackRequest, log: FilteringBoundLogger) -> None:
     if not request.is_failure_callback:
         log.warning(

airflow-core/src/airflow/example_dags/plugins/event_listener.py

@@ -222,3 +222,23 @@ def on_dag_run_running(dag_run: DagRun, msg: str):
 
 
 # [END howto_listen_dagrun_running_task]
+
+
+# [START howto_listen_dag_skipped_intervals]
+@hookimpl
+def on_dag_skipped_intervals(dag_id: str, summary):
+    """
+    Called when a Dag with catchup=False skips one or more scheduled intervals.
+
+    ``summary`` is a :class:`~airflow.timetables.base.SkippedIntervalsSummary` with
+    ``skipped_range``. Call
+    :meth:`~airflow.serialization.definitions.dag.SerializedDAG.iter_dagrun_infos_between`
+    with ``skipped_range.start`` and ``skipped_range.end`` if the full list of skipped
+    intervals is required.
+    """
+    print(
+        f"Dag {dag_id} skipped intervals in range: {summary.skipped_range.start} -> {summary.skipped_range.end}"
+    )
+
+
+# [END howto_listen_dag_skipped_intervals]