diff --git a/airflow-core/src/airflow/models/dagrun.py b/airflow-core/src/airflow/models/dagrun.py index 3173d51b54df4..323c5da7f7dc2 100644 --- a/airflow-core/src/airflow/models/dagrun.py +++ b/airflow-core/src/airflow/models/dagrun.py @@ -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( Log( event="paused", diff --git a/airflow-core/tests/unit/models/test_dag.py b/airflow-core/tests/unit/models/test_dag.py index ead765b31716d..b3a753271901a 100644 --- a/airflow-core/tests/unit/models/test_dag.py +++ b/airflow-core/tests/unit/models/test_dag.py @@ -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}, + ) + 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: 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..08f305c40c09b 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml +++ b/shared/observability/src/airflow_shared/observability/metrics/metrics_template.yaml @@ -234,6 +234,13 @@ metrics: legacy_name: "dag.serialization_writes.{dag_id}.{bundle_name}" name_variables: ["dag_id", "bundle_name"] + - name: "dag.auto_paused" + 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"