Skip to content

Commit 0b1b2d3

Browse files
author
deepinsight coder
committed
Add run_type tag to dagrun.duration.failed timeout metric
When a Dag run timed out, the scheduler emitted dagrun.duration.failed with only a dag_id tag, while every other dagrun.duration.* metric (emitted by DagRun._emit_duration_stats_for_finished_state) also carries run_type. This made the timeout path inconsistent and dropped run_type from dashboards and alerts for timed-out runs. Emit the metric with dag_run.stats_tags so it includes both dag_id and run_type, matching the canonical completion path. closes: #64765
1 parent 03c71e5 commit 0b1b2d3

3 files changed

Lines changed: 43 additions & 1 deletion

File tree

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
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.

airflow-core/src/airflow/jobs/scheduler_job_runner.py

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2457,7 +2457,9 @@ def _schedule_dag_run(
24572457
stats.timing(
24582458
"dagrun.duration.failed",
24592459
duration,
2460-
tags={"dag_id": dag_run.dag_id},
2460+
# Include ``run_type`` (via stats_tags) so the timeout path matches
2461+
# DagRun._emit_duration_stats_for_finished_state; see #64765.
2462+
tags=dag_run.stats_tags,
24612463
)
24622464
return callback_to_execute
24632465

airflow-core/tests/unit/jobs/test_scheduler_job.py

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3835,6 +3835,45 @@ def test_dagrun_timeout_fails_run(self, dag_maker):
38353835
session.rollback()
38363836
session.close()
38373837

3838+
@mock.patch("airflow._shared.observability.metrics.stats._get_backend")
3839+
def test_dagrun_timeout_duration_metric_has_run_type(self, mock_get_backend, dag_maker):
3840+
"""
3841+
The ``dagrun.duration.failed`` metric emitted when a Dag run times out must carry the
3842+
``run_type`` tag, matching the metric emitted on normal Dag run completion via
3843+
``DagRun._emit_duration_stats_for_finished_state``. Regression test for #64765.
3844+
"""
3845+
mock_stats = mock.MagicMock(spec=StatsLogger)
3846+
mock_get_backend.return_value = mock_stats
3847+
3848+
session = settings.Session()
3849+
with dag_maker(
3850+
dag_id="test_dagrun_timeout_duration_metric",
3851+
dagrun_timeout=datetime.timedelta(seconds=60),
3852+
session=session,
3853+
):
3854+
EmptyOperator(task_id="dummy")
3855+
3856+
dr = dag_maker.create_dagrun(start_date=timezone.utcnow() - datetime.timedelta(days=1))
3857+
3858+
scheduler_job = Job()
3859+
self.job_runner = SchedulerJobRunner(job=scheduler_job)
3860+
3861+
self.job_runner._schedule_dag_run(dr, session)
3862+
session.flush()
3863+
3864+
session.refresh(dr)
3865+
assert dr.state == State.FAILED
3866+
3867+
# The timing call must include both dag_id and run_type (dag_run.stats_tags), not just dag_id.
3868+
mock_stats.timing.assert_any_call(
3869+
"dagrun.duration.failed",
3870+
mock.ANY,
3871+
tags={"dag_id": dr.dag_id, "run_type": dr.run_type},
3872+
)
3873+
3874+
session.rollback()
3875+
session.close()
3876+
38383877
def test_dagrun_timeout_fails_run_and_update_next_dagrun(self, dag_maker):
38393878
"""
38403879
Test that dagrun timeout fails run and update the next dagrun

0 commit comments

Comments
 (0)