You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "nailo2c (via GitHub)" <gi...@apache.org> on 2025/07/03 17:20:26 UTC
[PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
nailo2c opened a new pull request, #52815:
URL: https://github.com/apache/airflow/pull/52815
Closes: #12368
# Why
Addresses #12368, expose a gauge metric `executor.running_dags` so operators can monitor how many DAGs are currently running.
# How
Add a lightweight metric emitter `_emit_running_dags_metric()`
# What
I started a local Breeze environment with StatsD enabled: `breeze start-airflow --integration statsd`
I then used the following example DAG to verify that the new metric is emitted:
```python
import pendulum
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.empty import EmptyOperator
def build_simple_dag(dag_id: str) -> DAG:
with DAG(
dag_id = dag_id,
schedule = "@daily",
start_date = pendulum.now("UTC").subtract(days=10),
catchup = True,
tags = ["statsd-demo"],
) as dag:
start = EmptyOperator(task_id="start")
sleep = BashOperator(task_id="sleep_5m", bash_command="sleep 300")
finish = EmptyOperator(task_id="finish")
start >> sleep >> finish
return dag
for i in range(2):
globals()[f"example_statsd_{i}"] = build_simple_dag(f"example_statsd_{i}")
```
StatsD: http://localhost:29102/metrics
Grafana: http://localhost:23000
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #52815:
URL: https://github.com/apache/airflow/pull/52815#issuecomment-3422360966
> Hi @vincbeck, after rebasing my PR, I'm seeing a failure in `test_bundles_with_team` (added ~4 weeks ago). I'm getting the following error and I'm not sure what I'm missing:
>
> ```
> FAILED airflow-core/tests/unit/dag_processing/test_manager.py::TestDagFileProcessorManager::test_bundles_with_team - airflow.exceptions.AirflowConfigException:
> Section `dag_processor` key `dag_bundle_config_list` cannot have a team name when multi-team mode is disabled.To enable multi-team, you need to update section `core` key `multi_team` in your config.
> ```
>
> Do you have any pointers on how I should address this in my PR? Thanks!
I do not have this failure. Can you try to rebase?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2273479802
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1018,6 +1022,14 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _emit_running_dags_metric(self):
+ """Emit executor.running_dags gauge."""
+ with create_session() as session:
+ running_dags = (
+ session.query(func.count(DagRun.dag_id)).filter(DagRun.state == DagRunState.RUNNING).scalar()
Review Comment:
Nice suggestion! Let me work on it later
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2462185588
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1024,6 +1024,13 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _is_metrics_enabled(self):
Review Comment:
It should be removed since I migrate the logic from `manager.py` to `scheduler_job_runner.py`, let me update 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2470347371
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2235,6 +2240,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.running_dags", running_dags)
Review Comment:
I like `scheduler.dagruns.running`!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on PR #52815:
URL: https://github.com/apache/airflow/pull/52815#issuecomment-3417672543
Hi @vincbeck, after rebasing my PR, I'm seeing a failure in `test_bundles_with_team` (added ~4 weeks ago). I'm getting the following error and I'm not sure what I'm missing:
```console
FAILED airflow-core/tests/unit/dag_processing/test_manager.py::TestDagFileProcessorManager::test_bundles_with_team - airflow.exceptions.AirflowConfigException:
Section `dag_processor` key `dag_bundle_config_list` cannot have a team name when multi-team mode is disabled.To enable multi-team, you need to update section `core` key `multi_team` in your config.
```
Do you have any pointers on how I should address this in my PR? Thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2462183397
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2235,6 +2240,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("executor.running_dags", running_dags)
Review Comment:
Thanks for the review. I kept the metric name `executor.running_dags` because it was used in the original issue. Since we all agree to emit it in `scheduler_job_runner.py`, it makes more sense to place it under `scheduler`. I'll update the PR later.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #52815:
URL: https://github.com/apache/airflow/pull/52815
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2448044801
##########
go-sdk/bin/.gitignore:
##########
Review Comment:
Not sure you want to delete 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on PR #52815:
URL: https://github.com/apache/airflow/pull/52815#issuecomment-3445244747
Thanks for the reviews!
Metric renamed to `airflow_scheduler_running_dags` and `dagrun_metrics_interval` added to `config.yml`. StatsD + Grafana now show the gauge correctly (see screenshot). Let me know if anything else is needed 🙌
StatsD
<img width="716" height="86" alt="image" src="https://github.com/user-attachments/assets/70876dba-8e58-4731-bbff-f281c4186a01" />
Grafana
<img width="1496" height="473" alt="image" src="https://github.com/user-attachments/assets/f7eef305-9754-4842-8397-6f8606cea00a" />
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2272232990
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1018,6 +1022,14 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _emit_running_dags_metric(self):
+ """Emit executor.running_dags gauge."""
+ with create_session() as session:
+ running_dags = (
+ session.query(func.count(DagRun.dag_id)).filter(DagRun.state == DagRunState.RUNNING).scalar()
Review Comment:
Let's change this query to use Sqlalchemy 2.0 style
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2314605866
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1014,6 +1018,13 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _emit_running_dags_metric(self):
+ """Emit executor.running_dags gauge."""
+ with create_session() as session:
+ stmt = select(func.count()).select_from(DagRun).where(DagRun.state == DagRunState.RUNNING)
+ running_dags = session.scalar(stmt)
+ Stats.gauge("executor.running_dags", running_dags)
+
Review Comment:
This will run a query even when a user does not intend to send metrics anywhere i.e. when none of the following is enabled:
- `[metrics] statsd_datadog_enabled`
- `[metrics] statsd_on`
- `[metrics] otel_on`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2462129969
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1021,6 +1025,22 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _is_metrics_enabled(self):
+ return (
+ conf.getboolean("metrics", "statsd_datadog_enabled", fallback=False)
+ or conf.getboolean("metrics", "statsd_on", fallback=False)
+ or conf.getboolean("metrics", "otel_on", fallback=False)
+ )
+
+ def _emit_running_dags_metric(self):
+ """Emit executor.running_dags gauge."""
+ if not self._is_metrics_enabled():
+ return
+ with create_session() as session:
+ stmt = select(func.count()).select_from(DagRun).where(DagRun.state == DagRunState.RUNNING)
+ running_dags = session.scalar(stmt)
+ Stats.gauge("executor.running_dags", running_dags)
Review Comment:
Hi @jason810496, thanks for the review. I've gone through the code again and I think you're right. Let me update the patch accordingly.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2328874618
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1014,6 +1018,13 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _emit_running_dags_metric(self):
+ """Emit executor.running_dags gauge."""
+ with create_session() as session:
+ stmt = select(func.count()).select_from(DagRun).where(DagRun.state == DagRunState.RUNNING)
+ running_dags = session.scalar(stmt)
+ Stats.gauge("executor.running_dags", running_dags)
+
Review Comment:
Thanks for the review! I've added a check for the metrics settings and now skip the query if none are enabled. PTAL :)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2448574140
##########
go-sdk/bin/.gitignore:
##########
Review Comment:
I didn't intend to delete this. It must have been dropped during the rebase. I'll restore 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "o-nikolas (via GitHub)" <gi...@apache.org>.
o-nikolas commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2462163036
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1315,6 +1315,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),
Review Comment:
Does this config exist? I don't see an update to config_templates/config.yml here?
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -7086,6 +7086,33 @@ 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."""
+ from airflow.utils.state import DagRunState
+
+ 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.dag_processing.manager.Stats.gauge", _fake_gauge, raising=True)
+
+ with conf_vars({("metrics", "statsd_on"): "True"}):
+ # mock_executor = MagicMock(spec=BaseExecutor)
+ # scheduler_job = Job(executor=mock_executor)
Review Comment:
Debugging commenting left?
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1024,6 +1024,13 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _is_metrics_enabled(self):
Review Comment:
Where is this used?
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2235,6 +2240,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("executor.running_dags", running_dags)
Review Comment:
Sorry, I'm late to this PR. Why are we emitting this metric under `executor` when executors have nothing to do with dag level scheduling? Shouldn't this be under `scheduler`? The config you're using above even says `scheduler.dagrun_metrics_interval`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2462193753
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -1315,6 +1315,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),
Review Comment:
My mistake, update it now. Thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2463009483
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2235,6 +2240,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.running_dags", running_dags)
Review Comment:
How about `scheduler.dagruns.running`? The metric is emitted by the scheduler, and according to the doc, we might follow the existing `scheduler.tasks.*` gauges.
https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/logging-monitoring/metrics.html#gauges
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2462402892
##########
airflow-core/src/airflow/jobs/scheduler_job_runner.py:
##########
@@ -2235,6 +2240,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.running_dags", running_dags)
Review Comment:
Based on naming convention of existing metrics, would somehow like `dagrun.running` be better?
Since most of the naming are like `<entity>.<status>` and the new naming might be more align with `dagrun_metrics_interval` config meaning.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2448871745
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1021,6 +1025,22 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _is_metrics_enabled(self):
+ return (
+ conf.getboolean("metrics", "statsd_datadog_enabled", fallback=False)
+ or conf.getboolean("metrics", "statsd_on", fallback=False)
+ or conf.getboolean("metrics", "otel_on", fallback=False)
+ )
+
+ def _emit_running_dags_metric(self):
+ """Emit executor.running_dags gauge."""
+ if not self._is_metrics_enabled():
+ return
+ with create_session() as session:
+ stmt = select(func.count()).select_from(DagRun).where(DagRun.state == DagRunState.RUNNING)
+ running_dags = session.scalar(stmt)
+ Stats.gauge("executor.running_dags", running_dags)
Review Comment:
Would it be better to place in `SchedulerJobRunner._run_scheduler_loop` instead of placing in DagProcessor?
Some place like:
https://github.com/apache/airflow/blob/f3ad20aad803196b6fc1c73c1a38560deaf31aee/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1286-L1290
Since if we search with `Stats.gauge` across the codebase, there are only `dag_processing` or `dag_bag` related metics in DagProcessor component but not `executor` related metrics.
Also `heartbeat`. IMO, Executor is more closer to Scheduler compare with DagProcessor.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2448871745
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1021,6 +1025,22 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _is_metrics_enabled(self):
+ return (
+ conf.getboolean("metrics", "statsd_datadog_enabled", fallback=False)
+ or conf.getboolean("metrics", "statsd_on", fallback=False)
+ or conf.getboolean("metrics", "otel_on", fallback=False)
+ )
+
+ def _emit_running_dags_metric(self):
+ """Emit executor.running_dags gauge."""
+ if not self._is_metrics_enabled():
+ return
+ with create_session() as session:
+ stmt = select(func.count()).select_from(DagRun).where(DagRun.state == DagRunState.RUNNING)
+ running_dags = session.scalar(stmt)
+ Stats.gauge("executor.running_dags", running_dags)
Review Comment:
Would it be better to place in `SchedulerJobRunner._run_scheduler_loop` instead of placing in DagProcessor?
Some place like:
https://github.com/apache/airflow/blob/f3ad20aad803196b6fc1c73c1a38560deaf31aee/airflow-core/src/airflow/jobs/scheduler_job_runner.py#L1286-L1290
Since if we search with `Stats.gauge` across the codebase, there are only `dag_processing` or `dag_bag` related metics in DagProcessor component but not `executor` related metrics.
IMO, Executor is more closer to Scheduler compare with DagProcessor.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on PR #52815:
URL: https://github.com/apache/airflow/pull/52815#issuecomment-3424740098
Thanks vincbeck, the issue was resolved after rebase.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2313362441
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -341,6 +343,8 @@ def _run_parsing_loop(self):
self.heartbeat()
+ self._emit_running_dags_metric()
Review Comment:
I will need another eye though on whether the dag processor is the right place to check this. cc @jedcunningham
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2279407426
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1018,6 +1022,14 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
self._add_files_to_queue(to_queue, False)
Stats.incr("dag_processing.file_path_queue_update_count")
+ def _emit_running_dags_metric(self):
+ """Emit executor.running_dags gauge."""
+ with create_session() as session:
+ running_dags = (
+ session.query(func.count(DagRun.dag_id)).filter(DagRun.state == DagRunState.RUNNING).scalar()
Review Comment:
done, I followed [this guide](https://docs.sqlalchemy.org/en/20/changelog/migration_20.html) to change the query to 2.0 style
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on code in PR #52815:
URL: https://github.com/apache/airflow/pull/52815#discussion_r2462184929
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -7086,6 +7086,33 @@ 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."""
+ from airflow.utils.state import DagRunState
+
+ 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.dag_processing.manager.Stats.gauge", _fake_gauge, raising=True)
+
+ with conf_vars({("metrics", "statsd_on"): "True"}):
+ # mock_executor = MagicMock(spec=BaseExecutor)
+ # scheduler_job = Job(executor=mock_executor)
Review Comment:
Oops, I forgot to delete that. I'll remove it now. Thanks for catching 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add `executor.running_dags` gauge to expose count of running DAGs (#12368) [airflow]
Posted by "nailo2c (via GitHub)" <gi...@apache.org>.
nailo2c commented on PR #52815:
URL: https://github.com/apache/airflow/pull/52815#issuecomment-3341971969
Hi, I'm running into a failing test and I’m not sure why:
```console
FAILED airflow-core/tests/unit/dag_processing/test_manager.py::TestDagFileProcessorManager::test_bundles_with_team - airflow.exceptions.AirflowConfigException:
Section `dag_processor` key `dag_bundle_config_list` cannot have a team name when multi-team mode is disabled.To enable multi-team, you need to update section `core` key `multi_team` in your config.
```
I've already rebased my branch to the latest `main`, and I can see the `multi_team` key defined in my feature branch’s config template:
https://github.com/nailo2c/airflow/blob/feature-12368_add_counter_for_number_of_running_dag/airflow-core/src/airflow/config_templates/config.yml#L512-L520
But the CI is still failing with the same error. Did I miss another place where `multi_team` needs to be defined or enabled?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org