(airflow) branch main updated: Prevent negative trigger queue delay metrics (#69971)

[shahar]

This is an automated email from the ASF dual-hosted git repository.

shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 5820c02124d Prevent negative trigger queue delay metrics (#69971)
5820c02124d is described below

commit 5820c02124d4c47e5386268385893180c395a9ac
Author: Vic Wen <vic...@apache.org>
AuthorDate: Sun Jul 19 04:06:23 2026 +0800

    Prevent negative trigger queue delay metrics (#69971)
---
 airflow-core/src/airflow/jobs/triggerer_job_runner.py |  6 ++++--
 airflow-core/tests/unit/jobs/test_triggerer_job.py    | 10 +++++++---
 2 files changed, 11 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 12f3f1d6068..50a899c2cf2 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 @@ class TriggerRunnerSupervisor(WatchedSubprocess):
         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
@@ -1401,9 +1401,11 @@ class TriggerRunner:
                     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.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 c9704c53c23..a67b6af5254 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 @@ class TestTriggerRunner:
         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}