diff --git a/airflow-core/newsfragments/71855.bugfix.rst b/airflow-core/newsfragments/71855.bugfix.rst new file mode 100644 index 0000000000000..f7b259e0bd87e --- /dev/null +++ b/airflow-core/newsfragments/71855.bugfix.rst @@ -0,0 +1 @@ +Fixed OTel metrics being exported twice, as two conflicting cumulative streams: a re-initialised pipeline no longer runs alongside the live one, and a forked child no longer exports the pipeline it inherited from its parent. 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 bc8b819f144f2..f64695092f55a 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -23,7 +23,7 @@ import random import warnings from collections.abc import Callable -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from opentelemetry import metrics from opentelemetry.sdk.metrics import MeterProvider @@ -47,6 +47,8 @@ ) if TYPE_CHECKING: + from collections.abc import Sequence + from opentelemetry.metrics import Instrument from opentelemetry.util.types import Attributes @@ -432,13 +434,98 @@ def record_histogram_value(self, name: str, value: float, tags: Attributes) -> N self.histograms[name].record(value, tags) +# The MeterProvider this module built, if any. A fork hands the child this reference, the atexit +# hook, and a live pipeline: PeriodicExportingMetricReader registers its own after_in_child hook +# that restarts the export thread, so the inherited readers resume exporting everything the parent +# accumulated. The child owns none of it. +_provider: MeterProvider | None = None + + def flush_otel_metrics(): - provider = metrics.get_meter_provider() - provider.force_flush() + if _provider is not None: + _provider.force_flush() -def atexit_register_metrics_flush(): - atexit.register(flush_otel_metrics) +def _find_own_readers(provider: MeterProvider) -> Sequence[Any] | None: + """ + Return the readers belonging to *provider*, across opentelemetry-sdk versions. + + ``_metric_readers`` exists from opentelemetry-sdk 1.44; before that the same list lives on + ``_sdk_config.metric_readers``. Airflow accepts ``opentelemetry-api>=1.27.0`` with no upper + bound, so reading only the newer attribute leaves the fix inert -- and silently so -- on a + supported install. + + Both attributes are per-provider. ``_all_metric_readers`` is deliberately not among them: it + is a class attribute shared with every provider in the process, including ones Airflow did not + build. + + Readers come back untyped because what the caller does with them -- replacing the collect + callback, setting the shutdown event -- reaches for private attributes the SDK does not + declare. + """ + readers = getattr(provider, "_metric_readers", None) + if readers is not None: + return readers + return getattr(getattr(provider, "_sdk_config", None), "metric_readers", None) + + +def _stop_inherited_pipeline(provider: MeterProvider) -> None: + """ + Stop the readers a fork handed us without letting them export the parent's state. + + The SDK's own ``after_in_child`` hook revives each reader's export thread, and the revived + ticker does one final collect on its way out, so telling a reader to stop is not enough by + itself: its collect callback has to stop producing the parent's measurements too. + """ + readers = _find_own_readers(provider) + if readers is None: + log.warning("Could not find the inherited metric readers; they may keep exporting.") + return + + for reader in readers: + try: + reader._collect = lambda *args, **kwargs: None + reader._shutdown_event.set() + except (AttributeError, TypeError): + log.warning( + "Could not stop a metric reader inherited across fork; it may keep exporting " + "metrics recorded before the fork.", + exc_info=True, + ) + + +def _drop_inherited_exit_handler(provider: object) -> None: + """ + Stop a provider this module did not build from flushing the parent's state at child exit. + + A ``MeterProvider`` built by the SDK — from ``OTEL_CONFIG_FILE`` or by an instrumentation + agent — defaults to ``shutdown_on_exit=True`` and registers its own atexit shutdown, which a + fork hands to the child along with everything the parent had accumulated. + """ + handler = getattr(provider, "_atexit_handler", None) + if handler is None: + return + try: + atexit.unregister(handler) + provider._atexit_handler = None # type: ignore[attr-defined] + except (AttributeError, TypeError): + log.warning( + "Could not drop the inherited shutdown hook of a MeterProvider this process did not " + "build; it may export metrics recorded before the fork.", + exc_info=True, + ) + + +def _reset_provider_after_fork() -> None: + global _provider + atexit.unregister(flush_otel_metrics) + _drop_inherited_exit_handler(metrics.get_meter_provider()) + if _provider is not None: + _stop_inherited_pipeline(_provider) + _provider = None + + +os.register_at_fork(after_in_child=_reset_provider_after_fork) def get_otel_logger( @@ -465,7 +552,16 @@ def get_otel_logger( 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. + configuration spec makes that file the sole source of SDK construction. Its lifecycle stays + the deployment's — no flush hook is registered for it here, since the SDK builds it with its + own atexit shutdown, and ``shutdown_on_exit=False`` is a deliberate opt-out to respect. Across + a fork the child only drops the inherited copy of that hook, so it cannot export what the + parent accumulated; the readers the SDK revives in the child are left running, because on this + path they are the only pipeline the child has. + + This module builds at most one pipeline: later calls return a logger over the provider it + already built rather than leaving a second one exporting alongside it. A forked child stops + the pipeline it inherited and builds its own. """ effective_prefix: str = prefix or DEFAULT_METRIC_NAME_PREFIX validator = get_validator(metrics_allow_list, metrics_block_list) @@ -473,11 +569,16 @@ def get_otel_logger( 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 ) + global _provider + if _provider is not None: + return SafeOtelLogger( + _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" @@ -514,8 +615,8 @@ def get_otel_logger( # 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. + # MeterProvider is not allowed", leaving the parent's provider installed globally for + # anything that reads it — carrying the parent's accumulated state, not the child's. # On first call (no fork), _done is already False so this is a no-op. # See: https://github.com/apache/airflow/issues/64690 try: @@ -526,23 +627,20 @@ def get_otel_logger( 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, - ), + _provider = MeterProvider( + resource=resource, + metric_readers=readers, + views=[ + View( + instrument_type=metrics.Histogram, + aggregation=ExponentialBucketHistogramAggregation(), + ) + ], + shutdown_on_exit=False, ) + metrics.set_meter_provider(_provider) # Register a hook that flushes any in-memory metrics at shutdown. - atexit_register_metrics_flush() + atexit.register(flush_otel_metrics) - return SafeOtelLogger( - metrics.get_meter_provider(), effective_prefix, validator, stat_name_handler, statsd_influxdb_enabled - ) + return SafeOtelLogger(_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 cc821771372e2..2b2390fb0346e 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -20,13 +20,16 @@ import os import subprocess import sys +import threading import time +from types import SimpleNamespace from unittest import mock import pytest from opentelemetry import metrics from opentelemetry.metrics import MeterProvider from opentelemetry.sdk.metrics import MeterProvider as SDKMeterProvider +from opentelemetry.sdk.metrics.export import ConsoleMetricExporter, PeriodicExportingMetricReader from opentelemetry.sdk.metrics.view import ( ExplicitBucketHistogramAggregation, ExponentialBucketHistogramAggregation, @@ -34,6 +37,7 @@ ) from airflow_shared.observability.common import get_otel_data_exporter +from airflow_shared.observability.metrics import otel_logger as otel_logger_module from airflow_shared.observability.metrics.otel_logger import ( OTEL_NAME_MAX_LENGTH, UP_DOWN_COUNTERS, @@ -41,6 +45,7 @@ SafeOtelLogger, _generate_key_name, _is_up_down_counter, + _stop_inherited_pipeline, full_name, get_otel_logger, ) @@ -52,6 +57,13 @@ from tests_common.test_utils.config import env_vars +# Long enough that only the shutdown flush exports, so an exported metric name appears in the +# output once per flush. +NO_PERIODIC_EXPORT_INTERVAL_MS = 600000 + +# Short enough that an inherited reader fires several times while the child is alive. +CHILD_EXPORT_INTERVAL_MS = 500 + INVALID_STAT_NAME_CASES = [ (None, "can not be None"), (42, "is not a string"), @@ -67,6 +79,22 @@ def name(): return "test_stats_run" +@pytest.fixture(autouse=True) +def reset_process_provider(): + """Clear the per-process provider cache so tests never inherit one another's pipeline.""" + + def discard() -> None: + # Stop the pipeline rather than just dropping the reference: its exporter threads would + # otherwise stay alive for the rest of the session. + if otel_logger_module._provider is not None: + otel_logger_module._stop_inherited_pipeline(otel_logger_module._provider) + otel_logger_module._provider = None + + discard() + yield + discard() + + @pytest.fixture def reset_meter_provider(): """Let a test install its own global MeterProvider, then restore the previous one. @@ -566,22 +594,7 @@ def test_atexit_flush_on_process_exit(self): The logger initialization registers an atexit hook. 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); " - "logger.incr('my_test_stat')" - ) - - proc = subprocess.run( - [sys.executable, "-c", function_call_str], - check=False, - env=os.environ.copy(), - capture_output=True, - text=True, - timeout=20, - ) - - assert proc.returncode == 0, f"Process failed\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}" + proc = run_service_helper("mock_service_run") assert "my_test_stat" in proc.stdout, ( "Expected the metric name to be present in the stdout but it wasn't.\n" @@ -589,50 +602,248 @@ def test_atexit_flush_on_process_exit(self): f"stderr:\n{proc.stderr}" ) - def test_reinit_after_fork_exports_metrics(self): - """Calling get_otel_logger() twice (simulating post-fork re-init) should still export metrics. + @pytest.mark.parametrize( + ("runner", "metric_name"), + [ + ("mock_service_fork_child_without_reinit", "parent_stat"), + ("mock_service_fork_child_with_reinit", "child_stat"), + ], + ) + def test_forked_child_does_not_duplicate_the_shutdown_flush(self, runner, metric_name): + """A fork must not turn one shutdown flush into two. - 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 - fresh MeterProvider. The fix resets the guard before each set_meter_provider() call. + The child inherits the parent's atexit hook along with its MeterProvider, so an + inherited hook either re-exports everything the parent accumulated (child that never + initializes its own logger) or runs twice over the child's own provider (child that does). """ - test_module_name = "tests.observability.metrics.test_otel_logger" - function_call_str = f"import {test_module_name} as m; m.mock_service_run_reinit()" - - proc = subprocess.run( - [sys.executable, "-c", function_call_str], - check=False, - env=os.environ.copy(), - capture_output=True, - text=True, - timeout=20, + proc = run_service_helper(runner) + + exports = proc.stdout.count(f'"name": "airflow.{metric_name}"') + assert exports == 1, ( + f"Expected 'airflow.{metric_name}' to be exported exactly once but it was " + f"exported {exports} times.\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}" ) - assert proc.returncode == 0, f"Process failed\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}" + @pytest.mark.parametrize( + ("runner", "own_metric"), + [ + ("mock_service_fork_idle_child", None), + ("mock_service_fork_reinitializing_child", "child_stat"), + ], + ) + def test_forked_child_stops_the_pipeline_it_inherited(self, tmp_path, runner, own_metric): + """A child must not keep exporting the metrics its parent accumulated. + + ``PeriodicExportingMetricReader`` registers its own ``after_in_child`` hook that restarts + the export thread, so without stopping the inherited provider the child republishes the + parent's cumulative totals on every interval for as long as it lives. A child that builds + its own pipeline — the case every worker that emits metrics reaches — must go on exporting + through that one on the same interval. + """ + child_output = tmp_path / "child-metrics.json" + run_service_helper(runner, child_output) - assert "post_fork_stat" in proc.stdout, ( - "Expected 'post_fork_stat' in stdout after re-initialization but it wasn't found. " - "This suggests set_meter_provider() failed due to the Once() guard.\n" - f"stdout:\n{proc.stdout}\n" - f"stderr:\n{proc.stderr}" + exported_in_child = child_output.read_text() + assert '"name": "airflow.parent_stat"' not in exported_in_child, ( + "The forked child kept exporting the parent's metrics from the inherited pipeline.\n" + f"child output:\n{exported_in_child}" + ) + if own_metric: + assert f'"name": "airflow.{own_metric}"' in exported_in_child, ( + "The forked child stopped exporting through the pipeline it built for itself.\n" + f"child output:\n{exported_in_child}" + ) + + @pytest.mark.parametrize( + "readers_attribute", + ["_metric_readers", "_sdk_config.metric_readers"], + ) + def test_inherited_readers_are_found_across_sdk_versions(self, readers_attribute): + """Test that a provider's own readers are found wherever the installed SDK keeps them. + + ``_metric_readers`` exists from opentelemetry-sdk 1.44; before that the same list lives on + ``_sdk_config.metric_readers``. Airflow accepts ``opentelemetry-api>=1.27.0`` with no upper + bound, so reading only the newer attribute leaves this inert on a supported install -- and + inert silently, since a pipeline nobody stops just goes on exporting. + """ + reader = SimpleNamespace( + _collect=lambda *args, **kwargs: "the parent's measurements", + _shutdown_event=threading.Event(), + ) + if readers_attribute == "_metric_readers": + provider = SimpleNamespace(_metric_readers=[reader]) + else: + provider = SimpleNamespace(_sdk_config=SimpleNamespace(metric_readers=[reader])) + + _stop_inherited_pipeline(provider) + + assert reader._shutdown_event.is_set() + assert reader._collect(reader) is None + + def test_forked_child_does_not_flush_a_provider_it_did_not_build(self, tmp_path): + """A provider supplied by the SDK brings its own atexit shutdown; a child must not run it. + + ``shutdown_on_exit=True`` is the SDK default for a ``MeterProvider`` built from + ``OTEL_CONFIG_FILE`` or by an instrumentation agent, and a fork hands that hook to the + child together with everything the parent accumulated. + """ + child_output = tmp_path / "child-metrics.json" + run_service_helper("mock_service_fork_child_under_foreign_provider", child_output) + + assert "foreign_counter" not in child_output.read_text(), ( + "The child ran the inherited provider's shutdown hook and exported the parent's state." ) + def test_forked_child_leaves_other_pipelines_alone(self, tmp_path): + """Stopping the inherited pipeline must not touch a provider Airflow did not build. + + ``MeterProvider._all_metric_readers`` is a class attribute shared by every provider in the + process, so reaching for that instead of this provider's own readers would silence an + ``opentelemetry-instrument`` pipeline in every forked child. + """ + child_output = tmp_path / "child-metrics.json" + run_service_helper("mock_service_fork_idle_child_beside_foreign_provider", child_output) + + exported_in_child = child_output.read_text() + assert "foreign_counter" in exported_in_child, ( + f"The child stopped a pipeline Airflow does not own.\nchild output:\n{exported_in_child}" + ) + assert '"name": "airflow.parent_stat"' not in exported_in_child + + def test_reinit_reuses_the_process_pipeline(self, reset_meter_provider): + """A second call must not leave a second pipeline exporting alongside the first. + + Every ``MeterProvider`` owns a ``PeriodicExportingMetricReader`` whose thread exports on + its own interval, and ``shutdown_on_exit=False`` means nothing reaps one that gets + replaced: its instruments stop being recorded to, so it republishes frozen cumulative + totals alongside the live stream for the same series. + """ + first = get_otel_logger(host="localhost", port=4318, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS) + readers_after_first = count_live_readers() + + second = get_otel_logger(host="localhost", port=4318, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS) + + assert second.otel is first.otel + assert count_live_readers() == readers_after_first + + +TEST_MODULE = "tests.observability.metrics.test_otel_logger" + + +def run_service_helper(runner: str, child_out=None) -> subprocess.CompletedProcess: + """Run one of this module's ``mock_service_*`` helpers in a fresh interpreter.""" + env = os.environ.copy() + if child_out is not None: + env["CHILD_METRICS_OUT"] = str(child_out) + + proc = subprocess.run( + [sys.executable, "-c", f"import {TEST_MODULE} as m; m.{runner}()"], + check=False, + env=env, + capture_output=True, + text=True, + timeout=30, + ) + assert proc.returncode == 0, f"{runner} failed\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}" + return proc + + +def count_live_readers() -> int: + """Number of running periodic exporter threads (a pipeline has one per configured reader).""" + return sum(1 for t in threading.enumerate() if t.name == "OtelPeriodicExportingMetricReader") + def mock_service_run(): logger = get_otel_logger(debug=True) logger.incr("my_test_stat") -def mock_service_run_reinit(): - """Simulate re-initialization after fork by calling get_otel_logger() twice. +def mock_service_fork_child_without_reinit(): + """Emit a metric, then fork a child that exits without initializing its own logger.""" + get_otel_logger(debug=True, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS).incr("parent_stat") + if os.fork() == 0: + sys.exit(0) + os.wait() + - The first call sets the global MeterProvider and the Once() guard. - The second call simulates what happens in a forked child: stats.py detects - a PID mismatch and calls the factory again. Without the fix, the second - set_meter_provider() silently fails and the child uses a stale provider. +def mock_service_fork_child_with_reinit(): + """Emit a metric, then fork a child that initializes its own logger and emits its own metric. + + The child also checks that it installed its own provider globally: the parent's + ``_METER_PROVIDER_SET_ONCE`` guard is inherited as already-done, so without the reset in + ``get_otel_logger()`` the child's ``set_meter_provider()`` silently leaves the parent's + provider in place for anything that reads the global one. See + https://github.com/apache/airflow/issues/64690. """ - # First init — sets Once._done = True - get_otel_logger(debug=True) - # Second init — simulates post-fork re-initialization - logger = get_otel_logger(debug=True) - logger.incr("post_fork_stat") + get_otel_logger(debug=True, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS).incr("parent_stat") + if os.fork() == 0: + child = get_otel_logger(debug=True, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS) + child.incr("child_stat") + if metrics.get_meter_provider() is not child.otel: + print("child did not install its own MeterProvider globally", file=sys.stderr) + sys.exit(1) + sys.exit(0) + os.wait() + + +def _fork_child_capturing_its_own_output(on_start=None): + """Fork a child that redirects its output to ``CHILD_METRICS_OUT`` and idles, then reap it.""" + if os.fork() == 0: + fd = os.open(os.environ["CHILD_METRICS_OUT"], os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + os.dup2(fd, 1) + os.dup2(fd, 2) + if on_start is not None: + on_start() + time.sleep(CHILD_EXPORT_INTERVAL_MS / 1000 * 5) + os._exit(0) + os.wait() + + +def mock_service_fork_idle_child(): + """Emit a metric, then fork a child that only idles while the export interval elapses.""" + get_otel_logger(debug=True, conf_interval=CHILD_EXPORT_INTERVAL_MS).incr("parent_stat") + _fork_child_capturing_its_own_output() + + +def mock_service_fork_reinitializing_child(): + """Emit a metric, then fork a child that builds its own pipeline and idles alongside it.""" + get_otel_logger(debug=True, conf_interval=CHILD_EXPORT_INTERVAL_MS).incr("parent_stat") + _fork_child_capturing_its_own_output( + on_start=lambda: get_otel_logger(debug=True, conf_interval=CHILD_EXPORT_INTERVAL_MS).incr( + "child_stat" + ) + ) + + +def mock_service_fork_idle_child_beside_foreign_provider(): + """Emit a metric with a provider Airflow does not own also running, then fork an idle child.""" + foreign = SDKMeterProvider( + metric_readers=[ + PeriodicExportingMetricReader( + ConsoleMetricExporter(), export_interval_millis=CHILD_EXPORT_INTERVAL_MS + ) + ], + shutdown_on_exit=False, + ) + foreign.get_meter("foreign").create_counter("foreign_counter").add(7) + get_otel_logger(debug=True, conf_interval=CHILD_EXPORT_INTERVAL_MS).incr("parent_stat") + _fork_child_capturing_its_own_output() + + +def mock_service_fork_child_under_foreign_provider(): + """Record into an SDK-built provider that keeps its own atexit shutdown, then fork and exit.""" + foreign = SDKMeterProvider( + metric_readers=[ + PeriodicExportingMetricReader( + ConsoleMetricExporter(), export_interval_millis=NO_PERIODIC_EXPORT_INTERVAL_MS + ) + ], + ) + metrics.set_meter_provider(foreign) + foreign.get_meter("foreign").create_counter("foreign_counter").add(7) + if os.fork() == 0: + fd = os.open(os.environ["CHILD_METRICS_OUT"], os.O_WRONLY | os.O_CREAT | os.O_TRUNC) + os.dup2(fd, 1) + os.dup2(fd, 2) + sys.exit(0) # normal exit: atexit hooks run + os.wait()