Respond to comments

Author: DylanRussell

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

@@ -199,6 +199,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)
@@ -210,7 +211,7 @@ def __init__(
 
         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"):
             os.register_at_fork(after_in_child=self._at_fork_reinit)  # pylint: disable=protected-access
@@ -224,15 +225,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",
@@ -247,15 +248,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:
@@ -285,7 +286,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)
@@ -294,15 +295,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

@@ -485,24 +485,19 @@ 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.11)
+        exporter.export.assert_called_once_with([EMPTY_LOG])
 
     def test_logs_flushed_before_shutdown_and_dropped_after_shutdown(self):
         exporter = Mock()
@@ -516,15 +511,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):
@@ -570,7 +566,9 @@ def test_batch_log_record_processor_fork(self):
             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)
 
@@ -590,7 +588,8 @@ def child(conn):
         process.start()
         self.assertTrue(parent_conn.recv())
         process.join()
-        self.assertTrue(len(exporter.get_finished_logs()) == 0)
+        log_record_processor.force_flush()
+        self.assertTrue(len(exporter.get_finished_logs()) == 10)
 
 
 class TestConsoleLogExporter(unittest.TestCase):