Test OpenTelemetry SDK 1.33 (#1124)

Author: alexmojaki

.github/workflows/main.yml

@@ -56,16 +56,21 @@ jobs:
           ALGOLIA_WRITE_API_KEY: ${{ secrets.ALGOLIA_WRITE_API_KEY }}
 
   test:
-    name: test on Python ${{ matrix.python-version }} and pydantic ${{ matrix.pydantic-version }}
+    name: test on Python ${{ matrix.python-version }}, pydantic ${{ matrix.pydantic-version }}, otel ${{ matrix.otel-version }}
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
       matrix:
         python-version: ["3.9", "3.10", "3.11", "3.12", "3.13"]
-        pydantic-version: ['2.11']
+        pydantic-version: ['2']  # i.e. the latest
+        otel-version: ['1']  # i.e. the latest
         include:
             - python-version: '3.12'
               pydantic-version: '2.4'
+              otel-version: '1'  # i.e. the latest
+            - python-version: '3.12'
+              pydantic-version: '2'  # i.e. the latest
+              otel-version: '1.33'
     env:
       PYTHON: ${{ matrix.python-version }}
     steps:
@@ -90,14 +95,18 @@ jobs:
         # installs the most recent patch on the minor version's track, ex 2.6.* -> 2.6.4
         run: uv pip install 'pydantic==${{ matrix.pydantic-version }}.*'
 
+      - name: Install OTel SDK ${{ matrix.otel-version }}
+        # installs the most recent patch on the minor version's track, ex 2.6.* -> 2.6.4
+        run: uv pip install 'opentelemetry-sdk==${{ matrix.otel-version }}.*'
+
       - run: mkdir coverage
       - run: uv run --no-sync coverage run -m pytest --junitxml=coverage/test-results.xml
         env:
-          COVERAGE_FILE: coverage/.coverage.${{ runner.os }}-py${{ matrix.python-version }}-pydantic-${{ matrix.pydantic-version }}
+          COVERAGE_FILE: coverage/.coverage.${{ runner.os }}-py${{ matrix.python-version }}-pydantic-${{ matrix.pydantic-version }}-otel-${{ matrix.otel-version }}
       - name: store coverage files
         uses: actions/upload-artifact@v4
         with:
-          name: coverage-py${{ matrix.python-version }}-pydantic-${{ matrix.pydantic-version }}
+          name: coverage-py${{ matrix.python-version }}-pydantic-${{ matrix.pydantic-version }}-otel-${{ matrix.otel-version }}
           path: coverage
           include-hidden-files: true
 

logfire/_internal/exporters/dynamic_batch.py

@@ -1,16 +1,17 @@
 from __future__ import annotations
 
 import os
-from typing import TYPE_CHECKING
 
 from opentelemetry.sdk.environment_variables import OTEL_BSP_SCHEDULE_DELAY
-from opentelemetry.sdk.trace import ReadableSpan, Span
+from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import BatchSpanProcessor, SpanExporter
 
 from logfire._internal.exporters.wrapper import WrapperSpanProcessor
 
-if TYPE_CHECKING:
+try:
     from opentelemetry.sdk._shared_internal import BatchProcessor
+except ImportError:
+    BatchProcessor = None
 
 
 class DynamicBatchSpanProcessor(WrapperSpanProcessor):
@@ -33,19 +34,40 @@ def __init__(self, exporter: SpanExporter) -> None:
     def on_end(self, span: ReadableSpan) -> None:
         self.num_processed += 1
         if self.num_processed == 10:
-            if hasattr(self.batch_processor, '_schedule_delay_millis'):
-                self.batch_processor._schedule_delay = self.final_delay / 1e3  # type: ignore
-            else:  # pragma: no cover
-                self.processor.schedule_delay_millis = self.final_delay  # type: ignore
+            self.schedule_delay_millis = self.final_delay
         super().on_end(span)
 
-    @property
-    def batch_processor(self) -> BatchSpanProcessor | BatchProcessor[Span]:
-        return getattr(self.processor, '_batch_processor', self.processor)
+    if BatchProcessor:
 
-    @property
-    def span_exporter(self) -> SpanExporter:
-        if isinstance(self.batch_processor, BatchSpanProcessor):  # pragma: no cover
-            return self.batch_processor.span_exporter  # type: ignore
-        else:
+        @property
+        def batch_processor(self):  # type: ignore
+            return self.processor._batch_processor  # type: ignore
+
+        @property
+        def span_exporter(self) -> SpanExporter:  # type: ignore
             return self.batch_processor._exporter  # type: ignore
+
+        @property
+        def schedule_delay_millis(self) -> float:  # type: ignore
+            return self.batch_processor._schedule_delay * 1000  # type: ignore
+
+        @schedule_delay_millis.setter
+        def schedule_delay_millis(self, value: float):  # type: ignore
+            self.batch_processor._schedule_delay = value / 1000  # type: ignore
+    else:
+
+        @property
+        def batch_processor(self):
+            return self.processor
+
+        @property
+        def span_exporter(self) -> SpanExporter:
+            return self.processor.span_exporter  # type: ignore
+
+        @property
+        def schedule_delay_millis(self) -> float:
+            return self.processor.schedule_delay_millis  # type: ignore
+
+        @schedule_delay_millis.setter
+        def schedule_delay_millis(self, value: float):
+            self.processor.schedule_delay_millis = value  # type: ignore

tests/exporters/test_dynamic_batch_span_processor.py

@@ -17,9 +17,10 @@ def test_dynamic_batch_span_processor(exporter: TestExporter, config_kwargs: dic
         logfire.info('test')
     assert processor.num_processed == 9
     assert isinstance(processor.processor, BatchSpanProcessor)
-    assert processor.batch_processor._schedule_delay == 0.1  # type: ignore
+    assert processor.schedule_delay_millis == 100
+
     logfire.info('test')
     assert processor.num_processed == 10
-    assert processor.batch_processor._schedule_delay == 0.5  # type: ignore
+    assert processor.schedule_delay_millis == 500
     logfire.force_flush()
     assert len(exporter.exported_spans) == 10

tests/test_configure.py

@@ -28,7 +28,7 @@
 from opentelemetry.propagators.composite import CompositePropagator
 from opentelemetry.sdk._logs import LogRecordProcessor
 from opentelemetry.sdk._logs._internal import SynchronousMultiLogRecordProcessor
-from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, SimpleLogRecordProcessor
+from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, LogExporter, SimpleLogRecordProcessor
 from opentelemetry.sdk.metrics.export import InMemoryMetricReader, PeriodicExportingMetricReader
 from opentelemetry.sdk.trace import ReadableSpan, SpanProcessor, SynchronousMultiSpanProcessor
 from opentelemetry.sdk.trace.export import (
@@ -572,6 +572,24 @@ def test_logfire_config_console_options() -> None:
         assert LogfireConfig().console == ConsoleOptions(verbose=False)
 
 
+def get_batch_span_exporter(processor: SpanProcessor) -> SpanExporter:
+    assert isinstance(processor, BatchSpanProcessor)
+    try:
+        exporter = processor._batch_processor._exporter  # type: ignore
+    except AttributeError:
+        exporter = processor.span_exporter  # type: ignore
+    return exporter  # type: ignore
+
+
+def get_batch_log_exporter(processor: LogRecordProcessor) -> LogExporter:
+    assert isinstance(processor, BatchLogRecordProcessor)
+    try:
+        exporter = processor._batch_processor._exporter  # type: ignore
+    except AttributeError:
+        exporter = processor._exporter  # type: ignore
+    return exporter  # type: ignore
+
+
 def test_configure_export_delay() -> None:
     class TrackingExporter(SpanExporter):
         def __init__(self) -> None:
@@ -602,8 +620,8 @@ def configure_tracking_exporter():
 
         dynamic_batch_span_processor, *_ = get_span_processors()
         assert isinstance(dynamic_batch_span_processor, DynamicBatchSpanProcessor)
-        ex = dynamic_batch_span_processor.batch_processor._exporter = TrackingExporter()  # type: ignore
-        assert dynamic_batch_span_processor.span_exporter is ex
+        batch_processor = dynamic_batch_span_processor.batch_processor
+        ex = batch_processor.span_exporter = batch_processor._exporter = TrackingExporter()  # type: ignore
         return ex
 
     def check_delays(exp: TrackingExporter, min_delay: float, max_delay: float) -> None:
@@ -612,7 +630,6 @@ def check_delays(exp: TrackingExporter, min_delay: float, max_delay: float) -> N
 
     # test the default behaviour
     exporter = configure_tracking_exporter()
-    assert get_span_processors()[0].batch_processor._schedule_delay == 0.1  # type: ignore
     for _ in range(10):
         logfire.info('test')
     sleep(0.1)
@@ -625,7 +642,6 @@ def check_delays(exp: TrackingExporter, min_delay: float, max_delay: float) -> N
             pass
         sleep(0.1)
     # After the first 10 spans, we increase to 500ms by default
-    assert get_span_processors()[0].batch_processor._schedule_delay == 0.5  # type: ignore
     check_delays(exporter, 0.4, 1.0)
 
     # test a very small value
@@ -1566,9 +1582,9 @@ def test_default_exporters(monkeypatch: pytest.MonkeyPatch):
     assert isinstance(console_log_processor._exporter, ConsoleLogExporter)  # type: ignore
     assert console_log_processor._exporter.span_exporter is console_span_processor.span_exporter  # type: ignore
 
-    assert isinstance(logfire_log_processor, BatchLogRecordProcessor)
-    assert isinstance(logfire_log_processor._batch_processor._exporter, QuietLogExporter)  # type: ignore
-    assert isinstance(logfire_log_processor._batch_processor._exporter.exporter, OTLPLogExporter)  # type: ignore
+    exporter = get_batch_log_exporter(logfire_log_processor)
+    assert isinstance(exporter, QuietLogExporter)
+    assert isinstance(exporter.exporter, OTLPLogExporter)
 
 
 def test_custom_exporters():
@@ -1600,8 +1616,7 @@ def test_otel_exporter_otlp_endpoint_env_var():
         logfire.configure(send_to_logfire=False, console=False)
 
     [otel_processor] = get_span_processors()
-    assert isinstance(otel_processor, BatchSpanProcessor)
-    exporter = otel_processor._batch_processor._exporter  # type: ignore
+    exporter = get_batch_span_exporter(otel_processor)
     assert isinstance(exporter, OTLPSpanExporter)
     assert exporter._endpoint == 'otel_endpoint/v1/traces'  # type: ignore
 
@@ -1611,9 +1626,9 @@ def test_otel_exporter_otlp_endpoint_env_var():
     assert otel_metric_reader._exporter._endpoint == 'otel_endpoint/v1/metrics'  # type: ignore
 
     [otel_log_processor] = get_log_record_processors()
-    assert isinstance(otel_log_processor, BatchLogRecordProcessor)
-    assert isinstance(otel_log_processor._batch_processor._exporter, OTLPLogExporter)  # type: ignore
-    assert otel_log_processor._batch_processor._exporter._endpoint == 'otel_endpoint/v1/logs'  # type: ignore
+    log_exporter = get_batch_log_exporter(otel_log_processor)
+    assert isinstance(log_exporter, OTLPLogExporter)
+    assert log_exporter._endpoint == 'otel_endpoint/v1/logs'  # type: ignore
 
 
 def test_otel_traces_exporter_env_var():
@@ -1644,8 +1659,7 @@ def test_otel_metrics_exporter_env_var():
         logfire.configure(send_to_logfire=False, console=False)
 
     [otel_processor] = get_span_processors()
-    assert isinstance(otel_processor, BatchSpanProcessor)
-    exporter = otel_processor._batch_processor._exporter  # type: ignore
+    exporter = get_batch_span_exporter(otel_processor)
     assert isinstance(exporter, OTLPSpanExporter)
     assert exporter._endpoint == 'otel_endpoint3/v1/traces'  # type: ignore
 
@@ -1658,8 +1672,7 @@ def test_otel_logs_exporter_env_var():
         logfire.configure(send_to_logfire=False, console=False)
 
     [otel_processor] = get_span_processors()
-    assert isinstance(otel_processor, BatchSpanProcessor)
-    exporter = otel_processor._batch_processor._exporter  # type: ignore
+    exporter = get_batch_span_exporter(otel_processor)
     assert isinstance(exporter, OTLPSpanExporter)
     assert exporter._endpoint == 'otel_endpoint4/v1/traces'  # type: ignore
 
@@ -1672,8 +1685,7 @@ def test_otel_exporter_otlp_traces_endpoint_env_var():
         logfire.configure(send_to_logfire=False, console=False)
 
     [otel_processor] = get_span_processors()
-    assert isinstance(otel_processor, BatchSpanProcessor)
-    exporter = otel_processor._batch_processor._exporter  # type: ignore
+    exporter = get_batch_span_exporter(otel_processor)
     assert isinstance(exporter, OTLPSpanExporter)
     assert exporter._endpoint == 'otel_traces_endpoint'  # type: ignore
 
@@ -1704,9 +1716,9 @@ def test_otel_exporter_otlp_logs_endpoint_env_var():
     assert len(list(get_metric_readers())) == 0
 
     [otel_log_processor] = get_log_record_processors()
-    assert isinstance(otel_log_processor, BatchLogRecordProcessor)
-    assert isinstance(otel_log_processor._batch_processor._exporter, OTLPLogExporter)  # type: ignore
-    assert otel_log_processor._batch_processor._exporter._endpoint == 'otel_logs_endpoint'  # type: ignore
+    exporter = get_batch_log_exporter(otel_log_processor)
+    assert isinstance(exporter, OTLPLogExporter)
+    assert exporter._endpoint == 'otel_logs_endpoint'  # type: ignore
 
 
 def test_metrics_false(monkeypatch: pytest.MonkeyPatch):