Fix TypeError in os.fork() for garbage-collected processors

.changelog/5453.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: fix `TypeError` in `os.fork()` when a `BatchProcessor` or `PeriodicExportingMetricReader` is garbage collected

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))

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 \

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"}
     )

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
     ):