Pebble32 commented on code in PR #69705:
URL: https://github.com/apache/airflow/pull/69705#discussion_r3657173009
##########
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:
Switched to hashlib_wrapper.md5 (FIPS-safe)
--
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]
