Skip to content

Emit metric dag auto paused #69078

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

Open
takayoshi-makabe wants to merge 3 commits into
base: main
Choose a base branch
from
+38 −0
Open

Emit metric dag auto paused #69078

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e1b28cc
Emit metric dag auto paused
takayoshi-makabe Jun 27, 2026
c701bff
Register dag.auto_paused metric in the metrics registry YAML
takayoshi-makabe Jun 27, 2026
98128f6
remove run_type
takayoshi-makabe Jul 30, 2026
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
1 change: 1 addition & 0 deletions airflow-core/src/airflow/models/dagrun.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1021,6 +1021,7 @@ def _check_last_n_dagruns_failed(self, dag_id, max_consecutive_failed_dag_runs,
.values(is_paused=True)
.execution_options(synchronize_session="fetch")
)
stats.incr("dag.auto_paused", tags={"dag_id": self.dag_id})
session.add(
Comment thread
takayoshi-makabe marked this conversation as resolved.
Log(
event="paused",
Expand Down
30 changes: 30 additions & 0 deletions airflow-core/tests/unit/models/test_dag.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1129,6 +1129,36 @@ def test_dag_not_paused_when_latest_by_run_after_succeeds(self, testing_dag_bund
session.expire_all()
assert not session.get(DagModel, dag.dag_id).is_paused

@mock.patch("airflow.models.dagrun.stats.incr")
@pytest.mark.db_test
def test_auto_pause_emits_metric(self, mock_stats_incr, testing_dag_bundle):
"""Verify stats.incr("dag.auto_paused") is emitted when the scheduler auto-pauses a DAG."""
dag_id = "dag_auto_pause_metric"
dag = DAG(dag_id, schedule=None, is_paused_upon_creation=False, max_consecutive_failed_dag_runs=1)
op1 = BashOperator(task_id="task", bash_command="exit 1;")
dag.add_task(op1)
session = settings.Session()
session.add(DagModel(dag_id=dag.dag_id, bundle_name="testing", is_stale=False))
session.flush()

scheduler_dag = sync_dag_to_db(dag, session=session)
self._add_dag_run(
scheduler_dag,
op1,
session,
run_id="run_fail",
logical_date=TEST_DATE,
run_after=TEST_DATE,
ti_state=TaskInstanceState.FAILED,
run_state=State.FAILED,
)

assert session.get(DagModel, dag.dag_id).is_paused
mock_stats_incr.assert_any_call(
"dag.auto_paused",
tags={"dag_id": dag_id},
)
Comment thread
takayoshi-makabe marked this conversation as resolved.

def test_dag_is_deactivated_upon_dagfile_deletion(self, dag_maker):
dag_id = "old_existing_dag"
with dag_maker(dag_id, schedule=None, is_paused_upon_creation=True) as dag:
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 @@ -234,6 +234,13 @@ metrics:
legacy_name: "dag.serialization_writes.{dag_id}.{bundle_name}"
name_variables: ["dag_id", "bundle_name"]

- name: "dag.auto_paused"

@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.

The declaration doesn't match what the code actually emits. stats.incr(..., tags=self.stats_tags) uses DagRun.stats_tags, which carries both dag_id and run_type — but this entry says name_variables: [] and the description mentions only "dag_id tagging".

For comparison, #70013 declares its tagged gauges as name_variables: ["dag_id"] with a matching legacy_name. Worth being consistent: either list both variables here with an appropriate legacy_name, or narrow the emission to {"dag_id": self.dag_id} if run_type isn't wanted on this counter. Given it counts a Dag-level event rather than a run-level one, dropping run_type may actually be the cleaner answer.

Also minor: the description's second line starts at the same indentation as the key, which makes it a slightly odd multi-line plain scalar. Indenting the continuation under description: reads better and matches the surrounding entries.

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

@takayoshi-makabe takayoshi-makabe Jul 30, 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

Thanks for the detailed review! I've dropped run_type from the tags so dag.auto_paused only carries dag_id (matching the "Dag-level event" reasoning you laid out), and fixed the description indentation to match the surrounding entries. CI is green now — happy to take another look whenever you get a chance.

commit: 98128f6

Drafted-by: Claude Code (Sonnet 5); reviewed by @takayoshi-makabe before posting

description: "Number of Dags automatically paused due to consecutive failures exceeding
the configured threshold. Metric with dag_id tagging."
type: "counter"
legacy_name: "-"
name_variables: []

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