potiuk commented on code in PR #70013:
URL: https://github.com/apache/airflow/pull/70013#discussion_r3682277086
##########
shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml:
##########
@@ -459,10 +459,16 @@ metrics:
name_variables: []
- name: "scheduler.dagruns.running"
- description: "Number of DAGs whose latest DagRun is currently in the
``RUNNING`` state"
+ description: "Number of DagRuns currently in the ``RUNNING`` state, tagged
by dag_id."
type: "gauge"
- legacy_name: "-"
- name_variables: []
+ legacy_name: "scheduler.dagruns.running.{dag_id}"
Review Comment:
`legacy_name` changes from `"-"` to `"scheduler.dagruns.running.{dag_id}"`,
which means on the legacy (non-tagged) StatsD path the **metric name itself**
now embeds the dag_id. Anyone with a dashboard or alert on
`scheduler.dagruns.running` stops receiving that series entirely — it isn't a
re-tagging, the old name ceases to exist.
That's a user-visible breaking change to a published metric and needs a
newsfragment in `airflow-core/newsfragments/` (probably `.significant.rst`)
spelling out what happens to existing dashboards and what to migrate to.
There's no newsfragment in the PR at present.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -3245,10 +3245,17 @@ def _emit_ti_metrics(self, *, session: Session =
NEW_SESSION) -> None:
self.previous_ti_metrics[state] = ti_metrics
@provide_session
- def _emit_running_dags_metric(self, *, session: Session = NEW_SESSION) ->
None:
- stmt = select(func.count()).select_from(DagRun).where(DagRun.state ==
DagRunState.RUNNING)
- running_dags = float(session.scalar(stmt) or 0)
- stats.gauge("scheduler.dagruns.running", running_dags)
+ def _emit_dag_runs_metric(self, *, session: Session = NEW_SESSION) -> None:
+ stmt = (
+ select(DagRun.dag_id, DagRun.state, func.count().label("count"))
+ .where(DagRun.state.in_([DagRunState.RUNNING, DagRunState.QUEUED]))
+ .group_by(DagRun.dag_id, DagRun.state)
+ )
+ for dag_id, state, count in session.execute(stmt).all():
Review Comment:
This emits one gauge per `(dag_id, state)` pair on **every scheduler loop**.
On a deployment with 10k Dags that's up to 20k time series per interval where
there was previously exactly one.
That's a well-known way to overwhelm a StatsD daemon or blow up Prometheus
cardinality, and the operators it hits hardest are precisely the large
deployments that can least afford it. The blast radius is also invisible from
the diff — nothing here scales with deployment size on the page, but it does at
runtime.
At minimum this wants to be opt-in behind a config flag (defaulting to the
current aggregate behaviour), or to keep emitting the untagged aggregate and
add the per-dag breakdown separately. Worth a note in the description about
expected cardinality either way.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
--
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]
