From 8a6475fd135ccea35dc79a0c0ecf4bb92d5557d2 Mon Sep 17 00:00:00 2001 From: Saiyam Jain Date: Wed, 28 Oct 2026 07:15:02 +0530 Subject: [PATCH 1/2] Reset fork reset once on BatchProcessor _at_fork_reinit --- .../sdk/_shared_internal/__init__.py | 1 + .../shared_internal/test_batch_processor.py | 25 +++++++++++++++++++ 2 files changed, 26 insertions(+) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 567bf41c347..c1ae46c378c 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -150,6 +150,7 @@ def _at_fork_reinit(self): ) self._worker_thread.start() self._pid = os.getpid() + self._bsp_reset_once = Once() def worker(self): while not self._shutdown: diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py index 4592f3146b9..77e682f483e 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -332,6 +332,31 @@ def test_shutdown_allows_1_export_to_finish(self, batch_processor_class, telemet assert exporter.sleep_interrupted is True assert 2 == exporter.num_export_calls + def test_repeated_fork_fallback_reinitializes(self, batch_processor_class, telemetry): + exporter = Mock() + processor = batch_processor_class( + exporter, + max_queue_size=200, + max_export_batch_size=10, + schedule_delay_millis=30000, + export_timeout_millis=500, + ) + bp = processor._batch_processor + initial_pid = bp._pid + + with unittest.mock.patch("os.getpid") as mock_getpid: + # First fork fallback + mock_getpid.return_value = initial_pid + 1 + bp.emit(telemetry) + assert bp._pid == initial_pid + 1 + + # Second fork fallback + mock_getpid.return_value = initial_pid + 2 + bp.emit(telemetry) + assert bp._pid == initial_pid + 2 + + processor.shutdown() + class TestCommonFuncs(unittest.TestCase): def test_duplicate_logs_filter_works(self): From da2b1622174d3f78676cd3c03749023d09e8c444 Mon Sep 17 00:00:00 2001 From: Saiyam Jain Date: Thu, 29 Oct 2026 13:01:18 +0530 Subject: [PATCH 2/2] fix: prevent concurrent fork reinit race and mock thread in fallback tests Signed-off-by: Saiyam Jain --- .../sdk/_shared_internal/__init__.py | 3 +- .../shared_internal/test_batch_processor.py | 86 ++++++++++++++----- 2 files changed, 68 insertions(+), 21 deletions(-) diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index c1ae46c378c..9453a6c0688 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -200,8 +200,9 @@ def emit(self, data: Telemetry) -> None: _logger.info("Shutdown called, ignoring %s.", self._exporting) self._metrics.drop_items(1, "already_shutdown") return + reset_once = self._bsp_reset_once if self._pid != os.getpid(): - self._bsp_reset_once.do_once(self._at_fork_reinit) + reset_once.do_once(self._at_fork_reinit) if len(self._queue) == self._max_queue_size: _logger.warning("Queue full, dropping %s.", self._exporting) self._metrics.drop_items(1) diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py index 77e682f483e..0772f0993e9 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -334,28 +334,74 @@ def test_shutdown_allows_1_export_to_finish(self, batch_processor_class, telemet def test_repeated_fork_fallback_reinitializes(self, batch_processor_class, telemetry): exporter = Mock() - processor = batch_processor_class( - exporter, - max_queue_size=200, - max_export_batch_size=10, - schedule_delay_millis=30000, - export_timeout_millis=500, - ) - bp = processor._batch_processor - initial_pid = bp._pid - - with unittest.mock.patch("os.getpid") as mock_getpid: - # First fork fallback - mock_getpid.return_value = initial_pid + 1 - bp.emit(telemetry) - assert bp._pid == initial_pid + 1 + with unittest.mock.patch( + "opentelemetry.sdk._shared_internal.threading.Thread" + ) as mock_thread: + processor = batch_processor_class( + exporter, + max_queue_size=200, + max_export_batch_size=10, + schedule_delay_millis=30000, + export_timeout_millis=500, + ) + bp = processor._batch_processor + initial_pid = bp._pid - # Second fork fallback - mock_getpid.return_value = initial_pid + 2 - bp.emit(telemetry) - assert bp._pid == initial_pid + 2 + try: + with unittest.mock.patch("os.getpid") as mock_getpid: + # First fork fallback + mock_getpid.return_value = initial_pid + 1 + bp.emit(telemetry) + assert bp._pid == initial_pid + 1 + + # Second fork fallback + mock_getpid.return_value = initial_pid + 2 + bp.emit(telemetry) + assert bp._pid == initial_pid + 2 + + assert mock_thread.call_count == 3 + finally: + processor.shutdown() + + def test_fork_fallback_concurrent_emit_race_does_not_duplicate_reinit( + self, batch_processor_class, telemetry + ): + exporter = Mock() + with unittest.mock.patch( + "opentelemetry.sdk._shared_internal.threading.Thread" + ) as mock_thread: + processor = batch_processor_class( + exporter, + max_queue_size=200, + max_export_batch_size=10, + schedule_delay_millis=30000, + export_timeout_millis=500, + ) + bp = processor._batch_processor + initial_pid = bp._pid + new_pid = initial_pid + 1 - processor.shutdown() + try: + # Capture the guard before the PID check to simulate a thread + # that checked the PID right before a concurrent emit completed recovery. + guard_before_fork = bp._bsp_reset_once + + with unittest.mock.patch("os.getpid", return_value=new_pid): + # Concurrent thread runs emit(), triggering _at_fork_reinit and enqueuing telemetry + bp.emit(telemetry) + assert len(bp._queue) == 1 + assert bp._pid == new_pid + assert bp._bsp_reset_once is not guard_before_fork + assert mock_thread.call_count == 2 + + # Resumed thread uses its captured guard instance: + guard_before_fork.do_once(bp._at_fork_reinit) + + # Ensure recovery is not re-executed: telemetry is preserved and no extra worker is spawned + assert len(bp._queue) == 1 + assert mock_thread.call_count == 2 + finally: + processor.shutdown() class TestCommonFuncs(unittest.TestCase):