diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index b96ec72b04e..917a58973cb 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -283,9 +283,9 @@ class LangfuseOtelLogger(OpenTelemetry): if langfuse_host: # If LANGFUSE_HOST is provided, construct OTEL endpoint from it - if not langfuse_host.startswith("http"): - langfuse_host = "https://" + langfuse_host - endpoint = f"{langfuse_host.rstrip('/')}/api/public/otel" + endpoint = LangfuseOtelLogger._construct_langfuse_otel_endpoint( + langfuse_host + ) verbose_logger.debug(f"Using Langfuse OTEL endpoint from host: {endpoint}") else: # Default to US cloud endpoint @@ -332,9 +332,9 @@ class LangfuseOtelLogger(OpenTelemetry): if langfuse_host: # If LANGFUSE_HOST is provided, construct OTEL endpoint from it - if not langfuse_host.startswith("http"): - langfuse_host = "https://" + langfuse_host - endpoint = f"{langfuse_host.rstrip('/')}/api/public/otel" + endpoint = LangfuseOtelLogger._construct_langfuse_otel_endpoint( + langfuse_host + ) verbose_logger.debug(f"Using Langfuse OTEL endpoint from host: {endpoint}") else: # Default to US cloud endpoint @@ -365,6 +365,13 @@ class LangfuseOtelLogger(OpenTelemetry): auth_header = base64.b64encode(auth_string.encode()).decode() return f"Basic {auth_header}" + @staticmethod + def _construct_langfuse_otel_endpoint(langfuse_host: str) -> str: + """Build the Langfuse OTLP base endpoint from a host (scheme-tolerant).""" + if not langfuse_host.startswith("http"): + langfuse_host = "https://" + langfuse_host + return f"{langfuse_host.rstrip('/')}/api/public/otel" + def construct_dynamic_otel_headers( self, standard_callback_dynamic_params: StandardCallbackDynamicParams ) -> Optional[dict]: @@ -393,6 +400,23 @@ class LangfuseOtelLogger(OpenTelemetry): return dynamic_headers + def construct_dynamic_otel_endpoint( + self, standard_callback_dynamic_params: StandardCallbackDynamicParams + ) -> Optional[str]: + """ + Construct a per-key/team Langfuse OTLP endpoint from the dynamic host. + + Per-key Langfuse credentials are only valid against the host that issued + them, so the host must travel with the credentials. Returns None when no + per-key host is set, falling back to the env-configured endpoint. + """ + dynamic_langfuse_host = standard_callback_dynamic_params.get("langfuse_host") + if not dynamic_langfuse_host: + return None + return LangfuseOtelLogger._construct_langfuse_otel_endpoint( + dynamic_langfuse_host + ) + def create_litellm_proxy_request_started_span( self, start_time: datetime, diff --git a/litellm/integrations/opentelemetry.py b/litellm/integrations/opentelemetry.py index ce5cfa2f525..9861824586d 100644 --- a/litellm/integrations/opentelemetry.py +++ b/litellm/integrations/opentelemetry.py @@ -858,9 +858,14 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): if dynamic_headers is not None: # Create spans using a temporary tracer with dynamic headers - tracer_to_use = self._get_tracer_with_dynamic_headers(dynamic_headers) + dynamic_endpoint = self._get_dynamic_otel_endpoint_from_kwargs(kwargs) + tracer_to_use = self._get_tracer_with_dynamic_headers( + dynamic_headers, dynamic_endpoint=dynamic_endpoint + ) verbose_logger.debug( - "[OTEL DEBUG] Using DYNAMIC tracer with headers: %s", dynamic_headers + "[OTEL DEBUG] Using DYNAMIC tracer with headers: %s endpoint: %s", + dynamic_headers, + dynamic_endpoint, ) else: # For langfuse_otel without dynamic headers, create a provider with env var credentials @@ -907,12 +912,29 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): return dynamic_headers if dynamic_headers else None - def _get_tracer_with_dynamic_headers(self, dynamic_headers: dict): + def _get_dynamic_otel_endpoint_from_kwargs(self, kwargs) -> Optional[str]: + """Extract a dynamic OTLP endpoint from kwargs if available.""" + standard_callback_dynamic_params: Optional[StandardCallbackDynamicParams] = ( + kwargs.get("standard_callback_dynamic_params") + ) + + if not standard_callback_dynamic_params: + return None + + return self.construct_dynamic_otel_endpoint( + standard_callback_dynamic_params=standard_callback_dynamic_params + ) + + def _get_tracer_with_dynamic_headers( + self, dynamic_headers: dict, dynamic_endpoint: Optional[str] = None + ): """Create a temporary tracer with dynamic headers for this request only.""" from opentelemetry.sdk.trace import TracerProvider # Prevents thread exhaustion by reusing providers for the same credential sets (e.g. per-team keys) cache_key = str(sorted(dynamic_headers.items())) + if dynamic_endpoint: + cache_key = f"{cache_key}|{dynamic_endpoint}" if cache_key in self._tracer_provider_cache: return self._tracer_provider_cache[cache_key].get_tracer( LITELLM_TRACER_NAME @@ -921,7 +943,9 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): # Create a temporary tracer provider with dynamic headers temp_provider = TracerProvider(resource=self._get_litellm_resource(self.config)) temp_provider.add_span_processor( - self._get_span_processor(dynamic_headers=dynamic_headers) + self._get_span_processor( + dynamic_headers=dynamic_headers, dynamic_endpoint=dynamic_endpoint + ) ) # Store in cache for reuse @@ -942,6 +966,17 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): """ return None + def construct_dynamic_otel_endpoint( + self, standard_callback_dynamic_params: StandardCallbackDynamicParams + ) -> Optional[str]: + """ + Construct a per-request OTLP endpoint from standard callback dynamic params. + + Override in subclasses (e.g. Langfuse OTEL) where per-key/team credentials + are bound to a per-key host. Returning None keeps the env-configured endpoint. + """ + return None + ######################################################### # End of Team/Key Based Logging Control Flow ######################################################### @@ -2761,7 +2796,11 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): ) return None, None - def _get_span_processor(self, dynamic_headers: Optional[dict] = None): + def _get_span_processor( + self, + dynamic_headers: Optional[dict] = None, + dynamic_endpoint: Optional[str] = None, + ): from opentelemetry.sdk.trace.export import ( BatchSpanProcessor, ConsoleSpanExporter, @@ -2827,7 +2866,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): self.OTEL_EXPORTER, ) normalized_endpoint = self._normalize_otel_endpoint( - self.OTEL_ENDPOINT, "traces" + dynamic_endpoint or self.OTEL_ENDPOINT, "traces" ) return BatchSpanProcessor( OTLPSpanExporterHTTP( @@ -2850,7 +2889,7 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger): self.OTEL_EXPORTER, ) normalized_endpoint = self._normalize_otel_endpoint( - self.OTEL_ENDPOINT, "traces" + dynamic_endpoint or self.OTEL_ENDPOINT, "traces" ) return BatchSpanProcessor( OTLPSpanExporterGRPC( diff --git a/tests/test_litellm/integrations/test_langfuse_otel.py b/tests/test_litellm/integrations/test_langfuse_otel.py index 44853d9dce5..30737650431 100644 --- a/tests/test_litellm/integrations/test_langfuse_otel.py +++ b/tests/test_litellm/integrations/test_langfuse_otel.py @@ -392,6 +392,33 @@ class TestLangfuseOtelIntegration: # Should return an empty dict assert result == {} + def test_construct_dynamic_otel_endpoint_with_host(self): + """Per-key langfuse_host is turned into a normalized OTLP base endpoint.""" + from litellm.types.utils import StandardCallbackDynamicParams + + logger = LangfuseOtelLogger() + + eu = logger.construct_dynamic_otel_endpoint( + StandardCallbackDynamicParams(langfuse_host="https://cloud.langfuse.com") + ) + assert eu == "https://cloud.langfuse.com/api/public/otel" + + # scheme-less + trailing slash normalize identically to the env path + no_scheme = logger.construct_dynamic_otel_endpoint( + StandardCallbackDynamicParams(langfuse_host="us.cloud.langfuse.com/") + ) + assert no_scheme == "https://us.cloud.langfuse.com/api/public/otel" + + def test_construct_dynamic_otel_endpoint_without_host(self): + """No per-key host -> None, so the env endpoint is used.""" + from litellm.types.utils import StandardCallbackDynamicParams + + logger = LangfuseOtelLogger() + assert ( + logger.construct_dynamic_otel_endpoint(StandardCallbackDynamicParams()) + is None + ) + def test_get_langfuse_otel_config_with_otel_host_priority(self): """LANGFUSE_OTEL_HOST should take priority over LANGFUSE_HOST.""" with patch.dict( diff --git a/tests/test_litellm/integrations/test_opentelemetry.py b/tests/test_litellm/integrations/test_opentelemetry.py index f6dee9b64c9..e348efbfa3c 100644 --- a/tests/test_litellm/integrations/test_opentelemetry.py +++ b/tests/test_litellm/integrations/test_opentelemetry.py @@ -1280,7 +1280,8 @@ class TestOpenTelemetry(unittest.TestCase): # Assertions mock_get_headers.assert_called_once_with(kwargs) mock_get_tracer.assert_called_once_with( - {"arize-space-id": "test-space", "api_key": "test-key"} + {"arize-space-id": "test-space", "api_key": "test-key"}, + dynamic_endpoint=None, ) self.assertEqual(result, mock_dynamic_tracer) @@ -1374,7 +1375,7 @@ class TestOpenTelemetry(unittest.TestCase): # Assertions mock_get_span_processor.assert_called_once_with( - dynamic_headers=dynamic_headers + dynamic_headers=dynamic_headers, dynamic_endpoint=None ) mock_provider_instance.add_span_processor.assert_called_once_with( mock_span_processor @@ -2104,6 +2105,56 @@ class TestOpenTelemetryProtocolSelection(unittest.TestCase): # Verify the exporter is the HTTP variant self.assertIsInstance(processor.span_exporter, OTLPSpanExporterHTTP) + def test_get_span_processor_uses_dynamic_endpoint_over_env(self): + """The dynamic per-key endpoint overrides self.OTEL_ENDPOINT (the core regression). + + Would FAIL on current code, where the exporter is always pinned to OTEL_ENDPOINT. + """ + config = OpenTelemetryConfig( + exporter="otlp_http", endpoint="https://cloud.langfuse.com/api/public/otel" + ) + otel = OpenTelemetry(config=config) + + processor = otel._get_span_processor( + dynamic_headers={"Authorization": "Basic abc"}, + dynamic_endpoint="https://us.cloud.langfuse.com/api/public/otel", + ) + # OTLP HTTP exporter stores the resolved endpoint on ._endpoint + endpoint = processor.span_exporter._endpoint + assert endpoint == "https://us.cloud.langfuse.com/api/public/otel/v1/traces" + + def test_get_span_processor_falls_back_to_env_endpoint(self): + """No dynamic endpoint -> env endpoint is used (no Arize/env-only regression).""" + config = OpenTelemetryConfig( + exporter="otlp_http", endpoint="https://cloud.langfuse.com/api/public/otel" + ) + otel = OpenTelemetry(config=config) + processor = otel._get_span_processor( + dynamic_headers={"Authorization": "Basic abc"} + ) + assert ( + processor.span_exporter._endpoint + == "https://cloud.langfuse.com/api/public/otel/v1/traces" + ) + + def test_dynamic_tracer_cache_key_separates_by_endpoint(self): + """Same creds + different host => distinct cached providers (cache-collision regression). + + Would FAIL if the host is dropped from the cache key: the 2nd host would reuse + the 1st host's provider/exporter. + """ + otel = OpenTelemetry() + headers = {"Authorization": "Basic abc"} + with patch.object(otel, "_get_span_processor", return_value=MagicMock()): + otel._get_tracer_with_dynamic_headers( + headers, dynamic_endpoint="https://cloud.langfuse.com/api/public/otel" + ) + otel._get_tracer_with_dynamic_headers( + headers, + dynamic_endpoint="https://us.cloud.langfuse.com/api/public/otel", + ) + assert len(otel._tracer_provider_cache) == 2 + def test_get_span_processor_uses_console_exporter_for_console(self): """Test that console protocol uses ConsoleSpanExporter""" from opentelemetry.sdk.trace.export import (