Add executor.running_dags gauge to expose count of running DAGs (#12368) (#52815)

* feature: introducing a counter for number of DAGs in running state

* wrap emit executor.running_dags gague as a function

* add test for _emit_running_dags_metric

* add executor.running_dags to metrics.rst

* use Sqlalchemy 2.0 style

* check for metrics configuration before emitting running dags metric

* Enable metrics configuration for running dags in DagFileProcessorManager test

* Add configuration for multi-team support in test_bundles_with_team

* add go-sdk/bin/.gitignore back

* Update .gitignore to include all files except itself

* Refactor running DAGs metric emission: move to SchedulerJobRunner and update tests

* Add running DAGs metric and remove unused metrics check

* Add dagrun_metrics_interval to config and update running DAGs metric to use 'scheduler' namespace

* Remove unused import of DagRunState in test_emit_running_dags_metric

* Fix Low dep tests:core

* Rename running DAGs metric to 'scheduler.dagruns.running' for consistency and update related test assertions

Author: nailo2c

airflow-core/docs/administration-and-deployment/logging-monitoring/metrics.rst

@@ -236,6 +236,7 @@ Name                                                 Description
 ``scheduler.tasks.executable``                       Number of tasks that are ready for execution (set to queued)
                                                      with respect to pool limits, Dag concurrency, executor state,
                                                      and priority.
+``scheduler.dagruns.running``                           Number of DAGs whose latest DagRun is currently in the ``RUNNING`` state
 ``executor.open_slots.<executor_class_name>``        Number of open slots on a specific executor. Only emitted when multiple executors are configured.
 ``executor.open_slots``                              Number of open slots on executor
 ``executor.queued_tasks.<executor_class_name>``      Number of queued tasks on on a specific executor. Only emitted when multiple executors are configured.

airflow-core/src/airflow/config_templates/config.yml

@@ -2088,6 +2088,14 @@ scheduler:
       type: float
       example: ~
       default: "30.0"
+    dagrun_metrics_interval:
+      description: |
+        How often (in seconds) the scheduler emits metrics on running DAG runs
+        to StatsD (if statsd_on is enabled)
+      version_added: 3.1.0
+      type: float
+      example: ~
+      default: "30.0"
     scheduler_health_check_threshold:
       description: |
         If the last scheduler heartbeat happened more than ``[scheduler] scheduler_health_check_threshold``

airflow-core/src/airflow/jobs/scheduler_job_runner.py

@@ -1339,6 +1339,11 @@ def _run_scheduler_loop(self) -> None:
                 self._emit_running_ti_metrics,
             )
 
+            timers.call_regular_interval(
+                conf.getfloat("scheduler", "dagrun_metrics_interval", fallback=30.0),
+                self._emit_running_dags_metric,
+            )
+
         timers.call_regular_interval(
             conf.getfloat("scheduler", "task_instance_heartbeat_timeout_detection_interval", fallback=10.0),
             self._find_and_purge_task_instances_without_heartbeats,
@@ -2276,6 +2281,12 @@ def _emit_running_ti_metrics(self, session: Session = NEW_SESSION) -> None:
 
         self.previous_ti_running_metrics = ti_running_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 = session.scalar(stmt)
+        Stats.gauge("scheduler.dagruns.running", running_dags)
+
     @provide_session
     def _emit_pool_metrics(self, session: Session = NEW_SESSION) -> None:
         from airflow.models.pool import Pool

airflow-core/tests/unit/jobs/test_scheduler_job.py

@@ -7165,6 +7165,29 @@ def test_process_expired_deadlines_no_deadlines_found(self, mock_handle_miss, se
         # The handler should not be called, but no exceptions should be raised either.`
         mock_handle_miss.assert_not_called()
 
+    def test_emit_running_dags_metric(self, dag_maker, monkeypatch):
+        """Test that the running_dags metric is emitted correctly."""
+        with dag_maker("metric_dag") as dag:
+            _ = dag
+        dag_maker.create_dagrun(run_id="run_1", state=DagRunState.RUNNING, logical_date=timezone.utcnow())
+        dag_maker.create_dagrun(
+            run_id="run_2", state=DagRunState.RUNNING, logical_date=timezone.utcnow() + timedelta(hours=1)
+        )
+
+        recorded: list[tuple[str, int]] = []
+
+        def _fake_gauge(metric: str, value: int, *_, **__):
+            recorded.append((metric, value))
+
+        monkeypatch.setattr("airflow.jobs.scheduler_job_runner.Stats.gauge", _fake_gauge, raising=True)
+
+        with conf_vars({("metrics", "statsd_on"): "True"}):
+            scheduler_job = Job()
+            self.job_runner = SchedulerJobRunner(scheduler_job)
+            self.job_runner._emit_running_dags_metric()
+
+        assert recorded == [("scheduler.dagruns.running", 2)]
+
 
 @pytest.mark.need_serialized_dag
 def test_schedule_dag_run_with_upstream_skip(dag_maker, session):