scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-12 23:01:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 88

fix(otel): route per-key langfuse_host to the OTLP exporter endpoint

Browse source
This commit is contained in:
michelligabriele 2026-06-03 14:16:25 +02:00
parent d45e9e4d56
commit 897f4dbdfb
No known key found for this signature in database
4 changed files with 156 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

36
litellm/integrations/langfuse/langfuse_otel.py
View file

@ -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,

53
litellm/integrations/opentelemetry.py
View file

@ -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(

27
tests/test_litellm/integrations/test_langfuse_otel.py
View file

@ -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(

55
tests/test_litellm/integrations/test_opentelemetry.py
View file

@ -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 (