Refactor BatchLogRecordProcessor and associated tests (open-telemetry#4535)

Author: DylanRussell

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

@@ -210,6 +210,7 @@ def __init__(
         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)
@@ -218,9 +219,10 @@ def __init__(
             target=self.worker,
             daemon=True,
         )
+
         self._shutdown = False
         self._export_lock = threading.Lock()
-        self._worker_sleep = threading.Event()
+        self._worker_awaken = threading.Event()
         self._worker_thread.start()
         if hasattr(os, "register_at_fork"):
             weak_reinit = weakref.WeakMethod(self._at_fork_reinit)
@@ -235,15 +237,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 == BatchLogExportStrategy.EXPORT_ALL:
+        if batch_strategy is BatchLogExportStrategy.EXPORT_ALL:
             return True
-        if batch_strategy == BatchLogExportStrategy.EXPORT_AT_LEAST_ONE_BATCH:
+        if batch_strategy is BatchLogExportStrategy.EXPORT_AT_LEAST_ONE_BATCH:
             return num_iterations == 0
         return False
 
     def _at_fork_reinit(self):
         self._export_lock = threading.Lock()
-        self._worker_sleep = threading.Event()
+        self._worker_awaken = threading.Event()
         self._queue.clear()
         self._worker_thread = threading.Thread(
             name="OtelBatchLogRecordProcessor",
@@ -258,15 +260,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_sleep.wait(self._schedule_delay)
+            sleep_interrupted = self._worker_awaken.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_sleep.clear()
+            self._worker_awaken.clear()
         self._export(BatchLogExportStrategy.EXPORT_ALL)
 
     def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
@@ -296,7 +298,7 @@ def _export(self, batch_strategy: BatchLogExportStrategy) -> None:
 
     def emit(self, log_data: LogData) -> None:
         if self._shutdown:
-            _logger.warning("Shutdown called, ignoring log.")
+            _logger.info("Shutdown called, ignoring log.")
             return
         if self._pid != os.getpid():
             _BSP_RESET_ONCE.do_once(self._at_fork_reinit)
@@ -305,15 +307,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_sleep.set()
+            self._worker_awaken.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_sleep.set()
+        self._worker_awaken.set()
         # Main worker loop should exit after one final export call with flush all strategy.
         self._worker_thread.join()
         self._exporter.shutdown()

opentelemetry-sdk/src/opentelemetry/sdk/environment_variables/__init__.py

@@ -87,6 +87,7 @@
 .. envvar:: OTEL_BLRP_EXPORT_TIMEOUT
 
 The :envvar:`OTEL_BLRP_EXPORT_TIMEOUT` represents the maximum allowed time to export data from the BatchLogRecordProcessor.
+This environment variable currently does nothing, see https://github.com/open-telemetry/opentelemetry-python/issues/4555.
 Default: 30000
 """
 

opentelemetry-sdk/tests/logs/test_export.py

@@ -486,24 +486,20 @@ 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.assertTrue((after_export - before_export) < 1e9)
+        self.assertLess(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,
         )
-        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)]
-        )
+        log_record_processor.emit(EMPTY_LOG)
+        time.sleep(0.2)
+        exporter.export.assert_called_once_with([EMPTY_LOG])
 
     def test_logs_flushed_before_shutdown_and_dropped_after_shutdown(self):
         exporter = Mock()
@@ -517,15 +513,16 @@ def test_logs_flushed_before_shutdown_and_dropped_after_shutdown(self):
         # This log should be flushed because it was written before shutdown.
         log_record_processor.emit(EMPTY_LOG)
         log_record_processor.shutdown()
+        exporter.export.assert_called_once_with([EMPTY_LOG])
         self.assertTrue(exporter._stopped)
 
-        with self.assertLogs(level="WARNING") as log:
+        with self.assertLogs(level="INFO") 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_with([EMPTY_LOG])
+        exporter.export.assert_called_once()
 
     # pylint: disable=no-self-use
     def test_force_flush_flushes_logs(self):
@@ -554,6 +551,7 @@ 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()
@@ -563,21 +561,53 @@ def bulk_log_and_flush(num_logs):
         hasattr(os, "fork"),
         "needs *nix",
     )
-    def test_batch_log_record_processor_fork(self):
+    def test_batch_log_record_processor_fork_clears_logs_from_child(self):
         exporter = InMemoryLogExporter()
         log_record_processor = BatchLogRecordProcessor(
             exporter,
             max_export_batch_size=64,
             schedule_delay_millis=30000,
         )
-        # These are not expected to be flushed. Calling fork clears any logs not flushed.
+        # 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.
         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):
-            for _ in range(100):
+            log_record_processor.force_flush()
+            logs = exporter.get_finished_logs()
+            conn.send(len(logs) == 0)
+            conn.close()
+
+        parent_conn, child_conn = multiprocessing.Pipe()
+        process = multiprocessing.Process(target=child, args=(child_conn,))
+        process.start()
+        self.assertTrue(parent_conn.recv())
+        process.join()
+        log_record_processor.force_flush()
+        self.assertTrue(len(exporter.get_finished_logs()) == 10)
+
+    @unittest.skipUnless(
+        hasattr(os, "fork"),
+        "needs *nix",
+    )
+    def test_batch_log_record_processor_fork_doesnot_deadlock(self):
+        exporter = InMemoryLogExporter()
+        log_record_processor = BatchLogRecordProcessor(
+            exporter,
+            max_export_batch_size=64,
+            schedule_delay_millis=30000,
+        )
+
+        def child(conn):
+            def _target():
                 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)
@@ -588,7 +618,6 @@ def child(conn):
         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
@@ -650,4 +679,5 @@ 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)