Skip to content

Fix missing task.queued_duration metric in Airflow 3 - #67592

Merged
kaxil merged 6 commits into
apache:mainfrom
myps6415:fix-queued-duration-metric-63503
Oct 14, 2026
Merged

kaxil merged 6 commits into
apache:mainfrom
myps6415:fix-queued-duration-metric-63503

Conversation

@myps6415

@myps6415 myps6415 commented May 27, 2026 •

Copy link
Copy Markdown
Contributor

Summary

task.queued_duration (and its registry-derived legacy name dag.<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 to RUNNING through the ti_run Execution 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_duration from ti_run, once per queue wait rather than once per try. The scheduler refreshes queued_dttm every 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_dttm is unset — runs that skip the scheduler's queueing, e.g. dag.test();
  • the request is a duplicate start from a client that already has the TI running.

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 by stats.timing via the metrics_template.yaml registry — 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_date was 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=True tasks defer straight from SCHEDULED, so the resume is the only queue wait they have.
  • test_ti_run_skips_queued_duration_metric_without_queued_dttm and test_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_name and test_ti_run_queued_duration_metric_carries_dag_tags — tags under multi-team and metrics.dag_tags_in_metrics.
  • All 205 runnable tests in airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py pass (4 skipped, needing a MySQL/Postgres backend).
  • ruff format / ruff check / mypy-airflow-core / prek run --from-ref upstream/main --stage pre-commit all green.

closes: #63503
closes: #66067


Was generative AI tooling used to co-author this PR?
  • Yes — Claude Code (Opus 4.7)

Generated-by: Claude Code (Opus 4.7) following the guidelines


Important

🛠️ Maintainer triage note for @myps6415 · by @potiuk · 2026-07-02 17:46 UTC

Some review feedback from @ashb, @henry3260 is waiting on you:

  • 3 unresolved review thread(s) from @ashb, @henry3260 need 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.

@boring-cyborg boring-cyborg Bot added area:API Airflow's REST/HTTP API area:task-sdk labels May 27, 2026
@boring-cyborg

boring-cyborg Bot commented May 27, 2026

Copy link
Copy Markdown

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
Here are some useful points:

  • Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
  • In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
  • Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
  • Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
  • Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
  • Be sure to read the Airflow Coding style.
  • Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
    Apache Airflow is a community-driven project and together we are making it better 🚀.
    In case of doubts contact the developers at:
    Mailing List: dev@airflow.apache.org
    Slack: https://s.apache.org/airflow-slack

@henry3260 henry3260 left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the fix!

@henry3260

Copy link
Copy Markdown
Contributor

@eladkal gentle ping! Could you help trigger CI?

@potiuk potiuk added the ready for maintainer review Set by triage when all criteria pass, or by the fork CI reconciler when fork CI is green. label May 28, 2026

@henry3260 henry3260 left a comment •

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

looks good from my side

@myps6415
myps6415 force-pushed the fix-queued-duration-metric-63503 branch 3 times, most recently from 0576c5e to 2988b04 Compare June 4, 2026 03:31
@myps6415

Copy link
Copy Markdown
Contributor Author

@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.

@myps6415
myps6415 force-pushed the fix-queued-duration-metric-63503 branch from 2988b04 to 5f57b80 Compare June 20, 2026 14:13
@myps6415

Copy link
Copy Markdown
Contributor Author

@ashb thanks for the review!

On the end_date guard (task_instances.py):
You're right. Since each retry is a separate TI row and the scheduler refreshes queued_dttm on every queueing, a retry genuinely waits in the queue and its utcnow() - queued_dttm is a valid sample (this also matches the legacy emit_state_change_metric, which fired on every transition to RUNNING). The end_date check was both fragile and semantically wrong, so I've dropped it — the guard is now just queued_dttm is not None and next_method is None, keeping only the deferral-resume skip. Also removed the now-unused TI.end_date from the select.

On the test:
Reworked it to assert the real invariant rather than the test setup: the emit case now has a retry parametrization where end_date is set on the row (the retry signature) and asserts the metric still fires — so a future regression that re-suppresses on end_date would fail here. The skip case now only covers deferral_resume and queued_dttm_missing.

@myps6415

myps6415 commented Jul 3, 2026

Copy link
Copy Markdown
Contributor Author

@ashb Gentle ping — I've addressed both your comments: dropped the end_date guard and reworked the test to assert the real invariant. Threads are resolved and CI is green. Would appreciate another look when you have a moment. Thanks!

@myps6415
myps6415 requested a review from ashb July 6, 2026 02:59
@myps6415
myps6415 force-pushed the fix-queued-duration-metric-63503 branch from 5f57b80 to 948cbcd Compare July 30, 2026 01:40
@myps6415

Copy link
Copy Markdown
Contributor Author

@kaxil all threads are resolved. The rebase reset the workflow approvals, so CI is sitting at action_required on 948cbcd — could you trigger it when you have a moment?

@myps6415
myps6415 force-pushed the fix-queued-duration-metric-63503 branch from 948cbcd to 0f3fb97 Compare August 4, 2026 03:20
@eladkal eladkal added this to the Airflow 3.3.1 milestone Aug 4, 2026
@eladkal
eladkal requested a review from kaxil August 4, 2026 04:59
@vatsrahul1001 vatsrahul1001 removed this from the Airflow 3.3.1 milestone Aug 5, 2026
@myps6415
myps6415 force-pushed the fix-queued-duration-metric-63503 branch from d78cd90 to 6cea295 Compare September 7, 2026 11:34
@myps6415

myps6415 commented Sep 7, 2026

Copy link
Copy Markdown
Contributor Author

@vatsrahul1001 done -- rebased onto main and all three threads addressed.

The conflict was with #70312, which reworked how ti_run resolves the owning team, so the diff actually got smaller: main now resolves the team itself, and the manual lookup and tag injection this PR was carrying are gone.

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 action_required on 6cea295. Could someone trigger it? I have no merged PR yet, so every run needs a manual approval.

@vatsrahul1001

Copy link
Copy Markdown
Contributor

LGTM! Can merge post review from code owners

@msumit

msumit commented Oct 1, 2026

Copy link
Copy Markdown
Contributor

LGTM! Can merge post review from code owners

@vatsrahul1001 who are the code owners here, can you tag them.

@kaxil kaxil left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.
@myps6415

myps6415 commented Oct 2, 2026

Copy link
Copy Markdown
Contributor Author

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?

@myps6415

Copy link
Copy Markdown
Contributor Author

@kaxil Updated the description as you suggested. Can this be merged?

@kaxil kaxil left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@kaxil
kaxil merged commit 21c858e into apache:main Oct 14, 2026
80 checks passed
@boring-cyborg

boring-cyborg Bot commented Oct 14, 2026

Copy link
Copy Markdown

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

@myps6415
myps6415 deleted the fix-queued-duration-metric-63503 branch October 14, 2026 02:12
myps6415 added a commit to myps6415/myps6415 that referenced this pull request Oct 14, 2026
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
myps6415 added a commit to myps6415/myps6415.github.io that referenced this pull request Oct 14, 2026
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>
@github-actions

Copy link
Copy Markdown
Contributor

Backport successfully created: v3-4-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status Branch Result
✅ v3-4-test PR Link

vatsrahul1001 pushed a commit that referenced this pull request Oct 18, 2026
) (#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>
vatsrahul1001 pushed a commit that referenced this pull request Oct 20, 2026
) (#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>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

area:API Airflow's REST/HTTP API area:task-sdk ready for maintainer review Set by triage when all criteria pass, or by the fork CI reconciler when fork CI is green. type:bug-fix Changelog: Bug Fixes

Projects

None yet

Development

Successfully merging this pull request may close these issues.

task.queued_duration metric not emitted in Airflow 3 (all executors) Metric dag.dag_id.task_id.queued_duration missing

9 participants