henry3260 commented on code in PR #67592:
URL: https://github.com/apache/airflow/pull/67592#discussion_r3311945730
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -230,6 +234,17 @@ def ti_run(
extra=json.dumps({"host_name": ti_run_payload.hostname}) if
ti_run_payload.hostname else None,
)
)
+ # Emit task.queued_duration on the first round of a try only — mirrors
the skip
+ # logic from TaskInstance.emit_state_change_metric (an existing
end_date means
+ # this is a deferral resume or similar, and the timing would be
misleading).
+ # The registry-based legacy name
dag.<dag_id>.<task_id>.queued_duration is
+ # emitted automatically by stats.timing via metrics_template.yaml.
+ if ti.queued_dttm is not None and ti.end_date is None:
Review Comment:
Could we include `ti.next_method is None` in the guard to avoid emitting
this metric again for deferred task resumes? In the Execution API path,
deferred tasks are marked by `next_method` / `next_kwargs`, and `end_date` may
still be `None`, so `end_date is None` alone does not reliably identify the
first run. wdyt?
--
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]
