diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 567bf41c347..9453a6c0688 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: @@ -199,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 4592f3146b9..0772f0993e9 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -332,6 +332,77 @@ 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() + 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 + + 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 + + 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): def test_duplicate_logs_filter_works(self):