diff --git a/.changelog/5790.fixed b/.changelog/5790.fixed new file mode 100644 index 00000000000..d368c241a7f --- /dev/null +++ b/.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 diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py index 4c2a484b004..1e7397f5c2f 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py +++ b/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,6 +751,7 @@ 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, ) @@ -753,6 +759,8 @@ 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 diff --git a/opentelemetry-sdk/tests/resources/test_resources.py b/opentelemetry-sdk/tests/resources/test_resources.py index db67f9bbc5e..47b3ed5c3eb 100644 --- a/opentelemetry-sdk/tests/resources/test_resources.py +++ b/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()) diff --git a/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py b/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py new file mode 100644 index 00000000000..cd575dbaa53 --- /dev/null +++ b/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() diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py index f5eb7128909..0703d6a3647 100644 --- a/opentelemetry-sdk/tests/trace/test_trace.py +++ b/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(