fix(sdk): bound get_aggregated_resources wait to per-detector timeout

.changelog/5545.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: bound `get_aggregated_resources()` wait to the timeout

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

@@ -610,7 +610,14 @@ def get_aggregated_resources(
     """
     detectors_merged_resource = initial_resource or Resource.create()
 
-    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
+    # The executor is not used as a context manager: exiting it would call
+    # `shutdown(wait=True)` and join still-running workers, so a detector that
+    # blocks longer than `timeout` would hold up the whole call regardless of
+    # `future.result(timeout=...)`. Shut down without waiting instead so the
+    # timeout actually bounds this call. A detector that outlives the timeout
+    # continues in its worker thread until detect() returns.
+    executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
+    try:
         futures = [executor.submit(detector.detect) for detector in detectors]
         for detector_ind, future in enumerate(futures):
             detector = detectors[detector_ind]
@@ -632,5 +639,7 @@ def get_aggregated_resources(
                 logger.warning("Exception %s in detector %s, ignoring", ex, detector)
             finally:
                 detectors_merged_resource = detectors_merged_resource.merge(detected_resource)
+    finally:
+        executor.shutdown(wait=False, cancel_futures=True)
 
     return detectors_merged_resource

opentelemetry-sdk/tests/resources/test_resources.py

@@ -6,6 +6,7 @@
 import os
 import subprocess
 import sys
+import time
 import unittest
 import uuid
 from concurrent.futures import TimeoutError
@@ -457,6 +458,28 @@ def test_resource_detector_raise_error(self):
         resource_detector.raise_on_error = True
         self.assertRaises(Exception, get_aggregated_resources, [resource_detector])
 
+    def test_aggregated_resources_timeout_bounds_wait(self):
+        # A detector that takes longer than the timeout must not hold up the
+        # call beyond the timeout: the skip warning fires, but the wait is bounded.
+        resource_detector = Mock(spec=ResourceDetector)
+        resource_detector.detect.side_effect = lambda: time.sleep(2)
+        resource_detector.raise_on_error = False
+
+        start = time.time()
+        with self.assertLogs(level=WARNING) as log_entry:
+            self.assertEqual(
+                get_aggregated_resources(
+                    [resource_detector],
+                    initial_resource=_DEFAULT_RESOURCE,
+                    timeout=0.1,
+                ),
+                _DEFAULT_RESOURCE,
+            )
+        elapsed = time.time() - start
+
+        self.assertLess(elapsed, 1.0)
+        self.assertIn("took longer than", log_entry.output[0])
+
     def test_resource_detector_is_not_process_dependent_by_default(self):
         self.assertFalse(DefaultResourceDetector().is_process_dependent())