Refactor BatchLogRecordProcessor

Author: DylanRussell

opentelemetry-sdk/src/opentelemetry/sdk/_logs/_internal/export/__init__.py

@@ -205,13 +205,11 @@ def __init__(
         BatchLogRecordProcessor._validate_arguments(
             max_queue_size, schedule_delay_millis, max_export_batch_size
         )
-
         self._exporter = exporter
         self._max_queue_size = max_queue_size
         self._schedule_delay = schedule_delay_millis / 1e3
         self._max_export_batch_size = max_export_batch_size
         # Not used. No way currently to pass timeout to export.
-        # TODO(https://github.com/open-telemetry/opentelemetry-python/issues/4555): figure out what this should do.
         self._export_timeout_millis = export_timeout_millis
         # Deque is thread safe.
         self._queue = collections.deque([], max_queue_size)
@@ -220,10 +218,9 @@ def __init__(
             target=self.worker,
             daemon=True,
         )
-
         self._shutdown = False
         self._export_lock = threading.Lock()
-        self._worker_awaken = threading.Event()
+        self._worker_sleep = threading.Event()
         self._worker_thread.start()
         if hasattr(os, "register_at_fork"):
             weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
@@ -238,15 +235,15 @@ def _should_export_batch(
         # Always continue to export while queue length exceeds max batch size.
         if len(self._queue) >= self._max_export_batch_size:
             return True
-        if batch_strategy is BatchLogExportStrategy.EXPORT_ALL:
+        if batch_strategy == BatchLogExportStrategy.EXPORT_ALL:
             return True
-        if batch_strategy is BatchLogExportStrategy.EXPORT_AT_LEAST_ONE_BATCH:
+        if batch_strategy == BatchLogExportStrategy.EXPORT_AT_LEAST_ONE_BATCH:
             return num_iterations == 0
         return False
 
     def _at_fork_reinit(self):
         self._export_lock = threading.Lock()
-        self._worker_awaken = threading.Event()
+        self._worker_sleep = threading.Event()
         self._queue.clear()
         self._worker_thread = threading.Thread(
             name="OtelBatchLogRecordProcessor",
@@ -261,15 +258,15 @@ def worker(self):
             # Lots of strategies in the spec for setting next timeout.
             # 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)
+            sleep_interrupted = self._worker_sleep.wait(self._schedule_delay)
             if self._shutdown:
                 break
             self._export(
                 BatchLogExportStrategy.EXPORT_WHILE_BATCH_EXCEEDS_THRESHOLD
                 if sleep_interrupted
                 else BatchLogExportStrategy.EXPORT_AT_LEAST_ONE_BATCH
             )
-            self._worker_awaken.clear()
+            self._worker_sleep.clear()
         self._export(BatchLogExportStrategy.EXPORT_ALL)
 
     def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
@@ -299,7 +296,7 @@ def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
 
     def emit(self, log_data: LogData) -> None:
         if self._shutdown:
-            _logger.info("Shutdown called, ignoring log.")
+            _logger.warning("Shutdown called, ignoring log.")
             return
         if self._pid != os.getpid():
             _BSP_RESET_ONCE.do_once(self._at_fork_reinit)
@@ -308,15 +305,15 @@ def emit(self, log_data: LogData) -> None:
             _logger.warning("Queue full, dropping log.")
         self._queue.appendleft(log_data)
         if len(self._queue) >= self._max_export_batch_size:
-            self._worker_awaken.set()
+            self._worker_sleep.set()
 
     def shutdown(self):
         if self._shutdown:
             return
         # Prevents emit and force_flush from further calling export.
         self._shutdown = True
         # Interrupts sleep in the worker, if it's sleeping.
-        self._worker_awaken.set()
+        self._worker_sleep.set()
         # Main worker loop should exit after one final export call with flush all strategy.
         self._worker_thread.join()
         self._exporter.shutdown()

opentelemetry-sdk/tests/logs/test_export.py

@@ -21,7 +21,7 @@
 import unittest
 import weakref
 from concurrent.futures import ThreadPoolExecutor
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, call, patch
 
 from opentelemetry._logs import SeverityNumber
 from opentelemetry.sdk import trace
@@ -46,7 +46,6 @@
 )
 from opentelemetry.sdk.resources import Resource as SDKResource
 from opentelemetry.sdk.util.instrumentation import InstrumentationScope
-from opentelemetry.test.concurrency_test import ConcurrencyTestBase
 from opentelemetry.trace import TraceFlags
 from opentelemetry.trace.span import INVALID_SPAN_CONTEXT
 
@@ -487,20 +486,24 @@ def test_logs_exported_once_batch_size_reached(self):
         exporter.export.assert_called_once()
         after_export = time.time_ns()
         # Shows the worker's 30 second sleep was interrupted within a second.
-        self.assertLess(after_export - before_export, 1e9)
+        self.assertTrue((after_export - before_export) < 1e9)
 
     # pylint: disable=no-self-use
     def test_logs_exported_once_schedule_delay_reached(self):
         exporter = Mock()
         log_record_processor = BatchLogRecordProcessor(
             exporter=exporter,
+            # Should not reach this during the test, instead export should be called when delay millis is hit.
             max_queue_size=15,
             max_export_batch_size=15,
             schedule_delay_millis=100,
         )
-        log_record_processor.emit(EMPTY_LOG)
-        time.sleep(0.2)
-        exporter.export.assert_called_once_with([EMPTY_LOG])
+        for _ in range(15):
+            log_record_processor.emit(EMPTY_LOG)
+            time.sleep(0.11)
+        exporter.export.assert_has_calls(
+            [call([EMPTY_LOG]) for _ in range(15)]
+        )
 
     def test_logs_flushed_before_shutdown_and_dropped_after_shutdown(self):
         exporter = Mock()
@@ -517,13 +520,13 @@ def test_logs_flushed_before_shutdown_and_dropped_after_shutdown(self):
         exporter.export.assert_called_once_with([EMPTY_LOG])
         self.assertTrue(exporter._stopped)
 
-        with self.assertLogs(level="INFO") as log:
+        with self.assertLogs(level="WARNING") as log:
             # This log should not be flushed.
             log_record_processor.emit(EMPTY_LOG)
             self.assertEqual(len(log.output), 1)
             self.assertEqual(len(log.records), 1)
             self.assertIn("Shutdown called, ignoring log.", log.output[0])
-        exporter.export.assert_called_once()
+        exporter.export.assert_called_once_with([EMPTY_LOG])
 
     # pylint: disable=no-self-use
     def test_force_flush_flushes_logs(self):
@@ -552,7 +555,6 @@ def bulk_log_and_flush(num_logs):
         with ThreadPoolExecutor(max_workers=69) as executor:
             for idx in range(69):
                 executor.submit(bulk_log_and_flush, idx + 1)
-
             executor.shutdown()
 
         finished_logs = exporter.get_finished_logs()
@@ -562,20 +564,16 @@ def bulk_log_and_flush(num_logs):
         hasattr(os, "fork"),
         "needs *nix",
     )
-    def test_batch_log_record_processor_fork_clears_logs_from_child(self):
+    def test_batch_log_record_processor_fork(self):
         exporter = InMemoryLogExporter()
         log_record_processor = BatchLogRecordProcessor(
             exporter,
             max_export_batch_size=64,
             schedule_delay_millis=30000,
         )
-        # These logs should be flushed only from the parent process.
-        # _at_fork_reinit should be called in the child process, to
-        # clear these logs in the child process.
+        # These are not expected to be flushed. Calling fork clears any logs not flushed.
         for _ in range(10):
             log_record_processor.emit(EMPTY_LOG)
-
-        # The below test also needs this, but it can only be set once.
         multiprocessing.set_start_method("fork")
 
         def child(conn):
@@ -605,10 +603,8 @@ def test_batch_log_record_processor_fork_doesnot_deadlock(self):
         )
 
         def child(conn):
-            def _target():
+            for _ in range(100):
                 log_record_processor.emit(EMPTY_LOG)
-
-            ConcurrencyTestBase.run_with_many_threads(_target, 100)
             log_record_processor.force_flush()
             logs = exporter.get_finished_logs()
             conn.send(len(logs) == 100)
@@ -619,6 +615,7 @@ def _target():
         process.start()
         self.assertTrue(parent_conn.recv())
         process.join()
+        self.assertTrue(len(exporter.get_finished_logs()) == 0)
 
     def test_batch_log_record_processor_gc(self):
         # Given a BatchLogRecordProcessor
@@ -680,5 +677,4 @@ def formatter(record):  # pylint: disable=unused-argument
         mock_stdout = Mock()
         exporter = ConsoleLogExporter(out=mock_stdout, formatter=formatter)
         exporter.export([EMPTY_LOG])
-
         mock_stdout.write.assert_called_once_with(mock_record_str)