Repository navigation
Reset fork reset once on BatchProcessor _at_fork_reinit - #5850
saiyam-jain-dev wants to merge 2 commits into
Conversation
|
|
Pull request dashboard statusWaiting on maintainers · refreshed 2026-11-01 11:15 UTC Merge when ready. Status above doesn't look right?
|
There was a problem hiding this comment.
🟡 Changes recommended
A concurrent fork-recovery race can discard queued telemetry and start duplicate workers, while PID-mocking tests can leave earlier worker threads running.
2 open issues
What changed in this PR
This PR addresses #5843 by resetting the shared BatchProcessor fork-recovery guard so repeated forks that bypass Python’s fork hooks can recover. This affects both batch span and log record processors.
Changes:
- Create a fresh
Onceguard after fork reinitialization. - Add a regression test that simulates two consecutive fork fallbacks.
| File | Description |
|---|---|
opentelemetry-sdk/tests/shared_internal/test_batch_processor.py |
Tests repeated fallback recovery using simulated PID changes. |
opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py |
Resets the recovery guard during fork reinitialization. |
Review effort: Balanced
| ) | ||
| self._worker_thread.start() | ||
| self._pid = os.getpid() | ||
| self._bsp_reset_once = Once() |
There was a problem hiding this comment.
Captured reset_once = self._bsp_reset_once before the PID check so concurrent callers seeing the PID mismatch execute on the same generation guard and don't re-trigger recovery. Also added test_fork_fallback_concurrent_emit_race_does_not_duplicate_reinit to cover this interleaving.
| 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() |
There was a problem hiding this comment.
Updated the fallback tests to patch threading.Thread so no real worker threads are stranded waiting during mocked PID changes, wrapped the execution in try...finally to guarantee shutdown() runs, and added assert mock_thread.call_count == 3 to verify worker lifecycle.
|
/dashboard route:reviewers |
|
@saiyam-jain-dev, your reviewer-routing request was recorded; this pull request has the approvals it needs and is now waiting on maintainers. The handoff remains active across pushes until newer actionable human feedback arrives. Top-level feedback through this request will not return; unresolved review threads remain open. |
…tests Signed-off-by: Saiyam Jain <jimisierra2@gmail.com>
7d98362 to
da2b162
Compare
Description
BatchProcessorhas a fallback for forks that bypassos.register_at_fork(e.g. native libraries callingfork()directly):Because
self._bsp_reset_onceis an instance ofOnce, it can only execute its target once over the lifetime of the object. When_at_fork_reinit()runs, it previously did not resetself._bsp_reset_once. A second bypassed fork in the same process lifetime would findself._bsp_reset_once._done == True, leavingself._pidstale, failing to spawn a new worker thread, and dropping subsequent exports.This PR re-initializes
self._bsp_reset_once = Once()inside_at_fork_reinit(), allowing subsequent fork fallbacks to trigger re-initialization as expected. A regression unit test is added totest_batch_processor.py.Fixes #5843
Type of change
Checklist: