This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 3fb080bf487 Add run_type tag to dagrun.duration.failed timeout metric
(#67765)
3fb080bf487 is described below
commit 3fb080bf4879ff7637dc24e236d4deafca9b9efa
Author: deepinsight coder <[email protected]>
AuthorDate: Thu Jul 30 02:10:22 2026 -0700
Add run_type tag to dagrun.duration.failed timeout metric (#67765)
---
airflow-core/newsfragments/67765.bugfix.rst | 1 +
.../src/airflow/jobs/scheduler_job_runner.py | 2 +-
airflow-core/tests/unit/jobs/test_scheduler_job.py | 38 ++++++++++++++++++++++
3 files changed, 40 insertions(+), 1 deletion(-)
diff --git a/airflow-core/newsfragments/67765.bugfix.rst
b/airflow-core/newsfragments/67765.bugfix.rst
new file mode 100644
index 00000000000..1838418be90
--- /dev/null
+++ b/airflow-core/newsfragments/67765.bugfix.rst
@@ -0,0 +1 @@
+Add the ``run_type`` tag to the ``dagrun.duration.failed`` metric emitted when
a Dag run times out, making it consistent with the same metric emitted on
normal Dag run completion.
diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
index 5f6d17ff360..b6a29995d59 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -2989,7 +2989,7 @@ class SchedulerJobRunner(BaseJobRunner, LoggingMixin):
duration,
tags=prune_dict(
{
- "dag_id": dag_run.dag_id,
+ **dag_run.stats_tags,
"team_name":
self._get_team_names_for_dag_ids([dag_run.dag_id], session).get(
dag_run.dag_id
)
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 1fb72d1df13..53ea5bc933f 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -4237,6 +4237,44 @@ class TestSchedulerJob:
session.rollback()
session.close()
+ @mock.patch("airflow._shared.observability.metrics.stats._get_backend")
+ def test_dagrun_timeout_duration_metric_has_run_type(self,
mock_get_backend, dag_maker):
+ """
+ The ``dagrun.duration.failed`` metric emitted when a Dag run times out
must carry the
+ ``run_type`` tag, matching the metric emitted on normal Dag run
completion via
+ ``DagRun._emit_duration_stats_for_finished_state``.
+ """
+ mock_stats = mock.MagicMock(spec=StatsLogger)
+ mock_get_backend.return_value = mock_stats
+
+ session = settings.Session()
+ with dag_maker(
+ dag_id="test_dagrun_timeout_duration_metric",
+ dagrun_timeout=datetime.timedelta(seconds=60),
+ session=session,
+ ):
+ EmptyOperator(task_id="dummy")
+
+ dr = dag_maker.create_dagrun(start_date=timezone.utcnow() -
datetime.timedelta(days=1))
+
+ scheduler_job = Job()
+ self.job_runner = SchedulerJobRunner(job=scheduler_job)
+
+ self.job_runner._schedule_dag_run(dr, session)
+ session.flush()
+
+ session.refresh(dr)
+ assert dr.state == State.FAILED
+
+ mock_stats.timing.assert_any_call(
+ "dagrun.duration.failed",
+ mock.ANY,
+ tags={"dag_id": dr.dag_id, "run_type": dr.run_type},
+ )
+
+ session.rollback()
+ session.close()
+
def test_dagrun_timeout_fails_run_and_update_next_dagrun(self, dag_maker):
"""
Test that dagrun timeout fails run and update the next dagrun
