From bfaf91a299dcd19039474c31667c03c62ace2c0f Mon Sep 17 00:00:00 2001
From: Jed Cunningham <jedcunningham@apache.org>
Date: Tue, 18 Aug 2026 15:46:18 -0600
Subject: [PATCH 1/6] 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 4d9289e3ed163af290a2146620fd14b69121eef9 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/6] 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 0076feb469c879a8f8949e4dc5ec823b87c9e630 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/6] 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 c196ae108cdbed2dd94129d2dae07a215de411a5 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/6] 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.
 

From b2b620d13edcb7308d897c3fbef588481789204a Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Mon, 24 Aug 2026 15:29:38 -0700
Subject: [PATCH 5/6] Build the metrics pipeline once per process, not once per
 module copy

The observability package is symlinked into both airflow-core and task-sdk, so
Python loads it twice and each copy gets its own module globals. A guard held in
one of those globals can only ever deduplicate within its own copy, so a
scheduler still ended up with two pipelines exporting alongside each other --
one per copy -- rather than the one intended.

The same split affects the fork cleanup: with the handle in module state, only
whichever copy happened to build the pipeline could recognise and stop it in the
child. Recording ownership on the provider instead puts it in the one place both
copies already share, the OTel global provider registry, so any copy can tell
whether the pipeline belongs to this process.

Measured over the scheduler's startup order, this takes the process from three
live MeterProviders and three exporter threads to one of each.
---
 airflow-core/newsfragments/71855.bugfix.rst   |   2 +-
 .../observability/metrics/otel_logger.py      |  72 +++++---
 .../observability/metrics/test_otel_logger.py | 158 +++++++++++++++++-
 3 files changed, 202 insertions(+), 30 deletions(-)

diff --git a/airflow-core/newsfragments/71855.bugfix.rst b/airflow-core/newsfragments/71855.bugfix.rst
index f7b259e0bd87e..e0a728aba7fd3 100644
--- a/airflow-core/newsfragments/71855.bugfix.rst
+++ b/airflow-core/newsfragments/71855.bugfix.rst
@@ -1 +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.
+Fixed OTel metrics being exported as several conflicting cumulative streams. A process now builds a single metrics pipeline and reuses it, instead of leaving replaced ones exporting frozen totals alongside the live one — a scheduler previously ran three at once, one per copy of the metrics module plus one per re-initialisation. 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 f64695092f55a..b491814117e4e 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, Any
+from typing import TYPE_CHECKING, Any, cast
 
 from opentelemetry import metrics
 from opentelemetry.sdk.metrics import MeterProvider
@@ -434,16 +434,44 @@ 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
+# Stamped on the MeterProvider this module builds, recording the pid that built it.
+#
+# It lives on the provider rather than in a module global because this package is imported twice --
+# once as ``airflow._shared.observability`` and once as ``airflow.sdk._shared.observability`` -- so
+# each copy has its own globals and neither can see the other's. A module global would only ever
+# deduplicate within one copy, and would leave the fork cleanup below working only in whichever
+# copy happened to build the pipeline. The OTel global provider registry is genuinely
+# process-wide, which makes the provider itself the only place both copies can agree on.
+#
+# The pid it records answers "did *this* process build it?". A fork hands the child the object
+# with the parent's pid still on it, along with 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, and the stale pid is what says so. See
+# https://github.com/apache/airflow/issues/64690; the OTel SDK keys ``service.instance.id`` off
+# the pid for the same question.
+_OWNER_PID_ATTR = "_airflow_owner_pid"
+
+
+def _find_provider_built_by_this_process() -> MeterProvider | None:
+    """Return the globally registered MeterProvider if this module built it in this process."""
+    if getattr(provider := metrics.get_meter_provider(), _OWNER_PID_ATTR, None) == os.getpid():
+        # Only a provider built by get_otel_logger() carries the marker, so this is an SDK one.
+        return cast("MeterProvider", provider)
+    return None
+
+
+def _find_pipeline_inherited_across_fork() -> MeterProvider | None:
+    """Return the globally registered MeterProvider if this module built it in a parent process."""
+    owner_pid = getattr(provider := metrics.get_meter_provider(), _OWNER_PID_ATTR, None)
+    if owner_pid is not None and owner_pid != os.getpid():
+        return cast("MeterProvider", provider)
+    return None
 
 
 def flush_otel_metrics():
-    if _provider is not None:
-        _provider.force_flush()
+    if (provider := _find_provider_built_by_this_process()) is not None:
+        provider.force_flush()
 
 
 def _find_own_readers(provider: MeterProvider) -> Sequence[Any] | None:
@@ -517,12 +545,10 @@ def _drop_inherited_exit_handler(provider: object) -> None:
 
 
 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
+    if (inherited := _find_pipeline_inherited_across_fork()) is not None:
+        _stop_inherited_pipeline(inherited)
 
 
 os.register_at_fork(after_in_child=_reset_provider_after_fork)
@@ -559,9 +585,12 @@ def get_otel_logger(
     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.
+    The process 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. That holds across both
+    copies of this package, since ownership is recorded on the provider rather than in module
+    state. Each caller still gets its own :class:`SafeOtelLogger`, so the prefix, allow/block
+    lists, name handler and InfluxDB flag stay per-caller; only the pipeline is shared. 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)
@@ -573,10 +602,10 @@ def get_otel_logger(
             configured_provider, effective_prefix, validator, stat_name_handler, statsd_influxdb_enabled
         )
 
-    global _provider
-    if _provider is not None:
+    own_provider = _find_provider_built_by_this_process()
+    if own_provider is not None:
         return SafeOtelLogger(
-            _provider, effective_prefix, validator, stat_name_handler, statsd_influxdb_enabled
+            own_provider, effective_prefix, validator, stat_name_handler, statsd_influxdb_enabled
         )
 
     otel_env_config = load_metrics_env_config()
@@ -627,7 +656,7 @@ def get_otel_logger(
     except (ImportError, AttributeError):
         pass
 
-    _provider = MeterProvider(
+    provider = MeterProvider(
         resource=resource,
         metric_readers=readers,
         views=[
@@ -638,9 +667,10 @@ def get_otel_logger(
         ],
         shutdown_on_exit=False,
     )
-    metrics.set_meter_provider(_provider)
+    setattr(provider, _OWNER_PID_ATTR, os.getpid())
+    metrics.set_meter_provider(provider)
 
     # Register a hook that flushes any in-memory metrics at shutdown.
     atexit.register(flush_otel_metrics)
 
-    return SafeOtelLogger(_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 2b2390fb0346e..0e7f6c9215b68 100644
--- a/shared/observability/tests/observability/metrics/test_otel_logger.py
+++ b/shared/observability/tests/observability/metrics/test_otel_logger.py
@@ -16,6 +16,7 @@
 # under the License.
 from __future__ import annotations
 
+import importlib.util
 import logging
 import os
 import subprocess
@@ -39,6 +40,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 (
+    _OWNER_PID_ATTR,
     OTEL_NAME_MAX_LENGTH,
     UP_DOWN_COUNTERS,
     MetricsMap,
@@ -81,14 +83,16 @@ def name():
 
 @pytest.fixture(autouse=True)
 def reset_process_provider():
-    """Clear the per-process provider cache so tests never inherit one another's pipeline."""
+    """Clear the per-process pipeline so tests never inherit one another's."""
 
     def discard() -> None:
-        # Stop the pipeline rather than just dropping the reference: its exporter threads would
+        provider = otel_logger_module._find_provider_built_by_this_process()
+        if provider is None:
+            return
+        # Stop the pipeline rather than just dropping the marker: 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
+        otel_logger_module._stop_inherited_pipeline(provider)
+        delattr(provider, _OWNER_PID_ATTR)
 
     discard()
     yield
@@ -727,17 +731,155 @@ def test_reinit_reuses_the_process_pipeline(self, reset_meter_provider):
         assert count_live_readers() == readers_after_first
 
 
-TEST_MODULE = "tests.observability.metrics.test_otel_logger"
+class TestOtelProviderIsProcessScoped:
+    """One MeterProvider per process, however many callers -- or module copies -- ask for one."""
+
+    def test_a_second_module_copy_reuses_one_provider(self, reset_meter_provider):
+        """Test that an independently imported copy of this module shares the same provider."""
+        second_copy = load_independent_module_copy()
+        assert second_copy is not otel_logger_module
+
+        first = get_otel_logger(host="localhost", port=4318, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS)
+        readers_after_first = count_live_readers()
+
+        second = second_copy.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
+
+    @pytest.mark.parametrize(
+        ("attribute", "first_value", "second_value"),
+        [
+            pytest.param("prefix", "alpha", "beta", id="prefix"),
+            pytest.param("statsd_influxdb_enabled", True, False, id="statsd_influxdb_enabled"),
+        ],
+    )
+    def test_sharing_a_provider_keeps_per_caller_configuration(
+        self, attribute, first_value, second_value, reset_meter_provider
+    ):
+        """Test that callers sharing one provider still get their own logger configuration."""
+        first = get_otel_logger(
+            host="localhost",
+            port=4318,
+            conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS,
+            **{attribute: first_value},
+        )
+        second = get_otel_logger(
+            host="localhost",
+            port=4318,
+            conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS,
+            **{attribute: second_value},
+        )
+
+        assert first.otel is second.otel
+        assert getattr(first, attribute) == first_value
+        assert getattr(second, attribute) == second_value
+
+    def test_a_provider_built_by_another_process_is_not_reused(self, reset_meter_provider):
+        """Test that a provider inherited from another process is replaced, not adopted."""
+        inherited = SDKMeterProvider(shutdown_on_exit=False)
+        setattr(inherited, _OWNER_PID_ATTR, os.getpid() + 1)
+        metrics.set_meter_provider(inherited)
+
+        logger = get_otel_logger(host="localhost", port=4318, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS)
+
+        assert logger.otel is not inherited
+
+    def test_both_module_copies_still_export_their_metrics(self):
+        """Test that sharing one provider does not stop either module copy emitting metrics."""
+        proc = subprocess.run(
+            [sys.executable, "-c", TWO_MODULE_COPIES_PROGRAM],
+            check=False,
+            env=os.environ.copy(),
+            capture_output=True,
+            text=True,
+            timeout=30,
+        )
+
+        assert proc.returncode == 0, f"Process failed\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}"
+
+        context = f"stdout:\n{proc.stdout}\nstderr:\n{proc.stderr}"
+        assert "COPIES_SHARE_PROVIDER=True" in proc.stdout, (
+            f"The two module copies each built their own provider.\n{context}"
+        )
+        assert "copyone.from_first_copy" in proc.stdout, (
+            f"The first module copy's metric was not exported.\n{context}"
+        )
+        assert "copytwo.from_second_copy" in proc.stdout, (
+            f"The second module copy's metric was not exported.\n{context}"
+        )
+
+
+def load_independent_module_copy():
+    """
+    Load a second, independent copy of otel_logger, as the two ``_shared`` symlinks do.
+
+    ``shared/observability`` is symlinked into both ``airflow/_shared`` and ``airflow/sdk/_shared``,
+    so the same file is executed twice under two names and each copy gets its own module globals.
+    The name below keeps the package prefix so the module's relative imports still resolve.
+    """
+    spec = importlib.util.spec_from_file_location(
+        "airflow_shared.observability.metrics._otel_logger_second_copy", otel_logger_module.__file__
+    )
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+# Runs in a fresh interpreter because it needs a process whose global MeterProvider starts out
+# unset. It imports only the installed package, so unlike the mock_service_* helpers it does not
+# depend on this test module being importable by name from the subprocess's working directory.
+TWO_MODULE_COPIES_PROGRAM = """
+import importlib.util
+
+from airflow_shared.observability.metrics import otel_logger as first_copy
+
+spec = importlib.util.spec_from_file_location(
+    "airflow_shared.observability.metrics._otel_logger_second_copy", first_copy.__file__
+)
+second_copy = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(second_copy)
+
+first_logger = first_copy.get_otel_logger(debug=True, prefix="copyone")
+second_logger = second_copy.get_otel_logger(debug=True, prefix="copytwo")
+
+print("COPIES_SHARE_PROVIDER=%s" % (first_logger.otel is second_logger.otel))
+
+first_logger.incr("from_first_copy")
+second_logger.incr("from_second_copy")
+
+# Flush both explicitly so that what was exported is asserted independently of whether the two
+# copies ended up sharing one provider.
+first_logger.otel.force_flush()
+second_logger.otel.force_flush()
+"""
 
 
 def run_service_helper(runner: str, child_out=None) -> subprocess.CompletedProcess:
-    """Run one of this module's ``mock_service_*`` helpers in a fresh interpreter."""
+    """
+    Run one of this module's ``mock_service_*`` helpers in a fresh interpreter.
+
+    The helper module is loaded by file path rather than imported by dotted name: ``tests`` is a
+    namespace package here, and any regular ``tests`` package that happens to be installed in the
+    environment would win the import and hide it.
+    """
     env = os.environ.copy()
     if child_out is not None:
         env["CHILD_METRICS_OUT"] = str(child_out)
 
+    program = (
+        "import importlib.util, sys;"
+        f"spec = importlib.util.spec_from_file_location('otel_service_helpers', {__file__!r});"
+        "m = importlib.util.module_from_spec(spec);"
+        "sys.modules['otel_service_helpers'] = m;"
+        "spec.loader.exec_module(m);"
+        f"m.{runner}()"
+    )
+
     proc = subprocess.run(
-        [sys.executable, "-c", f"import {TEST_MODULE} as m; m.{runner}()"],
+        [sys.executable, "-c", program],
         check=False,
         env=env,
         capture_output=True,

From e12d0b98cab77687f8d8741e2d1547db3b12af96 Mon Sep 17 00:00:00 2001
From: Daniel Standish <15932138+dstandish@users.noreply.github.com>
Date: Wed, 26 Aug 2026 14:50:21 -0700
Subject: [PATCH 6/6] Note delta temporality in the inherited-pipeline
 docstring

The surrounding reasoning is framed entirely around cumulative
temporality, so a reader running delta could conclude the fork
handling does not concern them.
---
 .../src/airflow_shared/observability/metrics/otel_logger.py  | 5 +++++
 1 file changed, 5 insertions(+)

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 b491814117e4e..db0b54746645b 100644
--- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py
+++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py
@@ -504,6 +504,11 @@ def _stop_inherited_pipeline(provider: MeterProvider) -> None:
     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.
+
+    Temporality does not change the need for this. Under cumulative the inherited readers
+    republish the parent's totals on every export; under delta they fall quiet, but their first
+    collect still emits the deltas the parent recorded before the fork -- which the parent
+    reports as well.
     """
     readers = _find_own_readers(provider)
     if readers is None: