Restore queued_duration metric in Airflow 3#67668
Open
wilmerdooley wants to merge 2 commits into
Open
Conversation
The dag.<dag_id>.<task_id>.queued_duration timer stopped being emitted in Airflow 3. In Airflow 2 it was sent from TaskInstance.emit_state_change_metric when the task moved to RUNNING, but in Airflow 3 that transition happens in the execution API ti_run handler rather than in the ORM, so the emit no longer fired. Its sibling scheduled_duration still works because it is emitted server side from the scheduler. Emit task.queued_duration from ti_run on the QUEUED to RUNNING transition (the metric template maps it to the legacy dag.<dag_id>.<task_id>.queued_duration name), guarded on queued_dttm being set and only on the first try (no prior end_date), matching the Airflow 2 behavior. Closes apache#63503
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The
dag.<dag_id>.<task_id>.queued_durationtimer stopped being emitted after the Airflow 3 upgrade (#63503).In Airflow 2 it was sent from
TaskInstance.emit_state_change_metricwhen the task moved to RUNNING. In Airflow 3 that transition happens in the execution APIti_runhandler rather than in the ORM, so the emit no longer fired. Its siblingscheduled_durationstill works because it is emitted server side from the scheduler on the QUEUED transition.This change emits
task.queued_durationfromti_runon the QUEUED to RUNNING transition. The metric template already mapstask.queued_durationto the legacydag.<dag_id>.<task_id>.queued_durationname, so both the tagged and statsd forms are restored. It is guarded onqueued_dttmbeing set and emitted only on the first try (no priorend_date), matching the Airflow 2 behavior.Unit tests cover the emit, the no-op when
queued_dttmis missing, the first-try-only guard, and that a duplicate run request does not double-emit.closes: #63503
Was generative AI tooling used to co-author this PR?
Generated-by: Claude Code following the guidelines