diff --git a/.changelog/5545.fixed b/.changelog/5545.fixed new file mode 100644 index 00000000000..17cb1cc67ad --- /dev/null +++ b/.changelog/5545.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: bound `get_aggregated_resources()` wait to the timeout diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py index a84c9f25fe9..7e711251414 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py +++ b/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 diff --git a/opentelemetry-sdk/tests/resources/test_resources.py b/opentelemetry-sdk/tests/resources/test_resources.py index 282611dd1e3..ca2f0746cd4 100644 --- a/opentelemetry-sdk/tests/resources/test_resources.py +++ b/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())