Skip to content

Add metrics for how often a dag gets serialised. #68906

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
potiuk merged 1 commit into from
Jul 30, 2026
+92 −3
Merged

Add metrics for how often a dag gets serialised. #68906

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions airflow-core/src/airflow/models/serialized_dag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
from sqlalchemy.orm import Mapped, backref, foreign, mapped_column, relationship
from sqlalchemy.sql.expression import func, literal

from airflow._shared.observability.metrics import stats
from airflow._shared.timezones import timezone
from airflow.configuration import conf
from airflow.models.asset import (
Expand Down Expand Up @@ -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})

@potiuk potiuk Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@xBis7 xBis7 Jul 31, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@potiuk There was a bug which was fixed in PR #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.

@Ei-Sandi Ei-Sandi Jul 31, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@potiuk I created a follow up PR #70838 with 2 different metrics names. Could you please have a look at it, thank you.

return True

dagv = DagVersion.write_dag(
Expand All @@ -784,6 +786,7 @@ 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})
return True

@classmethod
Expand Down
85 changes: 82 additions & 3 deletions airflow-core/tests/unit/models/test_serialized_dag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
from sqlalchemy import delete, func, select, update

import airflow.example_dags as example_dags_module
from airflow._shared.observability.metrics.base_stats_logger import StatsLogger
from airflow.dag_processing.dagbag import DagBag
from airflow.models.asset import AssetActive, AssetAliasModel, AssetModel
from airflow.models.dag import DagModel
Expand Down Expand Up @@ -90,6 +91,9 @@ def make_example_dags(module):
class TestSerializedDagModel:
"""Unit tests for SerializedDagModel."""

SERIALIZED_DAG_STATS = "airflow.models.serialized_dag.stats"
TEST_BUNDLE_NAME = "testing"

@pytest.fixture(
autouse=True,
params=[
Expand Down Expand Up @@ -161,7 +165,7 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
example_params_trigger_ui = example_dags.get("example_params_trigger_ui")
dag_updated = SDM.write_dag(
dag=LazyDeserializedDAG.from_dag(example_params_trigger_ui),
bundle_name="testing",
bundle_name=self.TEST_BUNDLE_NAME,
)
assert dag_updated is True

Expand All @@ -178,7 +182,7 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):
# column is not updated
dag_updated = SDM.write_dag(
dag=LazyDeserializedDAG.from_dag(example_params_trigger_ui),
bundle_name="testing",
bundle_name=self.TEST_BUNDLE_NAME,
)
s_dag_1 = SDM.get(example_params_trigger_ui.dag_id)

Expand All @@ -192,7 +196,7 @@ def test_serialized_dag_is_updated_if_dag_is_changed(self, testing_dag_bundle):

dag_updated = SDM.write_dag(
dag=LazyDeserializedDAG.from_dag(example_params_trigger_ui),
bundle_name="testing",
bundle_name=self.TEST_BUNDLE_NAME,
)
s_dag_2 = SDM.get(example_params_trigger_ui.dag_id)

Expand All @@ -201,6 +205,81 @@ 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

Comment thread
Ei-Sandi marked this conversation as resolved.
def test_serialization_metric_incremented_on_new_write(self, testing_dag_bundle):
"""A brand new serialized DAG write emits the ``dag.serialization_writes`` 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",
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."""
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

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 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", bundle_name=self.TEST_BUNDLE_NAME) 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=self.TEST_BUNDLE_NAME) is True

assert session.scalar(select(func.count()).select_from(DagVersion)) == 1
mock_stats.incr.assert_called_once_with(
"dag.serialization_writes",
tags={"dag_id": "metric_dag", "bundle_name": self.TEST_BUNDLE_NAME},
)

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", bundle_name=self.TEST_BUNDLE_NAME) 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=self.TEST_BUNDLE_NAME) is True

assert session.scalar(select(func.count()).select_from(DagVersion)) == 2
mock_stats.incr.assert_called_once_with(
"dag.serialization_writes",
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."""
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=self.TEST_BUNDLE_NAME) is True

mock_backend.incr.assert_has_calls(
[
mock.call(f"dag.serialization_writes.{dag.dag_id}.{self.TEST_BUNDLE_NAME}"),
mock.call(
"dag.serialization_writes",
tags={"dag_id": dag.dag_id, "bundle_name": self.TEST_BUNDLE_NAME},
),
]
)

def test_read_dags(self):
"""DAGs can be read from database."""
example_dags = self._write_example_dags()
Expand Down
7 changes: 7 additions & 0 deletions shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -227,6 +227,13 @@ 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."
type: "counter"
legacy_name: "dag.serialization_writes.{dag_id}.{bundle_name}"
name_variables: ["dag_id", "bundle_name"]

- name: "celery.task_timeout_error"
description: "Number of ``AirflowTaskTimeout`` errors raised when publishing Task to Celery Broker."
type: "counter"
Expand Down