Prevent negative trigger queue delay metrics

[viiccwen]

closes #69970

Trigger queue delay is a duration measured from when the supervisor queues a workload until the runner creates its trigger. Using wall-clock timestamps for that duration can produce negative or inflated metrics when the system clock is adjusted.

This changes both ends of the measurement to use the monotonic clock. The regression tests simulate the wall clock moving backward and cover both the supervisor timestamp and the runner's emitted metric.

---

Was generative AI tooling used to co-author this PR?

• Yes - GPT-5.6 Sol

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[shahar1]

This re-applies the time.monotonic() change that was removed during review of #67927 (queued_at is stamped in the supervisor but subtracted in the runner subprocess).
Please either implement a more robust solution for this issue, or pivot this PR to adding a comment for why we use time.time() here.

[viiccwen]

Hello @shahar1,
thanks for raising the concern. I took another look at the guarantees provided by Python and the execution model here.

Although the reference point of time.monotonic() is unspecified, current Python documentation explicitly states that the monotonic clock is the same for all processes since Python 3.5. Airflow currently supports Python 3.10+, and the supervisor / runner are local processes on the same host, so both calls use the same system-wide monotonic clock.

On Linux, this is backed by CLOCK_MONOTONIC; the value has an unspecified epoch, but values obtained by separate processes on the same host are still comparable. The unspecified reference point only means that the value cannot be interpreted as an absolute timestamp.

For this metric, we are measuring an elapsed duration, so time.monotonic() also avoids negative or inflated values caused by wall-clock adjustments such as NTP corrections or manual clock changes. (ref)

I see here:

airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py

Lines 272 to 278 in fe9d865

	self.trigger_runner = TriggerRunnerSupervisor.start(
	job=self.job,
	capacity=self.capacity,
	logger=log,
	queues=self.queues,
	team_name=self.team_name,
	)

then here:

airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py

Lines 547 to 554 in fe9d865

	proc = super().start(
	id=proc_id,
	job=job,
	target=cls.run_in_process,
	logger=logger,
	use_exec=supervisor._should_use_exec(),
	**kwargs,
	)

so the relation is like: Triggerer host/container -> Triggerer main process -> TriggerRunnerSupervisor -> local TriggerRunner subprocess

IMO, we can add a comment documenting the same-host and cross-process clock assumption, not just use max(0, time.time() - queued_at) to hide negative.

[shahar1]

Hello @shahar1, thanks for raising the concern. I took another look at the guarantees provided by Python and the execution model here.

Although the reference point of time.monotonic() is unspecified, current Python documentation explicitly states that the monotonic clock is the same for all processes since Python 3.5. Airflow currently supports Python 3.10+, and the supervisor / runner are local processes on the same host, so both calls use the same system-wide monotonic clock.

On Linux, this is backed by CLOCK_MONOTONIC; the value has an unspecified epoch, but values obtained by separate processes on the same host are still comparable. The unspecified reference point only means that the value cannot be interpreted as an absolute timestamp.

For this metric, we are measuring an elapsed duration, so time.monotonic() also avoids negative or inflated values caused by wall-clock adjustments such as NTP corrections or manual clock changes. (ref)

I see here:

airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py

Lines 272 to 278 in fe9d865

	self.trigger_runner = TriggerRunnerSupervisor.start(
	job=self.job,
	capacity=self.capacity,
	logger=log,
	queues=self.queues,
	team_name=self.team_name,
	)

then here:

airflow/airflow-core/src/airflow/jobs/triggerer_job_runner.py

Lines 547 to 554 in fe9d865

	proc = super().start(
	id=proc_id,
	job=job,
	target=cls.run_in_process,
	logger=logger,
	use_exec=supervisor._should_use_exec(),
	**kwargs,
	)

so the relation is like: Triggerer host/container -> Triggerer main process -> TriggerRunnerSupervisor -> local TriggerRunner subprocess

IMO, we can add a comment documenting the same-host and cross-process clock assumption, not just use max(0, time.time() - queued_at) to hide negative.

Thanks for challenging my claim - it seems like you've got a point :)
Worth adding a comment as suggested that it relies on the premise of same host.
I wonder if at a later point (if we want to allow executions on remote) we should measure it in the supervisor instead.

@SameerMesiah97 - FYI, I'll be happy for your feedback.