From cec28d92efc8e7d00cf4ca5c4597a50ebe2b1604 Mon Sep 17 00:00:00 2001 From: viiccwen Date: Thu, 16 Jul 2026 21:23:07 +0800 Subject: [PATCH 1/2] Prevent negative trigger queue delay metrics System clock adjustments can make trigger startup latency negative or inflate it, producing misleading observability data. Signed-off-by: viiccwen --- airflow-core/src/airflow/jobs/triggerer_job_runner.py | 4 ++-- airflow-core/tests/unit/jobs/test_triggerer_job.py | 10 +++++++--- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index 12f3f1d60685a..3b8abc4573316 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py @@ -979,7 +979,7 @@ def update_triggers(self, requested_trigger_ids: set[int]): if new_trigger_ids: workloads_to_create = self.build_trigger_workloads(new_trigger_ids) - queued_at = time.time() + queued_at = time.monotonic() for workload in workloads_to_create: workload.queued_at = queued_at @@ -1403,7 +1403,7 @@ async def create_triggers(self): if workload.queued_at is not None: stats.timing( "triggerer.trigger_queue_delay", - int((time.time() - workload.queued_at) * 1000), + int((time.monotonic() - workload.queued_at) * 1000), tags=prune_dict({"team_name": self.team_name}), ) diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py index c9704c53c23a8..a67b6af52548a 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py @@ -1582,9 +1582,9 @@ async def test_create_triggers_emits_queue_delay_metric( runner.team_name = team_name runner.to_create.append(workload) - with patch( - "airflow.jobs.triggerer_job_runner.time.time", - return_value=101.5, + with ( + patch("airflow.jobs.triggerer_job_runner.time.monotonic", return_value=101.5), + patch("airflow.jobs.triggerer_job_runner.time.time", return_value=90.0), ): await runner.create_triggers() @@ -2291,6 +2291,8 @@ def test_update_triggers_delegates_workload_creation(supervisor_builder, mocker) supervisor = supervisor_builder() supervisor.running_triggers = {1, 3} workload = workloads.RunTrigger(id=2, classpath="some.trigger", encrypted_kwargs="", ti=None) + monotonic = mocker.patch("airflow.jobs.triggerer_job_runner.time.monotonic", return_value=100.0) + mocker.patch("airflow.jobs.triggerer_job_runner.time.time", return_value=90.0) build_trigger_workloads = mocker.patch.object( TriggerRunnerSupervisor, "build_trigger_workloads", autospec=True, return_value=[workload] ) @@ -2299,6 +2301,8 @@ def test_update_triggers_delegates_workload_creation(supervisor_builder, mocker) build_trigger_workloads.assert_called_once_with(supervisor, {2}) assert list(supervisor.creating_triggers) == [workload] + assert workload.queued_at == 100.0 + monotonic.assert_called_once_with() assert supervisor.cancelling_triggers == {1} From ec37fcf7bf6049263c50f41b33bb2279f0f58271 Mon Sep 17 00:00:00 2001 From: viiccwen Date: Sat, 18 Jul 2026 19:41:11 +0800 Subject: [PATCH 2/2] Document trigger queue delay clock assumption Monotonic timestamps cross the supervisor and runner process boundary, so the same-host requirement needs to remain visible to future changes. Signed-off-by: viiccwen --- airflow-core/src/airflow/jobs/triggerer_job_runner.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index 3b8abc4573316..50a899c2cf256 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/airflow-core/src/airflow/jobs/triggerer_job_runner.py @@ -1401,6 +1401,8 @@ async def create_triggers(self): inlets=[Asset(name=name, uri=uri) for name, uri in workload.watched_assets.items()] ) if workload.queued_at is not None: + # `queued_at` is captured by the supervisor process and consumed by the runner process. + # Both run on the same host, so their monotonic timestamps are comparable. stats.timing( "triggerer.trigger_queue_delay", int((time.monotonic() - workload.queued_at) * 1000),