Only set OTLP providers when exporter endpoints are set (#497)

This PR makes sure DBOS only sets the global OTLP tracer/logger
providers if OTLP exporter endpoints are provided. The change is
necessary if users are using other providers like Logfire, which already
set up the exporters. Otherwise, setting providers again would generate
warnings.

Author: qianl15

dbos/_logger.py

@@ -68,30 +68,37 @@ def config_logger(config: "ConfigFile") -> None:
     )
     disable_otlp = config.get("telemetry", {}).get("disable_otlp", False)  # type: ignore
 
-    if not disable_otlp and otlp_logs_endpoints:
+    if not disable_otlp:
 
-        from opentelemetry._logs import set_logger_provider
+        from opentelemetry._logs import get_logger_provider, set_logger_provider
         from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
         from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
         from opentelemetry.sdk._logs.export import BatchLogRecordProcessor
         from opentelemetry.sdk.resources import Resource
         from opentelemetry.semconv.attributes.service_attributes import SERVICE_NAME
 
-        log_provider = LoggerProvider(
-            Resource.create(
-                attributes={
-                    SERVICE_NAME: config["name"],
-                }
-            )
-        )
-        set_logger_provider(log_provider)
-        for e in otlp_logs_endpoints:
-            log_provider.add_log_record_processor(
-                BatchLogRecordProcessor(
-                    OTLPLogExporter(endpoint=e),
-                    export_timeout_millis=5000,
+        # Only set up OTLP provider and exporter if endpoints are provided
+        log_provider = get_logger_provider()
+        if otlp_logs_endpoints is not None:
+            if not isinstance(log_provider, LoggerProvider):
+                log_provider = LoggerProvider(
+                    Resource.create(
+                        attributes={
+                            SERVICE_NAME: config["name"],
+                        }
+                    )
+                )
+                set_logger_provider(log_provider)
+
+            for e in otlp_logs_endpoints:
+                log_provider.add_log_record_processor(
+                    BatchLogRecordProcessor(
+                        OTLPLogExporter(endpoint=e),
+                        export_timeout_millis=5000,
+                    )
                 )
-            )
+
+        # Even if no endpoints are provided, we still need a LoggerProvider to create the LoggingHandler
         global _otlp_handler
         _otlp_handler = LoggingHandler(logger_provider=log_provider)
 

dbos/_tracer.py

@@ -25,6 +25,10 @@ def __init__(self) -> None:
     def config(self, config: ConfigFile) -> None:
         self.otlp_attributes = config.get("telemetry", {}).get("otlp_attributes", {})  # type: ignore
         self.disable_otlp = config.get("telemetry", {}).get("disable_otlp", False)  # type: ignore
+        otlp_traces_endpoints = (
+            config.get("telemetry", {}).get("OTLPExporter", {}).get("tracesEndpoint")  # type: ignore
+        )
+
         if not self.disable_otlp:
             from opentelemetry import trace
             from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
@@ -38,25 +42,26 @@ def config(self, config: ConfigFile) -> None:
             )
             from opentelemetry.semconv.attributes.service_attributes import SERVICE_NAME
 
-            if not isinstance(trace.get_tracer_provider(), TracerProvider):
-                resource = Resource(
-                    attributes={
-                        SERVICE_NAME: config["name"],
-                    }
-                )
-
-                provider = TracerProvider(resource=resource)
-                if os.environ.get("DBOS__CONSOLE_TRACES", None) is not None:
-                    processor = BatchSpanProcessor(ConsoleSpanExporter())
-                    provider.add_span_processor(processor)
-                otlp_traces_endpoints = (
-                    config.get("telemetry", {}).get("OTLPExporter", {}).get("tracesEndpoint")  # type: ignore
-                )
-                if otlp_traces_endpoints:
-                    for e in otlp_traces_endpoints:
-                        processor = BatchSpanProcessor(OTLPSpanExporter(endpoint=e))
-                        provider.add_span_processor(processor)
-                trace.set_tracer_provider(provider)
+            tracer_provider = trace.get_tracer_provider()
+
+            # Only set up OTLP provider and exporter if endpoints are provided
+            if otlp_traces_endpoints is not None:
+                if not isinstance(tracer_provider, TracerProvider):
+                    resource = Resource(
+                        attributes={
+                            SERVICE_NAME: config["name"],
+                        }
+                    )
+
+                    tracer_provider = TracerProvider(resource=resource)
+                    if os.environ.get("DBOS__CONSOLE_TRACES", None) is not None:
+                        processor = BatchSpanProcessor(ConsoleSpanExporter())
+                        tracer_provider.add_span_processor(processor)
+                    trace.set_tracer_provider(tracer_provider)
+
+                for e in otlp_traces_endpoints:
+                    processor = BatchSpanProcessor(OTLPSpanExporter(endpoint=e))
+                    tracer_provider.add_span_processor(processor)
 
     def set_provider(self, provider: "Optional[TracerProvider]") -> None:
         self.provider = provider

tests/test_queue.py

@@ -1476,7 +1476,7 @@ def parent(child_one: str, child_two: str) -> None:
             queue.enqueue(child)
 
     child_one, child_two = str(uuid.uuid4()), str(uuid.uuid4())
-    with SetWorkflowTimeout(1.0):
+    with SetWorkflowTimeout(2.0):
         queue.enqueue(parent, child_one, child_two).get_result()
 
     # Verify child one, which has a propagated timeout, is cancelled

tests/test_spans.py

@@ -5,7 +5,6 @@
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 from inline_snapshot import snapshot
-from opentelemetry._logs import set_logger_provider
 from opentelemetry.sdk import trace as tracesdk
 from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
 from opentelemetry.sdk._logs.export import BatchLogRecordProcessor, InMemoryLogExporter
@@ -17,6 +16,7 @@
 from dbos._logger import dbos_logger
 from dbos._tracer import dbos_tracer
 from dbos._utils import GlobalParams
+from tests.conftest import default_config
 
 
 @dataclass
@@ -65,7 +65,6 @@ def test_step() -> None:
     log_processor = BatchLogRecordProcessor(log_exporter)
     log_provider = LoggerProvider()
     log_provider.add_log_record_processor(log_processor)
-    set_logger_provider(log_provider)
     dbos_logger.addHandler(LoggingHandler(logger_provider=log_provider))
 
     test_workflow()
@@ -192,7 +191,6 @@ async def test_step() -> None:
     log_processor = BatchLogRecordProcessor(log_exporter)
     log_provider = LoggerProvider()
     log_provider.add_log_record_processor(log_processor)
-    set_logger_provider(log_provider)
     dbos_logger.addHandler(LoggingHandler(logger_provider=log_provider))
 
     await test_workflow()
@@ -303,7 +301,6 @@ def test_workflow_endpoint() -> str:
     log_processor = BatchLogRecordProcessor(log_exporter)
     log_provider = LoggerProvider()
     log_provider.add_log_record_processor(log_processor)
-    set_logger_provider(log_provider)
     dbos_logger.addHandler(LoggingHandler(logger_provider=log_provider))
 
     client = TestClient(app)
@@ -378,7 +375,6 @@ def test_step() -> None:
     log_processor = BatchLogRecordProcessor(log_exporter)
     log_provider = LoggerProvider()
     log_provider.add_log_record_processor(log_processor)
-    set_logger_provider(log_provider)
     dbos_logger.addHandler(LoggingHandler(logger_provider=log_provider))
 
     test_workflow()