logs: remove context from LogRecord and pass to on_emit

- Remove context attribute from LogRecord to prevent memory inflation.
- Update  signature to accept context separately.
- Update  and SDK processors to align with the new signature.
- Update tests to reflect the breaking changes.

This aligns the Python SDK with the OpenTelemetry specification and
resolves memory usage concerns in BatchLogRecordProcessor.

Author: ajuijas

opentelemetry-api/src/opentelemetry/_logs/_internal/__init__.py

@@ -118,7 +118,6 @@ def __init__(
         if observed_timestamp is None:
             observed_timestamp = time_ns()
         self.observed_timestamp = observed_timestamp
-        self.context = context
         self.trace_id = trace_id or span_context.trace_id
         self.span_id = span_id or span_context.span_id
         self.trace_flags = trace_flags or span_context.trace_flags

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

@@ -325,7 +325,7 @@ def on_emit(self, log_record: ReadWriteLogRecord):
     """
 
     @abc.abstractmethod
-    def on_emit(self, log_record: ReadWriteLogRecord):
+    def on_emit(self, log_record: ReadWriteLogRecord, context: Context):
         """Emits the ``ReadWriteLogRecord``.
 
         Implementers should handle any exceptions raised during log processing
@@ -374,9 +374,9 @@ def add_log_record_processor(
         with self._lock:
             self._log_record_processors += (log_record_processor,)
 
-    def on_emit(self, log_record: ReadWriteLogRecord) -> None:
+    def on_emit(self, log_record: ReadWriteLogRecord, context: Context) -> None:
         for lp in self._log_record_processors:
-            lp.on_emit(log_record)
+            lp.on_emit(log_record, context)
 
     def shutdown(self) -> None:
         """Shutdown the log processors one by one"""
@@ -448,8 +448,8 @@ def _submit_and_wait(
         for future in futures:
             future.result()
 
-    def on_emit(self, log_record: ReadWriteLogRecord):
-        self._submit_and_wait(lambda lp: lp.on_emit, log_record)
+    def on_emit(self, log_record: ReadWriteLogRecord, context: Context):
+        self._submit_and_wait(lambda lp: lp.on_emit, log_record, context)
 
     def shutdown(self):
         self._submit_and_wait(lambda lp: lp.shutdown)
@@ -675,6 +675,9 @@ def emit(
         """Emits the :class:`ReadWriteLogRecord` by setting instrumentation scope
         and forwarding to the processor.
         """
+        if context is None:
+            context = get_current()
+
         # If a record is provided, use it directly
         if record is not None:
             if not isinstance(record, ReadWriteLogRecord):
@@ -706,7 +709,7 @@ def emit(
             )
 
         self._logger_metrics.emit_log()
-        self._multi_log_record_processor.on_emit(writable_record)
+        self._multi_log_record_processor.on_emit(writable_record, context)
 
 
 class LoggerProvider(APILoggerProvider):

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

@@ -174,7 +174,7 @@ def __init__(self, exporter: LogRecordExporter):
         self._exporter = exporter
         self._shutdown = False
 
-    def on_emit(self, log_record: ReadWriteLogRecord):
+    def on_emit(self, log_record: ReadWriteLogRecord, context: Context):
         # Prevent entering a recursive loop.
         cnt = get_value(_ON_EMIT_RECURSION_COUNT_KEY) or 0
         # Recursive depth of 3 is sort of arbitrary. It's possible that an Exporter.export call
@@ -278,7 +278,7 @@ def __init__(
             "Log",
         )
 
-    def on_emit(self, log_record: ReadWriteLogRecord) -> None:
+    def on_emit(self, log_record: ReadWriteLogRecord, context: Context) -> None:
         # Convert ReadWriteLogRecord to ReadableLogRecord before passing to BatchProcessor
         # Note: resource should not be None at this point as it's set during Logger.emit()
         resource = (

opentelemetry-sdk/tests/logs/test_export.py

@@ -27,6 +27,7 @@
 from pytest import mark
 
 from opentelemetry._logs import LogRecord, SeverityNumber
+from opentelemetry.context import get_current
 from opentelemetry.sdk import trace
 from opentelemetry.sdk._logs import (
     LoggerProvider,
@@ -433,7 +434,7 @@ def test_with_multiple_threads(self):  # pylint: disable=no-self-use
 
         def bulk_emit(num_emit):
             for _ in range(num_emit):
-                batch_processor.on_emit(EMPTY_LOG)
+                batch_processor.on_emit(EMPTY_LOG, get_current())
 
         total_expected_logs = 0
         with ThreadPoolExecutor(max_workers=69) as executor:

opentelemetry-sdk/tests/logs/test_handler.py

@@ -32,8 +32,11 @@
 from opentelemetry.semconv.attributes import exception_attributes
 from opentelemetry.trace import (
     INVALID_SPAN_CONTEXT,
+    SpanContext,
+    TraceFlags,
     set_span_in_context,
 )
+from opentelemetry.context.context import Context
 
 
 # pylint: disable=too-many-public-methods
@@ -336,6 +339,7 @@ def test_log_record_trace_correlation(self):
                     logger.critical("Critical message within span")
 
                 record = processor.get_log_record(0)
+                context = processor.contexts_emitted[0]
 
                 self.assertEqual(
                     record.log_record.body,
@@ -346,7 +350,7 @@ def test_log_record_trace_correlation(self):
                     record.log_record.severity_number,
                     SeverityNumber.FATAL,
                 )
-                self.assertEqual(record.log_record.context, mock_context)
+                self.assertEqual(context, mock_context)
                 span_context = span.get_span_context()
                 self.assertEqual(
                     record.log_record.trace_id, span_context.trace_id
@@ -575,9 +579,11 @@ def set_up_test_logging(level, formatter=None, root_logger=False):
 class FakeProcessor(LogRecordProcessor):
     def __init__(self):
         self.log_data_emitted = []
+        self.contexts_emitted = []
 
-    def on_emit(self, log_record: ReadableLogRecord):
+    def on_emit(self, log_record: ReadableLogRecord, context: Context):
         self.log_data_emitted.append(log_record)
+        self.contexts_emitted.append(context)
 
     def shutdown(self):
         pass

opentelemetry-sdk/tests/logs/test_log_record.py

@@ -221,7 +221,6 @@ def test_log_record_from_api_log_record(self):
 
         self.assertEqual(record.log_record.timestamp, 1)
         self.assertEqual(record.log_record.observed_timestamp, 2)
-        self.assertEqual(record.log_record.context, get_current())
         # trace_id, span_id, and trace_flags come from the context's span
         self.assertEqual(record.log_record.trace_id, 0)
         self.assertEqual(record.log_record.span_id, 0)

opentelemetry-sdk/tests/logs/test_logs.py

@@ -163,9 +163,10 @@ def test_can_emit_logrecord(self):
 
         logger.emit(log_record)
         log_record_processor_mock.on_emit.assert_called_once()
-        log_data = log_record_processor_mock.on_emit.call_args.args[0]
+        log_data, context = log_record_processor_mock.on_emit.call_args.args
         self.assertTrue(isinstance(log_data.log_record, LogRecord))
         self.assertTrue(log_data.log_record is log_record)
+        self.assertIsNotNone(context)
 
     def test_can_emit_api_logrecord(self):
         logger, log_record_processor_mock = self._get_logger()
@@ -175,12 +176,12 @@ def test_can_emit_api_logrecord(self):
         )
         logger.emit(api_log_record)
         log_record_processor_mock.on_emit.assert_called_once()
-        log_data = log_record_processor_mock.on_emit.call_args.args[0]
+        log_data, context = log_record_processor_mock.on_emit.call_args.args
         log_record = log_data.log_record
         self.assertTrue(isinstance(log_record, LogRecord))
         self.assertEqual(log_record.timestamp, None)
         self.assertEqual(log_record.observed_timestamp, 0)
-        self.assertIsNotNone(log_record.context)
+        self.assertIsNotNone(context)
         self.assertEqual(log_record.severity_number, None)
         self.assertEqual(log_record.severity_text, None)
         self.assertEqual(log_record.body, "a log line")
@@ -203,12 +204,12 @@ def test_can_emit_with_keywords_arguments(self):
         )
         logger.emit(log_record)
         log_record_processor_mock.on_emit.assert_called_once()
-        log_data = log_record_processor_mock.on_emit.call_args.args[0]
+        log_data, context = log_record_processor_mock.on_emit.call_args.args
         result_log_record = log_data.log_record
         self.assertTrue(isinstance(result_log_record, LogRecord))
         self.assertEqual(result_log_record.timestamp, 100)
         self.assertEqual(result_log_record.observed_timestamp, 101)
-        self.assertIsNotNone(result_log_record.context)
+        self.assertIsNotNone(context)
         self.assertEqual(
             result_log_record.severity_number, SeverityNumber.WARN
         )

opentelemetry-sdk/tests/logs/test_multi_log_processor.py

@@ -22,6 +22,7 @@
 from unittest.mock import Mock
 
 from opentelemetry._logs import LogRecord, SeverityNumber
+from opentelemetry.context import Context, get_current
 from opentelemetry.sdk._logs._internal import (
     ConcurrentMultiLogRecordProcessor,
     LoggerProvider,
@@ -38,7 +39,7 @@ def __init__(self, exporter, logs_list):
         self._log_list = logs_list
         self._closed = False
 
-    def on_emit(self, log_record: ReadWriteLogRecord):
+    def on_emit(self, log_record: ReadWriteLogRecord, context: Context):
         if self._closed:
             return
         self._log_list.append(
@@ -123,9 +124,10 @@ def test_on_emit(self):
         for mock in mocks:
             multi_log_record_processor.add_log_record_processor(mock)
         record = self.make_record()
-        multi_log_record_processor.on_emit(record)
+        context = get_current()
+        multi_log_record_processor.on_emit(record, context)
         for mock in mocks:
-            mock.on_emit.assert_called_with(record)
+            mock.on_emit.assert_called_with(record, context)
         multi_log_record_processor.shutdown()
 
     def test_on_shutdown(self):