Add metric for scheduling delay between first run task & expected start time (#9544)

Acehaidrey · web-flow · commit aac3877ec374 · 2020-11-13T23:03:42.000+01:00
Co-authored-by: Ace Haidrey &lt;ahaidrey@pinterest.com&gt;
diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
@@ -480,6 +480,7 @@ def update_state(
         else:
             self.set_state(State.RUNNING)
 
+        self._emit_true_scheduling_delay_stats_for_finished_state(finished_tasks)
         self._emit_duration_stats_for_finished_state()
 
         session.merge(self)
@@ -565,6 +566,41 @@ def _are_premature_tis(
                 return True
         return False
 
+    def _emit_true_scheduling_delay_stats_for_finished_state(self, finished_tis):
+        """
+        This is a helper method to emit the true scheduling delay stats, which is defined as
+        the time when the first task in DAG starts minus the expected DAG run datetime.
+        This method will be used in the update_state method when the state of the DagRun
+        is updated to a completed status (either success or failure). The method will find the first
+        started task within the DAG and calculate the expected DagRun start time (based on
+        dag.execution_date & dag.schedule_interval), and minus these two values to get the delay.
+        The emitted data may contains outlier (e.g. when the first task was cleared, so
+        the second task's start_date will be used), but we can get rid of the the outliers
+        on the stats side through the dashboards tooling built.
+        Note, the stat will only be emitted if the DagRun is a scheduler triggered one
+        (i.e. external_trigger is False).
+        """
+        try:
+            if self.state == State.RUNNING:
+                return
+            if self.external_trigger:
+                return
+            if not finished_tis:
+                return
+            dag = self.get_dag()
+            ordered_tis_by_start_date = [ti for ti in finished_tis if ti.start_date]
+            ordered_tis_by_start_date.sort(key=lambda ti: ti.start_date, reverse=False)
+            first_start_date = ordered_tis_by_start_date[0].start_date
+            if first_start_date:
+                # dag.following_schedule calculates the expected start datetime for a scheduled dagrun
+                # i.e. a daily flow for execution date 1/1/20 actually runs on 1/2/20 hh:mm:ss,
+                # and ti.start_date will be 1/2/20 hh:mm:ss so the following schedule is comparison
+                true_delay = (first_start_date - dag.following_schedule(self.execution_date)).total_seconds()
+                if true_delay >= 0:
+                    Stats.timing(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)
+        except Exception as e:
+            self.log.warning(f'Failed to record first_task_scheduling_delay metric:\n{e}')
+
     def _emit_duration_stats_for_finished_state(self):
         if self.state == State.RUNNING:
             return
diff --git a/docs/logging-monitoring/metrics.rst b/docs/logging-monitoring/metrics.rst
@@ -133,16 +133,17 @@ Name                                                Description
 Timers
 ------
 
-=========================================== =================================================================
-Name                                        Description
-=========================================== =================================================================
-``dagrun.dependency-check.<dag_id>``        Milliseconds taken to check DAG dependencies
-``dag.<dag_id>.<task_id>.duration``         Milliseconds taken to finish a task
-``dag_processing.last_duration.<dag_file>`` Milliseconds taken to load the given DAG file
-``dagrun.duration.success.<dag_id>``        Milliseconds taken for a DagRun to reach success state
-``dagrun.duration.failed.<dag_id>``         Milliseconds taken for a DagRun to reach failed state
-``dagrun.schedule_delay.<dag_id>``          Milliseconds of delay between the scheduled DagRun start date and
-                                            the actual DagRun start date
-``scheduler.critical_section_duration``     Milliseconds spent in the critical section of scheduler loop --
-                                            only a single scheduler can enter this loop at a time
-=========================================== =================================================================
+=================================================== ========================================================================
+Name                                                Description
+=================================================== ========================================================================
+``dagrun.dependency-check.<dag_id>``                Milliseconds taken to check DAG dependencies
+``dag.<dag_id>.<task_id>.duration``                 Milliseconds taken to finish a task
+``dag_processing.last_duration.<dag_file>``         Milliseconds taken to load the given DAG file
+``dagrun.duration.success.<dag_id>``                Milliseconds taken for a DagRun to reach success state
+``dagrun.duration.failed.<dag_id>``                 Milliseconds taken for a DagRun to reach failed state
+``dagrun.schedule_delay.<dag_id>``                  Milliseconds of delay between the scheduled DagRun
+                                                    start date and the actual DagRun start date
+``scheduler.critical_section_duration``             Milliseconds spent in the critical section of scheduler loop --
+                                                    only a single scheduler can enter this loop at a time
+``dagrun.<dag_id>.first_task_scheduling_delay``     Seconds elapsed between first task start_date and dagrun expected start
+=================================================== ========================================================================
diff --git a/tests/models/test_dagrun.py b/tests/models/test_dagrun.py
@@ -19,6 +19,7 @@
 import datetime
 import unittest
 from unittest import mock
+from unittest.mock import call
 
 from parameterized import parameterized
 
@@ -27,8 +28,10 @@
 from airflow.models.dagrun import DagRun
 from airflow.operators.dummy_operator import DummyOperator
 from airflow.operators.python import ShortCircuitOperator
+from airflow.stats import Stats
 from airflow.utils import timezone
 from airflow.utils.callback_requests import DagCallbackRequest
+from airflow.utils.dates import days_ago
 from airflow.utils.state import State
 from airflow.utils.trigger_rule import TriggerRule
 from airflow.utils.types import DagRunType
@@ -714,3 +717,56 @@ def test_next_dagruns_to_examine_only_unpaused(self):
 
         session.rollback()
         session.close()
+
+    @mock.patch.object(Stats, 'timing')
+    def test_no_scheduling_delay_for_nonscheduled_runs(self, stats_mock):
+        """
+        Tests that dag scheduling delay stat is not called if the dagrun is not a scheduled run.
+        This case is manual run. Simple test for sanity check.
+        """
+        dag = DAG(dag_id='test_dagrun_stats', start_date=days_ago(1))
+        dag_task = DummyOperator(task_id='dummy', dag=dag)
+
+        initial_task_states = {
+            dag_task.task_id: State.SUCCESS,
+        }
+
+        dag_run = self.create_dag_run(dag=dag, state=State.RUNNING, task_states=initial_task_states)
+        dag_run.update_state()
+        self.assertNotIn(call(f'dagrun.{dag.dag_id}.first_task_scheduling_delay'), stats_mock.mock_calls)
+
+    @mock.patch.object(Stats, 'timing')
+    def test_emit_scheduling_delay(self, stats_mock):
+        """
+        Tests that dag scheduling delay stat is set properly once running scheduled dag.
+        dag_run.update_state() invokes the _emit_true_scheduling_delay_stats_for_finished_state method.
+        """
+        dag = DAG(dag_id='test_emit_dag_stats', start_date=days_ago(1))
+        dag_task = DummyOperator(task_id='dummy', dag=dag, owner='airflow')
+
+        session = settings.Session()
+        orm_dag = DagModel(
+            dag_id=dag.dag_id,
+            has_task_concurrency_limits=False,
+            next_dagrun=dag.start_date,
+            next_dagrun_create_after=dag.following_schedule(dag.start_date),
+            is_active=True,
+        )
+        session.add(orm_dag)
+        session.flush()
+        dag_run = dag.create_dagrun(
+            run_type=DagRunType.SCHEDULED,
+            state=State.SUCCESS,
+            execution_date=dag.start_date,
+            start_date=dag.start_date,
+            session=session,
+        )
+        ti = dag_run.get_task_instance(dag_task.task_id)
+        ti.set_state(State.SUCCESS, session)
+        session.commit()
+        session.close()
+        dag_run.update_state()
+        true_delay = (ti.start_date - dag.following_schedule(dag_run.execution_date)).total_seconds()
+        stats_mock.assert_called()
+        sched_delay_stat_call = call(f'dagrun.{dag.dag_id}.first_task_scheduling_delay', true_delay)
+        self.assertIn(sched_delay_stat_call, stats_mock.mock_calls)
