Support OpenTelemetry 1.39.0, drop support for earlier versions, stop using the OTel events API/SDK (#1562)

Author: alexmojaki

.github/workflows/daily_deps_test.yml

@@ -45,15 +45,10 @@ jobs:
             pydantic-version: '2.11'
             otel-version: '1'
           # OpenTelemetry versions
-          - python-version: '3.13'
-            pydantic-version: 'main'
-            otel-version: '1.37'
-          - python-version: '3.13'
-            pydantic-version: 'main'
-            otel-version: '1.36'
-          - python-version: '3.13'
-            pydantic-version: 'main'
-            otel-version: '1.35'
+          # Uncomment when we support multiple versions of OTel again
+#          - python-version: '3.13'
+#            pydantic-version: 'main'
+#            otel-version: '1.39'
     env:
       PYTHON: ${{ matrix.python-version }}
     steps:

.github/workflows/main.yml

@@ -68,9 +68,10 @@ jobs:
             - python-version: '3.12'
               pydantic-version: '2.4'
               otel-version: '1'  # i.e. the latest
-            - python-version: '3.13'
-              pydantic-version: '2'  # i.e. the latest
-              otel-version: '1.35'
+            # Uncomment when we support multiple versions of OTel again
+#            - python-version: '3.13'
+#              pydantic-version: '2'  # i.e. the latest
+#              otel-version: '1.39'
     env:
       PYTHON: ${{ matrix.python-version }}
     steps:

logfire/_internal/__init__.py

@@ -3,3 +3,12 @@
 We use the `_internal` module to discourage imports from outside the package,
 and thereby avoid causing breaking changes when refactoring the package.
 """
+
+from opentelemetry.sdk.resources import Resource
+
+from logfire._internal.utils import platform_is_emscripten
+
+if platform_is_emscripten():  # pragma: no cover
+    # Resource.create starts a thread, which is not supported in Emscripten.
+    # We have to patch it early like this because it gets called just by importing OTel logs modules.
+    Resource.create = Resource  # type: ignore

logfire/_internal/config.py

@@ -115,8 +115,6 @@
 if TYPE_CHECKING:
     from typing import TextIO
 
-    from opentelemetry._events import EventLoggerProvider
-
     from .main import Logfire
 
 
@@ -721,12 +719,6 @@ def __init__(
         # thus it "shuts down" when it's gc'ed
         self._meter_provider = ProxyMeterProvider(NoOpMeterProvider())
         self._logger_provider = ProxyLoggerProvider(NoOpLoggerProvider())
-        try:
-            from opentelemetry.sdk._events import EventLoggerProvider as SDKEventLoggerProvider
-
-            self._event_logger_provider = SDKEventLoggerProvider(self._logger_provider)  # type: ignore
-        except ImportError:
-            self._event_logger_provider = None
         # This ensures that we only call OTEL's global set_tracer_provider once to avoid warnings.
         self._has_set_providers = False
         self._initialized = False
@@ -824,11 +816,7 @@ def _initialize(self) -> None:
             ):
                 otel_resource_attributes[RESOURCE_ATTRIBUTES_CODE_WORK_DIR] = os.getcwd()
 
-            if emscripten:  # pragma: no cover
-                # Resource.create creates a thread pool which fails in Pyodide / Emscripten
-                resource = Resource(otel_resource_attributes)
-            else:
-                resource = Resource.create(otel_resource_attributes)
+            resource = Resource.create(otel_resource_attributes)
 
             # Set service instance ID to a random UUID if it hasn't been set already.
             # Setting it above would have also mostly worked and allowed overriding via OTEL_RESOURCE_ATTRIBUTES,
@@ -1111,12 +1099,7 @@ def fix_pid():  # pragma: no cover
             )
             logger_provider = SDKLoggerProvider(resource)
             logger_provider.add_log_record_processor(root_log_processor)
-
-            if self._event_logger_provider:
-                # This also shuts down the underlying self._logger_provider
-                self._event_logger_provider.shutdown()
-            else:
-                self._logger_provider.shutdown()
+            self._logger_provider.shutdown()
 
             self._logger_provider.set_provider(logger_provider)
             self._logger_provider.set_min_level(self.min_level)
@@ -1126,10 +1109,6 @@ def fix_pid():  # pragma: no cover
                 trace.set_tracer_provider(self._tracer_provider)
                 set_meter_provider(self._meter_provider)
                 set_logger_provider(self._logger_provider)
-                if self._event_logger_provider:
-                    from opentelemetry._events import set_event_logger_provider
-
-                    set_event_logger_provider(self._event_logger_provider)
 
             @atexit.register
             def exit_open_spans():  # pragma: no cover
@@ -1190,8 +1169,6 @@ def force_flush(self, timeout_millis: int = 30_000) -> bool:
         """
         self._meter_provider.force_flush(timeout_millis)
         self._logger_provider.force_flush(timeout_millis)
-        if self._event_logger_provider:
-            self._event_logger_provider.force_flush(timeout_millis)
         return self._tracer_provider.force_flush(timeout_millis)
 
     def get_tracer_provider(self) -> ProxyTracerProvider:
@@ -1224,16 +1201,6 @@ def get_logger_provider(self) -> ProxyLoggerProvider:
         """
         return self._logger_provider
 
-    def get_event_logger_provider(self) -> EventLoggerProvider | None:
-        """Get an event logger provider from this `LogfireConfig`.
-
-        This is used internally and should not be called by users of the SDK.
-
-        Returns:
-            The event logger provider.
-        """
-        return self._event_logger_provider
-
     def warn_if_not_initialized(self, message: str):
         ignore_no_config_env = os.getenv('LOGFIRE_IGNORE_NO_CONFIG', '')
         ignore_no_config = ignore_no_config_env.lower() in ('1', 'true', 't') or self.ignore_no_config

logfire/_internal/exporters/console.py

@@ -14,8 +14,9 @@
 from textwrap import indent as indent_text
 from typing import Any, Literal, TextIO, cast
 
-from opentelemetry.sdk._logs import LogData, LogRecord
-from opentelemetry.sdk._logs.export import LogExporter, LogExportResult
+from opentelemetry._logs import LogRecord
+from opentelemetry.sdk._logs import ReadableLogRecord
+from opentelemetry.sdk._logs.export import LogRecordExporter, LogRecordExportResult
 from opentelemetry.sdk.trace import Event, ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
 from rich.columns import Columns
@@ -492,14 +493,14 @@ def _pending_span_parent(attributes: Mapping[str, object]) -> int | None:
 
 
 @dataclass
-class ConsoleLogExporter(LogExporter):
+class ConsoleLogExporter(LogRecordExporter):
     span_exporter: SimpleConsoleSpanExporter
 
-    def export(self, batch: Sequence[LogData]) -> LogExportResult:  # type: ignore
+    def export(self, batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult:
         for log_data in batch:
             self.span_exporter.export_record(Record.from_log(log_data.log_record))
 
-        return LogExportResult.SUCCESS
+        return LogRecordExportResult.SUCCESS
 
     def shutdown(self):
         self.span_exporter.shutdown()

logfire/_internal/exporters/logs.py

@@ -1,6 +1,6 @@
 from dataclasses import dataclass
 
-from opentelemetry.sdk._logs import LogData
+from opentelemetry.sdk._logs import ReadWriteLogRecord
 
 import logfire
 from logfire._internal.exporters.wrapper import WrapperLogProcessor
@@ -14,17 +14,17 @@ class CheckSuppressInstrumentationLogProcessorWrapper(WrapperLogProcessor):
     Placed at the root of the tree of processors.
     """
 
-    def on_emit(self, log_data: LogData):
+    def on_emit(self, log_record: ReadWriteLogRecord):
         if is_instrumentation_suppressed():
             return None
         with logfire.suppress_instrumentation():
-            return super().on_emit(log_data)
+            return super().on_emit(log_record)
 
 
 @dataclass
 class MainLogProcessorWrapper(WrapperLogProcessor):
     scrubber: BaseScrubber
 
-    def on_emit(self, log_data: LogData):
-        log_data.log_record = self.scrubber.scrub_log(log_data.log_record)
-        return super().on_emit(log_data)
+    def on_emit(self, log_record: ReadWriteLogRecord):
+        log_record.log_record = self.scrubber.scrub_log(log_record.log_record)
+        return super().on_emit(log_record)

logfire/_internal/exporters/otlp.py

@@ -13,8 +13,8 @@
 
 import requests.exceptions
 from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
-from opentelemetry.sdk._logs import LogData
-from opentelemetry.sdk._logs._internal.export import LogExportResult
+from opentelemetry.sdk._logs import ReadableLogRecord
+from opentelemetry.sdk._logs._internal.export import LogRecordExportResult
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExportResult
 from requests import Session
@@ -263,9 +263,9 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
 class QuietLogExporter(WrapperLogExporter):
     """A LogExporter that catches request exceptions to prevent OTEL from logging a huge traceback."""
 
-    def export(self, batch: Sequence[LogData]):
+    def export(self, batch: Sequence[ReadableLogRecord]):
         try:
             return super().export(batch)
         except requests.exceptions.RequestException:
             # Rely on OTLPExporterHttpSession/DiskRetryer to log this kind of error periodically.
-            return LogExportResult.FAILURE
+            return LogRecordExportResult.FAILURE

logfire/_internal/exporters/test.py

@@ -12,9 +12,9 @@
 from typing import Any, cast
 
 from opentelemetry import trace
-from opentelemetry.sdk._logs import LogData
-from opentelemetry.sdk._logs._internal.export import LogExportResult
-from opentelemetry.sdk._logs.export import InMemoryLogExporter
+from opentelemetry.sdk._logs import ReadableLogRecord
+from opentelemetry.sdk._logs._internal.export import LogRecordExportResult
+from opentelemetry.sdk._logs.export import InMemoryLogRecordExporter
 from opentelemetry.sdk.trace import Event, ReadableSpan
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
 
@@ -178,7 +178,7 @@ def build_attributes(
     return attributes
 
 
-class TestLogExporter(InMemoryLogExporter):
+class TestLogExporter(InMemoryLogRecordExporter):
     """A LogExporter that stores exported logs in a list for asserting in tests."""
 
     # NOTE: Avoid test discovery by pytest.
@@ -188,7 +188,7 @@ def __init__(self, ns_timestamp_generator: typing.Callable[[], int]) -> None:
         super().__init__()
         self.ns_timestamp_generator = ns_timestamp_generator
 
-    def export(self, batch: typing.Sequence[LogData]) -> LogExportResult:
+    def export(self, batch: typing.Sequence[ReadableLogRecord]) -> LogRecordExportResult:
         for log in batch:
             log.log_record.timestamp = self.ns_timestamp_generator()
             log.log_record.observed_timestamp = self.ns_timestamp_generator()
@@ -211,7 +211,7 @@ def exported_logs_as_dicts(
             parse_json_attributes=parse_json_attributes,
         )
 
-        def build_log(log_data: LogData) -> dict[str, Any]:
+        def build_log(log_data: ReadableLogRecord) -> dict[str, Any]:
             log_record = log_data.log_record
             res = {
                 'body': log_record.body,
@@ -226,12 +226,12 @@ def build_log(log_data: LogData) -> dict[str, Any]:
             }
 
             if include_resources:  # pragma: no branch
-                resource_attributes = _build_attributes(log_record.resource.attributes)
+                resource_attributes = _build_attributes(log_data.resource.attributes)
                 res['resource'] = {
                     'attributes': resource_attributes,
                 }
 
-            if include_instrumentation_scope:  # pragma: no branch
+            if include_instrumentation_scope and log_data.instrumentation_scope:  # pragma: no branch
                 res['instrumentation_scope'] = log_data.instrumentation_scope.name
 
             return res

logfire/_internal/exporters/wrapper.py

@@ -2,11 +2,11 @@
 
 from collections.abc import Sequence
 from dataclasses import dataclass
-from typing import Any, cast
+from typing import Any
 
 from opentelemetry import context
-from opentelemetry.sdk._logs import LogData, LogRecordProcessor
-from opentelemetry.sdk._logs.export import LogExporter, LogExportResult
+from opentelemetry.sdk._logs import LogRecordProcessor, ReadableLogRecord, ReadWriteLogRecord
+from opentelemetry.sdk._logs.export import LogRecordExporter, LogRecordExportResult
 from opentelemetry.sdk.metrics.export import AggregationTemporality, MetricExporter, MetricExportResult, MetricsData
 from opentelemetry.sdk.metrics.view import Aggregation
 from opentelemetry.sdk.trace import ReadableSpan, Span, SpanProcessor
@@ -75,13 +75,13 @@ def force_flush(self, timeout_millis: int = 30000) -> bool:
 
 
 @dataclass
-class WrapperLogExporter(LogExporter):
+class WrapperLogExporter(LogRecordExporter):
     """A base class for LogExporters that wrap another exporter."""
 
-    exporter: LogExporter
+    exporter: LogRecordExporter
 
-    def export(self, batch: Sequence[LogData]) -> LogExportResult:  # type: ignore
-        return cast(LogExportResult, self.exporter.export(batch))
+    def export(self, batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult:
+        return self.exporter.export(batch)
 
     def shutdown(self):
         return self.exporter.shutdown()
@@ -93,8 +93,8 @@ class WrapperLogProcessor(LogRecordProcessor):
 
     processor: LogRecordProcessor
 
-    def on_emit(self, log_data: LogData) -> None:
-        return self.processor.on_emit(log_data)
+    def on_emit(self, log_record: ReadWriteLogRecord) -> None:
+        return self.processor.on_emit(log_record)
 
     def shutdown(self):
         return self.processor.shutdown()

logfire/_internal/integrations/pydantic_ai.py

@@ -29,7 +29,7 @@ def instrument_pydantic_ai(
         for k, v in dict(
             tracer_provider=logfire_instance.config.get_tracer_provider(),
             meter_provider=logfire_instance.config.get_meter_provider(),  # not in old versions
-            event_logger_provider=logfire_instance.config.get_event_logger_provider(),  # may be removed in the future
+            logger_provider=logfire_instance.config.get_logger_provider(),
         ).items()
         if k in expected_kwarg_names
     }