Add heartbeat metric for DAG processor (apache#42398)

--------- Signed-off-by: kalyanr <[email protected]>
dabla · Oct 1, 2024 · c05dae7 · c05dae7
1 parent b1346ac
commit c05dae7
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 6 deletions.
diff --git a/airflow/jobs/dag_processor_job_runner.py b/airflow/jobs/dag_processor_job_runner.py
@@ -17,18 +17,18 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING
 
 from airflow.jobs.base_job_runner import BaseJobRunner
 from airflow.jobs.job import Job, perform_heartbeat
+from airflow.stats import Stats
 from airflow.utils.log.logging_mixin import LoggingMixin
+from airflow.utils.session import NEW_SESSION, provide_session
 
 if TYPE_CHECKING:
-    from airflow.dag_processing.manager import DagFileProcessorManager
-
+    from sqlalchemy.orm import Session
 
-def empty_callback(_: Any) -> None:
-    pass
+    from airflow.dag_processing.manager import DagFileProcessorManager
 
 
 class DagProcessorJobRunner(BaseJobRunner, LoggingMixin):
@@ -52,7 +52,7 @@ def __init__(
         self.processor = processor
         self.processor.heartbeat = lambda: perform_heartbeat(
             job=self.job,
-            heartbeat_callback=empty_callback,
+            heartbeat_callback=self.heartbeat_callback,
             only_if_necessary=True,
         )
 
@@ -67,3 +67,7 @@ def _execute(self) -> int | None:
             self.processor.terminate()
             self.processor.end()
         return None
+
+    @provide_session
+    def heartbeat_callback(self, session: Session = NEW_SESSION) -> None:
+        Stats.incr("dag_processor_heartbeat", 1, 1)
diff --git a/chart/files/statsd-mappings.yml b/chart/files/statsd-mappings.yml
@@ -46,6 +46,12 @@ mappings:
     labels:
       type: counter
 
+  - match: airflow.dag_processor_heartbeat
+    match_type: regex
+    name: "airflow_dag_processor_heartbeat"
+    labels:
+      type: counter
+
   - match: airflow.dag.*.*.duration
     name: "airflow_task_duration"
     labels:

diff --git a/docs/apache-airflow/administration-and-deployment/logging-monitoring/metrics.rst b/docs/apache-airflow/administration-and-deployment/logging-monitoring/metrics.rst
@@ -159,6 +159,7 @@ Name                                                                   Descripti
 ``previously_succeeded``                                               Number of previously succeeded task instances. Metric with dag_id and task_id tagging.
 ``zombies_killed``                                                     Zombie tasks killed. Metric with dag_id and task_id tagging.
 ``scheduler_heartbeat``                                                Scheduler heartbeats
+``dag_processor_heartbeat``                                            Standalone DAG processor heartbeats
 ``dag_processing.processes``                                           Relative number of currently running DAG parsing processes (ie this delta
                                                                        is negative when, since the last metric was sent, processes have completed).
                                                                        Metric with file_path and action tagging.