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. 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..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,13 +433,79 @@ 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 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 + + 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 _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 + + +os.register_at_fork(after_in_child=_reset_provider_after_fork) def get_otel_logger( @@ -465,7 +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. + 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) @@ -473,11 +548,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 +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: @@ -526,23 +606,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..165b52a9394a6 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, @@ -52,6 +54,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 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"), @@ -76,9 +85,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 +601,128 @@ 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}" + ) + + @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. + + 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 +752,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 +775,74 @@ 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() + + +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)