kaxil commented on code in PR #69705:
URL: https://github.com/apache/airflow/pull/69705#discussion_r3625136618
##########
airflow-core/src/airflow/timetables/trigger.py:
##########
@@ -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()))
Review Comment:
The guard is `max_jitter > timedelta(0)` but the divisor is
`int(max_jitter.total_seconds())`, so any `0 < max_jitter < 1s` passes the
guard and then hits modulo by zero. `max_jitter=timedelta(milliseconds=500)`
raises `ZeroDivisionError` here at construction. The `int()` also silently
drops sub-second precision, so `timedelta(seconds=1.5)` truncates to a 1s
window and the offset can never exceed it. Flooring the guard at 1s, or
dropping the `int()` truncation, closes both.
##########
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)
[email protected]
+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="")
Review Comment:
With `seed` defaulting to `""`, a user who sets `max_jitter` but forgets
`seed` gets the same `md5("")` offset for every DAG, so they all fire at the
same shifted instant -- the herd just moves off the cron boundary instead of
spreading out, silently defeating the point of the feature. The no-op default
only makes sense when `max_jitter=0` too. Could this raise (or warn) when
`max_jitter > 0` and `seed == ""`? Note the core `__init__` already makes
`seed` required, so the two classes disagree on this today.
##########
airflow-core/src/airflow/timetables/trigger.py:
##########
@@ -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)
Review Comment:
`hashlib.md5(seed.encode())` called directly raises on FIPS-enabled Python
builds. The hash here isn't security-sensitive, so
`airflow.utils.hashlib_wrapper.md5` (which passes `usedforsecurity=False`)
keeps it working on FIPS, the same way `serialized_dag.py` and `dagcode.py`
already do. This runs scheduler-side on every deserialize, so on a FIPS build
every DAG using this timetable would fail to load.
##########
airflow-core/src/airflow/timetables/trigger.py:
##########
@@ -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.
Review Comment:
"Only the wall-clock fire time is shifted" reads as if `logical_date` stays
pinned to the cron boundary, but it doesn't. With the default `interval=0`,
`next_dagrun_info` returns `DagRunInfo.interval(next_start_time,
next_start_time)` where `next_start_time` is the jittered time, so
`logical_date` (= `data_interval.start`) moves by the offset too. The
`max_jitter` caveat further down ("the shift cannot push a run's `logical_date`
across a day or period boundary") already assumes this, so the two statements
contradict each other. Worth rewording to say the offset shifts
`logical_date`/`data_interval` as well, or deciding whether `logical_date`
should stay pinned to the boundary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
