kaxil commented on code in PR #67592:
URL: https://github.com/apache/airflow/pull/67592#discussion_r3718746713
##########
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:
On the `_team_name` point from your last reply: it is the convention the
scheduler already uses for exactly this, [`ti.dag_run._team_name =
team`](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L843)
plus two more sites, and [`stats_tags` reads it through
`getattr`](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/models/dagrun.py#L612).
Setting `dr._team_name = team_name` alongside the `dr.team_name = team_name`
above would make these three lines and the comment redundant, and keeps the tag
set derived in one place if `stats_tags` grows again. Optional, since what you
have produces the same tags today.
On the test: yes please, pin the multi-team leg with `conf_vars`. Your read
of the gap is right, and [`get_team_name_for_ti` returns `None` unless
`core.multi_team` is
on](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/api_fastapi/execution_api/security.py#L258-L259),
so today that key agrees for the wrong reason on both sides.
##########
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:
When `metrics.dag_tags_in_metrics` is on, `dr.stats_tags` costs two lazy
loads per task start here: [`dag_tags_for_stats` touches `self.dag_model` and
then
`dag_model.tags`](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/models/dagrun.py#L588-L599),
and the `dr` select above only eager-loads `consumed_asset_events`. The
earlier query does join `DagModel`, but it selects `DagModel.owners` as a
column rather than the entity, so nothing lands in the identity map and the
lazy load still fires.
[`get_running_dag_runs_to_examine` eager-loads exactly
this](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/models/dagrun.py#L772-L776)
to keep it out of the scheduler loop, and the comment on `dag_tags_for_stats`
calls the remaining paths low frequency. `ti_run` is once per task start on the
execution API, so it belongs with the first group. Adding the same eager load
to the `dr` select, conditional on the config so the join is not paid when the
feature is off, would cover it.
Non-blocking, and the config is off by default.
##########
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:
This guard came in at @henry3260's request and it does what it says, but it
leaves `task.queued_duration` and its sibling `task.scheduled_duration`
disagreeing about which transitions count, in opposite directions. Taking the
three ways a TI reaches RUNNING:
- First run: both metrics emit.
- Retry: only `queued_duration`.
[`schedule_tis`](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/models/dagrun.py#L2245-L2249)
sets `SCHEDULED`, `scheduled_dttm` and `try_number` without clearing
`end_date`, so the previous attempt's `end_date` is still on the row when the
scheduler queues the TI, and [`emit_state_change_metric` returns
early](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/models/taskinstance.py#L1497-L1500).
- Deferral resume: only `scheduled_duration`. The `DEFERRED` update never
touches `end_date` and `ti_run` already set it to `None`, so that guard passes,
and [the trigger refreshes
`scheduled_dttm`](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/models/trigger.py#L544-L545),
so the sample is a real measurement.
The resume's queue wait is equally real: [`queued_dttm` is refreshed on
every
queueing](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1045)
and [the critical section selects `SCHEDULED` with no `next_method`
filter](https://github.com/apache/airflow/blob/a424a811da80ed1e84d690adaeebe0804f57a958/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L712),
so a deferrable task sits in QUEUED again waiting for a worker slot before
`execute_complete` runs. Skipping it means that wait is never measured, and for
sensor-heavy deployments the resume leg is where most of the queue time lives.
I would drop `and ti.next_method is None` and emit per queue wait, which is
also where @ashb's `end_date` change pointed: one sample per real wait rather
than one per try. The per-try reading is defensible too, but then the retry
case should not emit either, and the comment should name the axis so the
asymmetry reads as deliberate.
Either answer works for me, and this is the only thing I would like settled
before merge. Adding these samples after release shifts percentiles for anyone
alerting on the timer, which is cheap to decide now and awkward to change later.
On why this did not come up in my last pass: I was looking at the tag set
then, and only walked the resume path through the scheduler this time.
--
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]
