You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Ei-Sandi (via GitHub)" <gi...@apache.org> on 2026/06/23 15:33:14 UTC
[PR] Add metrics for how often a dag gets serialised. [airflow]
Ei-Sandi opened a new pull request, #68906:
URL: https://github.com/apache/airflow/pull/68906
### Changes
- Adds a new metric to monitor how often a DAG gets serialised.
(It’s helpful to observe serialization frequency to ensure we aren't unnecessarily re-serializing DAGs. )
### Details
- Name: dag.serialization
- Type: Counter
- Variables/Tags: dag_id, bundle_name
### Tests
- Covered by unit tests.
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [X] Yes (please specify the tool below)
Co-authored-by: Claude Sonnet 4.6, Claude Opus 4.8 following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3461597872
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +201,59 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
+ def test_serialization_metric_incremented_on_new_write(self, testing_dag_bundle):
+ """A brand new serialized DAG write emits the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ with mock.patch("airflow.models.serialized_dag.stats") as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": dag.dag_id, "bundle_name": "testing"},
+ )
+
+ def test_serialization_metric_not_incremented_when_unchanged(self, testing_dag_bundle):
+ """Re-writing an unchanged DAG must not emit the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ with mock.patch("airflow.models.serialized_dag.stats") as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is False
+
+ mock_stats.incr.assert_not_called()
+
+ def test_serialization_metric_incremented_on_inplace_update(self, dag_maker, session):
+ """Updating a DAG version in place (no dag runs) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ # Change the DAG so the hash differs; with no dag runs this updates in place.
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch("airflow.models.serialized_dag.stats") as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker") is True
+
+ assert session.scalar(select(func.count()).select_from(DagVersion)) == 1
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": "metric_dag", "bundle_name": "dag_maker"},
+ )
+
+ def test_serialization_metric_incremented_on_new_version(self, dag_maker, session):
+ """Writing a new DAG version (existing run) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ dag_maker.create_dagrun(run_id="run1", logical_date=pendulum.datetime(2025, 1, 1))
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch("airflow.models.serialized_dag.stats") as mock_stats:
Review Comment:
You are reusing `airflow.models.serialized_dag.stats` in every test. This can be a common variable.
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +201,59 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
Review Comment:
Can you add another unit test that asserts both the new name and the legacy name are exported? You can do that by mocking the backend. See this test as an example
https://github.com/apache/airflow/blob/main/airflow-core/tests/unit/jobs/test_scheduler_job.py#L473-L524
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "Ei-Sandi (via GitHub)" <gi...@apache.org>.
Ei-Sandi commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3467207801
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +204,76 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
+ def test_serialization_metric_incremented_on_new_write(self, testing_dag_bundle):
+ """A brand new serialized DAG write emits the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": dag.dag_id, "bundle_name": "testing"},
+ )
+
+ def test_serialization_metric_not_incremented_when_unchanged(self, testing_dag_bundle):
+ """Re-writing an unchanged DAG must not emit the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is False
+
+ mock_stats.incr.assert_not_called()
+
+ def test_serialization_metric_incremented_on_inplace_update(self, dag_maker, session):
+ """Updating a DAG version in place (no dag runs) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ # Change the DAG so the hash differs; with no dag runs this updates in place.
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker") is True
Review Comment:
Ok, I will just change it to "testing" to make everything the same.
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3677642035
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -763,6 +764,7 @@ def write_dag(
session.merge(dag_version)
# Update the latest DagCode
DagCode.update_source_code(dag_id=dag.dag_id, fileloc=dag.fileloc, session=session)
+ stats.incr("dag.serialization", tags={"dag_id": dag.dag_id, "bundle_name": bundle_name})
Review Comment:
nit: I think the naming of the metric could be more explicit. Would something like `dag.serialization_writes` be a bit clearer?
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3466417764
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +204,76 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
+ def test_serialization_metric_incremented_on_new_write(self, testing_dag_bundle):
+ """A brand new serialized DAG write emits the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": dag.dag_id, "bundle_name": "testing"},
+ )
+
+ def test_serialization_metric_not_incremented_when_unchanged(self, testing_dag_bundle):
+ """Re-writing an unchanged DAG must not emit the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is False
+
+ mock_stats.incr.assert_not_called()
+
+ def test_serialization_metric_incremented_on_inplace_update(self, dag_maker, session):
+ """Updating a DAG version in place (no dag runs) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ # Change the DAG so the hash differs; with no dag runs this updates in place.
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker") is True
Review Comment:
Half of the tests are setting `bundle_name` as `testing` and the other half as `dag_maker`. This isn't an issue but it's a bit confusing because someone might think that the context changes, but I don't think that's the case.
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +204,76 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
+ def test_serialization_metric_incremented_on_new_write(self, testing_dag_bundle):
+ """A brand new serialized DAG write emits the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": dag.dag_id, "bundle_name": "testing"},
+ )
+
+ def test_serialization_metric_not_incremented_when_unchanged(self, testing_dag_bundle):
+ """Re-writing an unchanged DAG must not emit the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is False
+
+ mock_stats.incr.assert_not_called()
+
+ def test_serialization_metric_incremented_on_inplace_update(self, dag_maker, session):
+ """Updating a DAG version in place (no dag runs) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ # Change the DAG so the hash differs; with no dag runs this updates in place.
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker") is True
+
+ assert session.scalar(select(func.count()).select_from(DagVersion)) == 1
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": "metric_dag", "bundle_name": "dag_maker"},
+ )
+
+ def test_serialization_metric_incremented_on_new_version(self, dag_maker, session):
+ """Writing a new DAG version (existing run) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ dag_maker.create_dagrun(run_id="run1", logical_date=pendulum.datetime(2025, 1, 1))
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker") is True
+
+ assert session.scalar(select(func.count()).select_from(DagVersion)) == 2
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": "metric_dag", "bundle_name": "dag_maker"},
+ )
+
+ @mock.patch("airflow._shared.observability.metrics.stats._export_legacy_names", True)
+ @mock.patch("airflow._shared.observability.metrics.stats._get_backend")
+ def test_serialization_metric_exports_new_and_legacy_names(self, mock_get_backend, testing_dag_bundle):
+ """Serializing a DAG emits both the modern ``dag.serialization`` metric and its legacy name."""
+ mock_backend = mock.MagicMock(spec=StatsLogger)
+ mock_get_backend.return_value = mock_backend
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
Review Comment:
Nit: the `bundle_name` is also used at least twice for each test and it could be a common variable.
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3467184103
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +204,76 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
+ def test_serialization_metric_incremented_on_new_write(self, testing_dag_bundle):
+ """A brand new serialized DAG write emits the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": dag.dag_id, "bundle_name": "testing"},
+ )
+
+ def test_serialization_metric_not_incremented_when_unchanged(self, testing_dag_bundle):
+ """Re-writing an unchanged DAG must not emit the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is False
+
+ mock_stats.incr.assert_not_called()
+
+ def test_serialization_metric_incremented_on_inplace_update(self, dag_maker, session):
+ """Updating a DAG version in place (no dag runs) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ # Change the DAG so the hash differs; with no dag runs this updates in place.
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker") is True
Review Comment:
That's because you haven't passed a `bundle_name` when calling the `dag_maker` and it defaults to `dag_maker`.
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "xBis7 (via GitHub)" <gi...@apache.org>.
xBis7 commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3690593321
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -762,6 +763,7 @@ def write_dag(
session.merge(dag_version)
# Update the latest DagCode
DagCode.update_source_code(dag_id=dag.dag_id, fileloc=dag.fileloc, session=session)
+ stats.incr("dag.serialization_writes", tags={"dag_id": dag.dag_id, "bundle_name": bundle_name})
Review Comment:
@potiuk There was a bug which was fixed in PR https://github.com/apache/airflow/pull/63871
When the dag template contained a callable as a value, without any actual changes to the dag, every time that it was parsed, it would get reserialized. We noticed this issue by chance while monitoring the DB.
I think it's good to keep both metrics but distinguish between them. Having a metric for even when there isn't an actual write, will help us identify such issues sooner.
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "Ei-Sandi (via GitHub)" <gi...@apache.org>.
Ei-Sandi commented on PR #68906:
URL: https://github.com/apache/airflow/pull/68906#issuecomment-4782582058
@xBis7 The comments has been addressed. Can you please review it again ? Thank 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] Add metrics for how often a dag gets serialised. [airflow]
Posted by "vatsrahul1001 (via GitHub)" <gi...@apache.org>.
vatsrahul1001 commented on PR #68906:
URL: https://github.com/apache/airflow/pull/68906#issuecomment-5116132674
This looks more like a new feature and more suitable for minor 3.4.0
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "Ei-Sandi (via GitHub)" <gi...@apache.org>.
Ei-Sandi commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3466959693
##########
airflow-core/tests/unit/models/test_serialized_dag.py:
##########
@@ -201,6 +204,76 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
assert s_dag_2.data["dag"]["tags"] == ["example", "new_tag", "params"]
assert dag_updated is True
+ def test_serialization_metric_incremented_on_new_write(self, testing_dag_bundle):
+ """A brand new serialized DAG write emits the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ mock_stats.incr.assert_called_once_with(
+ "dag.serialization",
+ tags={"dag_id": dag.dag_id, "bundle_name": "testing"},
+ )
+
+ def test_serialization_metric_not_incremented_when_unchanged(self, testing_dag_bundle):
+ """Re-writing an unchanged DAG must not emit the ``dag.serialization`` metric."""
+ dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is True
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="testing") is False
+
+ mock_stats.incr.assert_not_called()
+
+ def test_serialization_metric_incremented_on_inplace_update(self, dag_maker, session):
+ """Updating a DAG version in place (no dag runs) emits the metric once."""
+ with dag_maker("metric_dag") as dag:
+ PythonOperator(task_id="task1", python_callable=lambda: None)
+ # Change the DAG so the hash differs; with no dag runs this updates in place.
+ PythonOperator(task_id="task2", python_callable=lambda: None, dag=dag)
+
+ with mock.patch(self.SERIALIZED_DAG_STATS) as mock_stats:
+ assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name="dag_maker") is True
Review Comment:
The bundle names are different because
Tests built with the dag_maker fixture use its default bundle name "dag_maker" according to https://github.com/apache/airflow/blob/main/devel-common/src/tests_common/pytest_plugin.py#L1366.
Tests built with make_example_dags register and write under a "testing" bundle according to https://github.com/apache/airflow/blob/main/airflow-core/tests/unit/models/test_serialized_dag.py#L81-L87.
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "Ei-Sandi (via GitHub)" <gi...@apache.org>.
Ei-Sandi commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3692662549
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -762,6 +763,7 @@ def write_dag(
session.merge(dag_version)
# Update the latest DagCode
DagCode.update_source_code(dag_id=dag.dag_id, fileloc=dag.fileloc, session=session)
+ stats.incr("dag.serialization_writes", tags={"dag_id": dag.dag_id, "bundle_name": bundle_name})
Review Comment:
@potiuk I created a follow up PR #70838 with 2 different metrics names. Could you please have a look at it, thank 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] Add metrics for how often a dag gets serialised. [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #68906:
URL: https://github.com/apache/airflow/pull/68906
--
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 metrics for how often a dag gets serialised. [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on code in PR #68906:
URL: https://github.com/apache/airflow/pull/68906#discussion_r3683039882
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -762,6 +763,7 @@ def write_dag(
session.merge(dag_version)
# Update the latest DagCode
DagCode.update_source_code(dag_id=dag.dag_id, fileloc=dag.fileloc, session=session)
+ stats.incr("dag.serialization_writes", tags={"dag_id": dag.dag_id, "bundle_name": bundle_name})
Review Comment:
This is the earlier of the two emission sites, and it fires on the path where the serialized Dag **did not change** — only `dag_version.bundle_name` / `bundle_version` / `version_data` were merged and `DagCode.update_source_code` refreshed. No new `SerializedDagModel` row is written here.
The metric description says "Number of times a Dag was serialized and written to the metadata DB", which doesn't match that path. Anyone using `dag.serialization_writes` to measure how often Dags actually re-serialize — the obvious use, and what the name suggests — will over-count every time a bundle version changes without the Dag changing.
Two options, either is fine:
1. Emit only from the second site (the real write), and drop this one.
2. Keep both but distinguish them — either a `reason`/`kind` tag (`"metadata_refresh"` vs `"new_version"`), or reword the description to say it counts write operations including version-metadata refreshes.
I'd lean towards (1) unless you specifically want visibility into the refresh path.
---
Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org