Remove the DualStatsManager and the Stats interfaces

airflow-core/src/airflow/__init__.py

@@ -89,7 +89,6 @@
     # Deprecated lazy imports
     "AirflowException": (".exceptions", "AirflowException", True),
     "Dataset": (".sdk", "Asset", True),
-    "Stats": (".observability.stats", "Stats", True),
     "Trace": (".observability.trace", "Trace", True),
     "metrics": (".observability.metrics", "", True),
     "traces": ("._shared.observability.traces", "", True),

airflow-core/src/airflow/assets/manager.py

@@ -25,7 +25,7 @@
 from sqlalchemy import exc, or_, select
 from sqlalchemy.orm import joinedload
 
-from airflow._shared.observability.metrics.stats import Stats
+from airflow._shared.observability.metrics import stats
 from airflow.configuration import conf
 from airflow.listeners.listener import get_listener_manager
 from airflow.listeners.types import AssetEvent as ListenerAssetEvent
@@ -283,7 +283,7 @@ def register_asset_change(
             )
         )
 
-        Stats.incr("asset.updates")
+        stats.incr("asset.updates")
 
         dags_to_queue = (
             dags_to_queue_from_asset | dags_to_queue_from_asset_alias | dags_to_queue_from_asset_ref

airflow-core/src/airflow/cli/commands/daemon_utils.py

@@ -74,10 +74,6 @@ def run_command_with_daemon_option(
             )
 
             with ctx:
-                # in daemon context stats client needs to be reinitialized.
-                from airflow._shared.observability.metrics.stats import Stats
-
-                Stats.instance = None
                 callback()
     else:
         signal.signal(signal.SIGINT, sigint_handler)

airflow-core/src/airflow/dag_processing/manager.py

@@ -45,7 +45,8 @@
 from tabulate import tabulate
 from uuid6 import uuid7
 
-from airflow._shared.observability.metrics.stats import Stats, normalize_name_for_stats
+from airflow._shared.observability.metrics import stats
+from airflow._shared.observability.metrics.stats import normalize_name_for_stats
 from airflow._shared.timezones import timezone
 from airflow.api_fastapi.execution_api.app import InProcessExecutionAPI
 from airflow.configuration import conf
@@ -344,8 +345,10 @@ def prepare_process_context(self) -> None:
         # selector implementations. Also see _StubSelector documentation.
         self.selector = selectors.DefaultSelector()
 
-        stats_factory = stats_utils.get_stats_factory(Stats)
-        Stats.initialize(factory=stats_factory)
+        stats.initialize(
+            factory=stats_utils.get_stats_factory(),
+            export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
+        )
 
     def prepare_bundles(self) -> None:
         """Sync bundle configuration to the DB and load bundles for parsing."""
@@ -649,7 +652,7 @@ def _add_callback_to_queue(self, request: CallbackRequest) -> None:
         )
         self._callback_to_execute[file_info].append(request)
         self._add_files_to_queue([file_info], mode="front")
-        Stats.incr("dag_processing.other_callback_count")
+        stats.incr("dag_processing.other_callback_count")
 
     @provide_session
     def get_bundle_state(self, bundle_name: str, *, session: Session = NEW_SESSION) -> BundleState | None:
@@ -934,7 +937,7 @@ def _log_file_processing_stats(self, known_files: dict[str, set[DagFileInfo]]):
                 last_run = stat.last_finish_time
                 if last_run:
                     seconds_ago = (utcnow - last_run).total_seconds()
-                    Stats.gauge(f"dag_processing.last_run.seconds_ago.{file_name}", seconds_ago)
+                    stats.gauge(f"dag_processing.last_run.seconds_ago.{file_name}", seconds_ago)
 
                 rows.append(
                     (
@@ -1012,7 +1015,7 @@ def handle_removed_files(self, known_files: dict[str, set[DagFileInfo]]):
     def purge_removed_files_from_queue(self, present: set[DagFileInfo]):
         """Remove from queue any files no longer observed locally."""
         self._file_queue = deque(x for x in self._file_queue if x in present)
-        Stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))
+        stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))
 
     def remove_orphaned_file_stats(self, present: set[DagFileInfo]):
         """Remove the stats for any dag files that don't exist anymore."""
@@ -1030,7 +1033,7 @@ def terminate_orphan_processes(self, present: set[DagFileInfo]):
                     continue
                 file_name = str(file.rel_path)
                 self.log.warning("Stopping processor for %s", file_name)
-                Stats.decr("dag_processing.processes", tags={"file_path": file_name, "action": "stop"})
+                stats.decr("dag_processing.processes", tags={"file_path": file_name, "action": "stop"})
                 processor.kill(signal.SIGKILL)
                 processor.logger_filehandle.close()
                 self._file_stats.pop(file, None)
@@ -1246,10 +1249,10 @@ def _start_new_processes(self):
                 continue
 
             processor = self._create_process(file)
-            Stats.incr("dag_processing.processes", tags={"file_path": str(file.rel_path), "action": "start"})
+            stats.incr("dag_processing.processes", tags={"file_path": str(file.rel_path), "action": "start"})
 
             self._processors[file] = processor
-            Stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))
+            stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))
 
     def _add_new_files_to_queue(self, known_files: dict[str, set[DagFileInfo]]):
         """
@@ -1326,7 +1329,7 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
         if self._parsing_start_time is not None:
             emit_metrics(
                 parse_time=time.perf_counter() - self._parsing_start_time,
-                stats=list(self._file_stats.values()),
+                dag_file_stats=list(self._file_stats.values()),
             )
             self._parsing_start_time = None
 
@@ -1379,7 +1382,7 @@ def prepare_file_queue(self, known_files: dict[str, set[DagFileInfo]]):
                 "\n\t".join(str(f.rel_path) for f in to_queue),
             )
         self._add_files_to_queue(to_queue, mode="back")
-        Stats.incr("dag_processing.file_path_queue_update_count")
+        stats.incr("dag_processing.file_path_queue_update_count")
 
     def _kill_timed_out_processors(self):
         """Kill any file processors that timeout to defend against process hangs."""
@@ -1396,8 +1399,8 @@ def _kill_timed_out_processors(self):
                     self.processor_timeout,
                 )
                 file_name = str(file.rel_path)
-                Stats.decr("dag_processing.processes", tags={"file_path": file_name, "action": "timeout"})
-                Stats.incr("dag_processing.processor_timeouts", tags={"file_path": file_name})
+                stats.decr("dag_processing.processes", tags={"file_path": file_name, "action": "timeout"})
+                stats.incr("dag_processing.processor_timeouts", tags={"file_path": file_name})
                 processor.kill(signal.SIGKILL)
 
                 processors_to_remove.append(file)
@@ -1445,7 +1448,7 @@ def _add_files_to_queue(
         if self._parsing_start_time is None and self._file_queue:
             self._parsing_start_time = time.perf_counter()
 
-        Stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))
+        stats.gauge("dag_processing.file_path_queue_size", len(self._file_queue))
 
     def max_runs_reached(self):
         """:return: whether all file paths have been processed max_runs times."""
@@ -1459,7 +1462,7 @@ def terminate(self):
         """Stop all running processors."""
         for file, processor in self._processors.items():
             # todo: AIP-66 what to do about file_path tag? replace with bundle name and rel path?
-            Stats.decr(
+            stats.decr(
                 "dag_processing.processes", tags={"file_path": str(file.rel_path), "action": "terminate"}
             )
             # SIGTERM, wait 5s, SIGKILL if still alive
@@ -1472,16 +1475,16 @@ def end(self):
             kill_child_processes_by_pids(pids_to_kill)
 
 
-def emit_metrics(*, parse_time: float, stats: Sequence[DagFileStat]):
+def emit_metrics(*, parse_time: float, dag_file_stats: Sequence[DagFileStat]):
     """
     Emit metrics about dag parsing summary.
 
     This is called once every time around the parsing "loop" - i.e. after
     all files have been parsed.
     """
-    Stats.gauge("dag_processing.total_parse_time", parse_time)
-    Stats.gauge("dagbag_size", sum(stat.num_dags for stat in stats))
-    Stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in stats))
+    stats.gauge("dag_processing.total_parse_time", parse_time)
+    stats.gauge("dagbag_size", sum(stat.num_dags for stat in dag_file_stats))
+    stats.gauge("dag_processing.import_errors", sum(stat.import_errors for stat in dag_file_stats))
 
 
 def process_parse_results(
@@ -1506,7 +1509,7 @@ def process_parse_results(
             last_duration=run_duration,
             run_count=run_count,  # Don't increment for callback-only processing
         )
-        Stats.incr("dag_processing.callback_only_count")
+        stats.incr("dag_processing.callback_only_count")
     else:
         # Actual DAG parsing or import error
         stat = DagFileStat(
@@ -1521,11 +1524,10 @@ def process_parse_results(
         file_name = normalize_name_for_stats(Path(relative_fileloc).stem)
         # bundle_name is included to distinguish files with the same name across different bundles
         normalized_bundle = normalize_name_for_stats(bundle_name)
-        Stats.timing(f"dag_processing.last_duration.{normalized_bundle}.{file_name}", stat.last_duration)
-        Stats.timing(
+        stats.timing(
             "dag_processing.last_duration",
             stat.last_duration,
-            tags={"file_name": file_name, "bundle_name": normalized_bundle},
+            tags={"bundle_name": normalized_bundle, "file_name": file_name},
         )
 
     if parsing_result is None:

airflow-core/src/airflow/dag_processing/processor.py

@@ -28,7 +28,7 @@
 import attrs
 from pydantic import BaseModel, Field, TypeAdapter
 
-from airflow._shared.observability.metrics.stats import Stats
+from airflow._shared.observability.metrics import stats
 from airflow.callbacks.callback_requests import (
     CallbackRequest,
     DagCallbackRequest,
@@ -366,7 +366,7 @@ def _execute_dag_callbacks(dagbag: DagBag, request: DagCallbackRequest, log: Fil
             callback(context)
         except Exception:
             log.exception("Callback failed", dag_id=request.dag_id)
-            Stats.incr("dag.callback_exceptions", tags={"dag_id": request.dag_id})
+            stats.incr("dag.callback_exceptions", tags={"dag_id": request.dag_id})
 
 
 def _execute_task_callbacks(dagbag: DagBag, request: TaskCallbackRequest, log: FilteringBoundLogger) -> None:

airflow-core/src/airflow/executors/base_executor.py

@@ -28,7 +28,7 @@
 
 import pendulum
 
-from airflow._shared.observability.metrics.stats import Stats
+from airflow._shared.observability.metrics import stats
 from airflow.cli.cli_config import DefaultHelpParser
 from airflow.configuration import conf
 from airflow.executors import workloads
@@ -204,8 +204,10 @@ def jwt_generator(self) -> JWTGenerator:
         return generator
 
     def __init__(self, parallelism: int = PARALLELISM, team_name: str | None = None):
-        stats_factory = stats_utils.get_stats_factory(Stats)
-        Stats.initialize(factory=stats_factory)
+        stats.initialize(
+            factory=stats_utils.get_stats_factory(),
+            export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
+        )
         super().__init__()
         # Ensure we set this now, so that each subprocess gets the same value
         from airflow.api_fastapi.auth.tokens import get_signing_args
@@ -371,20 +373,20 @@ def _emit_metrics(self, open_slots, num_running_tasks, num_queued_tasks):
         else:
             self.log.debug("%s open slots for executor %s", open_slots, name)
 
-        Stats.gauge(
+        stats.gauge(
             open_slots_metric_name,
             value=open_slots,
-            tags={"status": "open", "name": name},
+            tags={"status": "open", "executor_class_name": name},
         )
-        Stats.gauge(
+        stats.gauge(
             queued_tasks_metric_name,
             value=num_queued_tasks,
-            tags={"status": "queued", "name": name},
+            tags={"status": "queued", "executor_class_name": name},
         )
-        Stats.gauge(
+        stats.gauge(
             running_tasks_metric_name,
             value=num_running_tasks,
-            tags={"status": "running", "name": name},
+            tags={"status": "running", "executor_class_name": name},
         )
 
     def order_queued_tasks_by_priority(self) -> list[tuple[TaskInstanceKey, workloads.ExecuteTask]]:

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

@@ -19,7 +19,7 @@
 
 from typing import TYPE_CHECKING
 
-from airflow._shared.observability.metrics.stats import Stats
+from airflow._shared.observability.metrics import stats
 from airflow.jobs.base_job_runner import BaseJobRunner
 from airflow.jobs.job import Job, perform_heartbeat
 from airflow.utils.log.logging_mixin import LoggingMixin
@@ -68,4 +68,4 @@ def _execute(self) -> int | None:
         return None
 
     def heartbeat_callback(self, session: Session | None = None) -> None:
-        Stats.incr("dag_processor_heartbeat", 1, 1)
+        stats.incr("dag_processor_heartbeat", 1, 1)

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

@@ -29,7 +29,7 @@
 from sqlalchemy.orm import Mapped, backref, foreign, mapped_column, relationship
 from sqlalchemy.orm.session import make_transient
 
-from airflow._shared.observability.metrics.stats import Stats
+from airflow._shared.observability.metrics import stats
 from airflow._shared.timezones import timezone
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
@@ -245,7 +245,7 @@ def heartbeat(
             self.log.debug("[heartbeat]")
             self.heartbeat_failed = False
         except OperationalError:
-            Stats.incr(convert_camel_to_snake(self.__class__.__name__) + "_heartbeat_failure", 1, 1)
+            stats.incr(convert_camel_to_snake(self.__class__.__name__) + "_heartbeat_failure", 1, 1)
             if not self.heartbeat_failed:
                 self.log.exception("%s heartbeat failed with error", self.__class__.__name__)
                 self.heartbeat_failed = True
@@ -268,7 +268,7 @@ def heartbeat(
     @provide_session
     def prepare_for_execution(self, session: Session = NEW_SESSION):
         """Prepare the job for execution."""
-        Stats.incr(self.__class__.__name__.lower() + "_start", 1, 1)
+        stats.incr(self.__class__.__name__.lower() + "_start", 1, 1)
         self.state = JobState.RUNNING
         self.start_date = timezone.utcnow()
         session.add(self)
@@ -284,7 +284,7 @@ def complete_execution(self, session: Session = NEW_SESSION):
         self.end_date = timezone.utcnow()
         session.merge(self)
         session.commit()
-        Stats.incr(self.__class__.__name__.lower() + "_end", 1, 1)
+        stats.incr(self.__class__.__name__.lower() + "_end", 1, 1)
 
     @provide_session
     def most_recent_job(self, session: Session = NEW_SESSION) -> Job | None: