myps6415 commented on code in PR #67592:
URL: https://github.com/apache/airflow/pull/67592#discussion_r3679314503
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -233,6 +236,22 @@ def ti_run(
extra=json.dumps({"host_name": ti_run_payload.hostname}) if
ti_run_payload.hostname else None,
)
)
+ # Emit task.queued_duration on a real QUEUED -> RUNNING transition.
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 that
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 — this matches how the endpoint already detects deferral resumes
via
+ # next_kwargs / next_method.
+ # The registry-based legacy name
dag.<dag_id>.<task_id>.queued_duration is
+ # emitted automatically by stats.timing via metrics_template.yaml.
+ if ti.queued_dttm is not None and ti.next_method is None:
+ stats.timing(
+ "task.queued_duration",
+ timezone.utcnow() - ti.queued_dttm,
+ tags={"task_id": ti.task_id, "dag_id": ti.dag_id, "queue":
ti.queue},
Review Comment:
Worth flagging the cause, since it wasn't a design call here: this branch
predates the
`stats_tags` refactor. On its base, `TaskInstance.stats_tags` was
`{"dag_id", "task_id"}`
— so `{task_id, dag_id, queue}` *was* exactly `{**ti.stats_tags, "queue":
ti.queue}`
when written. Rebased onto main and fixed in 948cbcda3c.
Emitting after the team-name resolution as you suggested — and since `dr` is
a real
`DagRun` there, the tags are now `{**dr.stats_tags, "task_id": ti.task_id,
"queue":
ti.queue}`, equal to `{**ti.stats_tags, "queue": ti.queue}` by construction
and in
step if `stats_tags` grows again. No need to add `DR.run_type` to the select.
Two things fell out of moving it down:
- `dr.team_name = ...` writes a plain attribute while `stats_tags` reads
`_team_name`,
which nothing sets here — so `dr.stats_tags` alone would drop it under
multi-team.
The looked-up value is now added to the tags directly.
- The duplicate-start path (same hostname/unixname/pid replay) only logs and
falls
through, so it reached the emit site too. Guarded with a flag, plus a test.
Tag assertions now derive from `{**ti.stats_tags, "queue": ti.queue}`
instead of a
hardcoded dict. Gap worth naming: the multi-team leg is untested — with
`core.multi_team`
off both sides omit the tag, so tests pass either way on that key. Can add a
`conf_vars`
case if you want it pinned.
--
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]
