seanmuth commented on code in PR #73692:
URL: https://github.com/apache/airflow/pull/73692#discussion_r4145875929
##########
airflow-core/src/airflow/api_fastapi/core_api/datamodels/dags.py:
##########
@@ -221,6 +222,18 @@ class DAGDetailsResponse(DAGResponse):
owner_links: dict[str, str] | None = None
is_favorite: bool = False
active_runs_count: int = 0
+ is_at_max_active_runs: bool = Field(
+ description=(
+ "Whether this Dag currently has as many active runs as its
max_active_runs allows. "
+ "Counted differently from active_runs_count above: this counts
RUNNING and QUEUED "
+ "runs (excluding backfill runs), matching the scheduler's own
promotion check, while "
Review Comment:
Fixed the wording. It now says what the field is: whether the Dag is
currently at its `max_active_runs` limit, counting running and queued runs
(backfill runs excluded). The spec, TS clients and airflowctl are regenerated.
On whether it needs to ship: it's meant for external clients, not in-tree
readers. The main case is `schedule=None` Dags that are only triggered manually
or through the API, where the caller wants to know before triggering whether
another run would start or have to wait. RUNNING + QUEUED is the right count
for that, because queued runs already waiting get promoted first. In your
example (`max_active_runs=1`, one queued, none running), a newly triggered run
would wait, so `true` is the answer that caller needs.
The scheduler already tracks this exact state (`exceeds_max_non_backfill`,
persisted since 3.2.0). This just exposes it, computed fresh because the cached
copy goes stale on manual triggers. That costs one COUNT per details request
here, and nothing extra once #73693 folds it into its single count query.
The #73693 tooltip answers a different question, whether an existing queued
run is held back, which is why it doesn't read this field. Together the three
PRs cover each observability layer for #73686: logs (#73689), API (this PR), UI
(#73693).
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @seanmuth before posting
##########
airflow-core/src/airflow/dag_processing/collection.py:
##########
@@ -690,7 +692,7 @@ def update_dags(
dm.bundle_version = self.bundle_version
reference_run: DagRun | None = run_info.latest_run
- dm.exceeds_max_non_backfill = run_info.num_active_runs >=
dm.max_active_runs
+ dm.exceeds_max_non_backfill = active_run_counts.get(dag_id, 0) >=
dm.max_active_runs
Review Comment:
Agreed. I updated the docstrings in `test_collection.py` and `test_dag.py`.
The PR body now describes the dag-processor side as a batching change with no
scheduling effect for `schedule=None` Dags.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @seanmuth before posting
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py:
##########
@@ -1512,6 +1513,96 @@ def test_dag_details_includes_active_runs_count(self,
session, test_client):
assert isinstance(body["active_runs_count"], int)
assert body["active_runs_count"] == 0
+ def test_dag_details_includes_is_at_max_active_runs(self, session,
test_client):
+ """is_at_max_active_runs is computed fresh from real DagRuns, not a
stale cached column."""
+ dag_model = session.get(DagModel, DAG2_ID)
+ dag_model.max_active_runs = 1
+ session.add(
+ DagRun(
+ dag_id=DAG2_ID,
+ run_id="is_at_max_active_runs_running",
+ logical_date=datetime(2021, 6, 15, 4, 0, 0,
tzinfo=timezone.utc),
+ start_date=datetime(2021, 6, 15, 4, 0, 0, tzinfo=timezone.utc),
+ run_type=DagRunType.MANUAL,
+ state=DagRunState.RUNNING,
+ triggered_by=DagRunTriggeredByType.TEST,
+ )
+ )
+ session.commit()
+
+ response = test_client.get(f"/dags/{DAG2_ID}/details")
+ assert response.status_code == 200
+ body = response.json()
+
+ assert body["is_at_max_active_runs"] is True
+
+ # Test with a DAG that has not hit its max_active_runs
+ response = test_client.get(f"/dags/{DAG1_ID}/details")
+ assert response.status_code == 200
+ body = response.json()
+
+ assert body["is_at_max_active_runs"] is False
+
+ def test_dag_details_is_at_max_active_runs_counts_queued_runs_too(self,
session, test_client):
+ """A queued (non-backfill) run counts toward is_at_max_active_runs
even with 0 running."""
+ dag_model = session.get(DagModel, DAG2_ID)
+ dag_model.max_active_runs = 1
+ session.add(
+ DagRun(
+ dag_id=DAG2_ID,
+ run_id="is_at_max_active_runs_queued",
+ logical_date=datetime(2021, 6, 15, 4, 0, 0,
tzinfo=timezone.utc),
+ start_date=datetime(2021, 6, 15, 4, 0, 0, tzinfo=timezone.utc),
+ run_type=DagRunType.MANUAL,
+ state=DagRunState.QUEUED,
+ triggered_by=DagRunTriggeredByType.TEST,
+ )
+ )
+ session.commit()
+
+ response = test_client.get(f"/dags/{DAG2_ID}/details")
+ assert response.status_code == 200
+ body = response.json()
+
+ assert body["active_runs_count"] == 0
+ assert body["is_at_max_active_runs"] is True
+
+ def test_dag_details_is_at_max_active_runs_excludes_backfill_runs(self,
session, test_client):
+ """A running backfill run doesn't count toward the Dag's own
is_at_max_active_runs."""
+ from airflow.models.backfill import Backfill
Review Comment:
Dropped the `Backfill` row and its import. A `DagRun` with
`run_type=BACKFILL_JOB` covers it.
---
Drafted-by: Claude Code (Opus 5.5); reviewed by @seanmuth 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]
