[AIRFLOW-5274] dag loading duration metric name too long (apache#5890)

cfei18 · Aug 26, 2019 · 45176c8 · 45176c8
1 parent c921812
commit 45176c8
Show file tree

Hide file tree

Showing 3 changed files with 11 additions and 11 deletions.
diff --git a/UPDATING.md b/UPDATING.md
@@ -39,6 +39,11 @@ assists users migrating to a new version.
 <!-- END doctoc generated TOC please keep comment here to allow auto update -->
 ## Airflow Master
 
+### Change dag loading duration metric name
+Change DAG file loading duration metric from 
+`dag.loading-duration.<dag_id>` to `dag.loading-duration.<dag_file>`. This is to 
+better handle the case when a DAG file has multiple DAGs.
+
 ### Changes to ImapHook, ImapAttachmentSensor and ImapAttachmentToS3Operator
 
 ImapHook:

diff --git a/airflow/models/dagbag.py b/airflow/models/dagbag.py
@@ -374,8 +374,6 @@ def collect_dags(
 
         dag_folder = correct_maybe_zipped(dag_folder)
 
-        dags_by_name = {}
-
         for filepath in list_py_file_paths(dag_folder, safe_mode=safe_mode,
                                            include_examples=include_examples):
             try:
@@ -389,7 +387,6 @@ def collect_dags(
                 td = timezone.utcnow() - ts
                 td = td.total_seconds() + (
                     float(td.microseconds) / 1000000)
-                dags_by_name[dag_id_names] = dag_ids
                 stats.append(FileLoadStat(
                     filepath.replace(dag_folder, ''),
                     td,
@@ -408,13 +405,11 @@ def collect_dags(
         self.dagbag_stats = sorted(
             stats, key=lambda x: x.duration, reverse=True)
         for file_stat in self.dagbag_stats:
-            dag_ids = dags_by_name[file_stat.dags]
-            if file_stat.dag_num >= 1:
-                # if we found multiple dags per file, the stat is 'dag_id1 _ dag_id2'
-                dag_names = '_'.join(dag_ids)
-                Stats.timing('dag.loading-duration.{}'.
-                             format(dag_names),
-                             file_stat.duration)
+            # file_stat.file similar format: /subdir/dag_name.py
+            filename = file_stat.file.split('/')[-1].replace('.py', '')
+            Stats.timing('dag.loading-duration.{}'.
+                         format(filename),
+                         file_stat.duration)
 
     def dagbag_report(self):
         """Prints a report around DagBag loading stats"""

diff --git a/docs/metrics.rst b/docs/metrics.rst
@@ -84,7 +84,7 @@ Name                              Description
 ================================= =================================================
 dagrun.dependency-check.<dag_id>  Seconds taken to check DAG dependencies
 dag.<dag_id>.<task_id>.duration   Seconds taken to finish a task
-dag.loading-duration.<dag_id>     Seconds taken to load the given DAG
+dag.loading-duration.<dag_file>   Seconds taken to load the given DAG file
 dagrun.duration.success.<dag_id>  Seconds taken for a DagRun to reach success state
 dagrun.duration.failed.<dag_id>   Seconds taken for a DagRun to reach failed state
 dagrun.schedule_delay.<dag_id>    Seconds of delay between the scheduled DagRun