myps6415 commented on code in PR #67592:
URL: https://github.com/apache/airflow/pull/67592#discussion_r3949358805
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -242,6 +249,14 @@ def ti_run(
extra=json.dumps({"host_name": ti_run_payload.hostname}) if
ti_run_payload.hostname else None,
)
)
+ # The scheduler refreshes queued_dttm every time it queues a task, so
utcnow() - queued_dttm
+ # is a meaningful queue wait for first runs and retries alike (a retry
is a new try that
+ # genuinely waited in the queue) -- mirroring the legacy emit in
emit_state_change_metric,
+ # which fired on every transition to RUNNING. Only resumes from
deferral are skipped,
+ # identified by next_method (the trigger sets it on resume), to avoid
re-emitting within the
+ # same try. queued_dttm is None only in rare races and test setups.
+ emit_queued_duration = ti.queued_dttm is not None and ti.next_method
is None
Review Comment:
Going with your stated preference rather than waiting on a further
confirmation, since @vatsrahul1001 asked for the threads and the conflict to be
cleared:
> I would drop `and ti.next_method is None` and emit per queue wait
Done. The guard is now just `ti.queued_dttm is not None`, and the comment
names the axis and says that `task.scheduled_duration` still counts per try, so
the disagreement on retries reads as deliberate.
Two existing tests change with the axis, which is worth flagging since a
diff alone makes that look like tests bending to the code: `deferral_resume`
moves out of the skip parametrize into the emit one, and the
`start_from_trigger` test flips to a positive assertion. That second one is the
case with nothing else to measure, so it is now covered rather than merely
documented.
@henry3260 thanks for walking the resume path through the scheduler.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -297,7 +312,21 @@ def ti_run(
or 0
)
- dr.team_name = get_team_name_for_ti(task_instance_id, session)
+ team_name = get_team_name_for_ti(task_instance_id, session)
+ dr.team_name = team_name
+
+ if emit_queued_duration:
+ # Tag via dr.stats_tags so this stays sliceable the same way as
its sibling
+ # task.scheduled_duration, which emit_state_change_metric sends as
+ # {**ti.stats_tags, "queue": ti.queue} -- that is
dag_run.stats_tags plus task_id.
+ # Team lives on the Bundle rather than the DagRun schema, so
stats_tags cannot resolve
+ # it here; add the value looked up above instead. Falsy values are
pruned from
+ # stats_tags, so only set it when there is a team. The
registry-derived legacy name
+ # dag.<dag_id>.<task_id>.queued_duration is emitted by
stats.timing automatically.
+ tags = {**dr.stats_tags, "task_id": ti.task_id, "queue": ti.queue}
Review Comment:
Added, lifted from `get_running_dag_runs_to_examine` rather than rewritten:
`joinedload(DR.dag_model).selectinload(DagModel.tags)`, applied only when
`metrics.dag_tags_in_metrics` is on and the emit will actually happen.
To be accurate about the coverage: the new test asserts that dag tags reach
the metric through that route, not that the lazy load is gone -- pinning a
query count on `ti_run` would be brittle. It does mean the branch is exercised
at all, which it was not before, since no test turned the config on.
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py:
##########
@@ -297,7 +312,21 @@ def ti_run(
or 0
)
- dr.team_name = get_team_name_for_ti(task_instance_id, session)
+ team_name = get_team_name_for_ti(task_instance_id, session)
+ dr.team_name = team_name
+
+ if emit_queued_duration:
+ # Tag via dr.stats_tags so this stays sliceable the same way as
its sibling
+ # task.scheduled_duration, which emit_state_change_metric sends as
+ # {**ti.stats_tags, "queue": ti.queue} -- that is
dag_run.stats_tags plus task_id.
+ # Team lives on the Bundle rather than the DagRun schema, so
stats_tags cannot resolve
+ # it here; add the value looked up above instead. Falsy values are
pruned from
+ # stats_tags, so only set it when there is a team. The
registry-derived legacy name
+ # dag.<dag_id>.<task_id>.queued_duration is emitted by
stats.timing automatically.
+ tags = {**dr.stats_tags, "task_id": ti.task_id, "queue": ti.queue}
+ if team_name:
+ tags["team_name"] = team_name
Review Comment:
Your point outlived the code it was about. #70312 landed in the meantime:
`DagRun.team_name` is now a read-only property resolved from `dag_model`, and
`ti_run` already eager-loads the teams behind it, so the old `dr.team_name =
...` no longer even assigns -- that was the rebase conflict.
So this is now `dr._team_name = dr.team_name`, and the manual tag injection
is gone. `get_team_name_for_ti` is no longer called on this path at all, which
also drops a query per task start. The tag set is derived in one place as you
asked, and the `conf_vars` test still pins it.
--
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]
