diff --git a/airflow-core/src/airflow/api_fastapi/app.py b/airflow-core/src/airflow/api_fastapi/app.py index e4d1e40efbb3d..1d67aa5ef9041 100644 --- a/airflow-core/src/airflow/api_fastapi/app.py +++ b/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): diff --git a/airflow-core/src/airflow/dag_processing/manager.py b/airflow-core/src/airflow/dag_processing/manager.py index f523ca6c3e1e2..a194bba30ff1d 100644 --- a/airflow-core/src/airflow/dag_processing/manager.py +++ b/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() diff --git a/airflow-core/src/airflow/executors/base_executor.py b/airflow-core/src/airflow/executors/base_executor.py index 8030e9c14455e..36802ca2408cd 100644 --- a/airflow-core/src/airflow/executors/base_executor.py +++ b/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 diff --git a/airflow-core/src/airflow/jobs/scheduler_job_runner.py b/airflow-core/src/airflow/jobs/scheduler_job_runner.py index a9efa1c03d62c..36c0269632fbb 100644 --- a/airflow-core/src/airflow/jobs/scheduler_job_runner.py +++ b/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: diff --git a/airflow-core/src/airflow/jobs/triggerer_job_runner.py b/airflow-core/src/airflow/jobs/triggerer_job_runner.py index f851e491a69dd..8b9a5088043ae 100644 --- a/airflow-core/src/airflow/jobs/triggerer_job_runner.py +++ b/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 diff --git a/airflow-core/src/airflow/observability/metrics/otel_logger.py b/airflow-core/src/airflow/observability/metrics/otel_logger.py index 4e953df468d59..09e1d9e473e62 100644 --- a/airflow-core/src/airflow/observability/metrics/otel_logger.py +++ b/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), ) diff --git a/airflow-core/src/airflow/settings.py b/airflow-core/src/airflow/settings.py index 0a0e1c6619726..50cb3cc35b293 100644 --- a/airflow-core/src/airflow/settings.py +++ b/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. diff --git a/airflow-core/tests/unit/api_fastapi/test_app.py b/airflow-core/tests/unit/api_fastapi/test_app.py index 9fd9a3edad1ae..1e8817ef2431a 100644 --- a/airflow-core/tests/unit/api_fastapi/test_app.py +++ b/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() diff --git a/airflow-core/tests/unit/core/test_settings.py b/airflow-core/tests/unit/core/test_settings.py index c703ee05ab941..4b55ad5f1f890 100644 --- a/airflow-core/tests/unit/core/test_settings.py +++ b/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() diff --git a/airflow-core/tests/unit/dag_processing/test_manager.py b/airflow-core/tests/unit/dag_processing/test_manager.py index c61e29c591a2a..3b28a43f57eaf 100644 --- a/airflow-core/tests/unit/dag_processing/test_manager.py +++ b/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: diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py b/airflow-core/tests/unit/jobs/test_triggerer_job.py index 31a77b50b6592..41ff9cd5cc311 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/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" diff --git a/providers/common/ai/docs/observability.rst b/providers/common/ai/docs/observability.rst index 54d5f5f7e9a36..8d073b020f8da 100644 --- a/providers/common/ai/docs/observability.rst +++ b/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 --------------------------------------- diff --git a/scripts/ci/docker-compose/integration-otel.yml b/scripts/ci/docker-compose/integration-otel.yml index fd7f252c3f1a3..beb72e126958c 100644 --- a/scripts/ci/docker-compose/integration-otel.yml +++ b/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 diff --git a/shared/observability/src/airflow_shared/observability/common.py b/shared/observability/src/airflow_shared/observability/common.py index e912664b35953..64b4fb5527a55 100644 --- a/shared/observability/src/airflow_shared/observability/common.py +++ b/shared/observability/src/airflow_shared/observability/common.py @@ -17,17 +17,24 @@ # under the License. from __future__ import annotations -from typing import TYPE_CHECKING +import os -import structlog -from .otel_env_config import OtelDataType, OtelEnvConfig +def _resolve_otlp_protocol(specific_env_var: str) -> str: + """ + Return the OTLP transport per the OTel SDK environment-variable spec. -if TYPE_CHECKING: - from opentelemetry.sdk.metrics._internal.export import MetricExporter - from opentelemetry.sdk.trace.export import SpanExporter + Returns ``'http/protobuf'`` (default) or ``'grpc'``. ``specific_env_var`` is + the signal-specific override (e.g. ``OTEL_EXPORTER_OTLP_METRICS_PROTOCOL`` + or ``OTEL_EXPORTER_OTLP_TRACES_PROTOCOL``); falls back to the generic + ``OTEL_EXPORTER_OTLP_PROTOCOL`` and finally the default. -log = structlog.getLogger(__name__) + See: + https://opentelemetry.io/docs/specs/otel/protocol/exporter/#specify-protocol + """ + return ( + os.environ.get(specific_env_var) or os.environ.get("OTEL_EXPORTER_OTLP_PROTOCOL") or "http/protobuf" + ) def _format_url_host(host: str | None) -> str | None: @@ -43,94 +50,3 @@ def _format_url_host(host: str | None) -> str | None: if host is not None and ":" in host and not host.startswith("["): return f"[{host}]" return host - - -def get_otel_data_exporter( - *, - otel_env_config: OtelEnvConfig, - host: str | None = None, - port: int | None = None, - ssl_active: bool = False, -) -> SpanExporter | MetricExporter: - protocol = "https" if ssl_active else "http" - - # According to the OpenTelemetry Spec, specific config options like 'OTEL_EXPORTER_OTLP_TRACES_ENDPOINT' - # take precedence over generic ones like 'OTEL_EXPORTER_OTLP_ENDPOINT'. - env_exporter_protocol = ( - otel_env_config.type_specific_exporter_protocol or otel_env_config.exporter_protocol - ) - env_endpoint = otel_env_config.type_specific_endpoint or otel_env_config.base_endpoint - - # If the protocol env var isn't set, then it will be None, - # and it will default to an http/protobuf exporter. - # The grpc and http variants are incompatible types to mypy but functionally interchangeable here. - OTLPMetricExporter: type - OTLPSpanExporter: type - if env_endpoint and env_exporter_protocol == "grpc": - from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import ( - OTLPMetricExporter, # type: ignore[no-redef] - ) - from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( - OTLPSpanExporter, # type: ignore[no-redef] - ) - else: - from opentelemetry.exporter.otlp.proto.http.metric_exporter import ( - OTLPMetricExporter, # type: ignore[no-redef] - ) - from opentelemetry.exporter.otlp.proto.http.trace_exporter import ( - OTLPSpanExporter, # type: ignore[no-redef] - ) - - exporter: SpanExporter | MetricExporter - if env_endpoint: - if host is not None and port is not None: - log.warning( - "Both the standard OpenTelemetry environment variables and " - "the Airflow OpenTelemetry configs have been provided. " - "Using the OpenTelemetry environment variables. " - "The Airflow configs have been deprecated and will be removed in the future." - ) - - endpoint_str = env_endpoint - # The SDK will pick up all the values from the environment. - if otel_env_config.data_type == OtelDataType.TRACES: - exporter = OTLPSpanExporter() - else: - exporter = OTLPMetricExporter() - else: - if host is None or port is None: - # Since the configs have been deprecated, host and port could be None. - # Log a warning to steer the user towards configuring the environment variables - # and deliberately let it fail here without providing fallbacks. - log.warning( - "OpenTelemetry %s have been enabled but the endpoint settings haven't been configured. " - "The Airflow configs have been deprecated and will be removed in the future. " - "Configure the standard OpenTelemetry environment variables instead. " - "For more info, check the docs.", - otel_env_config.data_type.value, - ) - else: - log.warning( - "The Airflow OpenTelemetry configs have been deprecated and will be removed in the future. " - "OpenTelemetry is advised to be configured using the standard environment variables. " - "For more info, check the docs." - ) - # If the environment endpoint isn't set, then assume that the airflow config is used - # where protocol isn't specified, and it's always http/protobuf. - # In that case it should default to the full 'url_path' and set it directly. - - endpoint_suffix = "traces" if otel_env_config.data_type == OtelDataType.TRACES else "metrics" - - endpoint_str = f"{protocol}://{_format_url_host(host)}:{port}/v1/{endpoint_suffix}" - if otel_env_config.data_type == OtelDataType.TRACES: - exporter = OTLPSpanExporter(endpoint=endpoint_str) - else: - exporter = OTLPMetricExporter(endpoint=endpoint_str) - - exporter_name = ( - "OTLPSpanExporter" if otel_env_config.data_type == OtelDataType.TRACES else "OTLPMetricExporter" - ) - - log.info("[%s] Connecting to OpenTelemetry Collector at %s", exporter_name, endpoint_str) - - return exporter diff --git a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py index 9be7103a62b50..707a5a6d95e37 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -19,9 +19,11 @@ import atexit import datetime import logging +import os import random import warnings from collections.abc import Callable +from importlib.metadata import entry_points from typing import TYPE_CHECKING from opentelemetry import metrics @@ -33,9 +35,8 @@ from opentelemetry.sdk.metrics.view import ExponentialBucketHistogramAggregation, View from opentelemetry.sdk.resources import SERVICE_NAME, Resource -from ..common import get_otel_data_exporter +from ..common import _format_url_host, _resolve_otlp_protocol from ..exceptions import InvalidStatsNameException -from ..otel_env_config import load_metrics_env_config from .protocols import Timer from .validators import ( OTEL_NAME_MAX_LENGTH, @@ -47,6 +48,7 @@ if TYPE_CHECKING: from opentelemetry.metrics import Instrument + from opentelemetry.sdk.metrics._internal.export import MetricExporter from opentelemetry.util.types import Attributes from .protocols import DeltaType @@ -435,70 +437,130 @@ def atexit_register_metrics_flush(): atexit.register(flush_otel_metrics) -def get_otel_logger( +def _get_backcompat_config( *, - host: str | None = None, - port: int | None = None, - prefix: str | None = None, - ssl_active: bool = False, - conf_interval: float | None = None, - debug: bool = False, - service_name: str | None = None, - metrics_allow_list: str | None = None, - metrics_block_list: str | None = None, - stat_name_handler: Callable[[str], str] | None = None, - statsd_influxdb_enabled: bool = False, -) -> SafeOtelLogger: + host: str | None, + port: str | None, + ssl_active: bool, + service: str | None, + interval_ms: str | None, +) -> tuple[str | None, float | None, Resource | None]: + resource = None + if service and not os.environ.get("OTEL_SERVICE_NAME") and not os.environ.get("OTEL_RESOURCE_ATTRIBUTES"): + resource = Resource.create(attributes={SERVICE_NAME: service}) + + endpoint = None + if ( + host + and port + and not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") + and not os.environ.get("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") + ): + scheme = "https" if ssl_active else "http" + endpoint = f"{scheme}://{_format_url_host(host)}:{port}/v1/metrics" + + parsed_interval_ms: float | None = None + if interval_ms and not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL"): + try: + parsed_interval_ms = float(interval_ms) + except (TypeError, ValueError): + log.warning("Invalid metrics.otel_interval_milliseconds value: %r; ignoring.", interval_ms) + + return endpoint, parsed_interval_ms, resource + + +def _load_exporter_from_env() -> MetricExporter: """ - Build and return a :class:`SafeOtelLogger` backed by a configured :class:`MeterProvider`. + Pick a metric exporter per the OTel SDK environment-variable spec. - Histogram instruments (used for ``timing()`` / ``timer()`` metrics) are aggregated with - :class:`~opentelemetry.sdk.metrics.view.ExponentialBucketHistogramAggregation` - so that bucket boundaries adapt automatically to the observed data range. This avoids - the need to hand-tune explicit bucket boundaries for metrics that span very different - scales (milliseconds to hours). + ``OTEL_METRICS_EXPORTER`` selects the backend (``otlp`` default; ``console`` + for debugging; ``prometheus`` or custom values are looked up via entry + points). For ``otlp``, ``OTEL_EXPORTER_OTLP_METRICS_PROTOCOL`` then + ``OTEL_EXPORTER_OTLP_PROTOCOL`` selects the transport: ``http/protobuf`` + (default) or ``grpc``. + + See: + https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#exporter-selection + https://opentelemetry.io/docs/specs/otel/protocol/exporter/#specify-protocol """ - otel_env_config = load_metrics_env_config() + exporter_name = os.environ.get("OTEL_METRICS_EXPORTER", "otlp") + if exporter_name == "otlp": + protocol = _resolve_otlp_protocol("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL") + if protocol == "http/protobuf": + from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter + + return OTLPMetricExporter() + if protocol == "grpc": + from opentelemetry.exporter.otlp.proto.grpc.metric_exporter import ( # type: ignore[assignment] + OTLPMetricExporter, + ) - effective_service_name: str = otel_env_config.service_name or service_name or "airflow" - effective_prefix: str = prefix or DEFAULT_METRIC_NAME_PREFIX - resource = Resource.create(attributes={SERVICE_NAME: effective_service_name}) + return OTLPMetricExporter() + raise ValueError(f"Unsupported OTLP protocol {protocol!r}; expected 'grpc' or 'http/protobuf'.") + eps = entry_points(group="opentelemetry_metrics_exporter", name=exporter_name) + ep = next(iter(eps), None) + if ep is None: + raise RuntimeError( + f"No metric exporter found for OTEL_METRICS_EXPORTER={exporter_name!r}. " + f"Available: {[e.name for e in entry_points(group='opentelemetry_metrics_exporter')]}" + ) + return ep.load()() - # https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#periodic-exporting-metricreader - interval = otel_env_config.interval_ms or conf_interval - metric_exporter = get_otel_data_exporter( - otel_env_config=otel_env_config, +def configure_otel( + *, + host: str | None = None, + port: str | None = None, + ssl_active: bool = False, + service: str | None = None, + interval_ms: str | None = None, + debug: bool = False, + prefix: str = DEFAULT_METRIC_NAME_PREFIX, + allow_list: str | None = None, + block_list: str | None = None, + stat_name_handler: Callable[[str], str] | None = None, + statsd_influxdb_enabled: bool = False, +) -> SafeOtelLogger: + backcompat_endpoint, backcompat_interval_ms, resource = _get_backcompat_config( host=host, port=port, ssl_active=ssl_active, + service=service, + interval_ms=interval_ms, ) - readers = [ + if backcompat_endpoint: + os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = backcompat_endpoint + if backcompat_interval_ms is not None: + os.environ["OTEL_METRIC_EXPORT_INTERVAL"] = str(int(backcompat_interval_ms)) + + interval_str = os.environ.get("OTEL_METRIC_EXPORT_INTERVAL") + final_interval_ms = float(interval_str) if interval_str else None + + debug_on = debug or os.environ.get("OTEL_METRICS_EXPORTER") == "console" + + readers: list[PeriodicExportingMetricReader] = [ PeriodicExportingMetricReader( - exporter=metric_exporter, # type: ignore[arg-type] - export_interval_millis=interval, # type: ignore[arg-type] + exporter=_load_exporter_from_env(), # type: ignore[arg-type] + export_interval_millis=final_interval_ms, # type: ignore[arg-type] ) ] - - if otel_env_config.exporter: - debug = otel_env_config.exporter == "console" - - if debug: - export_to_console = PeriodicExportingMetricReader( - ConsoleMetricExporter(), - export_interval_millis=interval, + if debug_on: + readers.append( + PeriodicExportingMetricReader( + ConsoleMetricExporter(), + export_interval_millis=final_interval_ms, # type: ignore[arg-type] + ) ) - readers.append(export_to_console) # Reset the OTel SDK's Once() guard so set_meter_provider() can succeed. - # This is necessary when get_otel_logger() is called after a process fork: - # the parent's _METER_PROVIDER_SET_ONCE._done = True is inherited by the child, - # causing set_meter_provider() to silently fail with "Overriding of current - # MeterProvider is not allowed". The child then uses the parent's stale provider - # whose PeriodicExportingMetricReader thread is dead after fork. - # On first call (no fork), _done is already False so this is a no-op. - # See: https://github.com/apache/airflow/issues/64690 + # This is necessary when configure_otel() is called after a process fork: + # the parent's _METER_PROVIDER_SET_ONCE._done = True is inherited by the + # child, causing set_meter_provider() to silently fail with "Overriding of + # current MeterProvider is not allowed". The child then uses the parent's + # stale provider whose PeriodicExportingMetricReader thread is dead after + # fork. On first call (no fork), _done is already False so this is a + # no-op. See: https://github.com/apache/airflow/issues/64690 try: import opentelemetry.metrics._internal as _metrics_internal @@ -524,8 +586,10 @@ def get_otel_logger( # Register a hook that flushes any in-memory metrics at shutdown. atexit_register_metrics_flush() - validator = get_validator(metrics_allow_list, metrics_block_list) - return SafeOtelLogger( - metrics.get_meter_provider(), effective_prefix, validator, stat_name_handler, statsd_influxdb_enabled + metrics.get_meter_provider(), + prefix, + get_validator(allow_list, block_list), + stat_name_handler, + statsd_influxdb_enabled, ) diff --git a/shared/observability/src/airflow_shared/observability/otel_env_config.py b/shared/observability/src/airflow_shared/observability/otel_env_config.py deleted file mode 100644 index 61f2cf2e1f7af..0000000000000 --- a/shared/observability/src/airflow_shared/observability/otel_env_config.py +++ /dev/null @@ -1,114 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one -# or more contributor license agreements. See the NOTICE file -# distributed with this work for additional information -# regarding copyright ownership. The ASF licenses this file -# to you under the Apache License, Version 2.0 (the -# "License"); you may not use this file except in compliance -# with the License. You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, -# software distributed under the License is distributed on an -# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -# KIND, either express or implied. See the License for the -# specific language governing permissions and limitations -# under the License. - -from __future__ import annotations - -import os -from dataclasses import dataclass -from enum import Enum - -import structlog - -log = structlog.getLogger(__name__) - - -def _parse_kv_str_to_dict(str_var: str | None) -> dict[str, str]: - """ - Convert a string of key-value pairs to a dictionary. - - Environment variables like 'OTEL_RESOURCE_ATTRIBUTES' or 'OTEL_EXPORTER_OTLP_HEADERS' - accept values with the format "key1=value1,key2=value2,..." - """ - configs = {} - if str_var: - for pair in str_var.split(","): - if "=" in pair: - k, v = pair.split("=", 1) - configs[k.strip()] = v.strip() - return configs - - -class OtelDataType(str, Enum): - """Enum with the different telemetry data types.""" - - TRACES = "traces" - METRICS = "metrics" - - -@dataclass(frozen=True) -class OtelEnvConfig: - """Immutable class for holding OTel config environment variables.""" - - data_type: OtelDataType # traces | metrics - base_endpoint: str | None # base url - type_specific_endpoint: str | None # traces | metrics specific url - exporter_protocol: str | None # "grpc" | "http/protobuf" - type_specific_exporter_protocol: str | None # traces | metrics specific protocol - exporter: str | None # OTEL_TRACES_EXPORTER | OTEL_METRICS_EXPORTER - service_name: str | None - headers_kv_str: str | None - headers: dict[str, str] - resource_attributes_kv_str: str | None - resource_attributes: dict[str, str] - interval_ms: float | None - - -def load_otel_env_config(data_type: OtelDataType) -> OtelEnvConfig: - """Read OTel config env vars and return an OtelEnvConfig object.""" - exporter_protocol = os.getenv("OTEL_EXPORTER_OTLP_PROTOCOL") - service_name = os.getenv("OTEL_SERVICE_NAME") - headers_kv_str = os.getenv("OTEL_EXPORTER_OTLP_HEADERS") - resource_attributes_kv_str = os.getenv("OTEL_RESOURCE_ATTRIBUTES") - base_endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT") - - if data_type == OtelDataType.TRACES: - type_specific_endpoint = os.getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") - type_specific_exporter_protocol = os.getenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL") - exporter = os.getenv("OTEL_TRACES_EXPORTER") - interval_ms = None - else: - type_specific_endpoint = os.getenv("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") - type_specific_exporter_protocol = os.getenv("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL") - exporter = os.getenv("OTEL_METRICS_EXPORTER") - # Instead of directly providing a default value of float, - # use a value of str and convert to float to get rid of a static-code check error. - interval = os.getenv("OTEL_METRIC_EXPORT_INTERVAL") - interval_ms = float(interval) if interval else None - - return OtelEnvConfig( - data_type=data_type, - base_endpoint=base_endpoint, - type_specific_endpoint=type_specific_endpoint, - exporter_protocol=exporter_protocol, - type_specific_exporter_protocol=type_specific_exporter_protocol, - exporter=exporter, - service_name=service_name, - headers_kv_str=headers_kv_str, - headers=_parse_kv_str_to_dict(headers_kv_str), - resource_attributes_kv_str=resource_attributes_kv_str, - resource_attributes=_parse_kv_str_to_dict(resource_attributes_kv_str), - interval_ms=interval_ms, - ) - - -def load_traces_env_config() -> OtelEnvConfig: - return load_otel_env_config(OtelDataType.TRACES) - - -def load_metrics_env_config() -> OtelEnvConfig: - return load_otel_env_config(OtelDataType.METRICS) diff --git a/shared/observability/src/airflow_shared/observability/traces/__init__.py b/shared/observability/src/airflow_shared/observability/traces/__init__.py index ea866997db902..401feccaa176c 100644 --- a/shared/observability/src/airflow_shared/observability/traces/__init__.py +++ b/shared/observability/src/airflow_shared/observability/traces/__init__.py @@ -32,6 +32,8 @@ from opentelemetry.trace import NonRecordingSpan, Span, SpanContext, TraceFlags, TraceState from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator +from ..common import _resolve_otlp_protocol + if TYPE_CHECKING: from configparser import ConfigParser log = logging.getLogger(__name__) @@ -229,15 +231,32 @@ def _get_backcompat_config(conf: ConfigParser) -> tuple[str | None, Resource | N def _load_exporter_from_env() -> SpanExporter: """ - Load a span exporter using the OTEL_TRACES_EXPORTER env var. + Pick a span exporter per the OTel SDK environment-variable spec. + + ``OTEL_TRACES_EXPORTER`` selects the backend (``otlp`` default; ``console`` + for debugging; ``zipkin`` or custom values are looked up via entry points). + For ``otlp``, ``OTEL_EXPORTER_OTLP_TRACES_PROTOCOL`` then + ``OTEL_EXPORTER_OTLP_PROTOCOL`` selects the transport: ``http/protobuf`` + (default) or ``grpc``. - Mirrors the entry-point mechanism used by the OTEL SDK auto-instrumentation - configurator. Supported values (from installed packages): - - ``otlp`` (default) — OTLP/gRPC - - ``otlp_proto_http`` — OTLP/HTTP - - ``console`` — stdout (useful for debugging) + See: + https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#exporter-selection + https://opentelemetry.io/docs/specs/otel/protocol/exporter/#specify-protocol """ exporter_name = os.environ.get("OTEL_TRACES_EXPORTER", "otlp") + if exporter_name == "otlp": + protocol = _resolve_otlp_protocol("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL") + if protocol == "http/protobuf": + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + + return OTLPSpanExporter() + if protocol == "grpc": + from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import ( # type: ignore[assignment] + OTLPSpanExporter, + ) + + return OTLPSpanExporter() + raise ValueError(f"Unsupported OTLP protocol {protocol!r}; expected 'grpc' or 'http/protobuf'.") eps = entry_points(group="opentelemetry_traces_exporter", name=exporter_name) ep = next(iter(eps), None) if ep is None: diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index f1e90e5300311..c522c273df8b8 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -27,22 +27,22 @@ from opentelemetry.metrics import MeterProvider from opentelemetry.sdk.metrics.view import ExponentialBucketHistogramAggregation, View -from airflow_shared.observability.common import get_otel_data_exporter from airflow_shared.observability.metrics.otel_logger import ( OTEL_NAME_MAX_LENGTH, UP_DOWN_COUNTERS, MetricsMap, SafeOtelLogger, _generate_key_name, + _get_backcompat_config, _is_up_down_counter, + _load_exporter_from_env, + configure_otel, full_name, - get_otel_logger, ) from airflow_shared.observability.metrics.validators import ( BACK_COMPAT_METRIC_NAMES, MetricNameLengthExemptionWarning, ) -from airflow_shared.observability.otel_env_config import load_metrics_env_config from tests_common.test_utils.config import env_vars @@ -360,84 +360,47 @@ def test_timer_start_and_stop_manually_send_true(self, mock_time, name): self.meter.get_meter().create_histogram.assert_called_once_with(name=full_name(name), unit="ms") @pytest.mark.parametrize( - ( - "provided_env_vars", - "airflow_conf_host", - "airflow_conf_port", - "expected_endpoint", - "expected_exporter_module", - ), + ("provided_env_vars", "airflow_conf_host", "airflow_conf_port", "expected_endpoint"), [ + # When the standard OTel env var is set, the bridging is a no-op + # (the SDK reads the env var directly) — _get_backcompat_config + # returns endpoint=None. pytest.param( - { - "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:1234", - "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", - }, + {"OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:1234"}, "breeze-otel-collector", "4318", - "localhost:1234", - "grpc", - id="env_vars_with_grpc", - ), - pytest.param( - { - "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", - }, - "breeze-otel-collector", - "4318", - "http://breeze-otel-collector:4318/v1/metrics", - "http", - id="protocol_is_ignored_if_no_env_endpoint", + None, + id="env_endpoint_takes_precedence_no_bridging", ), pytest.param( - { - "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:1234", - "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", - }, + {"OTEL_EXPORTER_OTLP_METRICS_ENDPOINT": "http://localhost:2222"}, "breeze-otel-collector", "4318", - "http://localhost:1234/v1/metrics", - "http", - id="for_http_with_env_vars_otel_builds_full_url", + None, + id="env_metrics_endpoint_takes_precedence_no_bridging", ), + # No env endpoint set → bridge the deprecated Airflow conf into a URL. pytest.param( {}, "breeze-otel-collector", "4318", "http://breeze-otel-collector:4318/v1/metrics", - "http", - id="use_airflow_config", + id="airflow_conf_bridges_to_url", ), + # No env endpoint, no Airflow conf → no bridging. pytest.param( - { - "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:1234", - "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", - }, + {}, None, None, - "http://localhost:1234/v1/metrics", - "http", - id="only_env_vars", - ), - pytest.param( - { - "OTEL_EXPORTER_OTLP_ENDPOINT": "http://localhost:1234", - "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT": "http://localhost:2222", - "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", - "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "grpc", - }, - None, None, - "localhost:2222", - "grpc", - id="type_specific_vars_take_precedence", + id="neither_env_nor_airflow_conf", ), + # IPv6 hosts get bracketed per RFC 3986 §3.2.2. pytest.param( {}, "::1", "4318", "http://[::1]:4318/v1/metrics", - "http", id="airflow_config_ipv6_loopback_is_bracketed", ), pytest.param( @@ -445,7 +408,6 @@ def test_timer_start_and_stop_manually_send_true(self, mock_time, name): "2001:db8::1", "4318", "http://[2001:db8::1]:4318/v1/metrics", - "http", id="airflow_config_ipv6_literal_is_bracketed", ), pytest.param( @@ -453,7 +415,6 @@ def test_timer_start_and_stop_manually_send_true(self, mock_time, name): "[::1]", "4318", "http://[::1]:4318/v1/metrics", - "http", id="airflow_config_already_bracketed_ipv6_is_preserved", ), pytest.param( @@ -461,39 +422,33 @@ def test_timer_start_and_stop_manually_send_true(self, mock_time, name): "10.0.0.1", "4318", "http://10.0.0.1:4318/v1/metrics", - "http", id="airflow_config_ipv4_literal_passes_through_unchanged", ), ], ) - def test_config_priorities( - self, - provided_env_vars, - airflow_conf_host, - airflow_conf_port, - expected_endpoint, - expected_exporter_module, + def test_backcompat_endpoint_bridging( + self, provided_env_vars, airflow_conf_host, airflow_conf_port, expected_endpoint ): - with env_vars(provided_env_vars): - otel_env_config = load_metrics_env_config() + """The bridging helper returns the right backcompat endpoint URL. - otel_metric_exporter = get_otel_data_exporter( - otel_env_config=otel_env_config, + The contract: if the standard OTel env var is set, return ``None`` (the + SDK reads from env directly); otherwise, build the URL from the + deprecated Airflow conf primitives with proper IPv6 bracketing. + """ + with env_vars(provided_env_vars): + endpoint, _, _ = _get_backcompat_config( host=airflow_conf_host, port=airflow_conf_port, + ssl_active=False, + service=None, + interval_ms=None, ) - - assert otel_metric_exporter._endpoint == expected_endpoint - - assert ( - otel_metric_exporter.__class__.__module__ - == f"opentelemetry.exporter.otlp.proto.{expected_exporter_module}.metric_exporter" - ) + assert endpoint == expected_endpoint @mock.patch("airflow_shared.observability.metrics.otel_logger.metrics") @mock.patch("airflow_shared.observability.metrics.otel_logger.MeterProvider") - def test_get_otel_logger_uses_exponential_histogram_view(self, mock_provider, mock_metrics): - get_otel_logger(host="localhost", port=4318) + def test_configure_otel_uses_exponential_histogram_view(self, mock_provider, mock_metrics): + configure_otel(host="localhost", port="4318") call_kwargs = mock_provider.call_args.kwargs views = call_kwargs["views"] @@ -502,6 +457,158 @@ def test_get_otel_logger_uses_exponential_histogram_view(self, mock_provider, mo assert isinstance(view, View) assert isinstance(view._aggregation, ExponentialBucketHistogramAggregation) + @pytest.mark.parametrize( + ("provided_env_vars", "expected_module"), + [ + pytest.param( + {}, + "opentelemetry.exporter.otlp.proto.http.metric_exporter", + id="default_otlp_no_protocol_uses_http", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", + }, + "opentelemetry.exporter.otlp.proto.http.metric_exporter", + id="generic_protocol_http_protobuf_uses_http", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.metric_exporter", + id="generic_protocol_grpc_uses_grpc", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.metric_exporter", + id="metrics_specific_protocol_grpc_uses_grpc", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.metric_exporter", + id="metrics_specific_protocol_overrides_generic", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "http/protobuf", + }, + "opentelemetry.exporter.otlp.proto.http.metric_exporter", + id="metrics_specific_protocol_http_protobuf_uses_http", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "http/protobuf", + }, + "opentelemetry.exporter.otlp.proto.http.metric_exporter", + id="metrics_specific_http_overrides_generic_grpc", + ), + pytest.param( + { + "OTEL_METRICS_EXPORTER": "otlp", + }, + "opentelemetry.exporter.otlp.proto.http.metric_exporter", + id="explicit_otlp_no_protocol_uses_http", + ), + pytest.param( + { + "OTEL_METRICS_EXPORTER": "otlp", + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.metric_exporter", + id="explicit_otlp_with_generic_grpc_uses_grpc", + ), + pytest.param( + { + "OTEL_METRICS_EXPORTER": "otlp", + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.metric_exporter", + id="explicit_otlp_with_metrics_specific_grpc_uses_grpc", + ), + pytest.param( + { + "OTEL_METRICS_EXPORTER": "otlp", + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "http/protobuf", + }, + "opentelemetry.exporter.otlp.proto.grpc.metric_exporter", + id="traces_specific_protocol_does_not_affect_metrics", + ), + pytest.param( + { + "OTEL_METRICS_EXPORTER": "console", + }, + "opentelemetry.sdk.metrics._internal.export", + id="console_exporter_via_entry_point", + ), + ], + ) + def test_load_exporter_from_env_selects_correct_exporter(self, provided_env_vars, expected_module): + with env_vars(provided_env_vars): + exporter = _load_exporter_from_env() + assert exporter.__class__.__module__ == expected_module + + @pytest.mark.parametrize( + ("provided_env_vars", "expected_message"), + [ + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "not_existing_protocol", + }, + "Unsupported OTLP protocol 'not_existing_protocol'", + id="generic_not_existing_protocol", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "not_existing_protocol", + }, + "Unsupported OTLP protocol 'not_existing_protocol'", + id="metrics_specific_not_existing_protocol", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "not_existing_protocol", + }, + "Unsupported OTLP protocol 'not_existing_protocol'", + id="metrics_specific_not_existing_protocol_overrides_valid_generic", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "HTTP/PROTOBUF", + }, + "Unsupported OTLP protocol 'HTTP/PROTOBUF'", + id="case_sensitive_protocol_value", + ), + ], + ) + def test_load_exporter_from_env_raises_on_unsupported_otlp_protocol( + self, provided_env_vars, expected_message + ): + with env_vars(provided_env_vars): + with pytest.raises(ValueError, match=expected_message): + _load_exporter_from_env() + + def test_load_exporter_from_env_raises_on_unknown_exporter_name(self): + with env_vars({"OTEL_METRICS_EXPORTER": "no-such-exporter"}): + with pytest.raises(RuntimeError, match="No metric exporter found"): + _load_exporter_from_env() + + def test_load_exporter_from_env_resolves_protocol_lazily(self): + with env_vars({"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf"}): + first = _load_exporter_from_env() + assert first.__class__.__module__ == "opentelemetry.exporter.otlp.proto.http.metric_exporter" + with env_vars({"OTEL_EXPORTER_OTLP_PROTOCOL": "grpc"}): + second = _load_exporter_from_env() + assert second.__class__.__module__ == "opentelemetry.exporter.otlp.proto.grpc.metric_exporter" + def test_atexit_flush_on_process_exit(self): """ Run a process that initializes a logger, creates a stat and then exits. @@ -510,8 +617,10 @@ def test_atexit_flush_on_process_exit(self): Test that the hook runs and flushes the created stat at shutdown. """ function_call_str = ( - "from airflow_shared.observability.metrics.otel_logger import get_otel_logger; " - "logger = get_otel_logger(debug=True); " + "from airflow_shared.observability.metrics.otel_logger import configure_otel; " + "logger = configure_otel(" + "host='localhost', port='4318', debug=True" + "); " "logger.incr('my_test_stat')" ) @@ -533,7 +642,7 @@ def test_atexit_flush_on_process_exit(self): ) def test_reinit_after_fork_exports_metrics(self): - """Calling get_otel_logger() twice (simulating post-fork re-init) should still export metrics. + """Calling configure_otel() twice (simulating post-fork re-init) should still export metrics. Reproduces https://github.com/apache/airflow/issues/64690: the OTel SDK's Once() guard on set_meter_provider() survives fork, preventing the child from setting a @@ -562,12 +671,12 @@ def test_reinit_after_fork_exports_metrics(self): def mock_service_run(): - logger = get_otel_logger(debug=True) + logger = configure_otel(host="localhost", port="4318", debug=True) logger.incr("my_test_stat") def mock_service_run_reinit(): - """Simulate re-initialization after fork by calling get_otel_logger() twice. + """Simulate re-initialization after fork by calling configure_otel() twice. The first call sets the global MeterProvider and the Once() guard. The second call simulates what happens in a forked child: stats.py detects @@ -575,7 +684,7 @@ def mock_service_run_reinit(): set_meter_provider() silently fails and the child uses a stale provider. """ # First init — sets Once._done = True - get_otel_logger(debug=True) + configure_otel(host="localhost", port="4318", debug=True) # Second init — simulates post-fork re-initialization - logger = get_otel_logger(debug=True) + logger = configure_otel(host="localhost", port="4318", debug=True) logger.incr("post_fork_stat") diff --git a/shared/observability/tests/observability/test_traces.py b/shared/observability/tests/observability/test_traces.py index eb0f499a250e1..557eda8ca7004 100644 --- a/shared/observability/tests/observability/test_traces.py +++ b/shared/observability/tests/observability/test_traces.py @@ -32,11 +32,14 @@ from airflow_shared.observability.traces import ( DEFAULT_TASK_SPAN_DETAIL_LEVEL, TASK_SPAN_DETAIL_LEVEL_KEY, + _load_exporter_from_env, build_trace_state_entries, get_task_span_detail_level, new_dagrun_trace_carrier, ) +from tests_common.test_utils.config import env_vars + def _carrier_is_sampled(carrier: dict[str, str]) -> bool: ctx = TraceContextTextMapPropagator().extract(carrier) @@ -360,3 +363,157 @@ def test_roundtrip_via_carrier(self): span = trace.get_current_span(ctx) assert get_task_span_detail_level(span) == 3 + + +class TestLoadExporterFromEnv: + @pytest.mark.parametrize( + ("provided_env_vars", "expected_module"), + [ + pytest.param( + {}, + "opentelemetry.exporter.otlp.proto.http.trace_exporter", + id="default_otlp_no_protocol_uses_http", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", + }, + "opentelemetry.exporter.otlp.proto.http.trace_exporter", + id="generic_protocol_http_protobuf_uses_http", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.trace_exporter", + id="generic_protocol_grpc_uses_grpc", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.trace_exporter", + id="traces_specific_protocol_grpc_uses_grpc", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.trace_exporter", + id="traces_specific_protocol_overrides_generic", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "http/protobuf", + }, + "opentelemetry.exporter.otlp.proto.http.trace_exporter", + id="traces_specific_protocol_http_protobuf_uses_http", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "http/protobuf", + }, + "opentelemetry.exporter.otlp.proto.http.trace_exporter", + id="traces_specific_http_overrides_generic_grpc", + ), + pytest.param( + { + "OTEL_TRACES_EXPORTER": "otlp", + }, + "opentelemetry.exporter.otlp.proto.http.trace_exporter", + id="explicit_otlp_no_protocol_uses_http", + ), + pytest.param( + { + "OTEL_TRACES_EXPORTER": "otlp", + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.trace_exporter", + id="explicit_otlp_with_generic_grpc_uses_grpc", + ), + pytest.param( + { + "OTEL_TRACES_EXPORTER": "otlp", + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "grpc", + }, + "opentelemetry.exporter.otlp.proto.grpc.trace_exporter", + id="explicit_otlp_with_traces_specific_grpc_uses_grpc", + ), + pytest.param( + { + "OTEL_TRACES_EXPORTER": "otlp", + "OTEL_EXPORTER_OTLP_PROTOCOL": "grpc", + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "http/protobuf", + }, + "opentelemetry.exporter.otlp.proto.grpc.trace_exporter", + id="metrics_specific_protocol_does_not_affect_traces", + ), + pytest.param( + { + "OTEL_TRACES_EXPORTER": "console", + }, + "opentelemetry.sdk.trace.export", + id="console_exporter_via_entry_point", + ), + ], + ) + def test_load_exporter_from_env_selects_correct_exporter(self, provided_env_vars, expected_module): + with env_vars(provided_env_vars): + exporter = _load_exporter_from_env() + assert exporter.__class__.__module__ == expected_module + + @pytest.mark.parametrize( + ("provided_env_vars", "expected_message"), + [ + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "not_existing_protocol", + }, + "Unsupported OTLP protocol 'not_existing_protocol'", + id="generic_not_existing_protocol", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "not_existing_protocol", + }, + "Unsupported OTLP protocol 'not_existing_protocol'", + id="traces_specific_not_existing_protocol", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf", + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "not_existing_protocol", + }, + "Unsupported OTLP protocol 'not_existing_protocol'", + id="traces_specific_not_existing_protocol_overrides_valid_generic", + ), + pytest.param( + { + "OTEL_EXPORTER_OTLP_PROTOCOL": "HTTP/PROTOBUF", + }, + "Unsupported OTLP protocol 'HTTP/PROTOBUF'", + id="case_sensitive_protocol_value", + ), + ], + ) + def test_load_exporter_from_env_raises_on_unsupported_otlp_protocol( + self, provided_env_vars, expected_message + ): + with env_vars(provided_env_vars): + with pytest.raises(ValueError, match=expected_message): + _load_exporter_from_env() + + def test_load_exporter_from_env_raises_on_unknown_exporter_name(self): + with env_vars({"OTEL_TRACES_EXPORTER": "no-such-exporter"}): + with pytest.raises(RuntimeError, match="No span exporter found"): + _load_exporter_from_env() + + def test_load_exporter_from_env_resolves_protocol_lazily(self): + with env_vars({"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf"}): + first = _load_exporter_from_env() + assert first.__class__.__module__ == "opentelemetry.exporter.otlp.proto.http.trace_exporter" + with env_vars({"OTEL_EXPORTER_OTLP_PROTOCOL": "grpc"}): + second = _load_exporter_from_env() + assert second.__class__.__module__ == "opentelemetry.exporter.otlp.proto.grpc.trace_exporter" diff --git a/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py b/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py index ebf64d701b3f7..9ba26a854008d 100644 --- a/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py +++ b/task-sdk/src/airflow/sdk/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), )