This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new f558ff13273 Stop discarding a MeterProvider configured via
OTEL_CONFIG_FILE (#71529)
f558ff13273 is described below
commit f558ff13273c6fd6a81ca592a70b8c3761b748cd
Author: Stefan Wang <[email protected]>
AuthorDate: Tue Aug 18 12:12:44 2026 -0700
Stop discarding a MeterProvider configured via OTEL_CONFIG_FILE (#71529)
The OpenTelemetry declarative configuration spec makes OTEL_CONFIG_FILE the
sole source of SDK construction. Airflow force-resets the SDK's once-guard
and
installs its own MeterProvider unconditionally, so a deployment running
under
opentelemetry-instrument with a config file loses every view, reader and
exporter it declared, with no warning.
The once-guard reset exists for the post-fork case in #64690 and stays in
place
for every path where Airflow owns the provider.
Signed-off-by: 1fanwang <[email protected]>
Co-authored-by: Copilot App <[email protected]>
---
.../observability/metrics/otel_logger.py | 23 +++++++--
.../observability/metrics/test_otel_logger.py | 59 +++++++++++++++++++++-
2 files changed, 78 insertions(+), 4 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 9be7103a62b..bc8b819f144 100644
---
a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py
+++
b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py
@@ -19,6 +19,7 @@ from __future__ import annotations
import atexit
import datetime
import logging
+import os
import random
import warnings
from collections.abc import Callable
@@ -76,6 +77,11 @@ DEFAULT_METRIC_NAME_PREFIX = "airflow"
# Delimiter is placed between the universal metric prefix and the unique
metric name.
DEFAULT_METRIC_NAME_DELIMITER = "."
+# Not imported from ``opentelemetry.sdk.environment_variables``: the constant
postdates the
+# ``opentelemetry-api>=1.27.0`` floor this distribution declares, while the
name is spec-fixed.
+#
https://opentelemetry.io/docs/specs/otel/configuration/sdk/#declarative-configuration
+OTEL_CONFIG_FILE = "OTEL_CONFIG_FILE"
+
def full_name(name: str, *, prefix: str = DEFAULT_METRIC_NAME_PREFIX) -> str:
"""Assembles the prefix, delimiter, and name and returns it as a string."""
@@ -457,11 +463,24 @@ def get_otel_logger(
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).
+
+ A ``MeterProvider`` already built from ``OTEL_CONFIG_FILE`` is used as-is:
the declarative
+ configuration spec makes that file the sole source of SDK construction.
"""
+ effective_prefix: str = prefix or DEFAULT_METRIC_NAME_PREFIX
+ validator = get_validator(metrics_allow_list, metrics_block_list)
+
+ configured_provider = metrics.get_meter_provider()
+ if os.environ.get(OTEL_CONFIG_FILE) and isinstance(configured_provider,
MeterProvider):
+ log.info("%s is set; using the MeterProvider it built.",
OTEL_CONFIG_FILE)
+ atexit_register_metrics_flush()
+ return SafeOtelLogger(
+ configured_provider, effective_prefix, validator,
stat_name_handler, statsd_influxdb_enabled
+ )
+
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
@@ -524,8 +543,6 @@ def get_otel_logger(
# Register a hook that flushes any in-memory metrics at shutdown.
atexit_register_metrics_flush()
- validator = get_validator(metrics_allow_list, metrics_block_list)
-
return SafeOtelLogger(
metrics.get_meter_provider(), effective_prefix, validator,
stat_name_handler, statsd_influxdb_enabled
)
diff --git
a/shared/observability/tests/observability/metrics/test_otel_logger.py
b/shared/observability/tests/observability/metrics/test_otel_logger.py
index f1e90e53003..cc821771372 100644
--- a/shared/observability/tests/observability/metrics/test_otel_logger.py
+++ b/shared/observability/tests/observability/metrics/test_otel_logger.py
@@ -24,8 +24,14 @@ import time
from unittest import mock
import pytest
+from opentelemetry import metrics
from opentelemetry.metrics import MeterProvider
-from opentelemetry.sdk.metrics.view import
ExponentialBucketHistogramAggregation, View
+from opentelemetry.sdk.metrics import MeterProvider as SDKMeterProvider
+from opentelemetry.sdk.metrics.view import (
+ ExplicitBucketHistogramAggregation,
+ ExponentialBucketHistogramAggregation,
+ View,
+)
from airflow_shared.observability.common import get_otel_data_exporter
from airflow_shared.observability.metrics.otel_logger import (
@@ -61,6 +67,26 @@ def name():
return "test_stats_run"
[email protected]
+def reset_meter_provider():
+ """Let a test install its own global MeterProvider, then restore the
previous one.
+
+ ``set_meter_provider`` is guarded by a process-wide ``Once``, so tests
that install a
+ provider have to clear it the same way ``get_otel_logger`` does after a
fork.
+ """
+ import opentelemetry.metrics._internal as metrics_internal
+
+ def clear() -> None:
+ metrics_internal._METER_PROVIDER_SET_ONCE._done = False
+ metrics_internal._METER_PROVIDER = None
+
+ previous = metrics_internal._METER_PROVIDER
+ clear()
+ yield
+ clear()
+ metrics_internal._METER_PROVIDER = previous
+
+
class TestOtelMetrics:
def setup_method(self):
self.meter = mock.Mock(MeterProvider)
@@ -502,6 +528,37 @@ class TestOtelMetrics:
assert isinstance(view, View)
assert isinstance(view._aggregation,
ExponentialBucketHistogramAggregation)
+ def test_declaratively_configured_provider_is_not_replaced(self,
reset_meter_provider):
+ """A MeterProvider built from OTEL_CONFIG_FILE must survive
get_otel_logger().
+
+ The declarative configuration spec makes that file the sole source of
SDK
+ construction, so replacing its provider silently drops the
deployment's views.
+ See https://github.com/apache/airflow/issues/64690 for why the
provider is
+ otherwise force-replaced.
+ """
+ declarative_view = View(
+ instrument_name="*_duration",
+ aggregation=ExplicitBucketHistogramAggregation(boundaries=(0.5, 1,
2, 4, 8)),
+ )
+ configured_provider = SDKMeterProvider(views=[declarative_view],
shutdown_on_exit=False)
+ metrics.set_meter_provider(configured_provider)
+
+ with env_vars({"OTEL_CONFIG_FILE": "/tmp/otel-config.yaml"}):
+ logger = get_otel_logger(host="localhost", port=4318)
+
+ assert logger.otel is configured_provider
+ assert metrics.get_meter_provider() is configured_provider
+ assert list(configured_provider._sdk_config.views) ==
[declarative_view]
+
+ def test_provider_is_replaced_without_declarative_config(self,
reset_meter_provider):
+ """Without OTEL_CONFIG_FILE, Airflow still installs its own
provider."""
+ pre_existing = SDKMeterProvider(shutdown_on_exit=False)
+ metrics.set_meter_provider(pre_existing)
+
+ logger = get_otel_logger(host="localhost", port=4318)
+
+ assert logger.otel is not pre_existing
+
def test_atexit_flush_on_process_exit(self):
"""
Run a process that initializes a logger, creates a stat and then exits.
