diff --git a/.changelog/5706.fixed b/.changelog/5706.fixed new file mode 100644 index 00000000000..56bc9fd10b3 --- /dev/null +++ b/.changelog/5706.fixed @@ -0,0 +1 @@ +`opentelemetry-sdk`: fix thread leak bug in resource initialization logic for long running resource detectors diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py index 7c88f6145c5..9fa25576b9d 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py @@ -743,12 +743,10 @@ def get_aggregated_resources( """ detectors_merged_resource = initial_resource or Resource.create() - # 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. + # 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) try: futures = [executor.submit(detector.detect) for detector in detectors] @@ -773,6 +771,6 @@ def get_aggregated_resources( finally: detectors_merged_resource = detectors_merged_resource.merge(detected_resource) finally: - executor.shutdown(wait=False, cancel_futures=True) + 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 063770854c5..8b8612c8ff4 100644 --- a/opentelemetry-sdk/tests/resources/test_resources.py +++ b/opentelemetry-sdk/tests/resources/test_resources.py @@ -7,6 +7,7 @@ import platform import subprocess import sys +import threading import time import unittest import uuid @@ -472,14 +473,20 @@ 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. + def test_aggregated_resources_joins_detector_threads(self): + # A detector that takes longer than the timeout is skipped, but its + # worker thread must be joined before the call returns so no threads + # outlive resource creation. + worker_threads = [] + + def slow_detect(): + worker_threads.append(threading.current_thread()) + time.sleep(0.5) + resource_detector = Mock(spec=ResourceDetector) - resource_detector.detect.side_effect = lambda: time.sleep(2) + resource_detector.detect.side_effect = slow_detect resource_detector.raise_on_error = False - start = time.time() with self.assertLogs(level=WARNING) as log_entry: self.assertEqual( get_aggregated_resources( @@ -489,10 +496,10 @@ def test_aggregated_resources_timeout_bounds_wait(self): ), _DEFAULT_RESOURCE, ) - elapsed = time.time() - start - self.assertLess(elapsed, 1.0) self.assertIn("took longer than", log_entry.output[0]) + self.assertEqual(len(worker_threads), 1) + self.assertFalse(worker_threads[0].is_alive()) def test_resource_detector_is_not_process_dependent_by_default(self): self.assertFalse(DefaultResourceDetector().is_process_dependent())