From c709ee298d0b77bd7098cbb2c690e3fdd3a98f08 Mon Sep 17 00:00:00 2001 From: Jed Cunningham Date: Tue, 18 Aug 2026 15:20:54 -0600 Subject: [PATCH 1/3] Stop forked children re-exporting the parent's OTel metrics A forked child inherits both the atexit flush hook and the parent's MeterProvider, so at its own exit it flushed a pipeline it never recorded to: every metric the parent had accumulated was exported a second time, from a process that did not own it, as a second writer for the same cumulative series. Airflow forks per task, so that duplicate arrives once per fork. Rebuilding the pipeline on every get_otel_logger() call is the same root cause from the other side. The replaced provider's exporter thread keeps running under shutdown_on_exit=False while nothing records to its instruments, so it republishes frozen totals alongside the live stream. The scheduler reaches this because BaseExecutor.__init__ and SchedulerJobRunner each initialize stats in the same process. Making the provider per-process state answers both: only the process that built it ever flushes it, and no second pipeline is built to be left behind. --- .../observability/metrics/otel_logger.py | 58 +++++++++----- .../observability/metrics/test_otel_logger.py | 80 +++++++++++++++++++ 2 files changed, 117 insertions(+), 21 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..f8e5a73144f00 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,24 @@ def record_histogram_value(self, name: str, value: float, tags: Attributes) -> N self.histograms[name].record(value, tags) +# The MeterProvider this process built, if any. A fork hands the child this reference and the +# atexit hook, but the child owns neither: flushing a provider it inherited re-exports everything +# the parent accumulated, and the reader threads behind it no longer exist. +_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 _reset_provider_after_fork() -> None: + global _provider + atexit.unregister(flush_otel_metrics) + _provider = None + + +os.register_at_fork(after_in_child=_reset_provider_after_fork) def get_otel_logger( @@ -466,6 +477,9 @@ def get_otel_logger( 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. + + The pipeline is built once per process. Later calls return a logger over the same provider + rather than leaving a second one exporting alongside it; a fork resets that. """ effective_prefix: str = prefix or DEFAULT_METRIC_NAME_PREFIX validator = get_validator(metrics_allow_list, metrics_block_list) @@ -473,11 +487,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" @@ -526,23 +545,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..4079b1103190d 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 @@ -52,6 +53,10 @@ 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 + INVALID_STAT_NAME_CASES = [ (None, "can not be None"), (42, "is not a string"), @@ -76,9 +81,12 @@ def reset_meter_provider(): """ import opentelemetry.metrics._internal as metrics_internal + from airflow_shared.observability.metrics import otel_logger as otel_logger_module + def clear() -> None: metrics_internal._METER_PROVIDER_SET_ONCE._done = False metrics_internal._METER_PROVIDER = None + otel_logger_module._provider = None previous = metrics_internal._METER_PROVIDER clear() @@ -589,6 +597,56 @@ def test_atexit_flush_on_process_exit(self): f"stderr:\n{proc.stderr}" ) + @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. + + 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.{runner}()" + + 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}" + + 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}" + ) + + 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 + def test_reinit_after_fork_exports_metrics(self): """Calling get_otel_logger() twice (simulating post-fork re-init) should still export metrics. @@ -618,6 +676,11 @@ def test_reinit_after_fork_exports_metrics(self): ) +def count_live_readers() -> int: + """Number of running periodic exporter threads, i.e. how many pipelines are live.""" + 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") @@ -636,3 +699,20 @@ def mock_service_run_reinit(): # Second init — simulates post-fork re-initialization logger = get_otel_logger(debug=True) logger.incr("post_fork_stat") + + +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.""" + get_otel_logger(debug=True, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS).incr("parent_stat") + if os.fork() == 0: + get_otel_logger(debug=True, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS).incr("child_stat") + sys.exit(0) + os.wait() From 335e8899df58463e713f448ddfe0a2a449e80fa7 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Wed, 19 Aug 2026 12:53:46 -0700 Subject: [PATCH 2/3] Stop forked children exporting the OTel pipeline they inherited Owning the provider per process keeps the child from flushing what it inherited, but not from exporting it. The SDK registers register_at_fork(after_in_child=...) for every PeriodicExportingMetricReader, so the child restarts the exporter thread behind the inherited pipeline. Nothing in the child records to it, so it republishes the totals the parent held at the instant of the fork -- once per export interval, for as long as the child lives. A consumer sees two writers on one cumulative series: one climbing, one frozen. Airflow forks constantly, and the long-lived children make that permanent rather than momentary: LocalExecutor pool workers, the OpenLineage dag-state-change pool and the scheduler's log and health-check servers all fork from a scheduler whose pipeline is already live, then outlive many export cycles. Setting the reader's shutdown event alone is not enough, because the ticker publishes one last collection on its way out; dropping the collect callback keeps that final pass from carrying the parent's totals with it. A child that emits metrics of its own still builds its own pipeline, so this costs it nothing. A provider the SDK built for the deployment -- from OTEL_CONFIG_FILE or by an instrumentation agent -- reaches the child the same way, carrying the atexit shutdown that shutdown_on_exit=True registered for it. Its readers stay running, since on that path they are the only pipeline the child has, but the inherited copy of that hook is dropped so the child cannot dump the parent's state on the way out. --- .../observability/metrics/otel_logger.py | 79 +++++++++-- .../observability/metrics/test_otel_logger.py | 130 ++++++++++++++++++ 2 files changed, 200 insertions(+), 9 deletions(-) diff --git a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py index f8e5a73144f00..9babb90272fc2 100644 --- a/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py +++ b/shared/observability/src/airflow_shared/observability/metrics/otel_logger.py @@ -48,6 +48,7 @@ if TYPE_CHECKING: from opentelemetry.metrics import Instrument + from opentelemetry.sdk.metrics.export import MetricReader from opentelemetry.util.types import Attributes from .protocols import DeltaType @@ -432,9 +433,9 @@ def record_histogram_value(self, name: str, value: float, tags: Attributes) -> N self.histograms[name].record(value, tags) -# The MeterProvider this process built, if any. A fork hands the child this reference and the -# atexit hook, but the child owns neither: flushing a provider it inherited re-exports everything -# the parent accumulated, and the reader threads behind it no longer exist. +# The MeterProvider this process built, if any. A fork hands the child both this reference and the +# atexit hook, but the child owns neither: were it to flush a provider it inherited, everything the +# parent accumulated would be exported a second time, by a process that recorded none of it. _provider: MeterProvider | None = None @@ -443,9 +444,64 @@ def flush_otel_metrics(): _provider.force_flush() +def _collect_nothing(reader: MetricReader, timeout_millis: float = 10_000) -> None: + """Collect callback that yields no measurements, so the reader never reaches its exporter.""" + return None + + +def _stop_inherited_readers(provider: MeterProvider) -> None: + """ + Stop the exporter threads a fork handed this child. + + The SDK restarts every ``PeriodicExportingMetricReader`` ticker in the child + (``register_at_fork(after_in_child=...)``), so the pipeline a child inherits goes on exporting + the totals the parent held at the moment of the fork -- once per interval, for as long as the + child lives, as a second writer on a cumulative series it does not record to. A child that + emits metrics of its own builds its own pipeline in :func:`get_otel_logger`. + + Setting the shutdown event ends the restarted ticker at its first wait, and dropping the + collect callback keeps the one collection it makes on its way out from publishing those + totals. + + 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: + reader._collect = _collect_nothing + if (shutdown_event := getattr(reader, "_shutdown_event", None)) is not None: + shutdown_event.set() + + +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. Its readers are left + running: on that path they are the only pipeline the child has. + + A provider without that handler is either one built here (``shutdown_on_exit=False``) or one + that never registered it, so there is nothing to drop either way. + """ + handler = getattr(provider, "_atexit_handler", None) + if handler is None: + return + atexit.unregister(handler) + provider._atexit_handler = None # type: ignore[attr-defined] + + 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_readers(_provider) _provider = None @@ -476,10 +532,15 @@ 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. - - The pipeline is built once per process. Later calls return a logger over the same provider - rather than leaving a second one exporting alongside it; a fork resets that. + configuration spec makes that file the sole source of SDK construction. Its lifecycle stays the + deployment's -- the SDK builds it with its own atexit shutdown, so no flush hook is registered + for it here. Across a fork the child drops only 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 that path they are the only pipeline the child has. + + The pipeline built here is built once per process. Later calls return a logger over the same + provider rather than leaving a second one exporting alongside it; a forked child stops the one + it inherited and builds its own. """ effective_prefix: str = prefix or DEFAULT_METRIC_NAME_PREFIX validator = get_validator(metrics_allow_list, metrics_block_list) @@ -533,8 +594,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: diff --git a/shared/observability/tests/observability/metrics/test_otel_logger.py b/shared/observability/tests/observability/metrics/test_otel_logger.py index 4079b1103190d..165b52a9394a6 100644 --- a/shared/observability/tests/observability/metrics/test_otel_logger.py +++ b/shared/observability/tests/observability/metrics/test_otel_logger.py @@ -28,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, @@ -57,6 +58,9 @@ # output once per flush. NO_PERIODIC_EXPORT_INTERVAL_MS = 600000 +# Short enough that a live exporter thread publishes several times over the child's lifetime. +PERIODIC_EXPORT_INTERVAL_MS = 200 + INVALID_STAT_NAME_CASES = [ (None, "can not be None"), (42, "is not a string"), @@ -631,6 +635,78 @@ 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}" ) + @staticmethod + def run_forking_helper(runner: str, tmp_path) -> str: + """Run one of this module's forking helpers and return what its child exported.""" + child_output = tmp_path / "child_stdout.json" + test_module_name = "tests.observability.metrics.test_otel_logger" + function_call_str = f"import {test_module_name} as m; m.{runner}()" + + proc = subprocess.run( + [sys.executable, "-c", function_call_str], + check=False, + env={**os.environ, "CHILD_OUTPUT_FILE": str(child_output)}, + capture_output=True, + text=True, + timeout=20, + ) + + assert proc.returncode == 0, f"Process failed\nstdout:\n{proc.stdout}\nstderr:\n{proc.stderr}" + assert child_output.exists(), f"Child never redirected its output\nstderr:\n{proc.stderr}" + return child_output.read_text() + + @pytest.mark.parametrize( + ("runner", "own_metric"), + [ + ("mock_service_fork_child_periodic_no_reinit", None), + ("mock_service_fork_child_periodic_reinit", "child_stat"), + ], + ) + def test_forked_child_stops_exporting_the_inherited_pipeline(self, tmp_path, runner, own_metric): + """Test that a forked child stops exporting the pipeline it inherited, not the one it builds. + + The SDK restarts every ``PeriodicExportingMetricReader`` ticker in a forked child, so an + inherited pipeline keeps publishing the totals the parent had accumulated at the moment of + the fork -- once per interval, for the child's whole life, from a process that records + nothing to it. + """ + child_exports = self.run_forking_helper(runner, tmp_path) + + assert child_exports.count('"name": "airflow.parent_stat"') == 0, ( + f"Child re-exported the parent's metric:\n{child_exports}" + ) + if own_metric: + assert f'"name": "airflow.{own_metric}"' in child_exports, ( + f"Child stopped exporting its own metric:\n{child_exports}" + ) + + def test_forked_child_does_not_flush_a_provider_it_did_not_build(self, tmp_path): + """Test that a child does not run the atexit shutdown of a provider it inherited. + + ``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 along with everything the parent accumulated. + """ + child_exports = self.run_forking_helper("mock_service_fork_child_under_foreign_provider", tmp_path) + + assert "foreign_counter" not in child_exports, ( + f"Child ran the inherited provider's shutdown hook:\n{child_exports}" + ) + + def test_forked_child_leaves_other_pipelines_alone(self, tmp_path): + """Test that stopping the inherited pipeline spares 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_exports = self.run_forking_helper("mock_service_fork_child_beside_foreign_provider", tmp_path) + + assert "foreign_counter" in child_exports, ( + f"Child stopped a pipeline Airflow does not own:\n{child_exports}" + ) + assert child_exports.count('"name": "airflow.parent_stat"') == 0 + def test_reinit_reuses_the_process_pipeline(self, reset_meter_provider): """A second call must not leave a second pipeline exporting alongside the first. @@ -716,3 +792,57 @@ def mock_service_fork_child_with_reinit(): get_otel_logger(debug=True, conf_interval=NO_PERIODIC_EXPORT_INTERVAL_MS).incr("child_stat") sys.exit(0) os.wait() + + +def _redirect_child_output() -> None: + """Point fd 1 at ``CHILD_OUTPUT_FILE``, so a child's exports are told apart from its parent's.""" + os.dup2(os.open(os.environ["CHILD_OUTPUT_FILE"], os.O_WRONLY | os.O_CREAT | os.O_TRUNC), 1) + + +def _build_foreign_provider(export_interval_ms: float, *, shutdown_on_exit: bool) -> SDKMeterProvider: + """Build a pipeline Airflow does not own, as an instrumentation agent would.""" + provider = SDKMeterProvider( + metric_readers=[ + PeriodicExportingMetricReader(ConsoleMetricExporter(), export_interval_millis=export_interval_ms) + ], + shutdown_on_exit=shutdown_on_exit, + ) + provider.get_meter("foreign").create_counter("foreign_counter").add(7) + return provider + + +def _fork_child_writing_exports_to_file(reinit: bool) -> None: + """Emit a metric, then fork a child that outlives several export intervals.""" + get_otel_logger(debug=True, conf_interval=PERIODIC_EXPORT_INTERVAL_MS).incr("parent_stat") + if os.fork() == 0: + _redirect_child_output() + if reinit: + get_otel_logger(debug=True, conf_interval=PERIODIC_EXPORT_INTERVAL_MS).incr("child_stat") + time.sleep(PERIODIC_EXPORT_INTERVAL_MS * 5 / 1000) + os._exit(0) + os.wait() + + +def mock_service_fork_child_periodic_no_reinit(): + """Fork a child that never initializes a logger of its own.""" + _fork_child_writing_exports_to_file(reinit=False) + + +def mock_service_fork_child_periodic_reinit(): + """Fork a child that initializes its own logger and emits its own metric.""" + _fork_child_writing_exports_to_file(reinit=True) + + +def mock_service_fork_child_under_foreign_provider(): + """Fork under an SDK-built provider that kept its own atexit shutdown; the child exits cleanly.""" + metrics.set_meter_provider(_build_foreign_provider(NO_PERIODIC_EXPORT_INTERVAL_MS, shutdown_on_exit=True)) + if os.fork() == 0: + _redirect_child_output() + sys.exit(0) # a normal exit runs the atexit hooks + os.wait() + + +def mock_service_fork_child_beside_foreign_provider(): + """Fork an idle child while both Airflow's pipeline and one it does not own are exporting.""" + _build_foreign_provider(PERIODIC_EXPORT_INTERVAL_MS, shutdown_on_exit=False) + _fork_child_writing_exports_to_file(reinit=False) From 9a666dfadf16283be64a48d1807bfe3d563b98fc Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Wed, 19 Aug 2026 13:18:29 -0700 Subject: [PATCH 3/3] Add newsfragment for the duplicate OTel metric stream fix --- airflow-core/newsfragments/71804.bugfix.rst | 1 + 1 file changed, 1 insertion(+) create mode 100644 airflow-core/newsfragments/71804.bugfix.rst diff --git a/airflow-core/newsfragments/71804.bugfix.rst b/airflow-core/newsfragments/71804.bugfix.rst new file mode 100644 index 0000000000000..f7b259e0bd87e --- /dev/null +++ b/airflow-core/newsfragments/71804.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.