myps6415 commented on code in PR #67592:
URL: https://github.com/apache/airflow/pull/67592#discussion_r3719230958
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -242,6 +249,14 @@ def ti_run(
extra=json.dumps({"host_name": ti_run_payload.hostname}) if
ti_run_payload.hostname else None,
)
)
+ # The scheduler refreshes queued_dttm every time it queues a task, so
utcnow() - queued_dttm
+ # is a meaningful queue wait for first runs and retries alike (a retry
is a new try that
+ # genuinely waited in the queue) -- mirroring the legacy emit in
emit_state_change_metric,
+ # which fired on every transition to RUNNING. Only resumes from
deferral are skipped,
+ # identified by next_method (the trigger sets it on resume), to avoid
re-emitting within the
+ # same try. queued_dttm is None only in rare races and test setups.
+ emit_queued_duration = ti.queued_dttm is not None and ti.next_method
is None
Review Comment:
`start_from_trigger=True` operators are deferred straight from SCHEDULED,
never queued, so
the resume this guard skips is their only queue wait — they emit nothing at
all today.
Test added.
Also fixed my comment: legacy skipped retries and emitted on resume (it runs
before
`end_date` is cleared), the inverse of what I claimed.
Per queue wait is the only axis that measures those, so that's my vote —
your call.
--
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]
