diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index 12f3f1d60685a..50a899c2cf256 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 @@ -1401,9 +1401,11 @@ 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.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}