fix(sdk): run process-dependent resource detectors synchronously in fork handlers (#5789)

.changelog/5790.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: run process-dependent resource detectors synchronously in fork handlers to avoid post-fork deadlock when thread pool locks are held

opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py

@@ -54,6 +54,7 @@
 
 import abc
 import concurrent.futures
+import functools
 import logging
 import os
 import platform
@@ -182,6 +183,8 @@ def __init__(self, attributes: Attributes, schema_url: str | None = None):
     def create(
         attributes: Attributes | None = None,
         schema_url: str | None = None,
+        *,
+        _use_thread_pool: bool = True,
     ) -> "Resource":
         """Creates a new `Resource` from attributes.
 
@@ -198,9 +201,11 @@ def create(
         if not attributes:
             attributes = {}
 
-        resource = get_aggregated_resources(_build_resource_detectors(), _DEFAULT_RESOURCE).merge(
-            Resource(attributes, schema_url)
-        )
+        resource = get_aggregated_resources(
+            _build_resource_detectors(),
+            _DEFAULT_RESOURCE,
+            _use_thread_pool=_use_thread_pool,
+        ).merge(Resource(attributes, schema_url))
 
         if not resource.attributes.get(SERVICE_NAME, None):
             default_service_name = "unknown_service"
@@ -746,13 +751,16 @@ def _get_process_dependent_resource() -> Resource:  # pyright: ignore[reportUnus
     return get_aggregated_resources(
         [detector for detector in _build_resource_detectors() if detector.is_process_dependent()],
         Resource.get_empty(),
+        _use_thread_pool=False,
     )
 
 
 def get_aggregated_resources(
     detectors: Sequence["ResourceDetector"],
     initial_resource: Resource | None = None,
     timeout: int = 5,
+    *,
+    _use_thread_pool: bool = True,
 ) -> "Resource":
     """Retrieves resources from detectors in the order that they were passed
 
@@ -761,20 +769,26 @@ def get_aggregated_resources(
     :param timeout: Number of seconds to wait for each detector to return
     :return:
     """
-    detectors_merged_resource = initial_resource or Resource.create()
+    detectors_merged_resource = (
+        initial_resource if initial_resource is not None else Resource.create(_use_thread_pool=_use_thread_pool)
+    )
 
-    # A detector that exceeds `timeout` is skipped, but shutdown still joins
-    # its worker so no threads outlive this call. A detector that blocks
-    # indefinitely therefore blocks this call. `cancel_futures=True` drops
-    # detectors that have not started yet if a `raise_on_error` detector raises.
-    executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
+    executor: concurrent.futures.ThreadPoolExecutor | None = None
     try:
-        futures = [executor.submit(detector.detect) for detector in detectors]
-        for detector_ind, future in enumerate(futures):
-            detector = detectors[detector_ind]
+        if _use_thread_pool:
+            executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
+            futures = [executor.submit(detector.detect) for detector in detectors]
+            tasks = [
+                (detector, functools.partial(future.result, timeout=timeout))
+                for detector, future in zip(detectors, futures)
+            ]
+        else:
+            tasks = [(detector, detector.detect) for detector in detectors]
+
+        for detector, get_result in tasks:
             detected_resource: Resource = _EMPTY_RESOURCE
             try:
-                detected_resource = future.result(timeout=timeout)
+                detected_resource = get_result()
             except concurrent.futures.TimeoutError as ex:
                 if detector.raise_on_error:
                     raise ex
@@ -791,6 +805,7 @@ def get_aggregated_resources(
             finally:
                 detectors_merged_resource = detectors_merged_resource.merge(detected_resource)
     finally:
-        executor.shutdown(wait=True, cancel_futures=True)
+        if executor is not None:
+            executor.shutdown(wait=True, cancel_futures=True)
 
     return detectors_merged_resource

opentelemetry-sdk/tests/resources/test_resources.py

@@ -557,6 +557,68 @@ def slow_detect():
         self.assertEqual(len(worker_threads), 1)
         self.assertFalse(worker_threads[0].is_alive())
 
+    def test_get_aggregated_resources_synchronous(self):
+        resource_detector1 = Mock(spec=ResourceDetector)
+        resource_detector1.detect.return_value = Resource({"key1": "val1", "shared": "old"})
+        resource_detector1.raise_on_error = False
+
+        resource_detector2 = Mock(spec=ResourceDetector)
+        resource_detector2.detect.return_value = Resource({"key2": "val2", "shared": "new"})
+        resource_detector2.raise_on_error = False
+
+        res = get_aggregated_resources(
+            [resource_detector1, resource_detector2],
+            initial_resource=Resource({"initial": "init"}),
+            _use_thread_pool=False,
+        )
+        self.assertEqual(
+            res,
+            Resource({"initial": "init", "key1": "val1", "key2": "val2", "shared": "new"}),
+        )
+
+    @patch("concurrent.futures.ThreadPoolExecutor")
+    def test_get_aggregated_resources_synchronous_does_not_instantiate_thread_pool(self, mock_executor):
+        resource_detector = Mock(spec=ResourceDetector)
+        resource_detector.detect.return_value = Resource({"k": "v"})
+        resource_detector.raise_on_error = False
+
+        res = get_aggregated_resources(
+            [resource_detector],
+            initial_resource=Resource.get_empty(),
+            _use_thread_pool=False,
+        )
+        self.assertEqual(res, Resource({"k": "v"}))
+        mock_executor.assert_not_called()
+
+        res_default = get_aggregated_resources(
+            [resource_detector],
+            _use_thread_pool=False,
+        )
+        self.assertEqual(res_default.attributes["k"], "v")
+        mock_executor.assert_not_called()
+
+    def test_get_aggregated_resources_synchronous_error_logged(self):
+        resource_detector = Mock(spec=ResourceDetector)
+        resource_detector.detect.side_effect = RuntimeError("detector failure")
+        resource_detector.raise_on_error = False
+
+        with self.assertLogs(level=WARNING) as log_entry:
+            res = get_aggregated_resources(
+                [resource_detector],
+                initial_resource=Resource({"k": "v"}),
+                _use_thread_pool=False,
+            )
+            self.assertEqual(res, Resource({"k": "v"}))
+        self.assertIn("Exception detector failure in detector", log_entry.output[0])
+
+    def test_get_aggregated_resources_synchronous_raise_error(self):
+        resource_detector = Mock(spec=ResourceDetector)
+        resource_detector.detect.side_effect = RuntimeError("fatal error")
+        resource_detector.raise_on_error = True
+
+        with self.assertRaises(RuntimeError):
+            get_aggregated_resources([resource_detector], _use_thread_pool=False)
+
     def test_resource_detector_is_not_process_dependent_by_default(self):
         self.assertFalse(DefaultResourceDetector().is_process_dependent())
 

opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py

@@ -0,0 +1,24 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+import signal
+import subprocess
+import sys
+
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+
+
+def main() -> None:
+    _provider = TracerProvider(resource=Resource({"service.name": "test"}))
+    # pylint: disable=subprocess-popen-preexec-fn
+    with subprocess.Popen(
+        ["true"],
+        preexec_fn=lambda: signal.signal(signal.SIGPIPE, signal.SIG_DFL),  # noqa: PLW1509
+    ) as proc:
+        proc.wait()
+        sys.exit(proc.returncode)
+
+
+if __name__ == "__main__":
+    main()

opentelemetry-sdk/tests/trace/test_trace.py

@@ -210,6 +210,25 @@ def test_tracer_provider_updates_process_dependent_resource_after_fork(
         self.assertEqual(child_payload["new_tracer_pid"], child_payload["child_pid"])
         self.assertEqual(child_payload["span_pid"], child_payload["child_pid"])
 
+    @unittest.skipUnless(
+        hasattr(os, "fork") and hasattr(os, "register_at_fork"),
+        "requires os.fork and os.register_at_fork",
+    )
+    def test_tracer_provider_fork_with_preexec_fn_does_not_deadlock(self):
+        script_path = Path(__file__).parent / "scripts" / "tracer_provider_fork_preexec.py"
+        result = subprocess.run(
+            [sys.executable, str(script_path)],
+            capture_output=True,
+            check=False,
+            text=True,
+            timeout=5,
+        )
+        self.assertEqual(
+            result.returncode,
+            0,
+            f"stdout: {result.stdout}\nstderr: {result.stderr}",
+        )
+
     def test_get_tracer_sdk(self):
         tracer_provider = trace.TracerProvider()
         tracer = tracer_provider.get_tracer(