kaxil commented on code in PR #67668:
URL: https://github.com/apache/airflow/pull/67668#discussion_r3677728067
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -258,6 +262,20 @@ def ti_run(
result = session.execute(query)
log.info("Task instance state updated", rows_affected=getattr(result,
"rowcount", 0))
+ # queued_duration was historically emitted on the move to RUNNING; in
Airflow 3 that
+ # transition happens here rather than in the ORM, so emit it here
(mirrors scheduled_duration).
+ # Emit only on the first try (no prior end_date), matching the Airflow
2 behavior.
+ if (
+ previous_state in (TaskInstanceState.QUEUED,
TaskInstanceState.RESTARTING)
+ and ti.queued_dttm
+ and ti.end_date is None
Review Comment:
The mechanics work as written, and the comment above is accurate: a retry
really does still carry the previous attempt's `end_date` when `ti_run` is next
reached (`ti.end_date = ti_patch_payload.end_date` is set on the retry
transition, and `prepare_db_for_next_try` then rotates `id` via `uuid7()`
without clearing it), and Airflow 2 was first-try-only for the same reason --
`emit_state_change_metric` returns early on `if self.end_date`.
The issue is that @ashb made the opposite call on #67592, the other open fix
for #63503:
https://github.com/apache/airflow/pull/67592#discussion_r3435464094. The
scheduler refreshes `queued_dttm` on every enqueue, so a retry genuinely waited
in the queue and that sample is a real one worth keeping. That PR dropped the
guard in response. Worth aligning rather than shipping two different answers to
the same question. If the guard goes, `TI.end_date` comes back out of the
select and `test_ti_run_skips_queued_duration_on_retry` goes with it.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -258,6 +262,20 @@ def ti_run(
result = session.execute(query)
log.info("Task instance state updated", rows_affected=getattr(result,
"rowcount", 0))
+ # queued_duration was historically emitted on the move to RUNNING; in
Airflow 3 that
+ # transition happens here rather than in the ORM, so emit it here
(mirrors scheduled_duration).
+ # Emit only on the first try (no prior end_date), matching the Airflow
2 behavior.
+ if (
+ previous_state in (TaskInstanceState.QUEUED,
TaskInstanceState.RESTARTING)
+ and ti.queued_dttm
+ and ti.end_date 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:
These tags don't line up with `task.scheduled_duration`, which goes out as
`{**ti.stats_tags, "queue": ti.queue}`: `dag_id`, `task_id`, `queue`,
`run_type`, plus `team_name` under multi-team and Dag tags when that config is
enabled. In Airflow 2 both metrics shared that single emit path in
`emit_state_change_metric`, so `queued_duration` carried `run_type` as well. As
written the restored metric can't be sliced the same way as its sibling.
`DR.run_type` is a free add to the select since DR is already joined, and
`get_team_name_for_ti` is called about 40 lines below for `dr.team_name`, so
emitting after that point would reuse the value instead of paying for a second
query.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -258,6 +262,20 @@ def ti_run(
result = session.execute(query)
log.info("Task instance state updated", rows_affected=getattr(result,
"rowcount", 0))
+ # queued_duration was historically emitted on the move to RUNNING; in
Airflow 3 that
+ # transition happens here rather than in the ORM, so emit it here
(mirrors scheduled_duration).
+ # Emit only on the first try (no prior end_date), matching the Airflow
2 behavior.
Review Comment:
It isn't first-try-only in practice: a resume from deferral reaches this
again. The `DEFERRED` transition writes only `state`, `trigger_id`,
`next_method`, `next_kwargs` and `trigger_timeout`, and `ti_run` already
cleared `end_date` when the attempt started, so on resume `end_date` is `None`,
the scheduler has refreshed `queued_dttm`, and `previous_state` is `QUEUED`.
Every deferral cycle adds another sample inside the same try.
`ti.next_method` is already in the select and used further down, so `and
ti.next_method is None` covers it. @henry3260 raised the same point on #67592:
https://github.com/apache/airflow/pull/67592#discussion_r3311945730
--
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]
