diff --git a/airflow-core/docs/authoring-and-scheduling/timetable.rst b/airflow-core/docs/authoring-and-scheduling/timetable.rst index b19cc80211392..8a4802afc4471 100644 --- a/airflow-core/docs/authoring-and-scheduling/timetable.rst +++ b/airflow-core/docs/authoring-and-scheduling/timetable.rst @@ -191,6 +191,47 @@ The same optional ``interval`` argument as CronTriggerTimetable_ is also availab pass +.. _JitteredCronTimetable: + +JitteredCronTimetable +^^^^^^^^^^^^^^^^^^^^^ + +This is a CronTriggerTimetable_ that offsets each Dag's fire time by a deterministic, per-Dag jitter. +It is useful when many Dags share the same cron expression -- for example every ``@daily`` Dag resolves +to ``0 0 * * *`` and would otherwise fire at exactly midnight, creating a "thundering herd" that can +overload the scheduler and workers at that instant. Each Dag is shifted by a fixed offset drawn from +``[0, max_jitter)`` so the runs are spread out. + +The offset is derived from ``seed`` (a stable, unique-per-Dag string -- the Dag id is a natural choice), +so the same seed always maps to the same offset and runs stay predictable across scheduler restarts and +serialization. ``data_interval`` and ``logical_date`` semantics are inherited from CronTriggerTimetable_; +only the wall-clock fire time is shifted. + +.. code-block:: python + + from datetime import timedelta + + from airflow.timetables.trigger import JitteredCronTimetable + + + # Fires daily, offset by a fixed amount in [0, 1h) unique to this Dag. + @dag( + schedule=JitteredCronTimetable( + "0 0 * * *", + timezone="UTC", + seed="example_dag", + max_jitter=timedelta(hours=1), + ), + ..., + ) + def example_dag(): + pass + +Keep ``max_jitter`` smaller than the gap to the next cron boundary so the shift cannot push a run's +``logical_date`` across a day or period boundary. With the defaults (``seed=""``, ``max_jitter=0``) the +offset is zero and the timetable behaves exactly like CronTriggerTimetable_. + + .. _DeltaDataIntervalTimetable: DeltaDataIntervalTimetable diff --git a/airflow-core/newsfragments/69705.feature.rst b/airflow-core/newsfragments/69705.feature.rst new file mode 100644 index 0000000000000..a1ed4e23e145c --- /dev/null +++ b/airflow-core/newsfragments/69705.feature.rst @@ -0,0 +1 @@ +Add ``JitteredCronTimetable``, a ``CronTriggerTimetable`` that offsets each Dag's fire time by a deterministic, per-Dag jitter to spread out Dags sharing a cron expression. diff --git a/airflow-core/src/airflow/serialization/encoders.py b/airflow-core/src/airflow/serialization/encoders.py index ab37b78550d77..04b7b4f9aa180 100644 --- a/airflow-core/src/airflow/serialization/encoders.py +++ b/airflow-core/src/airflow/serialization/encoders.py @@ -47,6 +47,7 @@ FixedKeyMapper, HourWindow, IdentityMapper, + JitteredCronTimetable, MinimumCount, MonthWindow, MultipleCronTriggerTimetable, @@ -326,6 +327,7 @@ class _Serializer: DeltaDataIntervalTimetable: "airflow.timetables.interval.DeltaDataIntervalTimetable", DeltaTriggerTimetable: "airflow.timetables.trigger.DeltaTriggerTimetable", EventsTimetable: "airflow.timetables.events.EventsTimetable", + JitteredCronTimetable: "airflow.timetables.trigger.JitteredCronTimetable", MultipleCronTriggerTimetable: "airflow.timetables.trigger.MultipleCronTriggerTimetable", NullTimetable: "airflow.timetables.simple.NullTimetable", OnceTimetable: "airflow.timetables.simple.OnceTimetable", @@ -390,6 +392,17 @@ def _(self, timetable: CronTriggerTimetable) -> dict[str, Any]: "run_immediately": encode_run_immediately(timetable.run_immediately), } + @serialize_timetable.register + def _(self, timetable: JitteredCronTimetable) -> dict[str, Any]: + return { + "expression": timetable.expression, + "timezone": encode_timezone(timetable.timezone), + "interval": encode_interval(timetable.interval), + "run_immediately": encode_run_immediately(timetable.run_immediately), + "seed": timetable.seed, + "max_jitter": timetable.max_jitter.total_seconds(), + } + @serialize_timetable.register def _(self, timetable: CronPartitionTimetable) -> dict[str, Any]: return { diff --git a/airflow-core/src/airflow/timetables/trigger.py b/airflow-core/src/airflow/timetables/trigger.py index af3f1c23c4cf1..1ada56a43deb5 100644 --- a/airflow-core/src/airflow/timetables/trigger.py +++ b/airflow-core/src/airflow/timetables/trigger.py @@ -18,6 +18,7 @@ import datetime import functools +import hashlib import math import operator import time @@ -598,3 +599,83 @@ def generate_run_id( suffix = f"{suffix}__{partition_key}" suffix = f"{suffix}__{get_random_string()}" return run_type.generate_run_id(suffix=suffix) + + +class JitteredCronTimetable(CronTriggerTimetable): + """ + A :class:`CronTriggerTimetable` that offsets each run by a deterministic, per-DAG jitter. + + Behaves exactly like ``CronTriggerTimetable`` but shifts every fire time by a fixed offset + derived from ``seed`` and spread across ``[0, max_jitter)``. This avoids the "thundering + herd" where many DAGs sharing a cron expression (e.g. ``@daily`` -> ``0 0 * * *``) all fire + at the same instant and overload the scheduler and workers at that boundary. + + The offset is deterministic: the same ``seed`` always maps to the same offset, so runs are + stable and predictable across scheduler restarts and timetable serialization. DAGs with + different seeds land in different slots; collisions are possible but harmless (two DAGs + sharing one minute still beats every DAG firing at once). + + All ``data_interval`` and ``logical_date`` semantics are inherited from + ``CronTriggerTimetable`` -- only the wall-clock fire time is shifted. + + :param cron: cron string that defines the base schedule. + :param timezone: timezone used to interpret the cron string. + :param interval: timedelta that defines the data interval start (see ``CronTriggerTimetable``). + :param run_immediately: see ``CronTriggerTimetable``. + :param seed: stable, unique-per-DAG string that determines the offset. The DAG id is a + natural choice. + :param max_jitter: upper bound of the jitter window; the offset falls in ``[0, max_jitter)``. + Keep it smaller than the gap to the next cron boundary so the shift cannot push a run's + ``logical_date`` across a day or period boundary. + """ + + def __init__( + self, + cron: str, + *, + timezone: str | Timezone | FixedTimezone, + interval: datetime.timedelta | relativedelta = datetime.timedelta(), + run_immediately: bool | datetime.timedelta = False, + seed: str, + max_jitter: datetime.timedelta, + ) -> None: + super().__init__(cron, timezone=timezone, interval=interval, run_immediately=run_immediately) + h = int(hashlib.md5(seed.encode()).hexdigest(), 16) + self._offset = ( + datetime.timedelta(seconds=h % int(max_jitter.total_seconds())) + if max_jitter > datetime.timedelta(0) + else datetime.timedelta(0) + ) + self._seed = seed + self._max_jitter = max_jitter + + def _apply(self, t: DateTime) -> DateTime: + return t + self._offset + + def _strip(self, t: DateTime) -> DateTime: + return t - self._offset + + def _get_next(self, current: DateTime) -> DateTime: + return self._apply(super()._get_next(self._strip(current))) + + def _get_prev(self, current: DateTime) -> DateTime: + return self._apply(super()._get_prev(self._strip(current))) + + def serialize(self) -> dict[str, Any]: + data = super().serialize() + data["seed"] = self._seed + data["max_jitter"] = self._max_jitter.total_seconds() + return data + + @classmethod + def deserialize(cls, data: dict[str, Any]) -> Timetable: + from airflow.serialization.decoders import decode_interval, decode_run_immediately + + return cls( + data["expression"], + timezone=parse_timezone(data["timezone"]), + interval=decode_interval(data["interval"]), + run_immediately=decode_run_immediately(data.get("run_immediately", False)), + seed=data["seed"], + max_jitter=datetime.timedelta(seconds=data["max_jitter"]), + ) diff --git a/airflow-core/tests/unit/timetables/test_jittered_cron_timetable.py b/airflow-core/tests/unit/timetables/test_jittered_cron_timetable.py new file mode 100644 index 0000000000000..11b3ff42f7145 --- /dev/null +++ b/airflow-core/tests/unit/timetables/test_jittered_cron_timetable.py @@ -0,0 +1,136 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +from datetime import timedelta + +import pendulum +import pytest +import time_machine + +from airflow._shared.timezones.timezone import utc +from airflow.sdk import JitteredCronTimetable as SdkJitteredCronTimetable +from airflow.serialization.decoders import decode_timetable +from airflow.serialization.encoders import encode_timetable +from airflow.timetables.base import DataInterval, TimeRestriction +from airflow.timetables.trigger import CronTriggerTimetable, JitteredCronTimetable + +CRON = "0 0 * * *" # daily at midnight +START_DATE = pendulum.DateTime(2021, 9, 4, tzinfo=utc) +# seed/window known to produce a non-zero offset (used across the behavioural tests) +SEED = "my_dag" +MAX_JITTER = timedelta(hours=1) + + +def _catchup_run_afters(timetable, count, *, earliest): + """Return the ``run_after`` of the first ``count`` catchup runs, feeding each back in.""" + run_afters = [] + last = None + for _ in range(count): + info = timetable.next_dagrun_info( + last_automated_data_interval=last, + restriction=TimeRestriction(earliest=earliest, latest=None, catchup=True), + ) + assert info is not None + run_afters.append(info.run_after) + last = DataInterval.exact(info.run_after) + return run_afters + + +@pytest.mark.parametrize( + "timezone_name", + [ + pytest.param("UTC", id="utc"), + # Daily runs across the 2026-03-08 spring-forward in this zone: the offset must be + # applied identically on both sides of the transition. + pytest.param("America/New_York", id="dst-spring-forward"), + ], +) +def test_jittered_runs_equal_base_runs_plus_offset(timezone_name): + """Every jittered run is the plain-cron run shifted by the fixed per-DAG offset. + + Cron/DST correctness is delegated to ``CronTriggerTimetable``; this asserts the offset is + applied consistently across a catchup sequence (including a DST transition), i.e. the + strip -> cron -> apply overrides never drift. + """ + earliest = pendulum.datetime(2026, 3, 6, tz=timezone_name) + base = CronTriggerTimetable(CRON, timezone=timezone_name) + jittered = JitteredCronTimetable(CRON, timezone=timezone_name, seed=SEED, max_jitter=MAX_JITTER) + offset = jittered._offset + assert offset > timedelta(0), "seed/window must produce a real shift, else the test is vacuous" + + base_runs = _catchup_run_afters(base, 5, earliest=earliest) + jittered_runs = _catchup_run_afters(jittered, 5, earliest=earliest) + + assert jittered_runs == [run + offset for run in base_runs] + + +@pytest.mark.parametrize("catchup", [True, False]) +def test_zero_max_jitter_matches_cron_trigger(catchup): + """A zero window yields a zero offset, so the timetable behaves exactly like its parent.""" + base = CronTriggerTimetable(CRON, timezone=utc) + jittered = JitteredCronTimetable(CRON, timezone=utc, seed=SEED, max_jitter=timedelta(0)) + assert jittered._offset == timedelta(0) + + last = DataInterval.exact(pendulum.DateTime(2022, 7, 26, tzinfo=utc)) + restriction = TimeRestriction(earliest=START_DATE, latest=None, catchup=catchup) + # travel so the no-catchup branch (which reads utcnow()) is deterministic + with time_machine.travel(pendulum.DateTime(2022, 7, 27, 5, 30, tzinfo=utc)): + assert jittered.next_dagrun_info( + last_automated_data_interval=last, restriction=restriction + ) == base.next_dagrun_info(last_automated_data_interval=last, restriction=restriction) + + +def test_offsets_are_deterministic_bounded_and_spread(): + """Same seed -> same offset (stable hash, not process-salted); offsets stay in [0, max_jitter) and spread.""" + assert ( + JitteredCronTimetable(CRON, timezone=utc, seed="dag_a", max_jitter=MAX_JITTER)._offset + == JitteredCronTimetable(CRON, timezone=utc, seed="dag_a", max_jitter=MAX_JITTER)._offset + ) + + offsets = [ + JitteredCronTimetable(CRON, timezone=utc, seed=f"dag_{i}", max_jitter=MAX_JITTER)._offset + for i in range(25) + ] + assert all(timedelta(0) <= offset < MAX_JITTER for offset in offsets) + assert len(set(offsets)) > 1, "distinct seeds should not all collide on one slot" + + +def test_serialize_round_trip_preserves_offset(): + """The core ``serialize``/``deserialize`` round-trips seed + window (as seconds) and the derived offset.""" + tt = JitteredCronTimetable(CRON, timezone=utc, seed=SEED, max_jitter=MAX_JITTER) + data = tt.serialize() + assert data["seed"] == SEED + assert data["max_jitter"] == 3600.0 # serialized as plain seconds + + restored = JitteredCronTimetable.deserialize(data) + assert isinstance(restored, JitteredCronTimetable) + assert restored._seed == tt._seed + assert restored._max_jitter == tt._max_jitter + assert restored._offset == tt._offset + + +def test_encode_decode_round_trip_across_layers(): + """SDK class -> encode_timetable (dispatch keyed on the SDK class) -> decode_timetable -> core class.""" + sdk_tt = SdkJitteredCronTimetable(CRON, timezone="UTC", seed=SEED, max_jitter=MAX_JITTER) + + restored = decode_timetable(encode_timetable(sdk_tt)) + + assert isinstance(restored, JitteredCronTimetable) # rebuilt as the core scheduler-side class + assert restored._max_jitter == MAX_JITTER + expected_offset = JitteredCronTimetable(CRON, timezone=utc, seed=SEED, max_jitter=MAX_JITTER)._offset + assert restored._offset == expected_offset diff --git a/task-sdk/src/airflow/sdk/__init__.py b/task-sdk/src/airflow/sdk/__init__.py index 807019eade80e..db10ed11cd7d7 100644 --- a/task-sdk/src/airflow/sdk/__init__.py +++ b/task-sdk/src/airflow/sdk/__init__.py @@ -58,6 +58,7 @@ "FixedKeyMapper", "HourWindow", "IdentityMapper", + "JitteredCronTimetable", "Label", "Metadata", "MinimumCount", @@ -211,6 +212,7 @@ CronPartitionTimetable, CronTriggerTimetable, DeltaTriggerTimetable, + JitteredCronTimetable, MultipleCronTriggerTimetable, ) from airflow.sdk.definitions.variable import Variable @@ -261,6 +263,7 @@ "FixedKeyMapper": ".definitions.partition_mappers.fixed_key", "HourWindow": ".definitions.partition_mappers.window", "IdentityMapper": ".definitions.partition_mappers.identity", + "JitteredCronTimetable": ".definitions.timetables.trigger", "Label": ".definitions.edges", "Metadata": ".definitions.asset.metadata", "MinimumCount": ".definitions.partition_mappers.wait_policy", diff --git a/task-sdk/src/airflow/sdk/__init__.pyi b/task-sdk/src/airflow/sdk/__init__.pyi index d48bf614407f8..ef30f87e774ef 100644 --- a/task-sdk/src/airflow/sdk/__init__.pyi +++ b/task-sdk/src/airflow/sdk/__init__.pyi @@ -122,6 +122,7 @@ from airflow.sdk.definitions.timetables.trigger import ( CronPartitionTimetable, CronTriggerTimetable, DeltaTriggerTimetable, + JitteredCronTimetable, MultipleCronTriggerTimetable, ) from airflow.sdk.definitions.variable import Variable as Variable @@ -169,6 +170,7 @@ __all__ = [ "FixedKeyMapper", "HourWindow", "IdentityMapper", + "JitteredCronTimetable", "Label", "Metadata", "MinimumCount", diff --git a/task-sdk/src/airflow/sdk/definitions/timetables/trigger.py b/task-sdk/src/airflow/sdk/definitions/timetables/trigger.py index c2d86ffa85463..ea6fc4ce29dd5 100644 --- a/task-sdk/src/airflow/sdk/definitions/timetables/trigger.py +++ b/task-sdk/src/airflow/sdk/definitions/timetables/trigger.py @@ -86,6 +86,38 @@ class CronTriggerTimetable(CronMixin, BaseTimetable): run_immediately: bool | datetime.timedelta = attrs.field(kw_only=True, default=False) +@attrs.define +class JitteredCronTimetable(CronTriggerTimetable): + """ + A :class:`CronTriggerTimetable` that offsets each run by a deterministic, per-DAG jitter. + + Behaves like ``CronTriggerTimetable`` but shifts every fire time by a fixed offset derived + from ``seed`` and spread across ``[0, max_jitter)``. This avoids the "thundering herd" where + many DAGs sharing a cron expression (e.g. ``@daily`` -> ``0 0 * * *``) all fire at the same + instant and overload the scheduler and workers at that boundary. With the defaults + (``seed=""``, ``max_jitter=0``) the offset is zero, so it behaves exactly like + ``CronTriggerTimetable``. + + The offset is deterministic: the same ``seed`` always maps to the same offset, so runs stay + stable and predictable across scheduler restarts and timetable serialization. Different seeds + land in different slots; collisions are possible but harmless. ``data_interval`` and + ``logical_date`` semantics are inherited from ``CronTriggerTimetable`` -- only the wall-clock + fire time is shifted. + + :param seed: stable, unique-per-DAG string that determines the offset. The DAG id is a + natural choice. + :param max_jitter: upper bound of the jitter window; the offset falls in ``[0, max_jitter)``. + Keep it smaller than the gap to the next cron boundary so the shift cannot push a run's + ``logical_date`` across a day or period boundary. + + See :class:`CronTriggerTimetable` for ``cron``, ``timezone``, ``interval`` and + ``run_immediately``. + """ + + seed: str = attrs.field(kw_only=True, default="") + max_jitter: datetime.timedelta = attrs.field(kw_only=True, default=datetime.timedelta()) + + @attrs.define(init=False) class MultipleCronTriggerTimetable(BaseTimetable): """