diff --git a/airflow-core/newsfragments/70838.feature.rst b/airflow-core/newsfragments/70838.feature.rst new file mode 100644 index 0000000000000..e54e473fc4500 --- /dev/null +++ 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. diff --git a/airflow-core/src/airflow/models/serialized_dag.py b/airflow-core/src/airflow/models/serialized_dag.py index f3f5b78309680..cc7f83900b040 100644 --- a/airflow-core/src/airflow/models/serialized_dag.py +++ b/airflow-core/src/airflow/models/serialized_dag.py @@ -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 index fd3568f34619c..abbf179a714da 100644 --- 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}, ), ] diff --git a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml index 536e439f876e2..392b55d7c072f 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml +++ b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml @@ -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"