diff --git a/.changelog/batchprocessor-race.fixed b/.changelog/batchprocessor-race.fixed new file mode 100644 index 00000000000..e6e58334184 --- /dev/null +++ b/.changelog/batchprocessor-race.fixed @@ -0,0 +1 @@ +Fixed a missed wakeup race condition in `BatchProcessor` that could cause up to 5 seconds of export latency when the batch queue fills. diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py index 3e2b8a263a4..a019bfbc52e 100644 --- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py +++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py @@ -154,6 +154,7 @@ 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) + self._worker_awaken.clear() if self._shutdown: break self._export( @@ -161,7 +162,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..d6232a77959 100644 --- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py +++ b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py @@ -141,6 +141,47 @@ def test_telemetry_flushed_before_shutdown_and_dropped_after_shutdown( batch_processor._batch_processor.emit(telemetry) exporter.export.assert_called_once() + def test_telemetry_not_delayed_by_wakeup_race_condition( + self, batch_processor_class, telemetry + ): + exporter = Mock() + batch_processor = batch_processor_class( + exporter, + max_queue_size=20, + max_export_batch_size=10, + schedule_delay_millis=5000, + export_timeout_millis=500, + ) + bp = batch_processor._batch_processor + + # We hook the _worker_awaken.clear method to simulate a burst of telemetry + # arriving exactly after _export has finished but before the signal is cleared. + original_clear = bp._worker_awaken.clear + + def hooked_clear(): + # Emit a full batch right before the event is cleared. + # This simulates a concurrent thread filling the queue and calling .set() + for _ in range(10): + bp.emit(telemetry) + original_clear() + + bp._worker_awaken.clear = hooked_clear + + # Trigger the first export loop + for _ in range(10): + bp.emit(telemetry) + + # Wait for the worker to process the first batch, hit our hook, and go back to sleep. + # The worker will sleep for schedule_delay_millis (5 seconds) if the bug is present. + # We wait just a little bit (0.5s) to give it a chance to export the second batch. + time.sleep(0.5) + + # If the bug is present, the second batch is stuck in the queue and export is only called once. + # If the bug is fixed, the second batch is exported immediately. + assert exporter.export.call_count == 2, "Race condition detected: missed wakeup signal" + + batch_processor.shutdown() + # pylint: disable=no-self-use def test_force_flush_flushes_telemetry( self, batch_processor_class, telemetry