From 0fd98dcac1af92940a806c8820d2b68296f52915 Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Thu, 11 Jun 2026 17:04:49 +0200 Subject: [PATCH 01/11] refactor/simplify OTEL logger configuration. --- .../observability/metrics/otel_logger.py | 53 ----- .../observability/metrics/stats_utils.py | 4 +- .../airflow_shared/observability/common.py | 103 ---------- .../observability/metrics/__init__.py | 190 ++++++++++++++++++ .../observability/metrics/otel_logger.py | 106 ---------- .../observability/otel_env_config.py | 114 ----------- .../observability/metrics/test_otel_logger.py | 167 +++++++-------- .../sdk/observability/metrics/otel_logger.py | 53 ----- .../sdk/observability/metrics/stats_utils.py | 4 +- 9 files changed, 265 insertions(+), 529 deletions(-) delete mode 100644 airflow-core/src/airflow/observability/metrics/otel_logger.py delete mode 100644 shared/observability/src/airflow_shared/observability/otel_env_config.py delete mode 100644 task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py diff --git a/airflow-core/src/airflow/observability/metrics/otel_logger.py b/airflow-core/src/airflow/observability/metrics/otel_logger.py deleted file mode 100644 index 4e953df468d59..0000000000000 --- a/airflow-core/src/airflow/observability/metrics/otel_logger.py +++ /dev/null @@ -1,53 +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 - -from typing import TYPE_CHECKING - -from airflow._shared.observability.metrics import otel_logger -from airflow.configuration import conf - -if TYPE_CHECKING: - from airflow._shared.observability.metrics.otel_logger import SafeOtelLogger - - -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" - 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 - 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), - 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/observability/metrics/stats_utils.py b/airflow-core/src/airflow/observability/metrics/stats_utils.py index 9e0f1263bff66..10b561997a36c 100644 --- a/airflow-core/src/airflow/observability/metrics/stats_utils.py +++ b/airflow-core/src/airflow/observability/metrics/stats_utils.py @@ -33,7 +33,7 @@ def get_stats_factory() -> Callable: return statsd_logger.get_statsd_logger if conf.getboolean("metrics", "otel_on"): - from airflow.observability.metrics import otel_logger + from airflow._shared.observability.metrics import configure_otel - return otel_logger.get_otel_logger + return lambda: configure_otel(conf) return NoStatsLogger diff --git a/shared/observability/src/airflow_shared/observability/common.py b/shared/observability/src/airflow_shared/observability/common.py index e912664b35953..d8873a63d12d8 100644 --- a/shared/observability/src/airflow_shared/observability/common.py +++ b/shared/observability/src/airflow_shared/observability/common.py @@ -17,18 +17,6 @@ # under the License. from __future__ import annotations -from typing import TYPE_CHECKING - -import structlog - -from .otel_env_config import OtelDataType, OtelEnvConfig - -if TYPE_CHECKING: - from opentelemetry.sdk.metrics._internal.export import MetricExporter - from opentelemetry.sdk.trace.export import SpanExporter - -log = structlog.getLogger(__name__) - def _format_url_host(host: str | None) -> str | None: """ @@ -43,94 +31,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/__init__.py b/shared/observability/src/airflow_shared/observability/metrics/__init__.py index 13a83393a9124..6d4e9e61cfbc9 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/__init__.py +++ b/shared/observability/src/airflow_shared/observability/metrics/__init__.py @@ -14,3 +14,193 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. +from __future__ import annotations + +import logging +import os +from importlib.metadata import entry_points +from typing import TYPE_CHECKING + +from opentelemetry import metrics +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics._internal.export import ( + ConsoleMetricExporter, + PeriodicExportingMetricReader, +) +from opentelemetry.sdk.metrics.view import ExponentialBucketHistogramAggregation, View +from opentelemetry.sdk.resources import SERVICE_NAME, Resource + +from ..common import _format_url_host +from .otel_logger import ( + DEFAULT_METRIC_NAME_PREFIX, + SafeOtelLogger, + atexit_register_metrics_flush, +) +from .validators import get_validator + +if TYPE_CHECKING: + from configparser import ConfigParser + + from opentelemetry.sdk.metrics._internal.export import MetricExporter + +log = logging.getLogger(__name__) + + +def _get_backcompat_config( + conf: ConfigParser, +) -> tuple[str | None, float | None, Resource | None]: + """ + Possibly get deprecated Airflow configs for otel metrics. + + Ideally we return ``(None, None, None)`` here. But if the old configuration + is there, then we will use it. + """ + resource = None + if not os.environ.get("OTEL_SERVICE_NAME") and not os.environ.get("OTEL_RESOURCE_ATTRIBUTES"): + service_name = conf.get("metrics", "otel_service", fallback=None) + if service_name: + resource = Resource.create(attributes={SERVICE_NAME: service_name}) + + endpoint = None + if not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") and not os.environ.get( + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" + ): + 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) + if host and port: + scheme = "https" if ssl_active else "http" + endpoint = f"{scheme}://{_format_url_host(host)}:{port}/v1/metrics" + + interval_ms: float | None = None + if not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL") and conf.has_option( + "metrics", "otel_interval_milliseconds" + ): + interval_ms = conf.getfloat("metrics", "otel_interval_milliseconds") + + return endpoint, interval_ms, resource + + +def _load_exporter_from_env() -> MetricExporter: + """ + Load a metric exporter using the ``OTEL_METRICS_EXPORTER`` env var. + + 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) + """ + exporter_name = os.environ.get("OTEL_METRICS_EXPORTER", "otlp") + 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()() + + +def configure_otel(conf: ConfigParser) -> SafeOtelLogger | None: + """ + Configure the OpenTelemetry metrics pipeline from Airflow conf. + + Mirrors ``airflow_shared.observability.traces.configure_otel``: a single + conf-driven entry point that bridges deprecated Airflow-specific options + into the standard OTel environment variables, loads the exporter via the + SDK's entry-point mechanism, and installs the global meter provider. + + Returns the user-facing :class:`SafeOtelLogger` so callers (Stats) can + wrap their metrics. Returns ``None`` when ``metrics.otel_on`` is false. + """ + otel_on = conf.getboolean("metrics", "otel_on", fallback=False) + if not otel_on: + return None + + # Ideally all three are None here. + # They would only be something other than None if the user is still using + # the deprecated Airflow-defined otel configs. + backcompat_endpoint, backcompat_interval_ms, resource = _get_backcompat_config(conf) + + # Backcompat: bridge deprecated configs into the OTel env vars so the + # exporter (loaded below via entry points) picks them up automatically. + if backcompat_endpoint and not ( + os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or os.environ.get("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") + ): + os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = backcompat_endpoint + + if backcompat_interval_ms is not None and not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL"): + os.environ["OTEL_METRIC_EXPORT_INTERVAL"] = str(int(backcompat_interval_ms)) + + interval_str = os.environ.get("OTEL_METRIC_EXPORT_INTERVAL") + interval_ms = float(interval_str) if interval_str else None + + debug = conf.getboolean("metrics", "otel_debugging_on", fallback=False) or ( + os.environ.get("OTEL_METRICS_EXPORTER") == "console" + ) + + readers: list[PeriodicExportingMetricReader] = [ + PeriodicExportingMetricReader( + exporter=_load_exporter_from_env(), # type: ignore[arg-type] + export_interval_millis=interval_ms, # type: ignore[arg-type] + ) + ] + if debug: + readers.append( + PeriodicExportingMetricReader( + ConsoleMetricExporter(), + export_interval_millis=interval_ms, # type: ignore[arg-type] + ) + ) + + # Reset the OTel SDK's Once() guard so set_meter_provider() can succeed. + # 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 + + _metrics_internal._METER_PROVIDER_SET_ONCE._done = False + _metrics_internal._METER_PROVIDER = None + except (ImportError, AttributeError): + pass + + metrics.set_meter_provider( + MeterProvider( + resource=resource, + metric_readers=readers, + views=[ + View( + instrument_type=metrics.Histogram, + aggregation=ExponentialBucketHistogramAggregation(), + ) + ], + shutdown_on_exit=False, + ), + ) + + # Register a hook that flushes any in-memory metrics at shutdown. + atexit_register_metrics_flush() + + # ``getimport`` is an Airflow ``AirflowConfigParser`` extension; plain + # ``ConfigParser`` instances used in tests don't have it. Fall back to + # ``None`` so test fixtures that pass a plain ``ConfigParser`` still work. + stat_name_handler = None + if hasattr(conf, "getimport"): + stat_name_handler = conf.getimport("metrics", "stat_name_handler", fallback=None) + + return SafeOtelLogger( + metrics.get_meter_provider(), + conf.get("metrics", "otel_prefix", fallback=DEFAULT_METRIC_NAME_PREFIX), + get_validator( + conf.get("metrics", "metrics_allow_list", fallback=None), + conf.get("metrics", "metrics_block_list", fallback=None), + ), + stat_name_handler, + conf.getboolean("metrics", "statsd_influxdb_enabled", fallback=False), + ) 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 8d25b23372a10..b59c16190a45e 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -25,22 +25,12 @@ from typing import TYPE_CHECKING from opentelemetry import metrics -from opentelemetry.sdk.metrics import MeterProvider -from opentelemetry.sdk.metrics._internal.export import ( - ConsoleMetricExporter, - PeriodicExportingMetricReader, -) -from opentelemetry.sdk.metrics.view import ExponentialBucketHistogramAggregation, View -from opentelemetry.sdk.resources import SERVICE_NAME, Resource -from ..common import get_otel_data_exporter -from ..otel_env_config import load_metrics_env_config from .protocols import Timer from .validators import ( OTEL_NAME_MAX_LENGTH, ListValidator, PatternAllowListValidator, - get_validator, stat_name_otel_handler, ) @@ -415,99 +405,3 @@ def flush_otel_metrics(): def atexit_register_metrics_flush(): atexit.register(flush_otel_metrics) - - -def get_otel_logger( - *, - 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: - """ - Build and return a :class:`SafeOtelLogger` backed by a configured :class:`MeterProvider`. - - 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_env_config = load_metrics_env_config() - - 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}) - - # 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, - host=host, - port=port, - ssl_active=ssl_active, - ) - - readers = [ - PeriodicExportingMetricReader( - exporter=metric_exporter, # type: ignore[arg-type] - export_interval_millis=interval, # 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, - ) - 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 - try: - import opentelemetry.metrics._internal as _metrics_internal - - _metrics_internal._METER_PROVIDER_SET_ONCE._done = False - _metrics_internal._METER_PROVIDER = None - except (ImportError, AttributeError): - pass - - metrics.set_meter_provider( - MeterProvider( - resource=resource, - metric_readers=readers, - views=[ - View( - instrument_type=metrics.Histogram, - aggregation=ExponentialBucketHistogramAggregation(), - ) - ], - shutdown_on_exit=False, - ), - ) - - # 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 - ) 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/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index 21d27789def09..f49d9497cbb46 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -21,14 +21,15 @@ import subprocess import sys import time +from configparser import ConfigParser from unittest import mock import pytest 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.exceptions import InvalidStatsNameException +from airflow_shared.observability.metrics import _get_backcompat_config, configure_otel from airflow_shared.observability.metrics.otel_logger import ( OTEL_NAME_MAX_LENGTH, UP_DOWN_COUNTERS, @@ -37,13 +38,11 @@ _generate_key_name, _is_up_down_counter, 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 @@ -322,140 +321,99 @@ 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", + id="airflow_conf_ipv6_loopback_is_bracketed", ), pytest.param( {}, "2001:db8::1", "4318", "http://[2001:db8::1]:4318/v1/metrics", - "http", - id="airflow_config_ipv6_literal_is_bracketed", + id="airflow_conf_ipv6_literal_is_bracketed", ), pytest.param( {}, "[::1]", "4318", "http://[::1]:4318/v1/metrics", - "http", - id="airflow_config_already_bracketed_ipv6_is_preserved", + id="airflow_conf_already_bracketed_ipv6_is_preserved", ), pytest.param( {}, "10.0.0.1", "4318", "http://10.0.0.1:4318/v1/metrics", - "http", - id="airflow_config_ipv4_literal_passes_through_unchanged", + id="airflow_conf_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() - - otel_metric_exporter = get_otel_data_exporter( - otel_env_config=otel_env_config, - host=airflow_conf_host, - port=airflow_conf_port, - ) - - assert otel_metric_exporter._endpoint == expected_endpoint - - assert ( - otel_metric_exporter.__class__.__module__ - == f"opentelemetry.exporter.otlp.proto.{expected_exporter_module}.metric_exporter" - ) + """The bridging helper returns the right backcompat endpoint URL. - @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) + 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 with proper IPv6 bracketing. + """ + with env_vars(provided_env_vars): + conf = ConfigParser() + section = {"otel_on": "true"} + if airflow_conf_host: + section["otel_host"] = airflow_conf_host + if airflow_conf_port: + section["otel_port"] = airflow_conf_port + conf["metrics"] = section + + endpoint, _, _ = _get_backcompat_config(conf) + assert endpoint == expected_endpoint + + @mock.patch("airflow_shared.observability.metrics.metrics") + @mock.patch("airflow_shared.observability.metrics.MeterProvider") + def test_configure_otel_uses_exponential_histogram_view(self, mock_provider, mock_metrics): + conf = ConfigParser() + conf["metrics"] = {"otel_on": "true", "otel_host": "localhost", "otel_port": "4318"} + configure_otel(conf) call_kwargs = mock_provider.call_args.kwargs views = call_kwargs["views"] @@ -472,8 +430,14 @@ 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 configparser import ConfigParser; " + "from airflow_shared.observability.metrics import configure_otel; " + "c = ConfigParser(); " + "c['metrics'] = {" + "'otel_on': 'true', 'otel_debugging_on': 'true', " + "'otel_host': 'localhost', 'otel_port': '4318'" + "}; " + "logger = configure_otel(c); " "logger.incr('my_test_stat')" ) @@ -495,7 +459,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 @@ -523,13 +487,24 @@ def test_reinit_after_fork_exports_metrics(self): ) +def _debug_conf() -> ConfigParser: + c = ConfigParser() + c["metrics"] = { + "otel_on": "true", + "otel_debugging_on": "true", + "otel_host": "localhost", + "otel_port": "4318", + } + return c + + def mock_service_run(): - logger = get_otel_logger(debug=True) + logger = configure_otel(_debug_conf()) 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 @@ -537,7 +512,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(_debug_conf()) # Second init — simulates post-fork re-initialization - logger = get_otel_logger(debug=True) + logger = configure_otel(_debug_conf()) logger.incr("post_fork_stat") diff --git a/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py b/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py deleted file mode 100644 index ebf64d701b3f7..0000000000000 --- a/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py +++ /dev/null @@ -1,53 +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 - -from typing import TYPE_CHECKING - -from airflow.sdk._shared.observability.metrics import otel_logger -from airflow.sdk.configuration import conf - -if TYPE_CHECKING: - from airflow.sdk._shared.observability.metrics.otel_logger import SafeOtelLogger - - -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" - 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 - 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), - 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/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py b/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py index 9bd01adcaa033..d672f000d52ef 100644 --- a/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py +++ b/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py @@ -33,7 +33,7 @@ def get_stats_factory() -> Callable: return statsd_logger.get_statsd_logger if conf.getboolean("metrics", "otel_on"): - from airflow.sdk.observability.metrics import otel_logger + from airflow.sdk._shared.observability.metrics import configure_otel - return otel_logger.get_otel_logger + return lambda: configure_otel(conf) return NoStatsLogger From 790fd03a2c01db331c2b550ec12717e13dbc39b9 Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Thu, 11 Jun 2026 22:01:08 +0200 Subject: [PATCH 02/11] addressing PR remarks. --- .../observability/metrics/stats_utils.py | 2 +- .../observability/metrics/__init__.py | 190 ------------------ .../observability/metrics/otel_logger.py | 173 ++++++++++++++++ .../observability/metrics/test_otel_logger.py | 9 +- .../sdk/observability/metrics/stats_utils.py | 2 +- 5 files changed, 180 insertions(+), 196 deletions(-) diff --git a/airflow-core/src/airflow/observability/metrics/stats_utils.py b/airflow-core/src/airflow/observability/metrics/stats_utils.py index 10b561997a36c..80d04736da65e 100644 --- a/airflow-core/src/airflow/observability/metrics/stats_utils.py +++ b/airflow-core/src/airflow/observability/metrics/stats_utils.py @@ -33,7 +33,7 @@ def get_stats_factory() -> Callable: return statsd_logger.get_statsd_logger if conf.getboolean("metrics", "otel_on"): - from airflow._shared.observability.metrics import configure_otel + from airflow._shared.observability.metrics.otel_logger import configure_otel return lambda: configure_otel(conf) return NoStatsLogger diff --git a/shared/observability/src/airflow_shared/observability/metrics/__init__.py b/shared/observability/src/airflow_shared/observability/metrics/__init__.py index 6d4e9e61cfbc9..13a83393a9124 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/__init__.py +++ b/shared/observability/src/airflow_shared/observability/metrics/__init__.py @@ -14,193 +14,3 @@ # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. -from __future__ import annotations - -import logging -import os -from importlib.metadata import entry_points -from typing import TYPE_CHECKING - -from opentelemetry import metrics -from opentelemetry.sdk.metrics import MeterProvider -from opentelemetry.sdk.metrics._internal.export import ( - ConsoleMetricExporter, - PeriodicExportingMetricReader, -) -from opentelemetry.sdk.metrics.view import ExponentialBucketHistogramAggregation, View -from opentelemetry.sdk.resources import SERVICE_NAME, Resource - -from ..common import _format_url_host -from .otel_logger import ( - DEFAULT_METRIC_NAME_PREFIX, - SafeOtelLogger, - atexit_register_metrics_flush, -) -from .validators import get_validator - -if TYPE_CHECKING: - from configparser import ConfigParser - - from opentelemetry.sdk.metrics._internal.export import MetricExporter - -log = logging.getLogger(__name__) - - -def _get_backcompat_config( - conf: ConfigParser, -) -> tuple[str | None, float | None, Resource | None]: - """ - Possibly get deprecated Airflow configs for otel metrics. - - Ideally we return ``(None, None, None)`` here. But if the old configuration - is there, then we will use it. - """ - resource = None - if not os.environ.get("OTEL_SERVICE_NAME") and not os.environ.get("OTEL_RESOURCE_ATTRIBUTES"): - service_name = conf.get("metrics", "otel_service", fallback=None) - if service_name: - resource = Resource.create(attributes={SERVICE_NAME: service_name}) - - endpoint = None - if not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") and not os.environ.get( - "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" - ): - 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) - if host and port: - scheme = "https" if ssl_active else "http" - endpoint = f"{scheme}://{_format_url_host(host)}:{port}/v1/metrics" - - interval_ms: float | None = None - if not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL") and conf.has_option( - "metrics", "otel_interval_milliseconds" - ): - interval_ms = conf.getfloat("metrics", "otel_interval_milliseconds") - - return endpoint, interval_ms, resource - - -def _load_exporter_from_env() -> MetricExporter: - """ - Load a metric exporter using the ``OTEL_METRICS_EXPORTER`` env var. - - 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) - """ - exporter_name = os.environ.get("OTEL_METRICS_EXPORTER", "otlp") - 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()() - - -def configure_otel(conf: ConfigParser) -> SafeOtelLogger | None: - """ - Configure the OpenTelemetry metrics pipeline from Airflow conf. - - Mirrors ``airflow_shared.observability.traces.configure_otel``: a single - conf-driven entry point that bridges deprecated Airflow-specific options - into the standard OTel environment variables, loads the exporter via the - SDK's entry-point mechanism, and installs the global meter provider. - - Returns the user-facing :class:`SafeOtelLogger` so callers (Stats) can - wrap their metrics. Returns ``None`` when ``metrics.otel_on`` is false. - """ - otel_on = conf.getboolean("metrics", "otel_on", fallback=False) - if not otel_on: - return None - - # Ideally all three are None here. - # They would only be something other than None if the user is still using - # the deprecated Airflow-defined otel configs. - backcompat_endpoint, backcompat_interval_ms, resource = _get_backcompat_config(conf) - - # Backcompat: bridge deprecated configs into the OTel env vars so the - # exporter (loaded below via entry points) picks them up automatically. - if backcompat_endpoint and not ( - os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or os.environ.get("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") - ): - os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = backcompat_endpoint - - if backcompat_interval_ms is not None and not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL"): - os.environ["OTEL_METRIC_EXPORT_INTERVAL"] = str(int(backcompat_interval_ms)) - - interval_str = os.environ.get("OTEL_METRIC_EXPORT_INTERVAL") - interval_ms = float(interval_str) if interval_str else None - - debug = conf.getboolean("metrics", "otel_debugging_on", fallback=False) or ( - os.environ.get("OTEL_METRICS_EXPORTER") == "console" - ) - - readers: list[PeriodicExportingMetricReader] = [ - PeriodicExportingMetricReader( - exporter=_load_exporter_from_env(), # type: ignore[arg-type] - export_interval_millis=interval_ms, # type: ignore[arg-type] - ) - ] - if debug: - readers.append( - PeriodicExportingMetricReader( - ConsoleMetricExporter(), - export_interval_millis=interval_ms, # type: ignore[arg-type] - ) - ) - - # Reset the OTel SDK's Once() guard so set_meter_provider() can succeed. - # 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 - - _metrics_internal._METER_PROVIDER_SET_ONCE._done = False - _metrics_internal._METER_PROVIDER = None - except (ImportError, AttributeError): - pass - - metrics.set_meter_provider( - MeterProvider( - resource=resource, - metric_readers=readers, - views=[ - View( - instrument_type=metrics.Histogram, - aggregation=ExponentialBucketHistogramAggregation(), - ) - ], - shutdown_on_exit=False, - ), - ) - - # Register a hook that flushes any in-memory metrics at shutdown. - atexit_register_metrics_flush() - - # ``getimport`` is an Airflow ``AirflowConfigParser`` extension; plain - # ``ConfigParser`` instances used in tests don't have it. Fall back to - # ``None`` so test fixtures that pass a plain ``ConfigParser`` still work. - stat_name_handler = None - if hasattr(conf, "getimport"): - stat_name_handler = conf.getimport("metrics", "stat_name_handler", fallback=None) - - return SafeOtelLogger( - metrics.get_meter_provider(), - conf.get("metrics", "otel_prefix", fallback=DEFAULT_METRIC_NAME_PREFIX), - get_validator( - conf.get("metrics", "metrics_allow_list", fallback=None), - conf.get("metrics", "metrics_block_list", fallback=None), - ), - stat_name_handler, - conf.getboolean("metrics", "statsd_influxdb_enabled", fallback=False), - ) 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 a636f5aa11e69..4af94d387e3d9 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -19,24 +19,38 @@ 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 +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.metrics._internal.export import ( + ConsoleMetricExporter, + PeriodicExportingMetricReader, +) +from opentelemetry.sdk.metrics.view import ExponentialBucketHistogramAggregation, View +from opentelemetry.sdk.resources import SERVICE_NAME, Resource +from ..common import _format_url_host from ..exceptions import InvalidStatsNameException from .protocols import Timer from .validators import ( OTEL_NAME_MAX_LENGTH, ListValidator, PatternAllowListValidator, + get_validator, stat_name_otel_handler, ) if TYPE_CHECKING: + from configparser import ConfigParser + from opentelemetry.metrics import Instrument + from opentelemetry.sdk.metrics._internal.export import MetricExporter from opentelemetry.util.types import Attributes from .protocols import DeltaType @@ -423,3 +437,162 @@ def flush_otel_metrics(): def atexit_register_metrics_flush(): atexit.register(flush_otel_metrics) + + +def _get_backcompat_config( + conf: ConfigParser, +) -> tuple[str | None, float | None, Resource | None]: + """ + Possibly get deprecated Airflow configs for otel metrics. + + Ideally we return ``(None, None, None)`` here. But if the old configuration + is there, then we will use it. + """ + resource = None + if not os.environ.get("OTEL_SERVICE_NAME") and not os.environ.get("OTEL_RESOURCE_ATTRIBUTES"): + service_name = conf.get("metrics", "otel_service", fallback=None) + if service_name: + resource = Resource.create(attributes={SERVICE_NAME: service_name}) + + endpoint = None + if not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") and not os.environ.get( + "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" + ): + 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) + if host and port: + scheme = "https" if ssl_active else "http" + endpoint = f"{scheme}://{_format_url_host(host)}:{port}/v1/metrics" + + interval_ms: float | None = None + if not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL") and conf.has_option( + "metrics", "otel_interval_milliseconds" + ): + interval_ms = conf.getfloat("metrics", "otel_interval_milliseconds") + + return endpoint, interval_ms, resource + + +def _load_exporter_from_env() -> MetricExporter: + """ + Load a metric exporter using the ``OTEL_METRICS_EXPORTER`` env var. + + 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) + """ + exporter_name = os.environ.get("OTEL_METRICS_EXPORTER", "otlp") + 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()() + + +def configure_otel(conf: ConfigParser) -> SafeOtelLogger | None: + """ + Configure the OpenTelemetry metrics pipeline from Airflow conf. + + Bridges deprecated Airflow-specific options into the standard OTel + environment variables, loads the exporter via the SDK's entry-point + mechanism, and installs the global meter provider. + + Returns the user-facing :class:`SafeOtelLogger` so callers (Stats) can + wrap their metrics. Returns ``None`` when ``metrics.otel_on`` is false. + """ + otel_on = conf.getboolean("metrics", "otel_on", fallback=False) + if not otel_on: + return None + + # Ideally all three are None here. + # They would only be something other than None if the user is still using + # the deprecated Airflow-defined otel configs. + backcompat_endpoint, backcompat_interval_ms, resource = _get_backcompat_config(conf) + + # Backcompat: bridge deprecated configs into the OTel env vars so the + # exporter (loaded below via entry points) picks them up automatically. + if backcompat_endpoint and not ( + os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or os.environ.get("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") + ): + os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = backcompat_endpoint + + if backcompat_interval_ms is not None and not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL"): + os.environ["OTEL_METRIC_EXPORT_INTERVAL"] = str(int(backcompat_interval_ms)) + + interval_str = os.environ.get("OTEL_METRIC_EXPORT_INTERVAL") + interval_ms = float(interval_str) if interval_str else None + + debug = conf.getboolean("metrics", "otel_debugging_on", fallback=False) or ( + os.environ.get("OTEL_METRICS_EXPORTER") == "console" + ) + + readers: list[PeriodicExportingMetricReader] = [ + PeriodicExportingMetricReader( + exporter=_load_exporter_from_env(), # type: ignore[arg-type] + export_interval_millis=interval_ms, # type: ignore[arg-type] + ) + ] + if debug: + readers.append( + PeriodicExportingMetricReader( + ConsoleMetricExporter(), + export_interval_millis=interval_ms, # type: ignore[arg-type] + ) + ) + + # Reset the OTel SDK's Once() guard so set_meter_provider() can succeed. + # 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 + + _metrics_internal._METER_PROVIDER_SET_ONCE._done = False + _metrics_internal._METER_PROVIDER = None + except (ImportError, AttributeError): + pass + + metrics.set_meter_provider( + MeterProvider( + resource=resource, + metric_readers=readers, + views=[ + View( + instrument_type=metrics.Histogram, + aggregation=ExponentialBucketHistogramAggregation(), + ) + ], + shutdown_on_exit=False, + ), + ) + + # Register a hook that flushes any in-memory metrics at shutdown. + atexit_register_metrics_flush() + + # ``getimport`` is an Airflow ``AirflowConfigParser`` extension; plain + # ``ConfigParser`` instances used in tests don't have it. Fall back to + # ``None`` so test fixtures that pass a plain ``ConfigParser`` still work. + stat_name_handler = None + if hasattr(conf, "getimport"): + stat_name_handler = conf.getimport("metrics", "stat_name_handler", fallback=None) + + return SafeOtelLogger( + metrics.get_meter_provider(), + conf.get("metrics", "otel_prefix", fallback=DEFAULT_METRIC_NAME_PREFIX), + get_validator( + conf.get("metrics", "metrics_allow_list", fallback=None), + conf.get("metrics", "metrics_block_list", fallback=None), + ), + stat_name_handler, + conf.getboolean("metrics", "statsd_influxdb_enabled", fallback=False), + ) diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index b973465900ec4..dbb5f7be135f0 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -28,14 +28,15 @@ from opentelemetry.metrics import MeterProvider from opentelemetry.sdk.metrics.view import ExponentialBucketHistogramAggregation, View -from airflow_shared.observability.metrics import _get_backcompat_config, configure_otel 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, + configure_otel, full_name, ) from airflow_shared.observability.metrics.validators import ( @@ -446,8 +447,8 @@ def test_backcompat_endpoint_bridging( endpoint, _, _ = _get_backcompat_config(conf) assert endpoint == expected_endpoint - @mock.patch("airflow_shared.observability.metrics.metrics") - @mock.patch("airflow_shared.observability.metrics.MeterProvider") + @mock.patch("airflow_shared.observability.metrics.otel_logger.metrics") + @mock.patch("airflow_shared.observability.metrics.otel_logger.MeterProvider") def test_configure_otel_uses_exponential_histogram_view(self, mock_provider, mock_metrics): conf = ConfigParser() conf["metrics"] = {"otel_on": "true", "otel_host": "localhost", "otel_port": "4318"} @@ -469,7 +470,7 @@ def test_atexit_flush_on_process_exit(self): """ function_call_str = ( "from configparser import ConfigParser; " - "from airflow_shared.observability.metrics import configure_otel; " + "from airflow_shared.observability.metrics.otel_logger import configure_otel; " "c = ConfigParser(); " "c['metrics'] = {" "'otel_on': 'true', 'otel_debugging_on': 'true', " diff --git a/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py b/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py index d672f000d52ef..4dfe00aa5826d 100644 --- a/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py +++ b/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py @@ -33,7 +33,7 @@ def get_stats_factory() -> Callable: return statsd_logger.get_statsd_logger if conf.getboolean("metrics", "otel_on"): - from airflow.sdk._shared.observability.metrics import configure_otel + from airflow.sdk._shared.observability.metrics.otel_logger import configure_otel return lambda: configure_otel(conf) return NoStatsLogger From d2dd4ab18eb01e503409f1fbfbd35302d964cc7b Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Fri, 12 Jun 2026 15:34:06 +0200 Subject: [PATCH 03/11] addressing PR remarks. --- .../observability/metrics/stats_utils.py | 14 +- .../observability/metrics/otel_logger.py | 128 ++++++++---------- .../observability/metrics/test_otel_logger.py | 49 +++---- .../sdk/observability/metrics/stats_utils.py | 14 +- 4 files changed, 96 insertions(+), 109 deletions(-) diff --git a/airflow-core/src/airflow/observability/metrics/stats_utils.py b/airflow-core/src/airflow/observability/metrics/stats_utils.py index 80d04736da65e..97941fe2e7509 100644 --- a/airflow-core/src/airflow/observability/metrics/stats_utils.py +++ b/airflow-core/src/airflow/observability/metrics/stats_utils.py @@ -35,5 +35,17 @@ def get_stats_factory() -> Callable: if conf.getboolean("metrics", "otel_on"): from airflow._shared.observability.metrics.otel_logger import configure_otel - return lambda: configure_otel(conf) + return lambda: 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), + 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), + 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), + ) return NoStatsLogger 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 4af94d387e3d9..c1fca2c87cbd7 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -47,8 +47,6 @@ ) if TYPE_CHECKING: - from configparser import ConfigParser - from opentelemetry.metrics import Instrument from opentelemetry.sdk.metrics._internal.export import MetricExporter from opentelemetry.util.types import Attributes @@ -440,38 +438,35 @@ def atexit_register_metrics_flush(): def _get_backcompat_config( - conf: ConfigParser, + *, + host: str | None, + port: str | None, + ssl_active: bool, + service: str | None, + interval_ms: str | None, ) -> tuple[str | None, float | None, Resource | None]: - """ - Possibly get deprecated Airflow configs for otel metrics. - - Ideally we return ``(None, None, None)`` here. But if the old configuration - is there, then we will use it. - """ resource = None - if not os.environ.get("OTEL_SERVICE_NAME") and not os.environ.get("OTEL_RESOURCE_ATTRIBUTES"): - service_name = conf.get("metrics", "otel_service", fallback=None) - if service_name: - resource = Resource.create(attributes={SERVICE_NAME: service_name}) + 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 not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") and not os.environ.get( - "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT" + if ( + host + and port + and not os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") + and not os.environ.get("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") ): - 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) - if host and port: - scheme = "https" if ssl_active else "http" - endpoint = f"{scheme}://{_format_url_host(host)}:{port}/v1/metrics" - - interval_ms: float | None = None - if not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL") and conf.has_option( - "metrics", "otel_interval_milliseconds" - ): - interval_ms = conf.getfloat("metrics", "otel_interval_milliseconds") + 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, interval_ms, resource + return endpoint, parsed_interval_ms, resource def _load_exporter_from_env() -> MetricExporter: @@ -495,54 +490,49 @@ def _load_exporter_from_env() -> MetricExporter: return ep.load()() -def configure_otel(conf: ConfigParser) -> SafeOtelLogger | None: - """ - Configure the OpenTelemetry metrics pipeline from Airflow conf. - - Bridges deprecated Airflow-specific options into the standard OTel - environment variables, loads the exporter via the SDK's entry-point - mechanism, and installs the global meter provider. +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, + ) - Returns the user-facing :class:`SafeOtelLogger` so callers (Stats) can - wrap their metrics. Returns ``None`` when ``metrics.otel_on`` is false. - """ - otel_on = conf.getboolean("metrics", "otel_on", fallback=False) - if not otel_on: - return None - - # Ideally all three are None here. - # They would only be something other than None if the user is still using - # the deprecated Airflow-defined otel configs. - backcompat_endpoint, backcompat_interval_ms, resource = _get_backcompat_config(conf) - - # Backcompat: bridge deprecated configs into the OTel env vars so the - # exporter (loaded below via entry points) picks them up automatically. - if backcompat_endpoint and not ( - os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") or os.environ.get("OTEL_EXPORTER_OTLP_METRICS_ENDPOINT") - ): + if backcompat_endpoint: os.environ["OTEL_EXPORTER_OTLP_METRICS_ENDPOINT"] = backcompat_endpoint - - if backcompat_interval_ms is not None and not os.environ.get("OTEL_METRIC_EXPORT_INTERVAL"): + 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") - interval_ms = float(interval_str) if interval_str else None + final_interval_ms = float(interval_str) if interval_str else None - debug = conf.getboolean("metrics", "otel_debugging_on", fallback=False) or ( - os.environ.get("OTEL_METRICS_EXPORTER") == "console" - ) + debug_on = debug or os.environ.get("OTEL_METRICS_EXPORTER") == "console" readers: list[PeriodicExportingMetricReader] = [ PeriodicExportingMetricReader( exporter=_load_exporter_from_env(), # type: ignore[arg-type] - export_interval_millis=interval_ms, # type: ignore[arg-type] + export_interval_millis=final_interval_ms, # type: ignore[arg-type] ) ] - if debug: + if debug_on: readers.append( PeriodicExportingMetricReader( ConsoleMetricExporter(), - export_interval_millis=interval_ms, # type: ignore[arg-type] + export_interval_millis=final_interval_ms, # type: ignore[arg-type] ) ) @@ -579,20 +569,10 @@ def configure_otel(conf: ConfigParser) -> SafeOtelLogger | None: # Register a hook that flushes any in-memory metrics at shutdown. atexit_register_metrics_flush() - # ``getimport`` is an Airflow ``AirflowConfigParser`` extension; plain - # ``ConfigParser`` instances used in tests don't have it. Fall back to - # ``None`` so test fixtures that pass a plain ``ConfigParser`` still work. - stat_name_handler = None - if hasattr(conf, "getimport"): - stat_name_handler = conf.getimport("metrics", "stat_name_handler", fallback=None) - return SafeOtelLogger( metrics.get_meter_provider(), - conf.get("metrics", "otel_prefix", fallback=DEFAULT_METRIC_NAME_PREFIX), - get_validator( - conf.get("metrics", "metrics_allow_list", fallback=None), - conf.get("metrics", "metrics_block_list", fallback=None), - ), + prefix, + get_validator(allow_list, block_list), stat_name_handler, - conf.getboolean("metrics", "statsd_influxdb_enabled", fallback=False), + statsd_influxdb_enabled, ) diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index dbb5f7be135f0..84e5d0e523e29 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -21,7 +21,6 @@ import subprocess import sys import time -from configparser import ConfigParser from unittest import mock import pytest @@ -433,26 +432,22 @@ def test_backcompat_endpoint_bridging( 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 with proper IPv6 bracketing. + deprecated Airflow conf primitives with proper IPv6 bracketing. """ with env_vars(provided_env_vars): - conf = ConfigParser() - section = {"otel_on": "true"} - if airflow_conf_host: - section["otel_host"] = airflow_conf_host - if airflow_conf_port: - section["otel_port"] = airflow_conf_port - conf["metrics"] = section - - endpoint, _, _ = _get_backcompat_config(conf) + endpoint, _, _ = _get_backcompat_config( + host=airflow_conf_host, + port=airflow_conf_port, + ssl_active=False, + service=None, + interval_ms=None, + ) assert endpoint == expected_endpoint @mock.patch("airflow_shared.observability.metrics.otel_logger.metrics") @mock.patch("airflow_shared.observability.metrics.otel_logger.MeterProvider") def test_configure_otel_uses_exponential_histogram_view(self, mock_provider, mock_metrics): - conf = ConfigParser() - conf["metrics"] = {"otel_on": "true", "otel_host": "localhost", "otel_port": "4318"} - configure_otel(conf) + configure_otel(host="localhost", port="4318") call_kwargs = mock_provider.call_args.kwargs views = call_kwargs["views"] @@ -469,14 +464,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 configparser import ConfigParser; " "from airflow_shared.observability.metrics.otel_logger import configure_otel; " - "c = ConfigParser(); " - "c['metrics'] = {" - "'otel_on': 'true', 'otel_debugging_on': 'true', " - "'otel_host': 'localhost', 'otel_port': '4318'" - "}; " - "logger = configure_otel(c); " + "logger = configure_otel(" + "host='localhost', port='4318', debug=True" + "); " "logger.incr('my_test_stat')" ) @@ -526,19 +517,11 @@ def test_reinit_after_fork_exports_metrics(self): ) -def _debug_conf() -> ConfigParser: - c = ConfigParser() - c["metrics"] = { - "otel_on": "true", - "otel_debugging_on": "true", - "otel_host": "localhost", - "otel_port": "4318", - } - return c +_DEBUG_KWARGS = {"host": "localhost", "port": "4318", "debug": True} def mock_service_run(): - logger = configure_otel(_debug_conf()) + logger = configure_otel(**_DEBUG_KWARGS) logger.incr("my_test_stat") @@ -551,7 +534,7 @@ def mock_service_run_reinit(): set_meter_provider() silently fails and the child uses a stale provider. """ # First init — sets Once._done = True - configure_otel(_debug_conf()) + configure_otel(**_DEBUG_KWARGS) # Second init — simulates post-fork re-initialization - logger = configure_otel(_debug_conf()) + logger = configure_otel(**_DEBUG_KWARGS) logger.incr("post_fork_stat") diff --git a/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py b/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py index 4dfe00aa5826d..9cf752bc38942 100644 --- a/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py +++ b/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py @@ -35,5 +35,17 @@ def get_stats_factory() -> Callable: if conf.getboolean("metrics", "otel_on"): from airflow.sdk._shared.observability.metrics.otel_logger import configure_otel - return lambda: configure_otel(conf) + return lambda: 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), + 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), + 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), + ) return NoStatsLogger From 49cd4eaa625432f9ec4d66450c7baed837a534a3 Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Fri, 12 Jun 2026 16:29:13 +0200 Subject: [PATCH 04/11] addressing PR remarks. --- .../observability/metrics/otel_logger.py | 41 +++++++++++++++++++ .../observability/metrics/stats_utils.py | 16 +------- .../sdk/observability/metrics/otel_logger.py | 41 +++++++++++++++++++ .../sdk/observability/metrics/stats_utils.py | 16 +------- 4 files changed, 86 insertions(+), 28 deletions(-) create mode 100644 airflow-core/src/airflow/observability/metrics/otel_logger.py create mode 100644 task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py diff --git a/airflow-core/src/airflow/observability/metrics/otel_logger.py b/airflow-core/src/airflow/observability/metrics/otel_logger.py new file mode 100644 index 0000000000000..09e1d9e473e62 --- /dev/null +++ b/airflow-core/src/airflow/observability/metrics/otel_logger.py @@ -0,0 +1,41 @@ +# 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 + +from typing import TYPE_CHECKING + +from airflow._shared.observability.metrics import otel_logger +from airflow.configuration import conf + +if TYPE_CHECKING: + from airflow._shared.observability.metrics.otel_logger import SafeOtelLogger + + +def get_otel_logger() -> SafeOtelLogger: + 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), + 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), + 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/observability/metrics/stats_utils.py b/airflow-core/src/airflow/observability/metrics/stats_utils.py index 97941fe2e7509..9e0f1263bff66 100644 --- a/airflow-core/src/airflow/observability/metrics/stats_utils.py +++ b/airflow-core/src/airflow/observability/metrics/stats_utils.py @@ -33,19 +33,7 @@ def get_stats_factory() -> Callable: return statsd_logger.get_statsd_logger if conf.getboolean("metrics", "otel_on"): - from airflow._shared.observability.metrics.otel_logger import configure_otel + from airflow.observability.metrics import otel_logger - return lambda: 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), - 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), - 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), - ) + return otel_logger.get_otel_logger return NoStatsLogger diff --git a/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py b/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py new file mode 100644 index 0000000000000..9ba26a854008d --- /dev/null +++ b/task-sdk/src/airflow/sdk/observability/metrics/otel_logger.py @@ -0,0 +1,41 @@ +# 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 + +from typing import TYPE_CHECKING + +from airflow.sdk._shared.observability.metrics import otel_logger +from airflow.sdk.configuration import conf + +if TYPE_CHECKING: + from airflow.sdk._shared.observability.metrics.otel_logger import SafeOtelLogger + + +def get_otel_logger() -> SafeOtelLogger: + 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), + 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), + 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/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py b/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py index 9cf752bc38942..9bd01adcaa033 100644 --- a/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py +++ b/task-sdk/src/airflow/sdk/observability/metrics/stats_utils.py @@ -33,19 +33,7 @@ def get_stats_factory() -> Callable: return statsd_logger.get_statsd_logger if conf.getboolean("metrics", "otel_on"): - from airflow.sdk._shared.observability.metrics.otel_logger import configure_otel + from airflow.sdk.observability.metrics import otel_logger - return lambda: 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), - 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), - 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), - ) + return otel_logger.get_otel_logger return NoStatsLogger From 677e1fad01ca4b1ba1dd2653472bf2cedb7b10ba Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Fri, 12 Jun 2026 18:39:45 +0200 Subject: [PATCH 05/11] addressing PR remarks. --- .../src/airflow_shared/observability/metrics/otel_logger.py | 1 - .../src/airflow_shared/observability/traces/__init__.py | 1 - 2 files changed, 2 deletions(-) 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 c1fca2c87cbd7..efa98af209389 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -476,7 +476,6 @@ def _load_exporter_from_env() -> MetricExporter: 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) """ exporter_name = os.environ.get("OTEL_METRICS_EXPORTER", "otlp") diff --git a/shared/observability/src/airflow_shared/observability/traces/__init__.py b/shared/observability/src/airflow_shared/observability/traces/__init__.py index 8fd4e127b77c8..71fbc23b0d0f6 100644 --- a/shared/observability/src/airflow_shared/observability/traces/__init__.py +++ b/shared/observability/src/airflow_shared/observability/traces/__init__.py @@ -158,7 +158,6 @@ def _load_exporter_from_env() -> SpanExporter: 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) """ exporter_name = os.environ.get("OTEL_TRACES_EXPORTER", "otlp") From 6c7367f0796d979baa0e2d70f501943a190e2b0d Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Mon, 15 Jun 2026 17:27:18 +0200 Subject: [PATCH 06/11] addressing PR remarks. --- .../observability/metrics/otel_logger.py | 32 ++++++++++++++++--- .../observability/metrics/test_otel_logger.py | 8 ++--- 2 files changed, 31 insertions(+), 9 deletions(-) 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 efa98af209389..573b0a32030c6 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -471,14 +471,36 @@ def _get_backcompat_config( def _load_exporter_from_env() -> MetricExporter: """ - Load a metric exporter using the ``OTEL_METRICS_EXPORTER`` env var. + Pick a metric exporter per the OTel SDK environment-variable spec. - Mirrors the entry-point mechanism used by the OTEL SDK auto-instrumentation - configurator. Supported values (from installed packages): - - ``otlp`` (default) -- OTLP/gRPC - - ``console`` -- stdout (useful for debugging) + ``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 """ exporter_name = os.environ.get("OTEL_METRICS_EXPORTER", "otlp") + if exporter_name == "otlp": + protocol = ( + os.environ.get("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL") + or os.environ.get("OTEL_EXPORTER_OTLP_PROTOCOL") + or "http/protobuf" + ) + 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, + ) + + 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: diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index 84e5d0e523e29..b36262c419c0a 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -400,28 +400,28 @@ def test_timer_start_and_stop_manually_send_true(self, mock_time, name): "::1", "4318", "http://[::1]:4318/v1/metrics", - id="airflow_conf_ipv6_loopback_is_bracketed", + id="airflow_config_ipv6_loopback_is_bracketed", ), pytest.param( {}, "2001:db8::1", "4318", "http://[2001:db8::1]:4318/v1/metrics", - id="airflow_conf_ipv6_literal_is_bracketed", + id="airflow_config_ipv6_literal_is_bracketed", ), pytest.param( {}, "[::1]", "4318", "http://[::1]:4318/v1/metrics", - id="airflow_conf_already_bracketed_ipv6_is_preserved", + id="airflow_config_already_bracketed_ipv6_is_preserved", ), pytest.param( {}, "10.0.0.1", "4318", "http://10.0.0.1:4318/v1/metrics", - id="airflow_conf_ipv4_literal_passes_through_unchanged", + id="airflow_config_ipv4_literal_passes_through_unchanged", ), ], ) From 957461bb3476fbb4c84099791dc136917bc1b467 Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Tue, 16 Jun 2026 16:25:40 +0200 Subject: [PATCH 07/11] added more tests. --- .../observability/metrics/test_otel_logger.py | 54 +++++++++++++++++++ 1 file changed, 54 insertions(+) diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index b36262c419c0a..512af6a6a912a 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -35,6 +35,7 @@ _generate_key_name, _get_backcompat_config, _is_up_down_counter, + _load_exporter_from_env, configure_otel, full_name, ) @@ -456,6 +457,59 @@ def test_configure_otel_uses_exponential_histogram_view(self, mock_provider, moc 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_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 + + def test_load_exporter_from_env_raises_on_unsupported_otlp_protocol(self): + with env_vars({"OTEL_EXPORTER_OTLP_PROTOCOL": "bogus"}): + with pytest.raises(ValueError, match="Unsupported OTLP protocol 'bogus'"): + _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_atexit_flush_on_process_exit(self): """ Run a process that initializes a logger, creates a stat and then exits. From 097dfc2971efdd975412ce064b4d9c1dbc04eaf9 Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Tue, 16 Jun 2026 18:07:51 +0200 Subject: [PATCH 08/11] added more tests. refactor and move into a common. --- .../airflow_shared/observability/common.py | 19 ++++++ .../observability/metrics/otel_logger.py | 8 +-- .../observability/traces/__init__.py | 30 ++++++++-- .../tests/observability/test_traces.py | 59 +++++++++++++++++++ 4 files changed, 105 insertions(+), 11 deletions(-) diff --git a/shared/observability/src/airflow_shared/observability/common.py b/shared/observability/src/airflow_shared/observability/common.py index d8873a63d12d8..64b4fb5527a55 100644 --- a/shared/observability/src/airflow_shared/observability/common.py +++ b/shared/observability/src/airflow_shared/observability/common.py @@ -17,6 +17,25 @@ # under the License. from __future__ import annotations +import os + + +def _resolve_otlp_protocol(specific_env_var: str) -> str: + """ + Return the OTLP transport per the OTel SDK environment-variable spec. + + 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. + + 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: """ 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 573b0a32030c6..707a5a6d95e37 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -35,7 +35,7 @@ from opentelemetry.sdk.metrics.view import ExponentialBucketHistogramAggregation, View from opentelemetry.sdk.resources import SERVICE_NAME, Resource -from ..common import _format_url_host +from ..common import _format_url_host, _resolve_otlp_protocol from ..exceptions import InvalidStatsNameException from .protocols import Timer from .validators import ( @@ -485,11 +485,7 @@ def _load_exporter_from_env() -> MetricExporter: """ exporter_name = os.environ.get("OTEL_METRICS_EXPORTER", "otlp") if exporter_name == "otlp": - protocol = ( - os.environ.get("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL") - or os.environ.get("OTEL_EXPORTER_OTLP_PROTOCOL") - or "http/protobuf" - ) + protocol = _resolve_otlp_protocol("OTEL_EXPORTER_OTLP_METRICS_PROTOCOL") if protocol == "http/protobuf": from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter diff --git a/shared/observability/src/airflow_shared/observability/traces/__init__.py b/shared/observability/src/airflow_shared/observability/traces/__init__.py index 71fbc23b0d0f6..ee6dec79030fb 100644 --- a/shared/observability/src/airflow_shared/observability/traces/__init__.py +++ b/shared/observability/src/airflow_shared/observability/traces/__init__.py @@ -31,6 +31,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__) @@ -153,14 +155,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 - - ``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/test_traces.py b/shared/observability/tests/observability/test_traces.py index b21cc3c876173..6d5c2196bb5da 100644 --- a/shared/observability/tests/observability/test_traces.py +++ b/shared/observability/tests/observability/test_traces.py @@ -17,17 +17,21 @@ # under the License. from __future__ import annotations +import pytest from opentelemetry.trace import NonRecordingSpan, SpanContext, TraceFlags, TraceState from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator 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 + class TestBuildTraceStateEntries: def test_with_integer_level(self): @@ -102,3 +106,58 @@ 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_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 + + def test_load_exporter_from_env_raises_on_unsupported_otlp_protocol(self): + with env_vars({"OTEL_EXPORTER_OTLP_PROTOCOL": "bogus"}): + with pytest.raises(ValueError, match="Unsupported OTLP protocol 'bogus'"): + _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() From 237aa96f6564a65051b1e2c4ed4d5159bfe42612 Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Tue, 16 Jun 2026 19:34:05 +0200 Subject: [PATCH 09/11] minor changes. --- .../tests/observability/metrics/test_otel_logger.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index 512af6a6a912a..4d576a6c15282 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -571,11 +571,8 @@ def test_reinit_after_fork_exports_metrics(self): ) -_DEBUG_KWARGS = {"host": "localhost", "port": "4318", "debug": True} - - def mock_service_run(): - logger = configure_otel(**_DEBUG_KWARGS) + logger = configure_otel(host="localhost", port="4318", debug=True) logger.incr("my_test_stat") @@ -588,7 +585,7 @@ def mock_service_run_reinit(): set_meter_provider() silently fails and the child uses a stale provider. """ # First init — sets Once._done = True - configure_otel(**_DEBUG_KWARGS) + configure_otel(host="localhost", port="4318", debug=True) # Second init — simulates post-fork re-initialization - logger = configure_otel(**_DEBUG_KWARGS) + logger = configure_otel(host="localhost", port="4318", debug=True) logger.incr("post_fork_stat") From 1db648398ca8b4d987b29dc00082e565e5adac63 Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Wed, 17 Jun 2026 16:25:38 +0200 Subject: [PATCH 10/11] add more tests. a bit of cleanup. --- providers/common/ai/docs/observability.rst | 9 +- .../ci/docker-compose/integration-otel.yml | 2 +- .../observability/metrics/test_otel_logger.py | 113 ++++++++++++++++-- .../tests/observability/test_traces.py | 113 ++++++++++++++++-- 4 files changed, 219 insertions(+), 18 deletions(-) diff --git a/providers/common/ai/docs/observability.rst b/providers/common/ai/docs/observability.rst index 6669419c8aac7..9de5523a2d052 100644 --- a/providers/common/ai/docs/observability.rst +++ b/providers/common/ai/docs/observability.rst @@ -70,11 +70,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 8b98699fccbfd..fba80da5e3e9c 100644 --- a/scripts/ci/docker-compose/integration-otel.yml +++ b/scripts/ci/docker-compose/integration-otel.yml @@ -74,7 +74,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/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index 4d576a6c15282..2455719ef17d2 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -466,17 +466,23 @@ def test_configure_otel_uses_exponential_histogram_view(self, mock_provider, moc id="default_otlp_no_protocol_uses_http", ), pytest.param( - {"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf"}, + { + "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"}, + { + "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"}, + { + "OTEL_EXPORTER_OTLP_METRICS_PROTOCOL": "grpc", + }, "opentelemetry.exporter.otlp.proto.grpc.metric_exporter", id="metrics_specific_protocol_grpc_uses_grpc", ), @@ -489,7 +495,56 @@ def test_configure_otel_uses_exponential_histogram_view(self, mock_provider, moc id="metrics_specific_protocol_overrides_generic", ), pytest.param( - {"OTEL_METRICS_EXPORTER": "console"}, + { + "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", ), @@ -500,9 +555,45 @@ def test_load_exporter_from_env_selects_correct_exporter(self, provided_env_vars exporter = _load_exporter_from_env() assert exporter.__class__.__module__ == expected_module - def test_load_exporter_from_env_raises_on_unsupported_otlp_protocol(self): - with env_vars({"OTEL_EXPORTER_OTLP_PROTOCOL": "bogus"}): - with pytest.raises(ValueError, match="Unsupported OTLP protocol 'bogus'"): + @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): @@ -510,6 +601,14 @@ def test_load_exporter_from_env_raises_on_unknown_exporter_name(self): 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. diff --git a/shared/observability/tests/observability/test_traces.py b/shared/observability/tests/observability/test_traces.py index 6d5c2196bb5da..2694a3c073d6e 100644 --- a/shared/observability/tests/observability/test_traces.py +++ b/shared/observability/tests/observability/test_traces.py @@ -118,17 +118,23 @@ class TestLoadExporterFromEnv: id="default_otlp_no_protocol_uses_http", ), pytest.param( - {"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf"}, + { + "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"}, + { + "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"}, + { + "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL": "grpc", + }, "opentelemetry.exporter.otlp.proto.grpc.trace_exporter", id="traces_specific_protocol_grpc_uses_grpc", ), @@ -141,7 +147,56 @@ class TestLoadExporterFromEnv: id="traces_specific_protocol_overrides_generic", ), pytest.param( - {"OTEL_TRACES_EXPORTER": "console"}, + { + "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", ), @@ -152,12 +207,56 @@ def test_load_exporter_from_env_selects_correct_exporter(self, provided_env_vars exporter = _load_exporter_from_env() assert exporter.__class__.__module__ == expected_module - def test_load_exporter_from_env_raises_on_unsupported_otlp_protocol(self): - with env_vars({"OTEL_EXPORTER_OTLP_PROTOCOL": "bogus"}): - with pytest.raises(ValueError, match="Unsupported OTLP protocol 'bogus'"): + @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" From cc55a2dcf8ae3cdc36c30e4a7a9ddddaa022437f Mon Sep 17 00:00:00 2001 From: dsuhinin Date: Fri, 24 Jul 2026 17:14:47 +0200 Subject: [PATCH 11/11] move stats.initialize under settings.initialize --- airflow-core/src/airflow/api_fastapi/app.py | 24 ---------- .../src/airflow/dag_processing/manager.py | 8 +--- .../src/airflow/executors/base_executor.py | 5 -- .../src/airflow/jobs/scheduler_job_runner.py | 8 ---- .../src/airflow/jobs/triggerer_job_runner.py | 5 -- airflow-core/src/airflow/settings.py | 20 ++++++++ .../tests/unit/api_fastapi/test_app.py | 44 ----------------- airflow-core/tests/unit/core/test_settings.py | 47 +++++++++++++++++++ .../tests/unit/dag_processing/test_manager.py | 14 +----- .../tests/unit/jobs/test_triggerer_job.py | 37 +-------------- 10 files changed, 70 insertions(+), 142 deletions(-) diff --git a/airflow-core/src/airflow/api_fastapi/app.py b/airflow-core/src/airflow/api_fastapi/app.py index a4dbc3c9b722e..8931840c8807c 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 3f60cb3f4c4de..e9708732013c2 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 ebd850ad944b1..44e56aa5a3831 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 cb4c8f645520c..45ab48872e309 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 759735242ab1d..07053317ae33d 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/settings.py b/airflow-core/src/airflow/settings.py index 54150dd8bfd5b..ea15442558f5e 100644 --- a/airflow-core/src/airflow/settings.py +++ b/airflow-core/src/airflow/settings.py @@ -804,6 +804,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() @@ -815,6 +834,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 74da895bca30a..9717666e65103 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 fa4d68f5fff6a..5a24b415902c0 100644 --- a/airflow-core/tests/unit/jobs/test_triggerer_job.py +++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py @@ -2482,38 +2482,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.""" @@ -2534,10 +2502,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"