You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "gingeekrishna (via GitHub)" <gi...@apache.org> on 2026/05/26 19:27:13 UTC
[PR] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
gingeekrishna opened a new pull request, #67572:
URL: https://github.com/apache/airflow/pull/67572
## Summary
Fixes #66796
`SELECT COUNT()` on an aggregate column always returns an integer — `None` from `session.scalar()` is only possible on a transient DB connectivity failure, not when the table is empty (which returns `0`). Using `or 0` silently swallows that failure and emits a zero count, which can trigger false on-call pages ("all DAGs disappeared!") while hiding the real problem.
- Add `SerializedDagModel.get_count()` classmethod that raises `RuntimeError` when `session.scalar()` returns `None`, instead of masking it with `or 0`
- Emit a `serialized_dag.count` gauge in `dag_processing/manager.py`'s `emit_metrics()` using the new method, wrapped in `try/except` so a DB error logs a warning but never kills the parse loop
## Changes
- `airflow-core/src/airflow/models/serialized_dag.py` — add `get_count()` classmethod with None-guard
- `airflow-core/src/airflow/dag_processing/manager.py` — import `SerializedDagModel`, add module logger, emit `serialized_dag.count` metric with error isolation
## Test plan
- [ ] `tests/unit/models/test_serialized_dag.py::TestSerializedDagModel::test_get_count_returns_zero_on_empty_table`
- [ ] `tests/unit/models/test_serialized_dag.py::TestSerializedDagModel::test_get_count_returns_correct_value`
- [ ] `tests/unit/models/test_serialized_dag.py::TestSerializedDagModel::test_get_count_raises_on_none_result`
- [ ] `tests/unit/dag_processing/test_manager.py::TestEmitMetrics::test_emit_metrics_emits_serialized_dag_count`
- [ ] `tests/unit/dag_processing/test_manager.py::TestEmitMetrics::test_emit_metrics_does_not_raise_on_db_error`
--
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] feat: add SerializedDagModel.get_count() and serialized_dag.count StatsD gauge [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb closed pull request #67572: feat: add SerializedDagModel.get_count() and serialized_dag.count StatsD gauge
URL: https://github.com/apache/airflow/pull/67572
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3306371720
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -875,6 +875,23 @@ def has_dag(cls, dag_id: str, session: Session = NEW_SESSION) -> bool:
"""
return session.scalar(select(literal(True)).where(cls.dag_id == dag_id).limit(1)) is not None
+ @classmethod
+ @provide_session
+ def get_count(cls, session: Session = NEW_SESSION) -> int:
+ """
+ Return the total number of serialized DAGs stored in the database.
+
+ :param session: ORM Session
+ :raises RuntimeError: if the database returns None for the COUNT query, which indicates
+ a transient connectivity issue rather than an empty table (COUNT always returns an int).
+ """
+ result = session.scalar(select(func.count()).select_from(cls))
+ if result is None:
+ raise RuntimeError(
+ "COUNT query on serialized_dag returned None - possible database connectivity issue"
+ )
+ return result
Review Comment:
`COUNT(*)` queries should always return a row with an integer (including 0), and DB connectivity failures should typically surface as `SQLAlchemyError` rather than `None`. Consider switching to an API that enforces “exactly one row” (e.g., executing the statement and using `scalar_one()`), and then returning `int(result)` to guarantee the declared return type. This removes an effectively-unreachable `None` branch and clarifies the error model.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,11 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except (RuntimeError, SQLAlchemyError):
+ log.exception("Failed to emit serialized_dag.count metric")
Review Comment:
Emitting `serialized_dag.count` via a `COUNT(*)` each parsing loop can be expensive on large tables (many databases will scan the table for an exact count), and this code runs on a periodic cadence. Consider throttling (e.g., emit every N loops / with a time-based cache), or sourcing the count from an already-available in-process value updated during serialization/sync, to avoid adding frequent full-count queries to the metadata DB.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -2790,3 +2790,32 @@ def test_bundle_version_data_stored_after_refresh(self, session):
assert manager._bundle_versions["mock_bundle"] == "newhash"
assert manager._bundle_version_data["mock_bundle"] == test_data
+
+
+class TestEmitMetrics:
+ """Tests for the emit_metrics module-level function."""
+
+ def test_emit_metrics_emits_serialized_dag_count(self):
+ """emit_metrics emits the serialized_dag.count gauge with the value from get_count."""
+ from airflow.dag_processing.manager import emit_metrics
+
+ with mock.patch(
+ "airflow.dag_processing.manager.SerializedDagModel.get_count", return_value=3
+ ):
+ with mock.patch("airflow.dag_processing.manager.stats") as mock_stats:
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
+
+ calls = {call[0][0]: call[0][1] for call in mock_stats.gauge.call_args_list}
+ assert "serialized_dag.count" in calls
+ assert calls["serialized_dag.count"] == 3
+
+ def test_emit_metrics_does_not_raise_on_db_error(self):
+ """emit_metrics logs and swallows RuntimeError from get_count on DB failure."""
+ from airflow.dag_processing.manager import emit_metrics
+
+ with mock.patch(
+ "airflow.dag_processing.manager.SerializedDagModel.get_count",
+ side_effect=RuntimeError("COUNT query on serialized_dag returned None - possible"),
+ ):
+ with mock.patch("airflow.dag_processing.manager.stats"):
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
Review Comment:
This test currently only verifies “no exception”. Since the docstring claims it “logs and swallows”, it should also assert the logging side-effect (e.g., patch `airflow.dag_processing.manager.log` and assert `log.exception` called once) and/or assert the `serialized_dag.count` gauge was not emitted when `get_count()` fails. This will make the test actually validate the new error-handling behavior.
--
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] feat: add SerializedDagModel.get_count() and serialized_dag.count StatsD gauge [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3441694476
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2574,6 +2574,16 @@ scheduler:
type: float
example: ~
default: "30.0"
+ emit_serialized_dag_count_metric:
Review Comment:
How is this different to the existing `dagbag_size` metric?
##########
airflow-core/newsfragments/67572.feature.rst:
##########
Review Comment:
`SerializedDagModel.get_count()` is entirely an internal impl detail that is not exposed to users; it shouldn't be in newsfragments.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -3214,3 +3214,60 @@ def test_add_callback_to_queue_includes_team_name(self, mock_incr, multi_team, t
manager._add_callback_to_queue(request)
mock_incr.assert_called_once_with("dag_processing.other_callback_count", tags=expected_tags)
+
+
+class TestEmitMetrics:
Review Comment:
Please no extra test class -- add methods into the existing TestDagProcessorManager.
##########
airflow-core/src/airflow/config_templates/config.yml:
##########
@@ -2574,6 +2574,16 @@ scheduler:
type: float
example: ~
default: "30.0"
+ emit_serialized_dag_count_metric:
Review Comment:
If we even need this (I'm not sure) it should be under the `dag_processor` section, not `scheduler`.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3306644821
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,17 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # COUNT(*) on the serialized_dag table adds one DB round-trip per parse loop.
+ # This can be expensive on large deployments; throttling this metric is a
+ # straightforward follow-up if the round-trip becomes a bottleneck.
+ # On failure, an error counter is incremented so dashboards can alert on
+ # missing samples rather than silently showing a stale last value.
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except SQLAlchemyError:
+ log.exception("Failed to emit serialized_dag.count metric")
+ stats.incr("serialized_dag.count_error")
Review Comment:
This introduces a new `COUNT(*)` query and session creation on every parse loop iteration, which can become a measurable overhead at scale. Consider throttling/sampling (e.g., emit every N loops or every T seconds), or making it configurable so large deployments can disable/reduce the query cost while still keeping the metric available when desired.
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -912,3 +912,28 @@ def test_deadline_name_change_updates_db_and_returns_true(self, testing_dag_bund
# The name must have been updated in the DB.
assert updated_alert.name == "updated name"
+
+ def test_get_count_returns_zero_on_empty_table(self, session):
+ """get_count() returns 0 when no serialized DAGs are stored."""
+ db.clear_db_serialized_dags()
+ assert SDM.get_count(session=session) == 0
+
+ def test_get_count_returns_correct_value(self, dag_maker, session):
+ """get_count() returns the exact number of serialized DAGs in the table."""
+ baseline = SDM.get_count(session=session)
+ with dag_maker("dag_count_1"):
+ pass
+ with dag_maker("dag_count_2"):
+ pass
+ # dag_maker writes SerializedDagModel rows on context exit; flush to make
+ # them visible within the same session before asserting.
+ session.flush()
+ assert SDM.get_count(session=session) == baseline + 2
+
+ def test_get_count_propagates_db_error(self, session):
+ """get_count() lets OperationalError propagate so callers can handle DB failures."""
+ from sqlalchemy.exc import OperationalError
+
+ with mock.patch.object(session, "execute", side_effect=OperationalError("db failure", None, None)):
Review Comment:
`OperationalError` is typically constructed with a DBAPI `orig` exception; passing `None` can lead to odd behavior in stringification/inspection in some SQLAlchemy versions. Prefer providing an `Exception(...)` (or a minimal DBAPI-like exception instance) as the `orig` argument to better match real-world error objects.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -2790,3 +2790,36 @@ def test_bundle_version_data_stored_after_refresh(self, session):
assert manager._bundle_versions["mock_bundle"] == "newhash"
assert manager._bundle_version_data["mock_bundle"] == test_data
+
+
+class TestEmitMetrics:
+ """Tests for the emit_metrics module-level function."""
+
+ def test_emit_metrics_emits_serialized_dag_count(self):
+ """emit_metrics emits the serialized_dag.count gauge with the value from get_count."""
+ from airflow.dag_processing.manager import emit_metrics
+
+ with mock.patch(
+ "airflow.dag_processing.manager.SerializedDagModel.get_count", return_value=3
+ ):
+ with mock.patch("airflow.dag_processing.manager.stats") as mock_stats:
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
Review Comment:
`emit_metrics()` opens a real DB session via `create_session()` even though this test only cares about metric emission. To keep this a true unit test (faster and less environment-dependent), patch `airflow.dag_processing.manager.create_session` with a lightweight context manager stub so the test doesn’t require DB session creation.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -2790,3 +2790,36 @@ def test_bundle_version_data_stored_after_refresh(self, session):
assert manager._bundle_versions["mock_bundle"] == "newhash"
assert manager._bundle_version_data["mock_bundle"] == test_data
+
+
+class TestEmitMetrics:
+ """Tests for the emit_metrics module-level function."""
+
+ def test_emit_metrics_emits_serialized_dag_count(self):
+ """emit_metrics emits the serialized_dag.count gauge with the value from get_count."""
+ from airflow.dag_processing.manager import emit_metrics
+
+ with mock.patch(
+ "airflow.dag_processing.manager.SerializedDagModel.get_count", return_value=3
+ ):
+ with mock.patch("airflow.dag_processing.manager.stats") as mock_stats:
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
+
+ mock_stats.gauge.assert_any_call("serialized_dag.count", 3)
+
+ def test_emit_metrics_logs_and_swallows_db_error(self):
+ """emit_metrics logs, increments error counter, and swallows SQLAlchemyError from get_count."""
+ from sqlalchemy.exc import SQLAlchemyError
+
+ from airflow.dag_processing.manager import emit_metrics
+
+ with mock.patch(
+ "airflow.dag_processing.manager.SerializedDagModel.get_count",
+ side_effect=SQLAlchemyError("db failure"),
+ ):
+ with mock.patch("airflow.dag_processing.manager.stats") as mock_stats:
+ with mock.patch("airflow.dag_processing.manager.log") as mock_log:
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
Review Comment:
Same as above: this test will still create a real session via `create_session()` before `get_count()` raises. Patch `create_session` to a stub context manager so the test isolates the error-handling path without relying on DB session creation.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #67572:
URL: https://github.com/apache/airflow/pull/67572#issuecomment-4566729222
@gingeekrishna A few things need addressing before review — see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
- :x: **Pre-commit / static checks**. See [docs](https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst).
No rush.
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3306406589
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,14 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # COUNT(*) on the serialized_dag table adds one DB round-trip per parse loop.
+ # On large installations this is typically fast (index scan on the PK), but
+ # we isolate the call so that a transient DB error never kills the parse loop.
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except SQLAlchemyError:
+ log.exception("Failed to emit serialized_dag.count metric")
Review Comment:
Emitting `COUNT(*)` every parse loop can become expensive on large `serialized_dag` tables and may add sustained DB load (often a full table scan depending on the DB/visibility state). Consider reducing frequency (e.g., only every N loops / time-based throttle), caching the value, or using an estimated count mechanism where appropriate to avoid making parsing throughput dependent on this query.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,14 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # COUNT(*) on the serialized_dag table adds one DB round-trip per parse loop.
+ # On large installations this is typically fast (index scan on the PK), but
+ # we isolate the call so that a transient DB error never kills the parse loop.
Review Comment:
The comment asserts the count is 'typically fast (index scan on the PK)'. `COUNT(*)` is not guaranteed to use a PK index scan and often uses a sequential scan; some engines can use index-only scans under specific conditions. Suggest rewording to avoid implying a specific query plan (e.g., 'may be fast, but can be expensive on large tables').
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -2790,3 +2790,37 @@ def test_bundle_version_data_stored_after_refresh(self, session):
assert manager._bundle_versions["mock_bundle"] == "newhash"
assert manager._bundle_version_data["mock_bundle"] == test_data
+
+
+class TestEmitMetrics:
+ """Tests for the emit_metrics module-level function."""
+
+ def test_emit_metrics_emits_serialized_dag_count(self):
+ """emit_metrics emits the serialized_dag.count gauge with the value from get_count."""
+ from airflow.dag_processing.manager import emit_metrics
+
+ with mock.patch(
+ "airflow.dag_processing.manager.SerializedDagModel.get_count", return_value=3
+ ):
+ with mock.patch("airflow.dag_processing.manager.stats") as mock_stats:
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
+
+ calls = {call[0][0]: call[0][1] for call in mock_stats.gauge.call_args_list}
+ assert "serialized_dag.count" in calls
+ assert calls["serialized_dag.count"] == 3
Review Comment:
This assertion pattern can accidentally mask duplicate emissions because it folds the call list into a dict (later calls overwrite earlier ones). It’s more precise to assert the expected metric call directly (e.g., assert that `gauge` was called with `(\"serialized_dag.count\", 3)` among its calls) instead of post-processing `call_args_list`.
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -912,3 +912,27 @@ def test_deadline_name_change_updates_db_and_returns_true(self, testing_dag_bund
# The name must have been updated in the DB.
assert updated_alert.name == "updated name"
+
+ def test_get_count_returns_zero_on_empty_table(self, session):
+ """get_count() returns 0 when no serialized DAGs are stored."""
+ assert SDM.get_count(session=session) == 0
+
+ def test_get_count_returns_correct_value(self, dag_maker, session):
+ """get_count() returns the exact number of serialized DAGs in the table."""
+ with dag_maker("dag_count_1"):
+ pass
+ with dag_maker("dag_count_2"):
+ pass
+ # dag_maker writes SerializedDagModel rows on context exit; flush to make
+ # them visible within the same session before asserting.
+ session.flush()
+ count = SDM.get_count(session=session)
+ assert count == 2
Review Comment:
This test assumes the `serialized_dag` table is empty at the start of the test (`assert count == 2`). If other fixtures/tests insert rows (even implicitly), this can be flaky. A more robust approach is to record a baseline (`start = SDM.get_count(...)`) and assert `count == start + 2`, or explicitly clear the table within the test setup.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3306574975
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,16 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # COUNT(*) on the serialized_dag table adds one DB round-trip per parse loop.
+ # This can be expensive on large deployments (query plan is DB-dependent and
+ # may involve a full table scan). The call is isolated so that a transient
+ # DB error never kills the parse loop; throttling this metric in the future
+ # is a straightforward follow-up if the round-trip becomes a bottleneck.
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except OperationalError:
+ log.exception("Failed to emit serialized_dag.count metric")
Review Comment:
The comment/docstring intent is to prevent DB errors from killing the parse loop, but this currently only swallows `OperationalError`. Other SQLAlchemy DB failures (e.g. disconnects, timeouts mapped differently, or broader `SQLAlchemyError` subclasses) can still bubble up and terminate the loop. Consider catching `sqlalchemy.exc.SQLAlchemyError` instead (and importing that), or update the comment/tests to explicitly state only `OperationalError` is swallowed.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -41,6 +41,7 @@
import attrs
import structlog
from sqlalchemy import select, update
+from sqlalchemy.exc import OperationalError
Review Comment:
The comment/docstring intent is to prevent DB errors from killing the parse loop, but this currently only swallows `OperationalError`. Other SQLAlchemy DB failures (e.g. disconnects, timeouts mapped differently, or broader `SQLAlchemyError` subclasses) can still bubble up and terminate the loop. Consider catching `sqlalchemy.exc.SQLAlchemyError` instead (and importing that), or update the comment/tests to explicitly state only `OperationalError` is swallowed.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "1fanwang (via GitHub)" <gi...@apache.org>.
1fanwang commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3358893848
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1568,6 +1573,16 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # Gated by config so large deployments can opt out of the extra DB round-trip.
+ # On failure an error counter is incremented so dashboards can alert on
+ # missing samples rather than silently showing a stale last value.
+ if conf.getboolean("scheduler", "emit_serialized_dag_count_metric", fallback=True):
Review Comment:
Flagging something on scope before you put more time in.
Current `main` doesn't actually have an existing `SerializedDagModel.get_count()` or the `or 0` / `serialized_dag.count` emit that #66796 describes, so this PR is really *adding* a new metric (this gauge, plus the config and error counter) rather than the one-line None-safety fix the issue framed. That's on me — #66796 read like a small fix when the groundwork it references isn't in `main`.
Which raises the question worth settling first: is the `serialized_dag.count` gauge itself wanted? It runs a DB `COUNT` every parse loop, and nothing emits it today — so the None-masking the issue worried about only appears once we add this. If the goal is just None-safe counting, the footprint could be much smaller.
(Also fyi the branch has a merge conflict and a failing static check, so it isn't mergeable as-is — I'd settle "is this needed" before chasing those.)
If a leaner version turns out worth having, it could be a separate PR to compare here. Either way, appreciate you picking it up.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3306672622
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,16 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # Gated by config so large deployments can opt out of the extra DB round-trip.
+ # On failure an error counter is incremented so dashboards can alert on
+ # missing samples rather than silently showing a stale last value.
+ if conf.getboolean("scheduler", "emit_serialized_dag_count_metric", fallback=True):
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except SQLAlchemyError:
+ log.exception("Failed to emit serialized_dag.count metric")
+ stats.incr("serialized_dag.count_error")
Review Comment:
The new config-gated branch isn’t covered: there should be a unit test asserting that when `conf.getboolean(...)` returns False, `SerializedDagModel.get_count` is not called and no `serialized_dag.count` gauge is emitted. This prevents regressions where the setting is ignored or accidentally inverted.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -2790,3 +2790,38 @@ def test_bundle_version_data_stored_after_refresh(self, session):
assert manager._bundle_versions["mock_bundle"] == "newhash"
assert manager._bundle_version_data["mock_bundle"] == test_data
+
+
+class TestEmitMetrics:
+ """Tests for the emit_metrics module-level function."""
+
+ def test_emit_metrics_emits_serialized_dag_count(self):
+ """emit_metrics emits the serialized_dag.count gauge with the value from get_count."""
+ from airflow.dag_processing.manager import emit_metrics
+
+ with mock.patch("airflow.dag_processing.manager.create_session"):
+ with mock.patch(
+ "airflow.dag_processing.manager.SerializedDagModel.get_count", return_value=3
+ ):
+ with mock.patch("airflow.dag_processing.manager.stats") as mock_stats:
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
+
+ mock_stats.gauge.assert_any_call("serialized_dag.count", 3)
+
+ def test_emit_metrics_logs_and_swallows_db_error(self):
+ """emit_metrics logs, increments error counter, and swallows SQLAlchemyError from get_count."""
+ from sqlalchemy.exc import SQLAlchemyError
+
+ from airflow.dag_processing.manager import emit_metrics
+
+ with mock.patch("airflow.dag_processing.manager.create_session"):
+ with mock.patch(
+ "airflow.dag_processing.manager.SerializedDagModel.get_count",
+ side_effect=SQLAlchemyError("db failure"),
+ ):
+ with mock.patch("airflow.dag_processing.manager.stats") as mock_stats:
+ with mock.patch("airflow.dag_processing.manager.log") as mock_log:
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
Review Comment:
These tests assume the runtime config leaves `scheduler.emit_serialized_dag_count_metric` enabled (fallback is True, but an explicit config in the test environment could disable it), which can make the assertions flaky. Patch `airflow.dag_processing.manager.conf.getboolean` to return True in this test (and return False in a dedicated test) so the behavior under test is deterministic.
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -912,3 +912,30 @@ def test_deadline_name_change_updates_db_and_returns_true(self, testing_dag_bund
# The name must have been updated in the DB.
assert updated_alert.name == "updated name"
+
+ def test_get_count_returns_zero_on_empty_table(self, session):
+ """get_count() returns 0 when no serialized DAGs are stored."""
+ db.clear_db_serialized_dags()
+ assert SDM.get_count(session=session) == 0
+
+ def test_get_count_returns_correct_value(self, dag_maker, session):
+ """get_count() returns the exact number of serialized DAGs in the table."""
+ baseline = SDM.get_count(session=session)
+ with dag_maker("dag_count_1"):
+ pass
+ with dag_maker("dag_count_2"):
+ pass
+ # dag_maker writes SerializedDagModel rows on context exit; flush to make
+ # them visible within the same session before asserting.
+ session.flush()
+ assert SDM.get_count(session=session) == baseline + 2
+
+ def test_get_count_propagates_db_error(self, session):
+ """get_count() lets OperationalError propagate so callers can handle DB failures."""
+ from sqlalchemy.exc import OperationalError
+
+ with mock.patch.object(
+ session, "execute", side_effect=OperationalError("db failure", None, Exception("db failure"))
Review Comment:
Constructing `OperationalError` with `params=None` can be brittle because some SQLAlchemy error formatting/representation paths expect an empty dict/tuple. Consider using `{}` (or a realistic params structure) for the second argument to reduce the chance of the test failing due to error stringification rather than the intended propagation behavior.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3306338748
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1570,11 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except Exception:
+ log.exception("Failed to emit serialized_dag.count metric")
Review Comment:
Catching `Exception` here is overly broad and can mask non-DB programming/configuration errors inside `emit_metrics()` (including failures in `stats.gauge`). Since the intended behavior (per tests/docstrings) is to swallow DB/count failures, narrow this to the expected exception(s) (e.g., `RuntimeError` from `get_count`, and optionally the relevant SQLAlchemy exception type) and let unexpected exceptions propagate.
##########
airflow-core/tests/unit/dag_processing/test_manager.py:
##########
@@ -2790,3 +2790,33 @@ def test_bundle_version_data_stored_after_refresh(self, session):
assert manager._bundle_versions["mock_bundle"] == "newhash"
assert manager._bundle_version_data["mock_bundle"] == test_data
+
+
+class TestEmitMetrics:
+ """Tests for the emit_metrics module-level function."""
+
+ @pytest.mark.db_test
+ def test_emit_metrics_emits_serialized_dag_count(self, dag_maker, session):
+ """emit_metrics emits the serialized_dag.count gauge with the DB count."""
+ from airflow.dag_processing.manager import emit_metrics
+
+ with dag_maker("emit_count_dag"):
+ pass
+
+ with mock.patch("airflow.dag_processing.manager.stats") as mock_stats:
+ emit_metrics(parse_time=1.0, dag_file_stats=[])
+
+ calls = {call[0][0]: call[0][1] for call in mock_stats.gauge.call_args_list}
+ assert "serialized_dag.count" in calls
+ assert calls["serialized_dag.count"] == 1
Review Comment:
Similar to the `get_count()` unit test, this asserts `serialized_dag.count == 1` after creating a DAG with `dag_maker`, but `emit_metrics()` queries the `serialized_dag` table. If the test setup doesn’t actually serialize and persist the DAG into `serialized_dag`, the gauge value will be `0` and the assertion will fail. Make the test deterministic by explicitly writing a `SerializedDagModel` row for the created DAG (or inserting one directly) before calling `emit_metrics()`.
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -912,3 +912,22 @@ def test_deadline_name_change_updates_db_and_returns_true(self, testing_dag_bund
# The name must have been updated in the DB.
assert updated_alert.name == "updated name"
+
+ def test_get_count_returns_zero_on_empty_table(self, session):
+ """get_count() returns 0 when no serialized DAGs are stored."""
+ assert SDM.get_count(session=session) == 0
+
+ def test_get_count_returns_correct_value(self, dag_maker, session):
+ """get_count() returns the exact number of serialized DAGs in the table."""
+ with dag_maker("dag_count_1"):
+ pass
+ with dag_maker("dag_count_2"):
+ pass
Review Comment:
This test creates DAGs via `dag_maker`, but `SerializedDagModel.get_count()` counts rows in the `serialized_dag` table. Unless `dag_maker` (in this test environment) also writes `SerializedDagModel` rows, the expected count of `2` is not guaranteed and may be `0`, making the test incorrect/flaky. To make the test deterministic, explicitly create serialized DAG entries (e.g., call the appropriate `SerializedDagModel` write method for each created DAG, or insert rows directly) and flush/commit before asserting the count.
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -875,6 +875,23 @@ def has_dag(cls, dag_id: str, session: Session = NEW_SESSION) -> bool:
"""
return session.scalar(select(literal(True)).where(cls.dag_id == dag_id).limit(1)) is not None
+ @classmethod
+ @provide_session
+ def get_count(cls, session: Session = NEW_SESSION) -> int:
+ """
+ Return the total number of serialized DAGs stored in the database.
+
+ :param session: ORM Session
+ :raises RuntimeError: if the database returns None for the COUNT query, which indicates
+ a transient connectivity issue rather than an empty table (COUNT always returns an int).
+ """
+ result = session.scalar(select(func.count()).select_from(cls))
+ if result is None:
+ raise RuntimeError(
+ "COUNT query on serialized_dag returned None — possible database connectivity issue"
Review Comment:
The error message contains a Unicode em dash (`—`). To avoid encoding/rendering issues in some logging/terminal environments and to keep messages consistent/easy to grep, consider using plain ASCII punctuation (e.g., `-`) in this exception string.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on PR #67572:
URL: https://github.com/apache/airflow/pull/67572#issuecomment-4643793756
Thanks for the thorough flag, @1fanwang — and for being upfront that the issue framing was ambiguous.
Re-reading #66796, I now see I misread it: the issue describes `get_count()` and the `serialized_dag.count` emit as if they already existed and needed a one-line fix. Since they don't exist in `main`, I effectively added them from scratch — that's a scope mismatch from what was intended.
On the core question of whether the `serialized_dag.count` gauge is worth having: I do think there's operational value in being able to monitor how many serialized DAGs are in the DB (it's a useful signal for debugging serialization issues, DAG disappearance alerts, etc.). But I understand the concern about a `COUNT(*)` on every parse loop — the performance trade-off is real, especially at scale.
Happy to take any of these directions based on your guidance:
1. **Keep the PR** — address the performance concern by adding throttling (emit once every N loops), which would make the DB cost negligible
2. **Narrow the scope** — if the issue is closed as "no change needed" (since `get_count()` doesn't exist, there's nothing to fix), I can close this PR too
3. **Split it** — keep `SerializedDagModel.get_count()` as a utility method but leave the metric emission for a separate RFC/discussion
Also, FYI — the merge conflicts have been resolved and the branch is now up to date with `main`.
--
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] feat: add SerializedDagModel.get_count() and serialized_dag.count StatsD gauge [airflow]
Posted by "ashb (via GitHub)" <gi...@apache.org>.
ashb commented on PR #67572:
URL: https://github.com/apache/airflow/pull/67572#issuecomment-4750614088
Closing this as won't fix, since the original issue that this purports to fix is closed as not needed anymore.. If you think this metric is useful feel free to tell me why and we can reopen this PR.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on PR #67572:
URL: https://github.com/apache/airflow/pull/67572#issuecomment-4568030422
> @gingeekrishna A few things need addressing before review — see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
>
> * ❌ **Pre-commit / static checks**. See [docs](https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst).
>
> No rush.
>
> _Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
Thanks @potiuk , I have update the PR now
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3306538333
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -912,3 +912,27 @@ def test_deadline_name_change_updates_db_and_returns_true(self, testing_dag_bund
# The name must have been updated in the DB.
assert updated_alert.name == "updated name"
+
+ def test_get_count_returns_zero_on_empty_table(self, session):
+ """get_count() returns 0 when no serialized DAGs are stored."""
Review Comment:
This test assumes the `serialized_dag` table is empty but doesn’t actively ensure it (e.g., by clearing the table within the test). To make the test order-independent and more robust across fixture changes, consider explicitly deleting/truncating `SerializedDagModel` rows at the start of the test (or using a fixture that guarantees an empty table for this model).
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -875,6 +875,21 @@ def has_dag(cls, dag_id: str, session: Session = NEW_SESSION) -> bool:
"""
return session.scalar(select(literal(True)).where(cls.dag_id == dag_id).limit(1)) is not None
+ @classmethod
+ @provide_session
+ def get_count(cls, session: Session = NEW_SESSION) -> int:
+ """
+ Return the total number of serialized DAGs stored in the database.
+
+ Uses ``scalar_one()`` so that a DB connectivity failure surfaces as a
+ ``SQLAlchemyError`` rather than a silent ``None`` return. ``COUNT(*)``
+ always produces exactly one row, so ``NoResultFound`` is never raised
+ under normal conditions.
Review Comment:
The docstring rationale is misleading: DB connectivity failures will raise regardless of `scalar()` vs `scalar_one()`, so the “silent `None` return” claim isn’t accurate here. Consider updating the docstring to justify `scalar_one()` in terms of enforcing exactly-one-row semantics (and raising if the query shape changes unexpectedly), rather than connectivity behavior.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,16 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # COUNT(*) on the serialized_dag table adds one DB round-trip per parse loop.
+ # This can be expensive on large deployments (query plan is DB-dependent and
+ # may involve a full table scan). The call is isolated so that a transient
+ # DB error never kills the parse loop; throttling this metric in the future
+ # is a straightforward follow-up if the round-trip becomes a bottleneck.
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except SQLAlchemyError:
+ log.exception("Failed to emit serialized_dag.count metric")
Review Comment:
Catching `SQLAlchemyError` is very broad and can also swallow non-transient issues (e.g., programming/config/schema errors) that you might prefer to surface/test-fail rather than silently skip emitting the metric. Consider narrowing the exception to DB connectivity/DBAPI-related errors (e.g., `DBAPIError`/`OperationalError`) if the intent is specifically “don’t fail the parse loop on transient DB failures,” while still letting unexpected SQLAlchemy usage errors be noticed earlier.
--
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] fix: add SerializedDagModel.get_count() that raises on None instead of silently returning 0 [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #67572:
URL: https://github.com/apache/airflow/pull/67572#discussion_r3306618928
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,16 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # COUNT(*) on the serialized_dag table adds one DB round-trip per parse loop.
+ # This can be expensive on large deployments (query plan is DB-dependent and
+ # may involve a full table scan). The call is isolated so that a transient
+ # DB error never kills the parse loop; throttling this metric in the future
+ # is a straightforward follow-up if the round-trip becomes a bottleneck.
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except SQLAlchemyError:
+ log.exception("Failed to emit serialized_dag.count metric")
Review Comment:
This adds a DB round-trip (and potentially a full table scan) once per parse loop, which can materially increase scheduler/manager load at scale. Consider throttling (e.g., emit once per N loops or per time interval), gating behind a config/feature flag, or reusing an existing session/connection from the parse loop to avoid extra connection churn.
##########
airflow-core/src/airflow/dag_processing/manager.py:
##########
@@ -1566,6 +1571,16 @@ def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
stats.gauge("dag_processing.total_parse_time", parse_time)
stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
+ # COUNT(*) on the serialized_dag table adds one DB round-trip per parse loop.
+ # This can be expensive on large deployments (query plan is DB-dependent and
+ # may involve a full table scan). The call is isolated so that a transient
+ # DB error never kills the parse loop; throttling this metric in the future
+ # is a straightforward follow-up if the round-trip becomes a bottleneck.
+ try:
+ with create_session() as session:
+ stats.gauge("serialized_dag.count", SerializedDagModel.get_count(session=session))
+ except SQLAlchemyError:
Review Comment:
When a DB error occurs, the gauge is not emitted at all, which can leave dashboards showing a stale last value and potentially mask ongoing failure. Consider also emitting an explicit failure signal (e.g., a counter like `serialized_dag.count_error` or a sentinel gauge value) so monitoring can alert reliably on missing/failed samples.
--
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