Simplify OTEL metrics config to mirror traces pattern

airflow-core/src/airflow/api_fastapi/app.py

@@ -71,32 +71,8 @@ class _AuthManagerState:
     _lock = threading.Lock()
 
 
-def _initialize_api_server_stats() -> None:
-    """
-    Initialize the ``Stats`` singleton in the API server process.
-
-    Initialization is guarded so a metrics misconfiguration can never prevent the API server
-    from starting.
-    """
-    try:
-        from airflow._shared.observability.metrics import stats
-        from airflow.observability.metrics import stats_utils
-
-        stats.initialize(
-            factory=stats_utils.get_stats_factory(),
-            export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
-        )
-    except Exception:
-        log.warning(
-            "Failed to initialize API server Stats in the API server; metrics emitted through the "
-            "API server Stats singleton will not be recorded.",
-            exc_info=True,
-        )
-
-
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    _initialize_api_server_stats()
     async with AsyncExitStack() as stack:
         for route in app.routes:
             if isinstance(route, Mount) and isinstance(route.app, FastAPI):

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

@@ -64,7 +64,6 @@
 from airflow.models.dagwarning import DagWarning
 from airflow.models.db_callback_request import DbCallbackRequest
 from airflow.models.errors import ParseImportError
-from airflow.observability.metrics import stats_utils
 from airflow.sdk import SecretCache
 from airflow.sdk.log import init_log_file, logging_processors
 from airflow.typing_compat import assert_never
@@ -387,16 +386,11 @@ def prepare_server_process_context(self) -> None:
         os.environ["_AIRFLOW_PROCESS_CONTEXT"] = "server"
 
     def prepare_process_context(self) -> None:
-        """Initialize transport-neutral process state (selector, stats) before the parsing loop starts."""
+        """Initialize transport-neutral process state (selector) before the parsing loop starts."""
         # Initialization is delayed until here to avoid fork issues in some
         # selector implementations. Also see _StubSelector documentation.
         self.selector = selectors.DefaultSelector()
 
-        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."""
         self.sync_bundles()

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

@@ -39,7 +39,6 @@
 from airflow.executors.workloads.types import state_class_for_key
 from airflow.models import Log
 from airflow.models.taskinstancekey import TaskInstanceKey
-from airflow.observability.metrics import stats_utils
 from airflow.utils.helpers import prune_dict
 from airflow.utils.log.logging_mixin import LoggingMixin
 
@@ -211,10 +210,6 @@ def jwt_generator(self) -> JWTGenerator:
         return generator
 
     def __init__(self, parallelism: int = PARALLELISM, team_name: str | None = None):
-        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

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

@@ -108,7 +108,6 @@
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.models.team import Team
 from airflow.models.trigger import TRIGGER_FAIL_REPR, Trigger, TriggerFailureReason, handle_event_submit
-from airflow.observability.metrics import stats_utils
 from airflow.partition_mappers.base import is_rollup
 from airflow.serialization.definitions.assets import SerializedAssetUniqueKey
 from airflow.serialization.definitions.notset import NOTSET
@@ -1668,13 +1667,6 @@ def _execute(self) -> int | None:
                 executor.callback_sink = callback_sink
                 executor.start()
 
-            # local import due to type_checking.
-
-            stats.initialize(
-                factory=stats_utils.get_stats_factory(),
-                export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
-            )
-
             self._run_scheduler_loop()
 
             if settings.Session is not None:

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

@@ -56,7 +56,6 @@
 from airflow.jobs.job import perform_heartbeat
 from airflow.models.dagbag import DBDagBag
 from airflow.models.trigger import Trigger
-from airflow.observability.metrics import stats_utils
 from airflow.sdk.api.datamodels._generated import HITLDetailResponse
 from airflow.sdk.definitions.asset import Asset
 from airflow.sdk.execution_time import supervisor
@@ -262,10 +261,6 @@ def _execute(self) -> int | None:
         os.environ["_AIRFLOW_PROCESS_CONTEXT"] = "server"
         self.log.info("Starting the triggerer")
         self.register_signals()
-        stats.initialize(
-            factory=stats_utils.get_stats_factory(),
-            export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
-        )
         self.trigger_runner = None
         try:
             # Kick off runner sub-process without DB access

airflow-core/src/airflow/observability/metrics/otel_logger.py

@@ -26,28 +26,16 @@
 
 
 def get_otel_logger() -> SafeOtelLogger:
-    # The config values have been deprecated and therefore,
-    # if the user hasn't added them to the config, the default values won't be used.
-    # A fallback is needed to avoid an exception.
-    port = None
-    if conf.has_option("metrics", "otel_port"):
-        port = conf.getint("metrics", "otel_port")
-
-    conf_interval = None
-    if conf.has_option("metrics", "otel_interval_milliseconds"):
-        conf_interval = conf.getfloat("metrics", "otel_interval_milliseconds")
-
-    return otel_logger.get_otel_logger(
-        host=conf.get("metrics", "otel_host", fallback=None),  # ex: "breeze-otel-collector"
-        port=port,  # ex: 4318
-        prefix=conf.get("metrics", "otel_prefix", fallback=None),  # ex: "airflow"
+    return otel_logger.configure_otel(
+        host=conf.get("metrics", "otel_host", fallback=None),
+        port=conf.get("metrics", "otel_port", fallback=None),
         ssl_active=conf.getboolean("metrics", "otel_ssl_active", fallback=False),
-        # PeriodicExportingMetricReader will default to an interval of 60000 millis.
-        conf_interval=conf_interval,  # ex: 30000
+        service=conf.get("metrics", "otel_service", fallback=None),
+        interval_ms=conf.get("metrics", "otel_interval_milliseconds", fallback=None),
         debug=conf.getboolean("metrics", "otel_debugging_on", fallback=False),
-        service_name=conf.get("metrics", "otel_service", fallback=None),
-        metrics_allow_list=conf.get("metrics", "metrics_allow_list", fallback=None),
-        metrics_block_list=conf.get("metrics", "metrics_block_list", fallback=None),
+        prefix=conf.get("metrics", "otel_prefix", fallback="airflow"),
+        allow_list=conf.get("metrics", "metrics_allow_list", fallback=None),
+        block_list=conf.get("metrics", "metrics_block_list", fallback=None),
         stat_name_handler=conf.getimport("metrics", "stat_name_handler", fallback=None),
         statsd_influxdb_enabled=conf.getboolean("metrics", "statsd_influxdb_enabled", fallback=False),
     )

airflow-core/src/airflow/settings.py

@@ -805,6 +805,25 @@ def import_local_settings():
         log.info("Loaded airflow_local_settings from %s .", airflow_local_settings.__file__)
 
 
+def _initialize_stats() -> None:
+    """
+    Initialize the ``Stats`` singleton for this process.
+
+    Initialization is guarded so a metrics misconfiguration can never prevent ``import airflow``
+    from succeeding.
+    """
+    try:
+        from airflow._shared.observability.metrics import stats
+        from airflow.observability.metrics import stats_utils
+
+        stats.initialize(
+            factory=stats_utils.get_stats_factory(),
+            export_legacy_names=conf.getboolean("metrics", "legacy_names_on"),
+        )
+    except Exception:
+        log.warning("Failed to initialize Stats; metrics will not be recorded.", exc_info=True)
+
+
 def initialize():
     """Initialize Airflow with all the settings from this file."""
     configure_vars()
@@ -816,6 +835,7 @@ def initialize():
     import_local_settings()
     configure_logging()
     configure_otel(conf)
+    _initialize_stats()
     configure_adapters()
     # The webservers import this file from models.py with the default settings.
 

airflow-core/tests/unit/api_fastapi/test_app.py

@@ -168,47 +168,3 @@ def call_create_auth_manager():
     assert call_count == 1
 
     app_module.purge_cached_app()
-
-
-class TestInitializeApiServerStats:
-    """
-    Ensure that stats subsystem is properly initialized in API server.
-    """
-
-    def test_initializes_api_server_stats_with_factory(self):
-        """It initializes the Stats singleton in the API server using the configured factory."""
-        sentinel_factory = object()
-        with (
-            mock.patch("airflow._shared.observability.metrics.stats") as mock_stats,
-            mock.patch(
-                "airflow.observability.metrics.stats_utils.get_stats_factory",
-                return_value=sentinel_factory,
-            ) as mock_get_factory,
-        ):
-            app_module._initialize_api_server_stats()
-
-            mock_get_factory.assert_called_once_with()
-            mock_stats.initialize.assert_called_once()
-            _, kwargs = mock_stats.initialize.call_args
-            assert kwargs["factory"] is sentinel_factory
-            assert isinstance(kwargs["export_legacy_names"], bool)
-
-    def test_stats_failure_does_not_block_startup(self):
-        """A metrics misconfiguration must not prevent the API server from starting."""
-        with (
-            mock.patch("airflow._shared.observability.metrics.stats") as mock_stats,
-            mock.patch("airflow.observability.metrics.stats_utils.get_stats_factory"),
-            mock.patch("airflow.api_fastapi.app.log") as mock_logger,
-        ):
-            mock_stats.initialize.side_effect = RuntimeError("boom")
-
-            # Must not raise.
-            app_module._initialize_api_server_stats()
-
-        mock_logger.warning.assert_called_once()
-
-    def test_stats_initialized_during_lifespan(self, client):
-        """_initialize_api_server_stats must be called as part of the app lifespan, not just defined."""
-        with mock.patch.object(app_module, "_initialize_api_server_stats") as mock_init:
-            with client():
-                mock_init.assert_called_once()

airflow-core/tests/unit/core/test_settings.py

@@ -533,3 +533,50 @@ def test_early_return_when_all_none(self):
             settings.dispose_orm(do_log=False)
 
         mock_close.assert_not_called()
+
+
+class TestInitializeStats:
+    """
+    Ensure the Stats singleton is initialized once for the process via settings.initialize().
+    """
+
+    def test_initializes_stats_with_factory(self):
+        """It initializes the Stats singleton using the configured factory."""
+        sentinel_factory = object()
+        with (
+            mock.patch("airflow._shared.observability.metrics.stats") as mock_stats,
+            mock.patch(
+                "airflow.observability.metrics.stats_utils.get_stats_factory",
+                return_value=sentinel_factory,
+            ) as mock_get_factory,
+        ):
+            settings._initialize_stats()
+
+            mock_get_factory.assert_called_once_with()
+            mock_stats.initialize.assert_called_once()
+            _, kwargs = mock_stats.initialize.call_args
+            assert kwargs["factory"] is sentinel_factory
+            assert isinstance(kwargs["export_legacy_names"], bool)
+
+    def test_stats_failure_does_not_break_initialize(self):
+        """A metrics misconfiguration must not prevent process initialization."""
+        with (
+            mock.patch("airflow._shared.observability.metrics.stats") as mock_stats,
+            mock.patch("airflow.observability.metrics.stats_utils.get_stats_factory"),
+            mock.patch("airflow.settings.log") as mock_logger,
+        ):
+            mock_stats.initialize.side_effect = RuntimeError("boom")
+
+            # Must not raise.
+            settings._initialize_stats()
+
+        mock_logger.warning.assert_called_once()
+
+    @mock.patch("airflow.settings.prepare_syspath_for_config_and_plugins")
+    @mock.patch("airflow.settings.import_local_settings")
+    @mock.patch("airflow.settings._initialize_stats")
+    def test_stats_initialized_during_initialize(self, mock_initialize_stats, _, __):
+        """settings.initialize() must call _initialize_stats, not just define it."""
+        settings.initialize()
+
+        mock_initialize_stats.assert_called_once()

airflow-core/tests/unit/dag_processing/test_manager.py

@@ -2592,18 +2592,6 @@ def test_create_process_passes_bundle_name_to_process_start(
         call_kwargs = mock_process_start.call_args.kwargs
         assert call_kwargs["bundle_name"] == "testing"
 
-    @mock.patch("airflow.dag_processing.manager.stats.initialize")
-    def test_stats_initialize_called_on_run(self, stats_init_mock, tmp_path, configure_testing_dag_bundle):
-        """Test that stats.initialize() is called when DagFileProcessorManager.run() is executed."""
-        with configure_testing_dag_bundle(tmp_path):
-            manager = DagFileProcessorManager(max_runs=1)
-            manager.run()
-
-        # Verify stats.initialize was called with the expected configuration parameters
-        stats_init_mock.assert_called_once()
-        call_kwargs = stats_init_mock.call_args.kwargs
-        assert "factory" in call_kwargs
-
     def test_run_invokes_before_and_after_hooks(self, tmp_path, configure_testing_dag_bundle):
         """`run()` should call `before_run` then `after_run`, even if the loop raises."""
         with configure_testing_dag_bundle(tmp_path):
@@ -2635,7 +2623,7 @@ def test_after_run_runs_when_parsing_loop_raises(self, tmp_path, configure_testi
             after_run_mock.assert_called_once_with()
 
     def test_prepare_server_process_context_can_be_skipped(self, tmp_path, configure_testing_dag_bundle):
-        """API-backed subclasses can skip server-context setup without losing selector/stats init."""
+        """API-backed subclasses can skip server-context setup without losing selector init."""
         with configure_testing_dag_bundle(tmp_path):
             manager = DagFileProcessorManager(max_runs=1)
             with mock.patch.object(manager, "prepare_server_process_context") as server_ctx_mock:

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

@@ -2490,38 +2490,6 @@ def test_update_triggers_skips_when_ti_has_no_dag_version(session, supervisor_bu
 
 
 class TestTriggererJobRunner:
-    @patch("airflow.jobs.triggerer_job_runner.stats.initialize")
-    @patch.object(TriggerRunnerSupervisor, "start")
-    def test_stats_initialize_called_on_execute(self, mock_supervisor_start, stats_init_mock, session):
-        """Test that stats.initialize() is called when TriggererJobRunner._execute() is executed."""
-        # Setup mock supervisor to immediately stop
-        mock_supervisor = MagicMock()
-        mock_supervisor.stop = False
-        mock_supervisor._exit_code = None
-        mock_supervisor.is_alive.return_value = True
-        mock_supervisor.run.side_effect = lambda: setattr(mock_supervisor, "stop", True)
-        mock_supervisor_start.return_value = mock_supervisor
-
-        job = Job()
-        session.add(job)
-        session.flush()
-
-        job_runner = TriggererJobRunner(job)
-        job_runner.trigger_runner = mock_supervisor
-        mock_supervisor.stop = True  # Stop immediately
-
-        # We don't need to run the full _execute, just verify stats.initialize is called
-        # before TriggerRunnerSupervisor.start
-        with patch.object(job_runner, "register_signals"):
-            # We expect this to fail since we're mocking
-            with contextlib.suppress(Exception):
-                job_runner._execute()
-
-        # Verify stats.initialize was called with the expected configuration parameters
-        stats_init_mock.assert_called_once()
-        call_kwargs = stats_init_mock.call_args.kwargs
-        assert "factory" in call_kwargs
-
     @patch.object(TriggerRunnerSupervisor, "start")
     def test_execute_sets_server_process_context(self, mock_supervisor_start, session, monkeypatch):
         """_execute marks triggerer as server context for secrets backend detection."""
@@ -2542,10 +2510,7 @@ def capture_env(*args, **kwargs):
         monkeypatch.delenv("_AIRFLOW_PROCESS_CONTEXT", raising=False)
         job_runner = TriggererJobRunner(job)
 
-        with (
-            patch.object(job_runner, "register_signals"),
-            patch("airflow.jobs.triggerer_job_runner.stats.initialize"),
-        ):
+        with patch.object(job_runner, "register_signals"):
             job_runner._execute()
 
         assert captured_context["value"] == "server"

providers/common/ai/docs/observability.rst

@@ -78,11 +78,14 @@ variables, for example:
 
 .. code-block:: bash
 
-    # Core tracing defaults the exporter to OTLP/gRPC. For an OTLP/HTTP
-    # endpoint (port 4318, ``/v1/traces`` path) also select the HTTP exporter:
-    export OTEL_TRACES_EXPORTER="otlp_proto_http"
+    # Core tracing defaults to OTLP over HTTP/protobuf (port 4318,
+    # ``/v1/traces`` path):
     export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT="http://otel-collector:4318/v1/traces"
 
+    # To switch to OTLP/gRPC (port 4317, no path) set the protocol explicitly:
+    export OTEL_EXPORTER_OTLP_TRACES_PROTOCOL="grpc"
+    export OTEL_EXPORTER_OTLP_TRACES_ENDPOINT="http://otel-collector:4317"
+
 Capturing prompt and completion content
 ---------------------------------------
 

scripts/ci/docker-compose/integration-otel.yml

@@ -77,7 +77,7 @@ services:
       - INTEGRATION_OTEL=true
       - OTEL_SERVICE_NAME=test
       - OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf
-      - OTEL_TRACES_EXPORTER=otlp_proto_http
+      - OTEL_TRACES_EXPORTER=otlp
       - OTEL_METRICS_EXPORTER=otlp
       - OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://breeze-otel-collector:4318/v1/traces
       - OTEL_EXPORTER_OTLP_METRICS_ENDPOINT=http://breeze-otel-collector:4318/v1/metrics