apache · Ei-Sandi · Jul 31, 2026
diff --git a/airflow-core/newsfragments/70838.feature.rst b/airflow-core/newsfragments/70838.feature.rst
@@ -0,0 +1 @@
+Added the ``dag.serialization.version_created`` and ``dag.serialization.version_updated`` metrics, which count Dag serializations in each of the two situations where serialization happens: when a new Dag version is created, and when the latest Dag version is updated in place.
@@ -763,7 +763,10 @@ 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})
+            stats.incr(
+                "dag.serialization.version_updated",
+                tags={"dag_id": dag.dag_id, "bundle_name": bundle_name},
+            )
             return True
 
         dagv = DagVersion.write_dag(
@@ -786,7 +789,10 @@ def write_dag(
         cls._create_deadline_alert_records(new_serialized_dag, deadline_uuid_mapping)
         log.debug("DAG: %s written to the DB", dag.dag_id)
         DagCode.write_code(dagv, dag.fileloc, session=session)
-        stats.incr("dag.serialization_writes", tags={"dag_id": dag.dag_id, "bundle_name": bundle_name})
+        stats.incr(
+            "dag.serialization.version_created",
+            tags={"dag_id": dag.dag_id, "bundle_name": bundle_name},
+        )
         return True
 
     @classmethod

diff --git a/airflow-core/tests/unit/models/test_serialized_dag.py b/airflow-core/tests/unit/models/test_serialized_dag.py
@@ -206,18 +206,18 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
         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_writes`` metric."""
+        """A brand new serialized DAG write emits the ``dag.serialization.version_created`` 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=self.TEST_BUNDLE_NAME) is True
 
         mock_stats.incr.assert_called_once_with(
-            "dag.serialization_writes",
+            "dag.serialization.version_created",
             tags={"dag_id": dag.dag_id, "bundle_name": self.TEST_BUNDLE_NAME},
         )
 
     def test_serialization_metric_not_incremented_when_unchanged(self, testing_dag_bundle):
-        """Re-writing an unchanged DAG must not emit the ``dag.serialization_writes`` metric."""
+        """Re-writing an unchanged DAG must not emit any serialization metric."""
         dag = make_example_dags(example_dags_module).get("example_params_trigger_ui")
         assert SDM.write_dag(LazyDeserializedDAG.from_dag(dag), bundle_name=self.TEST_BUNDLE_NAME) is True
 
@@ -240,7 +240,7 @@ def test_serialization_metric_incremented_on_inplace_update(self, dag_maker, ses
 
         assert session.scalar(select(func.count()).select_from(DagVersion)) == 1
         mock_stats.incr.assert_called_once_with(
-            "dag.serialization_writes",
+            "dag.serialization.version_updated",
             tags={"dag_id": "metric_dag", "bundle_name": self.TEST_BUNDLE_NAME},
         )
 
@@ -256,14 +256,14 @@ def test_serialization_metric_incremented_on_new_version(self, dag_maker, sessio
 
         assert session.scalar(select(func.count()).select_from(DagVersion)) == 2
         mock_stats.incr.assert_called_once_with(
-            "dag.serialization_writes",
+            "dag.serialization.version_created",
             tags={"dag_id": "metric_dag", "bundle_name": self.TEST_BUNDLE_NAME},
         )
 
     @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_writes`` metric and its legacy name."""
+        """Serializing a DAG emits both the tagged 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")
@@ -272,9 +272,9 @@ def test_serialization_metric_exports_new_and_legacy_names(self, mock_get_backen
 
         mock_backend.incr.assert_has_calls(
             [
-                mock.call(f"dag.serialization_writes.{dag.dag_id}.{self.TEST_BUNDLE_NAME}"),
+                mock.call(f"dag.serialization.version_created.{dag.dag_id}.{self.TEST_BUNDLE_NAME}"),
                 mock.call(
-                    "dag.serialization_writes",
+                    "dag.serialization.version_created",
                     tags={"dag_id": dag.dag_id, "bundle_name": self.TEST_BUNDLE_NAME},
                 ),
             ]

@@ -227,11 +227,19 @@ metrics:
     legacy_name: "-"
     name_variables: []
 
-  - name: "dag.serialization_writes"
-    description: "Number of times a Dag was serialized and written to the metadata DB.
-    Metric with dag_id and bundle_name tagging."
+  - name: "dag.serialization.version_created"
+    description: "Number of times a Dag was serialized into a new Dag version, on its first serialization
+    or when its latest version already has task instances. Metric with dag_id and bundle_name tagging."
     type: "counter"
-    legacy_name: "dag.serialization_writes.{dag_id}.{bundle_name}"
+    legacy_name: "dag.serialization.version_created.{dag_id}.{bundle_name}"
+    name_variables: ["dag_id", "bundle_name"]
+
+  - name: "dag.serialization.version_updated"
+    description: "Number of times a Dag was serialized over its latest Dag version in place, without
+    creating a new version, because that version has no task instances. Metric with dag_id and
+    bundle_name tagging."
+    type: "counter"
+    legacy_name: "dag.serialization.version_updated.{dag_id}.{bundle_name}"
     name_variables: ["dag_id", "bundle_name"]
 
   - name: "celery.task_timeout_error"
Original file line number	Diff line number	Diff line change
		@@ -0,0 +1 @@
		Added the ``dag.serialization.version_created`` and ``dag.serialization.version_updated`` metrics, which count Dag serializations in each of the two situations where serialization happens: when a new Dag version is created, and when the latest Dag version is updated in place.