This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 21c858eb67b Fix missing task.queued_duration metric in Airflow 3
(#67592)
21c858eb67b is described below
commit 21c858eb67bbdd9115d5a5dff64c1c720030d58e
Author: Tung, Hsiao-Yu <[email protected]>
AuthorDate: Wed Oct 14 08:29:54 2026 +0800
Fix missing task.queued_duration metric in Airflow 3 (#67592)
* 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.
---
airflow-core/newsfragments/67592.bugfix.rst | 1 +
.../execution_api/routes/task_instances.py | 39 ++-
.../versions/head/test_task_instances.py | 360 +++++++++++++++++++++
3 files changed, 399 insertions(+), 1 deletion(-)
diff --git a/airflow-core/newsfragments/67592.bugfix.rst
b/airflow-core/newsfragments/67592.bugfix.rst
new file mode 100644
index 00000000000..e9b1604aa6a
--- /dev/null
+++ b/airflow-core/newsfragments/67592.bugfix.rst
@@ -0,0 +1 @@
+Restore the ``task.queued_duration`` metric, which stopped being emitted when
Airflow 3 workers moved to the Task SDK, and record it once per queue wait.
Airflow 2 skipped retries and reschedule-mode sensor pokes, because
``end_date`` was still set from the previous attempt or poke, and both now emit
a sample of their own: a sensor poking every minute for an hour reports about
60 samples where Airflow 2 reported one. Alerts and dashboards carried over
from Airflow 2 that aggregate this t [...]
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 3f462882ed9..3f234b4c40b 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -43,6 +43,7 @@ from sqlalchemy.sql import select
from sqlalchemy.sql.dml import Update
from structlog.contextvars import bind_contextvars
+from airflow._shared.observability.metrics import stats
from airflow._shared.observability.traces import override_ids
from airflow._shared.state import TaskScope
from airflow._shared.timezones import timezone
@@ -184,6 +185,8 @@ def ti_run(
TI.try_number,
TI.max_tries,
TI.start_date,
+ TI.queue,
+ TI.queued_dttm,
TI.next_method,
TI.hostname,
TI.unixname,
@@ -225,6 +228,10 @@ def ti_run(
query = update(TI).where(TI.id == task_instance_id).values(data)
previous_state = ti.state
+ # Set on a genuine QUEUED -> RUNNING transition so task.queued_duration is
emitted once the
+ # DagRun is loaded below (its stats_tags supply the tags). A duplicate
start request falls
+ # through the branch below without raising, so the flag keeps it from
emitting twice.
+ emit_queued_duration = False
if previous_state == TaskInstanceState.RESTARTING and (ti.hostname,
ti.unixname, ti.pid) != (
ti_run_payload.hostname,
@@ -274,6 +281,13 @@ def ti_run(
extra=json.dumps({"host_name": ti_run_payload.hostname}) if
ti_run_payload.hostname else None,
)
)
+ # One sample per queue wait, not per try: the scheduler refreshes
queued_dttm on every
+ # queueing, so a retry and a resume from deferral each waited for a
slot of their own.
+ # task.scheduled_duration skips retries (emit_state_change_metric
returns early while
+ # end_date is set), so the two disagree on retries by design.
+ # queued_dttm is unset only for runs that skip the scheduler's
queueing, e.g. dag.test().
+ emit_queued_duration = ti.queued_dttm is not None
+
# Ensure there is no end date set and clear retry policy overrides from
the previous attempt.
query = query.values(
end_date=None,
@@ -290,11 +304,23 @@ def ti_run(
result = session.execute(query)
log.info("Task instance state updated", rows_affected=getattr(result,
"rowcount", 0))
+ # stats_tags lazy-loads dag_model.tags when dag tags are emitted as
metric tags, and ti_run
+ # runs once per task start -- load them up front as the scheduler loop
does.
+ dag_tag_options = (
+ (joinedload(DR.dag_model).selectinload(DagModel.tags),)
+ if emit_queued_duration and conf.getboolean("metrics",
"dag_tags_in_metrics", fallback=False)
+ else ()
+ )
+
dr = (
session.scalars(
select(DR)
.filter_by(dag_id=ti.dag_id, run_id=ti.run_id)
- .options(joinedload(DR.consumed_asset_events),
*eager_load_teams(DR.dag_model))
+ .options(
+ joinedload(DR.consumed_asset_events),
+ *eager_load_teams(DR.dag_model),
+ *dag_tag_options,
+ )
)
.unique()
.one_or_none()
@@ -393,6 +419,17 @@ def ti_run(
if token.claims.scope == "workload":
issue_execution_token(services, response, sub=str(task_instance_id))
+ if emit_queued_duration:
+ # Emitted last so a 5xx cannot double-count the wait: the SDK retries
those, and the
+ # rollback returns the TI to QUEUED for the retry to sample it again.
Only the session
+ # commit can still fail past this point.
+ # Tags mirror the sibling task.scheduled_duration, which
emit_state_change_metric sends
+ # as {**ti.stats_tags, "queue": ti.queue}; stats_tags reads the team
off the transient
+ # _team_name. stats.timing also emits the legacy dotted name from the
metrics registry.
+ dr._team_name = dr.team_name
+ tags = {**dr.stats_tags, "task_id": ti.task_id, "queue": ti.queue}
+ stats.timing("task.queued_duration", timezone.utcnow() -
ti.queued_dttm, tags=tags)
+
return context
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 40b7f029a50..2a6e703e098 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -1886,6 +1886,364 @@ class TestTIRunState:
assert logs[0].owner == ti.task.owner
assert logs[0].extra == '{"host_name": "random-hostname"}'
+ @pytest.mark.parametrize(
+ "scenario",
+ ["first_run", "retry", "deferral_resume"],
+ )
+ def test_ti_run_emits_queued_duration_metric(
+ self, client, session, create_task_instance, time_machine, scenario
+ ):
+ """task.queued_duration is emitted once per queue wait.
+
+ The scheduler refreshes queued_dttm every time it queues a task, so a
retry and a
+ resume from deferral each waited for a worker slot of their own and
must emit too. A
+ retry still carries the previous attempt's end_date on the row when
ti_run is reached,
+ so this also asserts the emit does not depend on end_date being unset.
+ """
+ queued_at = timezone.parse("2024-09-30T12:00:00Z")
+ run_at = queued_at.add(seconds=42)
+
+ ti = create_task_instance(
+ task_id=f"test_ti_run_emits_queued_duration_metric_{scenario}",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=queued_at,
+ dag_id=str(uuid4()),
+ )
+ ti.queued_dttm = queued_at
+ ti.queue = "default"
+ if scenario == "retry":
+ # A retried TI still has the previous attempt's end_date set on
the row until
+ # ti_run clears it; the metric must fire regardless.
+ ti.end_date = queued_at.add(seconds=10)
+ elif scenario == "deferral_resume":
+ ti.next_method = "execute_complete"
+ session.commit()
+
+ # The metric has to stay sliceable the same way as its sibling
task.scheduled_duration,
+ # which emit_state_change_metric sends as {**ti.stats_tags, "queue":
ti.queue}. Deriving
+ # the expectation from that same expression makes the two drift apart
only if this fails.
+ expected_tags = {**ti.stats_tags, "queue": ti.queue}
+ assert "run_type" in expected_tags
+
+ time_machine.move_to(run_at, tick=False)
+
+ with mock.patch(
+ "airflow.api_fastapi.execution_api.routes.task_instances.stats",
autospec=True
+ ) as mock_stats:
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "random-hostname",
+ "unixname": "random-unixname",
+ "pid": 100,
+ "start_date": run_at.isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ mock_stats.timing.assert_called_once_with(
+ "task.queued_duration",
+ run_at - queued_at,
+ tags=expected_tags,
+ )
+
+ @mock.patch(
+
"airflow.api_fastapi.execution_api.routes.task_instances.get_arg_bindings",
+ autospec=True,
+ return_value=[{"name": "country", "kind": "hologram", "value": "uk"}],
+ )
+ def test_ti_run_skips_queued_duration_metric_when_the_response_fails(
+ self, _, client, session, create_task_instance, time_machine
+ ):
+ """The SDK retries a 5xx, and the rollback leaves the TI in QUEUED for
that retry to pick
+ up, so a sample sent before the response is settled counts one queue
wait twice."""
+ queued_at = timezone.parse("2024-09-30T12:00:00Z")
+ run_at = queued_at.add(seconds=42)
+ time_machine.move_to(run_at, tick=False)
+
+ ti = create_task_instance(
+
task_id="test_ti_run_skips_queued_duration_metric_when_the_response_fails",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=queued_at,
+ dag_id=str(uuid4()),
+ )
+ ti.queued_dttm = queued_at
+ session.commit()
+
+ with mock.patch(
+ "airflow.api_fastapi.execution_api.routes.task_instances.stats",
autospec=True
+ ) as mock_stats:
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "random-hostname",
+ "unixname": "random-unixname",
+ "pid": 100,
+ "start_date": run_at.isoformat(),
+ },
+ )
+
+ assert response.status_code == 500
+ assert response.json()["detail"]["reason"] == "invalid_arg_bindings"
+ mock_stats.timing.assert_not_called()
+
+ def test_ti_run_skips_queued_duration_metric_without_queued_dttm(
+ self, client, session, create_task_instance, time_machine
+ ):
+ """queued_dttm is what the wait is measured from, so a row without one
(a run that skips
+ the scheduler's queueing, e.g. dag.test()) has nothing to report."""
+ queued_at = timezone.parse("2024-09-30T12:00:00Z")
+ run_at = queued_at.add(seconds=42)
+ time_machine.move_to(run_at, tick=False)
+
+ ti = create_task_instance(
+
task_id="test_ti_run_skips_queued_duration_metric_without_queued_dttm",
+ state=State.QUEUED,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=queued_at,
+ dag_id=str(uuid4()),
+ )
+ ti.queued_dttm = None
+ session.commit()
+
+ with mock.patch(
+ "airflow.api_fastapi.execution_api.routes.task_instances.stats",
autospec=True
+ ) as mock_stats:
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "random-hostname",
+ "unixname": "random-unixname",
+ "pid": 100,
+ "start_date": run_at.isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ mock_stats.timing.assert_not_called()
+
+ def test_ti_run_skips_queued_duration_metric_on_duplicate_start(
+ self, client, session, create_task_instance, time_machine
+ ):
+ """A start request replayed by the same worker after a network glitch
returns the same
+ 200 without re-running the transition, so it must not double-count the
queue wait."""
+ queued_at = timezone.parse("2024-09-30T12:00:00Z")
+ run_at = queued_at.add(seconds=42)
+ time_machine.move_to(run_at, tick=False)
+
+ ti = create_task_instance(
+
task_id="test_ti_run_skips_queued_duration_metric_on_duplicate_start",
+ state=State.RUNNING,
+ dagrun_state=DagRunState.RUNNING,
+ session=session,
+ start_date=queued_at,
+ dag_id=str(uuid4()),
+ )
+ ti.queued_dttm = queued_at
+ ti.hostname = "random-hostname"
+ ti.unixname = "random-unixname"
+ ti.pid = 100
+ session.commit()
+
+ with mock.patch(
+ "airflow.api_fastapi.execution_api.routes.task_instances.stats",
autospec=True
+ ) as mock_stats:
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "random-hostname",
+ "unixname": "random-unixname",
+ "pid": 100,
+ "start_date": run_at.isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ mock_stats.timing.assert_not_called()
+
+ def test_ti_run_queued_duration_metric_is_tagged_with_team_name(
+ self, client, session, dag_maker, time_machine
+ ):
+ """The team tag has to be pinned with multi-team on: with it off both
the metric and any
+ expectation derived from stats_tags omit the key, so they would agree
for the wrong reason.
+ """
+ from airflow.models.dagbundle import DagBundleModel
+ from airflow.models.team import Team
+
+ queued_at = timezone.parse("2024-09-30T12:00:00Z")
+ run_at = queued_at.add(seconds=42)
+ time_machine.move_to(queued_at, tick=False)
+
+ dag_id = str(uuid4())
+ with dag_maker(dag_id=dag_id, session=session):
+ EmptyOperator(task_id="task")
+ dr = dag_maker.create_dagrun(
+ run_id="test", logical_date=queued_at, state=DagRunState.RUNNING,
start_date=queued_at
+ )
+ ti = dr.get_task_instance(task_id="task")
+ session.execute(
+ update(TaskInstance)
+ .where(TaskInstance.id == ti.id)
+ .values(state=State.QUEUED, queued_dttm=queued_at, queue="default")
+ )
+
+ bundle_name = f"bundle-{dag_id}"
+ team_name = f"team-{dag_id[:8]}"
+ bundle = DagBundleModel(name=bundle_name)
+ bundle.teams.append(Team(name=team_name))
+ session.add(bundle)
+ session.flush()
+ session.execute(update(DagModel).where(DagModel.dag_id ==
dag_id).values(bundle_name=bundle_name))
+ session.commit()
+
+ time_machine.move_to(run_at, tick=False)
+
+ with conf_vars({("core", "multi_team"): "True"}):
+ with mock.patch(
+
"airflow.api_fastapi.execution_api.routes.task_instances.stats", autospec=True
+ ) as mock_stats:
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "random-hostname",
+ "unixname": "random-unixname",
+ "pid": 100,
+ "start_date": run_at.isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ mock_stats.timing.assert_called_once()
+ assert mock_stats.timing.call_args.kwargs["tags"]["team_name"] ==
team_name
+
+ def test_ti_run_queued_duration_metric_carries_dag_tags(self, client,
session, dag_maker, time_machine):
+ """With dag_tags_in_metrics on, stats_tags would lazy-load
dag_model.tags per task start;
+ the select eager-loads them instead, so the tags have to survive that
route."""
+ queued_at = timezone.parse("2024-09-30T12:00:00Z")
+ run_at = queued_at.add(seconds=42)
+ time_machine.move_to(queued_at, tick=False)
+
+ dag_id = str(uuid4())
+ with dag_maker(dag_id=dag_id, tags=["critical", "env:prod"],
session=session):
+ EmptyOperator(task_id="task")
+ dr = dag_maker.create_dagrun(
+ run_id="test", logical_date=queued_at, state=DagRunState.RUNNING,
start_date=queued_at
+ )
+ ti = dr.get_task_instance(task_id="task")
+ session.execute(
+ update(TaskInstance)
+ .where(TaskInstance.id == ti.id)
+ .values(state=State.QUEUED, queued_dttm=queued_at, queue="default")
+ )
+ session.commit()
+
+ time_machine.move_to(run_at, tick=False)
+
+ with conf_vars({("metrics", "dag_tags_in_metrics"): "True"}):
+ with mock.patch(
+
"airflow.api_fastapi.execution_api.routes.task_instances.stats", autospec=True
+ ) as mock_stats:
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "random-hostname",
+ "unixname": "random-unixname",
+ "pid": 100,
+ "start_date": run_at.isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ mock_stats.timing.assert_called_once()
+ tags = mock_stats.timing.call_args.kwargs["tags"]
+ assert (tags["critical"], tags["env"]) == ("", "prod")
+
+ def test_ti_run_emits_queued_duration_for_start_from_trigger(
+ self, client, session, dag_maker, time_machine
+ ):
+ """A start_from_trigger operator is deferred straight from SCHEDULED
and only reaches the
+ queue once the trigger fires, so that resume is the one wait it ever
has to report.
+ """
+ from airflow.sdk import BaseOperator
+ from airflow.triggers.base import StartTriggerArgs
+
+ class StartFromTriggerOperator(BaseOperator):
+ start_trigger_args = StartTriggerArgs(
+ trigger_cls="airflow.triggers.testing.SuccessTrigger",
+ trigger_kwargs={},
+ next_method="execute_complete",
+ timeout=None,
+ )
+ start_from_trigger = True
+
+ def execute_complete(self):
+ pass
+
+ queued_at = timezone.parse("2024-09-30T12:00:00Z")
+ run_at = queued_at.add(seconds=42)
+ time_machine.move_to(queued_at, tick=False)
+
+ with dag_maker(dag_id=str(uuid4()), session=session):
+ StartFromTriggerOperator(task_id="task")
+ dr = dag_maker.create_dagrun(run_id="test", logical_date=queued_at,
state=DagRunState.RUNNING)
+
+ ti = dr.get_task_instance(task_id="task")
+ ti.task = dr.dag.get_task("task")
+ assert dr.schedule_tis((ti,), session=session) == 0
+ session.merge(ti)
+ session.commit()
+
+ # Read the row back rather than the in-memory TI: what ti_run sees is
the persisted state,
+ # and the point of this test is that the deferral lands without the TI
ever being queued.
+ state, next_method, queued_dttm = session.execute(
+ select(TaskInstance.state, TaskInstance.next_method,
TaskInstance.queued_dttm).where(
+ TaskInstance.id == ti.id
+ )
+ ).one()
+ assert (state, next_method, queued_dttm) ==
(TaskInstanceState.DEFERRED, "execute_complete", None)
+
+ # The trigger fires and the scheduler queues it -- the first and only
real queue wait.
+ session.execute(
+ update(TaskInstance)
+ .where(TaskInstance.id == ti.id)
+ .values(state=State.QUEUED, queued_dttm=queued_at, queue="default")
+ )
+ session.commit()
+
+ time_machine.move_to(run_at, tick=False)
+
+ with mock.patch(
+ "airflow.api_fastapi.execution_api.routes.task_instances.stats",
autospec=True
+ ) as mock_stats:
+ response = client.patch(
+ f"/execution/task-instances/{ti.id}/run",
+ json={
+ "state": "running",
+ "hostname": "random-hostname",
+ "unixname": "random-unixname",
+ "pid": 100,
+ "start_date": run_at.isoformat(),
+ },
+ )
+
+ assert response.status_code == 200
+ # Assert off the call rather than the in-memory TI, whose queue is
stale after the
+ # raw update above.
+ mock_stats.timing.assert_called_once()
+ assert mock_stats.timing.call_args.args == ("task.queued_duration",
run_at - queued_at)
+ assert mock_stats.timing.call_args.kwargs["tags"]["queue"] == "default"
+
class TestTIUpdateState:
def test_loop_continues_while_earlier_body_branch_is_running(self, client,
session, dag_maker):
@@ -3358,6 +3716,8 @@ class TestTIUpdateState:
try_number=1,
max_tries=0,
start_date=None,
+ queue="default",
+ queued_dttm=timezone.utcnow(),
next_method=None,
hostname=None,
unixname=None,
