diff --git a/.changelog/5453.fixed b/.changelog/5453.fixed new file mode 100644 index 00000000000..be06e3e8263 --- /dev/null +++ b/.changelog/5453.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: fix `TypeError` in `os.fork()` when a `BatchProcessor` or `PeriodicExportingMetricReader` is garbage collected diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 3e2b8a263a4..5a2a57f8fc0 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -116,7 +116,12 @@ def __init__( self._worker_thread.start() if hasattr(os, "register_at_fork"): weak_reinit = weakref.WeakMethod(self._at_fork_reinit) - os.register_at_fork(after_in_child=lambda: weak_reinit()()) # pyright: ignore[reportOptionalCall] pylint: disable=unnecessary-lambda + + def _after_in_child() -> None: + if reinit := weak_reinit(): + reinit() + + os.register_at_fork(after_in_child=_after_in_child) self._pid = os.getpid() metrics.register_queue_size(lambda: len(self._queue)) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py index aee356f78d2..8cce878cdba 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/metrics/_internal/export/__init__.py @@ -531,9 +531,11 @@ def __init__( if hasattr(os, "register_at_fork"): weak_at_fork = weakref.WeakMethod(self._at_fork_reinit) - os.register_at_fork( - after_in_child=lambda: weak_at_fork()() # pylint: disable=unnecessary-lambda - ) + def _after_in_child() -> None: + if at_fork := weak_at_fork(): + at_fork() + + os.register_at_fork(after_in_child=_after_in_child) elif self._export_interval_millis <= 0: raise ValueError( f"interval value {self._export_interval_millis} is invalid \ diff --git a/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py b/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py index ab8877c21c3..1a2a1b319f3 100644 --- a/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py +++ b/opentelemetry-sdk/tests/metrics/test_periodic_exporting_metric_reader.py @@ -5,6 +5,8 @@ import gc import math +import os +import sys import weakref from logging import WARNING from time import sleep, time_ns @@ -307,6 +309,68 @@ def test_metric_exporer_gc(self): "The PeriodicExportingMetricReader object created by this test wasn't garbage collected", ) + @pytest.mark.skipif( + not hasattr(os, "fork") or not hasattr(os, "register_at_fork"), + reason="needs fork and register_at_fork", + ) + def test_garbage_collected_processor_does_not_crash_on_fork(self): + exporter = FakeMetricsExporter( + preferred_aggregation={ + Counter: LastValueAggregation(), + }, + ) + processor = PeriodicExportingMetricReader(exporter) + w_ref = weakref.ref(processor) + processor.shutdown() + del processor + gc.collect() + + self.assertIsNone(w_ref()) + + # The bug causes an unraisable exception to be printed to stderr. + # We redirect stderr to a pipe before fork to capture it. + r_fd, w_fd = os.pipe() + + # Save original stderr and redirect it to the pipe + # We also temporarily restore the default sys.unraisablehook. Pytest overrides + # it to capture exceptions in memory, which would be lost on os._exit(0). + old_fd = os.dup(sys.stderr.fileno()) + old_hook = sys.unraisablehook + + try: + os.dup2(w_fd, sys.stderr.fileno()) + sys.unraisablehook = sys.__unraisablehook__ + + pid = os.fork() + if pid == 0: + os.close(r_fd) + # os.fork() has already run the at_fork hooks. + os._exit(0) + + _, status = os.waitpid(pid, 0) + self.assertEqual(status, 0) + finally: + # Restore original stderr and hook in parent + sys.unraisablehook = old_hook + os.dup2(old_fd, sys.stderr.fileno()) + os.close(old_fd) + os.close(w_fd) + + os.set_blocking(r_fd, False) + child_stderr = b"" + while True: + try: + chunk = os.read(r_fd, 4096) + if not chunk: + break + child_stderr += chunk + except BlockingIOError: + break + child_stderr = child_stderr.decode("utf-8") + os.close(r_fd) + + self.assertNotIn("TypeError", child_stderr) + @patch.dict( "os.environ", {OTEL_PYTHON_SDK_INTERNAL_METRICS_ENABLED: "true"} ) diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py index 02337757073..91500155cdb 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -6,6 +6,7 @@ import logging import multiprocessing import os +import sys import threading import time import unittest @@ -215,6 +216,65 @@ def test_record_processor_is_garbage_collected( # Then the reference to the processor should no longer exist assert weak_ref() is None + @unittest.skipUnless( + hasattr(os, "fork"), + "needs *nix", + ) + def test_garbage_collected_processor_does_not_crash_on_fork( + self, batch_processor_class, telemetry + ): + exporter = Mock() + processor = batch_processor_class(exporter) + w_ref = weakref.ref(processor) + processor.shutdown() + del processor + gc.collect() + + assert w_ref() is None + + # The bug causes an unraisable exception to be printed to stderr. + # We redirect stderr to a pipe before fork to capture it. + r_fd, w_fd = os.pipe() + + # Save original stderr and redirect it to the pipe + # We also temporarily restore the default sys.unraisablehook. Pytest overrides + # it to capture exceptions in memory, which would be lost on os._exit(0). + old_fd = os.dup(sys.stderr.fileno()) + old_hook = sys.unraisablehook + + try: + os.dup2(w_fd, sys.stderr.fileno()) + sys.unraisablehook = sys.__unraisablehook__ + + pid = os.fork() + if pid == 0: + os.close(r_fd) + # os.fork() has already run the at_fork hooks. + os._exit(0) + + os.waitpid(pid, 0) + finally: + # Restore original stderr and hook in parent + sys.unraisablehook = old_hook + os.dup2(old_fd, sys.stderr.fileno()) + os.close(old_fd) + os.close(w_fd) + + os.set_blocking(r_fd, False) + child_stderr = b"" + while True: + try: + chunk = os.read(r_fd, 4096) + if not chunk: + break + child_stderr += chunk + except BlockingIOError: + break + child_stderr = child_stderr.decode("utf-8") + os.close(r_fd) + + assert "TypeError" not in child_stderr + def test_shutdown_allows_1_export_to_finish( self, batch_processor_class, telemetry ):