Repository navigation
Fix missing task.queued_duration metric in Airflow 3 - #67592
Conversation
|
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
|
|
@eladkal gentle ping! Could you help trigger CI? |
0576c5e to
2988b04
Compare
|
@ashb Gentle ping! This has approval from @henry3260, would appreciate a look when you have a moment. The 4 failing checks are unrelated flaky test_wasb failures. |
2988b04 to
5f57b80
Compare
|
@ashb thanks for the review! On the On the test: |
|
@ashb Gentle ping — I've addressed both your comments: dropped the |
5f57b80 to
948cbcd
Compare
948cbcd to
0f3fb97
Compare
d78cd90 to
6cea295
Compare
|
@vatsrahul1001 done -- rebased onto main and all three threads addressed. The conflict was with #70312, which reworked how The remaining semantics question is settled the way @kaxil said they would settle it -- one sample per queue wait -- with @henry3260 agreeing on the thread. One ask: the push reset the workflow approvals, so CI is sitting at |
|
LGTM! Can merge post review from code owners |
@vatsrahul1001 who are the code owners here, can you tag them. |
kaxil
left a comment
There was a problem hiding this comment.
The code looks right to me now, and the inline comments are non-blocking.
The PR description still describes the old guard. It says retries (end_date set) and deferral resumes (next_method) are skipped, and it names a test_ti_run_skips_queued_duration_metric that no longer exists. The description becomes the squash commit message, so could you update it to say the metric now emits once per queue wait (first run, retry, deferral resume, reschedule poke) and skips only when queued_dttm is unset or on a duplicate start?
The SDK retries a 5xx from this route, and the rollback returns the task instance to QUEUED for that retry to pick up, so a sample taken before the response was settled counted a single queue wait twice. Airflow 2 already reported the wait on deferral resumes; what changes for anyone upgrading is retries and reschedule-mode pokes, so the newsfragment names those instead.
|
Pushed 0353c4f with all five threads addressed, and rewrote the description so the squash message matches the final behaviour. CI is sitting at action_required again — could a maintainer approve the run? |
|
@kaxil Updated the description as you suggested. Can this be merged? |
kaxil
left a comment
There was a problem hiding this comment.
Thanks for the follow-up. Emitting after the request can no longer 5xx, with a test pinning it, plus the corrected comments, newsfragment and description cover everything from my last pass. LGTM.
|
Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions. |
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
New row in the Skills grid (no template change); both languages still print to one A4 page. Bumps meta.updated to 2026-10-14. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Backport successfully created: v3-4-testNote: As of Merging PRs targeted for Airflow 3.X In matter of doubt please ask in #release-management Slack channel.
|
) (#74965) * Fix missing task.queued_duration metric in Airflow 3 task.queued_duration (and its registry-derived legacy name dag.<dag_id>.<task_id>.queued_duration) stopped firing after the Airflow 3 worker switched to the Task SDK / supervisor / Execution API. The only emit site was TaskInstance.emit_state_change_metric, reachable solely from the legacy LocalTaskJob path, while Airflow 3 workers flip the TI to RUNNING through the ti_run Execution API endpoint instead. Emit it from ti_run on the genuine QUEUED -> RUNNING transition, tagged via DagRun.stats_tags so the restored metric stays sliceable the same way as its sibling task.scheduled_duration (dag_id, task_id, queue, run_type, team_name under multi-team, and Dag tags when configured). Resumes from deferral are skipped so a deferral cycle does not add a second sample inside the same try. closes: #63503 closes: #66067 * Pin the team_name tag on task.queued_duration under multi-team With multi-team off the tag is absent from both the metric and any expectation derived from stats_tags, so the assertion held for the wrong reason and would not have caught the tag going missing. * Describe what the queued_duration guard actually does The comment claimed to mirror a legacy emit that fired on every transition to RUNNING; that emit skipped retries and fired on deferral resumes, so the claim was inverted on both paths. It also credited the trigger with setting next_method, which the DEFERRED transition does, and asserted one sample per try without accounting for operators deferred before they are ever queued. * Show start_from_trigger tasks emit no queued_duration These operators are deferred straight from SCHEDULED without being queued, so the resume the next_method guard skips is the only queue wait they ever have. * Count task.queued_duration once per queue wait A resume from deferral waits for a worker slot exactly as a first run does, and an operator with start_from_trigger=True never reaches the queue until that resume, so suppressing it left the only wait those tasks have unmeasured. Dag tags are eager-loaded because this path runs once per task start, where the lazy load stats_tags otherwise falls back to costs a query per task. * Emit task.queued_duration only once the run request cannot fail The SDK retries a 5xx from this route, and the rollback returns the task instance to QUEUED for that retry to pick up, so a sample taken before the response was settled counted a single queue wait twice. Airflow 2 already reported the wait on deferral resumes; what changes for anyone upgrading is retries and reschedule-mode pokes, so the newsfragment names those instead. (cherry picked from commit 21c858e) Co-authored-by: Tung, Hsiao-Yu <myps6415@gmail.com>
) (#74965) * Fix missing task.queued_duration metric in Airflow 3 task.queued_duration (and its registry-derived legacy name dag.<dag_id>.<task_id>.queued_duration) stopped firing after the Airflow 3 worker switched to the Task SDK / supervisor / Execution API. The only emit site was TaskInstance.emit_state_change_metric, reachable solely from the legacy LocalTaskJob path, while Airflow 3 workers flip the TI to RUNNING through the ti_run Execution API endpoint instead. Emit it from ti_run on the genuine QUEUED -> RUNNING transition, tagged via DagRun.stats_tags so the restored metric stays sliceable the same way as its sibling task.scheduled_duration (dag_id, task_id, queue, run_type, team_name under multi-team, and Dag tags when configured). Resumes from deferral are skipped so a deferral cycle does not add a second sample inside the same try. closes: #63503 closes: #66067 * Pin the team_name tag on task.queued_duration under multi-team With multi-team off the tag is absent from both the metric and any expectation derived from stats_tags, so the assertion held for the wrong reason and would not have caught the tag going missing. * Describe what the queued_duration guard actually does The comment claimed to mirror a legacy emit that fired on every transition to RUNNING; that emit skipped retries and fired on deferral resumes, so the claim was inverted on both paths. It also credited the trigger with setting next_method, which the DEFERRED transition does, and asserted one sample per try without accounting for operators deferred before they are ever queued. * Show start_from_trigger tasks emit no queued_duration These operators are deferred straight from SCHEDULED without being queued, so the resume the next_method guard skips is the only queue wait they ever have. * Count task.queued_duration once per queue wait A resume from deferral waits for a worker slot exactly as a first run does, and an operator with start_from_trigger=True never reaches the queue until that resume, so suppressing it left the only wait those tasks have unmeasured. Dag tags are eager-loaded because this path runs once per task start, where the lazy load stats_tags otherwise falls back to costs a query per task. * Emit task.queued_duration only once the run request cannot fail The SDK retries a 5xx from this route, and the rollback returns the task instance to QUEUED for that retry to pick up, so a sample taken before the response was settled counted a single queue wait twice. Airflow 2 already reported the wait on deferral resumes; what changes for anyone upgrading is retries and reschedule-mode pokes, so the newsfragment names those instead. (cherry picked from commit 21c858e) Co-authored-by: Tung, Hsiao-Yu <myps6415@gmail.com>
Summary
task.queued_duration(and its registry-derived legacy namedag.<dag_id>.<task_id>.queued_duration) stopped firing entirely after the Airflow 3 worker switched to the Task SDK / supervisor / Execution API.The metric was only emitted by
TaskInstance.emit_state_change_metric, which is only reachable from_check_and_change_state_before_execution— the legacy LocalTaskJob path. Airflow 3 workers flip TI state toRUNNINGthrough theti_runExecution API endpoint instead, which bypasses the emit site.This is the same regression pattern as #62019 (missing
ti.start/ti.finish).Fix
Emit
task.queued_durationfromti_run, once per queue wait rather than once per try. The scheduler refreshesqueued_dttmevery time it queues a task, so each transition into RUNNING that followed a queueing reports a wait of its own: first run, retry, resume from deferral, and reschedule-mode poke.The emit is skipped only when:
queued_dttmis unset — runs that skip the scheduler's queueing, e.g.dag.test();It is sent after everything in the request that can fail with a 5xx: the SDK retries those, and the rollback returns the TI to QUEUED for the retry, which would otherwise sample one wait twice.
Tags mirror the sibling
task.scheduled_duration({**ti.stats_tags, "queue": ti.queue}), so the two stay sliceable the same way. The legacy dotted name is emitted automatically bystats.timingvia themetrics_template.yamlregistry — no manual second call needed.Upgrading from Airflow 2: 2.x already reported the wait on deferral resumes, but skipped retries and reschedule pokes because
end_datewas still set. Both now emit, so a reschedule sensor poking every minute for an hour reports about 60 samples where 2.x reported one. The newsfragment calls this out.Test plan
test_ti_run_emits_queued_duration_metric— parametrized over first run / retry / deferral resume, each verified to fail without the production change.test_ti_run_emits_queued_duration_for_start_from_trigger—start_from_trigger=Truetasks defer straight from SCHEDULED, so the resume is the only queue wait they have.test_ti_run_skips_queued_duration_metric_without_queued_dttmandtest_ti_run_skips_queued_duration_metric_on_duplicate_start— the two skip conditions.test_ti_run_skips_queued_duration_metric_when_the_response_fails— no sample when the request ends in a 500; fails if the emit moves back ahead of that code.test_ti_run_queued_duration_metric_is_tagged_with_team_nameandtest_ti_run_queued_duration_metric_carries_dag_tags— tags under multi-team andmetrics.dag_tags_in_metrics.airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.pypass (4 skipped, needing a MySQL/Postgres backend).ruff format/ruff check/mypy-airflow-core/prek run --from-ref upstream/main --stage pre-commitall green.closes: #63503
closes: #66067
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code (Opus 4.7) following the guidelines
Important
🛠️ Maintainer triage note for @myps6415 · by
@potiuk· 2026-07-02 17:46 UTCSome review feedback from
@ashb,@henry3260is waiting on you:@ashb,@henry3260need a reply or a fix.The ball is in your court — you've been assigned to this PR. Reply or push a fix in each thread, then mark them resolved.
Automated triage — may be imperfect; a maintainer takes the next look.