From 8fcfc6460233f62f420c5cf45da717c724900323 Mon Sep 17 00:00:00 2001 From: Jed Cunningham Date: Tue, 18 Aug 2026 15:46:18 -0600 Subject: [PATCH 1/4] Stop forked children re-exporting the parent's OTel metrics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A forked child inherited the atexit flush hook, the MeterProvider, and — because the SDK restarts its reader threads in the child — a live pipeline over the parent's accumulated state, so it re-exported metrics it never recorded. Building the pipeline once per process rather than once per call closes the same duplication on the re-initialization path. --- .../observability/metrics/otel_logger.py | 123 +++++++-- .../observability/metrics/test_otel_logger.py | 253 ++++++++++++++---- 2 files changed, 304 insertions(+), 72 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 bc8b819f144f2..7c72c9af95c65 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -432,13 +432,77 @@ 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 _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. Only this + provider's own readers are touched — ``_all_metric_readers`` is a class attribute shared with + every provider in the process, including ones Airflow did not build. + """ + readers = getattr(provider, "_metric_readers", None) + 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 +529,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 +546,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 +592,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 +604,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..c720913c67c0c 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -20,6 +20,7 @@ import os import subprocess import sys +import threading import time from unittest import mock @@ -27,6 +28,7 @@ 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 +36,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, @@ -52,6 +55,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 +77,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 +592,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,33 +600,115 @@ 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}" + def test_forked_child_stops_the_pipeline_it_inherited(self, tmp_path): + """A child must not keep exporting the metrics its parent accumulated. - 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}" + ``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. + """ + child_output = tmp_path / "child-metrics.json" + run_service_helper("mock_service_fork_idle_child", child_output) + + 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}" + ) + + 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(): @@ -623,16 +716,80 @@ def mock_service_run(): 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() + + +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 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. + 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_idle_child_capturing_its_own_output(): + """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) + 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_idle_child_capturing_its_own_output() + + +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_idle_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() From 16cada64accc67547fa9ba8ac55043d2bbfef4fb Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Wed, 19 Aug 2026 13:47:14 -0700 Subject: [PATCH 2/4] Cover the forked child that builds its own pipeline The idle child shows the inherited pipeline goes quiet, but every worker that emits metrics reaches a different state: it stops the pipeline it inherited and then exports through one of its own, on the same interval, for the rest of its life. Nothing pinned that the second half still works -- dropping the reset of the module's provider reference leaves the child recording into the pipeline it just silenced, and its own metrics disappear with no test noticing. --- .../observability/metrics/test_otel_logger.py | 38 ++++++++++++++++--- 1 file changed, 32 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 c720913c67c0c..c754cd56912bc 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -622,21 +622,35 @@ def test_forked_child_does_not_duplicate_the_shutdown_flush(self, runner, metric f"exported {exports} times.\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}" ) - def test_forked_child_stops_the_pipeline_it_inherited(self, tmp_path): + @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. + 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("mock_service_fork_idle_child", child_output) + run_service_helper(runner, child_output) 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}" + ) 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. @@ -744,12 +758,14 @@ def mock_service_fork_child_with_reinit(): os.wait() -def _fork_idle_child_capturing_its_own_output(): +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() @@ -758,7 +774,17 @@ def _fork_idle_child_capturing_its_own_output(): 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_idle_child_capturing_its_own_output() + _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(): @@ -773,7 +799,7 @@ def mock_service_fork_idle_child_beside_foreign_provider(): ) 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_idle_child_capturing_its_own_output() + _fork_child_capturing_its_own_output() def mock_service_fork_child_under_foreign_provider(): From 6e45951fffc2d3c6a7c7d6e341be9f4e1a89c031 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Wed, 19 Aug 2026 13:47:24 -0700 Subject: [PATCH 3/4] Add newsfragment for the duplicate OTel metric stream fix --- airflow-core/newsfragments/71855.bugfix.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 airflow-core/newsfragments/71855.bugfix.rst 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. From b35eb95a3c01419cd0b68bbbdb943db717dd71b6 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Thu, 20 Aug 2026 10:02:32 -0700 Subject: [PATCH 4/4] Find the inherited readers on older opentelemetry-sdk versions too ``_metric_readers`` is only an attribute of ``MeterProvider`` from opentelemetry-sdk 1.44; before that the same per-provider list lives on ``_sdk_config.metric_readers``. Airflow asks for ``opentelemetry-api>=1.27.0`` with no upper bound, so a supported install can easily be on a version where reaching only for the newer attribute finds nothing and the child goes on exporting its parent's totals. That failure is silent apart from one warning per fork, which is what makes it worth guarding: nothing else about a pipeline nobody stopped looks wrong. Found by building this onto Astro Runtime 3.3-2, which ships opentelemetry-sdk 1.42.1: the child kept re-exporting the parent's metrics with the fix in place. --- .../observability/metrics/otel_logger.py | 33 ++++++++++++++++--- .../observability/metrics/test_otel_logger.py | 28 ++++++++++++++++ 2 files changed, 56 insertions(+), 5 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 7c72c9af95c65..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 @@ -444,17 +446,38 @@ def flush_otel_metrics(): _provider.force_flush() +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. Only this - provider's own readers are touched — ``_all_metric_readers`` is a class attribute shared with - every provider in the process, including ones Airflow did not build. + itself: its collect callback has to stop producing the parent's measurements too. """ - readers = getattr(provider, "_metric_readers", None) + readers = _find_own_readers(provider) if readers is None: log.warning("Could not find the inherited metric readers; they may keep exporting.") return diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index c754cd56912bc..2b2390fb0346e 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -22,6 +22,7 @@ import sys import threading import time +from types import SimpleNamespace from unittest import mock import pytest @@ -44,6 +45,7 @@ SafeOtelLogger, _generate_key_name, _is_up_down_counter, + _stop_inherited_pipeline, full_name, get_otel_logger, ) @@ -652,6 +654,32 @@ def test_forked_child_stops_the_pipeline_it_inherited(self, tmp_path, runner, ow 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.