This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-3-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-3-test by this push:
new 07653ca3b83 [v3-3-test] Add run_type tag to dagrun.duration.failed
timeout metric (#67765) (#70731)
07653ca3b83 is described below
commit 07653ca3b832ad7e829e156f1b08c618fab30901
Author: Jarek Potiuk <[email protected]>
AuthorDate: Thu Jul 30 11:28:37 2026 +0200
[v3-3-test] Add run_type tag to dagrun.duration.failed timeout metric
(#67765) (#70731)
(cherry picked from commit 3fb080bf4879ff7637dc24e236d4deafca9b9efa)
Co-authored-by: deepinsight coder
<[email protected]>
---
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 c37887c8c4b..9114ff979b9 100644
--- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py
+++ b/airflow-core/src/airflow/jobs/scheduler_job_runner.py
@@ -2876,7 +2876,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 f5aeceb34b3..439869d33c7 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -4113,6 +4113,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
