diff --git a/.changelog/5409.fixed b/.changelog/5409.fixed new file mode 100644 index 00000000000..e6ea86159ea --- /dev/null +++ b/.changelog/5409.fixed @@ -0,0 +1 @@ +Fix missed wakeup race in `BatchProcessor` that could delay span/log export by a full schedule delay when a batch filled while the worker was clearing its wakeup event diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 3e2b8a263a4..cefe41aae20 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -154,6 +154,12 @@ def worker(self): # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/sdk.md#batching-processor. # Shutdown will interrupt this sleep. Emit will interrupt this sleep only if the queue is bigger then threshold. sleep_interrupted = self._worker_awaken.wait(self._schedule_delay) + # Clear the event before exporting, not after. A wakeup signaled by + # `emit` between the export loop finishing and a late `clear` would + # be erased, leaving a full batch stuck in the queue for up to + # `schedule_delay`. Clearing first means any `set` that arrives + # while exporting is kept for the next `wait`. + self._worker_awaken.clear() if self._shutdown: break self._export( @@ -161,7 +167,6 @@ def worker(self): if sleep_interrupted else BatchExportStrategy.EXPORT_AT_LEAST_ONE_BATCH ) - self._worker_awaken.clear() self._export(BatchExportStrategy.EXPORT_ALL) def _export(self, batch_strategy: BatchExportStrategy) -> None: diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py index 02337757073..c0648b5d5c6 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -119,6 +119,56 @@ def test_telemetry_exported_once_schedule_delay_reached( exporter.export.assert_called_once_with([telemetry]) batch_processor.shutdown() + # pylint: disable=no-self-use + def test_wakeup_not_lost_when_batch_fills_during_event_clear( + self, batch_processor_class, telemetry + ): + # Regression test for a missed wakeup race: if `emit` filled a batch + # and set `_worker_awaken` after the worker's export loop drained the + # queue but before the worker cleared the event, the signal was erased + # and the batch sat in the queue for a full schedule delay. + exporter = Mock() + batch_processor = batch_processor_class( + exporter, + max_queue_size=100, + max_export_batch_size=10, + # Will not be reached during the test; if the wakeup is lost the + # asserts below fail long before this delay elapses. + schedule_delay_millis=30000, + export_timeout_millis=500, + ) + processor = batch_processor._batch_processor + original_clear = processor._worker_awaken.clear + raced = threading.Event() + + def emit_batch_then_clear(): + # Fill a full batch (which calls `_worker_awaken.set()`) at the + # exact point the worker is about to clear the event, simulating + # an application thread that raced with the worker. + if not raced.is_set(): + raced.set() + for _ in range(10): + processor.emit(telemetry) + original_clear() + + processor._worker_awaken.clear = emit_batch_then_clear + # Trigger the first export; the worker wakes, exports this batch, and + # then races with the batch emitted from the clear hook. + for _ in range(10): + processor.emit(telemetry) + + deadline = time.time() + 5 + while time.time() < deadline: + exported = sum( + len(call.args[0]) for call in exporter.export.call_args_list + ) + if exported == 20: + break + time.sleep(0.01) + assert exported == 20 + assert len(processor._queue) == 0 + batch_processor.shutdown() + def test_telemetry_flushed_before_shutdown_and_dropped_after_shutdown( self, batch_processor_class, telemetry ):