Make BatchProcessor a member of BLRP instead of having BLRP subclass it

Author: DylanRussell

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

@@ -18,7 +18,7 @@
 import logging
 import sys
 from os import environ, linesep
-from typing import IO, Callable, Sequence
+from typing import IO, Callable, Optional, Sequence
 
 from opentelemetry.context import (
     _SUPPRESS_INSTRUMENTATION_KEY,
@@ -131,7 +131,7 @@ def force_flush(self, timeout_millis: int = 30000) -> bool:  # pylint: disable=n
         return True
 
 
-class BatchLogRecordProcessor(BatchProcessor, LogRecordProcessor):
+class BatchLogRecordProcessor(LogRecordProcessor):
     """This is an implementation of LogRecordProcessor which creates batches of
     received logs in the export-friendly LogData representation and
     send to the configured LogExporter, as soon as they are emitted.
@@ -177,7 +177,7 @@ def __init__(
             max_queue_size, schedule_delay_millis, max_export_batch_size
         )
         # Initializes BatchProcessor
-        super().__init__(
+        self._batch_processor = BatchProcessor(
             exporter,
             schedule_delay_millis,
             max_export_batch_size,
@@ -186,6 +186,15 @@ def __init__(
             "Log",
         )
 
+    def emit(self, log_data: LogData) -> None:
+        return self._batch_processor.emit(log_data)
+
+    def shutdown(self):
+        return self._batch_processor.shutdown()
+
+    def force_flush(self, timeout_millis: Optional[int] = None):
+        return self._batch_processor.force_flush(timeout_millis)
+
     @staticmethod
     def _default_max_queue_size():
         try:

opentelemetry-sdk/tests/logs/test_export.py

@@ -351,11 +351,11 @@ def test_args(self):
             max_export_batch_size=256,
             export_timeout_millis=15000,
         )
-        self.assertEqual(log_record_processor._exporter, exporter)
-        self.assertEqual(log_record_processor._max_queue_size, 1024)
-        self.assertEqual(log_record_processor._schedule_delay, 2.5)
-        self.assertEqual(log_record_processor._max_export_batch_size, 256)
-        self.assertEqual(log_record_processor._export_timeout_millis, 15000)
+        self.assertEqual(log_record_processor._batch_processor._exporter, exporter)
+        self.assertEqual(log_record_processor._batch_processor._max_queue_size, 1024)
+        self.assertEqual(log_record_processor._batch_processor._schedule_delay, 2.5)
+        self.assertEqual(log_record_processor._batch_processor._max_export_batch_size, 256)
+        self.assertEqual(log_record_processor._batch_processor._export_timeout_millis, 15000)
 
     @patch.dict(
         "os.environ",
@@ -369,20 +369,20 @@ def test_args(self):
     def test_env_vars(self):
         exporter = InMemoryLogExporter()
         log_record_processor = BatchLogRecordProcessor(exporter)
-        self.assertEqual(log_record_processor._exporter, exporter)
-        self.assertEqual(log_record_processor._max_queue_size, 1024)
-        self.assertEqual(log_record_processor._schedule_delay, 2.5)
-        self.assertEqual(log_record_processor._max_export_batch_size, 256)
-        self.assertEqual(log_record_processor._export_timeout_millis, 15000)
+        self.assertEqual(log_record_processor._batch_processor._exporter, exporter)
+        self.assertEqual(log_record_processor._batch_processor._max_queue_size, 1024)
+        self.assertEqual(log_record_processor._batch_processor._schedule_delay, 2.5)
+        self.assertEqual(log_record_processor._batch_processor._max_export_batch_size, 256)
+        self.assertEqual(log_record_processor._batch_processor._export_timeout_millis, 15000)
 
     def test_args_defaults(self):
         exporter = InMemoryLogExporter()
         log_record_processor = BatchLogRecordProcessor(exporter)
-        self.assertEqual(log_record_processor._exporter, exporter)
-        self.assertEqual(log_record_processor._max_queue_size, 2048)
-        self.assertEqual(log_record_processor._schedule_delay, 5)
-        self.assertEqual(log_record_processor._max_export_batch_size, 512)
-        self.assertEqual(log_record_processor._export_timeout_millis, 30000)
+        self.assertEqual(log_record_processor._batch_processor._exporter, exporter)
+        self.assertEqual(log_record_processor._batch_processor._max_queue_size, 2048)
+        self.assertEqual(log_record_processor._batch_processor._schedule_delay, 5)
+        self.assertEqual(log_record_processor._batch_processor._max_export_batch_size, 512)
+        self.assertEqual(log_record_processor._batch_processor._export_timeout_millis, 30000)
 
     @patch.dict(
         "os.environ",
@@ -398,11 +398,11 @@ def test_args_env_var_value_error(self):
         _logger.disabled = True
         log_record_processor = BatchLogRecordProcessor(exporter)
         _logger.disabled = False
-        self.assertEqual(log_record_processor._exporter, exporter)
-        self.assertEqual(log_record_processor._max_queue_size, 2048)
-        self.assertEqual(log_record_processor._schedule_delay, 5)
-        self.assertEqual(log_record_processor._max_export_batch_size, 512)
-        self.assertEqual(log_record_processor._export_timeout_millis, 30000)
+        self.assertEqual(log_record_processor._batch_processor._exporter, exporter)
+        self.assertEqual(log_record_processor._batch_processor._max_queue_size, 2048)
+        self.assertEqual(log_record_processor._batch_processor._schedule_delay, 5)
+        self.assertEqual(log_record_processor._batch_processor._max_export_batch_size, 512)
+        self.assertEqual(log_record_processor._batch_processor._export_timeout_millis, 30000)
 
     def test_args_none_defaults(self):
         exporter = InMemoryLogExporter()
@@ -413,11 +413,11 @@ def test_args_none_defaults(self):
             max_export_batch_size=None,
             export_timeout_millis=None,
         )
-        self.assertEqual(log_record_processor._exporter, exporter)
-        self.assertEqual(log_record_processor._max_queue_size, 2048)
-        self.assertEqual(log_record_processor._schedule_delay, 5)
-        self.assertEqual(log_record_processor._max_export_batch_size, 512)
-        self.assertEqual(log_record_processor._export_timeout_millis, 30000)
+        self.assertEqual(log_record_processor._batch_processor._exporter, exporter)
+        self.assertEqual(log_record_processor._batch_processor._max_queue_size, 2048)
+        self.assertEqual(log_record_processor._batch_processor._schedule_delay, 5)
+        self.assertEqual(log_record_processor._batch_processor._max_export_batch_size, 512)
+        self.assertEqual(log_record_processor._batch_processor._export_timeout_millis, 30000)
 
     def test_validation_negative_max_queue_size(self):
         exporter = InMemoryLogExporter()

opentelemetry-sdk/tests/shared_internal/test_batch_processor.py

@@ -103,7 +103,7 @@ def test_telemetry_flushed_before_shutdown_and_dropped_after_shutdown(
         batch_processor.emit(telemetry)
         batch_processor.shutdown()
         exporter.export.assert_called_once_with([telemetry])
-        assert batch_processor._shutdown is True
+        assert batch_processor._batch_processor._shutdown is True
 
         # This should not be flushed.
         batch_processor.emit(telemetry)