Merge e6464ed663 into 3746ba58d7

litellm/integrations/langfuse/langfuse_otel.py

@@ -316,6 +316,23 @@ class LangfuseOtelLogger(OpenTelemetry):
             langfuse_host=LangfuseOtelLogger._get_langfuse_otel_host(),
         )
 
+    @staticmethod
+    def get_langfuse_otel_endpoint(langfuse_host: str | None) -> str:
+        """
+        The Langfuse OTLP endpoint for langfuse_host, or the US cloud endpoint when
+        no host is given.
+
+        Shared by the env-resolved config and by per-key/team routing so the two can
+        never derive a different endpoint from the same host.
+        """
+        if not langfuse_host:
+            verbose_logger.debug("Using Langfuse US cloud endpoint: %s", LANGFUSE_CLOUD_US_ENDPOINT)
+            return LANGFUSE_CLOUD_US_ENDPOINT
+        normalized_host: Final = langfuse_host if langfuse_host.startswith("http") else f"https://{langfuse_host}"
+        endpoint: Final = f"{normalized_host.rstrip('/')}/api/public/otel"
+        verbose_logger.debug("Using Langfuse OTEL endpoint from host: %s", endpoint)
+        return endpoint
+
     @staticmethod
     def _build_langfuse_otel_config(
         public_key: str, secret_key: str, langfuse_host: str | None
@@ -324,14 +341,7 @@ class LangfuseOtelLogger(OpenTelemetry):
         Builds an OTLP HTTP config pointing at the Langfuse OTEL endpoint for the
         given host (US cloud when no host is provided), authorized with the given keys.
         """
-        if langfuse_host:
-            normalized_host: Final = langfuse_host if langfuse_host.startswith("http") else f"https://{langfuse_host}"
-            endpoint = f"{normalized_host.rstrip('/')}/api/public/otel"
-            verbose_logger.debug("Using Langfuse OTEL endpoint from host: %s", endpoint)
-        else:
-            endpoint = LANGFUSE_CLOUD_US_ENDPOINT
-            verbose_logger.debug("Using Langfuse US cloud endpoint: %s", endpoint)
-
+        endpoint: Final = LangfuseOtelLogger.get_langfuse_otel_endpoint(langfuse_host)
         auth_header: Final = LangfuseOtelLogger._get_langfuse_authorization_header(
             public_key=public_key, secret_key=secret_key
         )

litellm/integrations/otel/presets/__init__.py

@@ -16,6 +16,7 @@ from litellm.integrations.otel.presets.agentops import agentops_preset
 from litellm.integrations.otel.presets.arize import arize_dynamic_headers, arize_preset
 from litellm.integrations.otel.presets.base import Preset
 from litellm.integrations.otel.presets.langfuse import (
+    langfuse_dynamic_endpoint,
     langfuse_dynamic_headers,
     langfuse_preset,
 )
@@ -64,12 +65,16 @@ DYNAMIC_HEADERS_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynami
 )
 
 #: Callback name → per-request OTLP endpoint resolver. Only integrations whose
-#: destination host varies per tenant (from a fixed region table, never a
-#: caller-supplied URL) appear here; for everyone else the preset's endpoint is
-#: authoritative.
+#: destination host varies per tenant appear here; for everyone else the preset's
+#: endpoint is authoritative. Two shapes exist: newrelic picks from a fixed region
+#: table, and langfuse_otel reads the host the key or team was configured with,
+#: which the proxy resolved at auth. Neither takes a URL straight off the request:
+#: a client-supplied langfuse_host is rejected unless an admin opts in with
+#: general_settings.allow_client_side_credentials.
 DYNAMIC_ENDPOINT_BY_CALLBACK: Final[Mapping[str, Callable[[StandardCallbackDynamicParams], str | None]]] = (
     MappingProxyType(
         {
+            "langfuse_otel": langfuse_dynamic_endpoint,
             "newrelic": newrelic_dynamic_endpoint,
         }
     )

litellm/integrations/otel/presets/langfuse.py

@@ -46,3 +46,16 @@ def langfuse_dynamic_headers(params: StandardCallbackDynamicParams) -> dict[str,
             _V1Langfuse._get_langfuse_authorization_header(public_key=public_key, secret_key=secret_key)
         )
     return {}
+
+
+def langfuse_dynamic_endpoint(params: StandardCallbackDynamicParams) -> str | None:
+    """Per-request Langfuse OTLP endpoint when the key or team pins its own host.
+
+    ``None`` means the request does not move the destination, so the preset's
+    env-resolved endpoint stands (V1 parity: ``construct_dynamic_otel_config``
+    falls back to the env host when the dynamic params carry no ``langfuse_host``).
+    """
+    host: Final = params.get("langfuse_host")
+    if not host:
+        return None
+    return _V1Langfuse.get_langfuse_otel_endpoint(host)

tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py

@@ -544,3 +544,68 @@ def test_newrelic_key_only_team_routes_to_us_not_operator_region(monkeypatch):
     )
     owned = next(e for e in new_cfg.exporters if e.owner == "newrelic")
     assert owned.endpoint == "https://otlp.nr-data.net"
+
+
+# --- a key or team pinning its own Langfuse host moves the destination ------ #
+
+
+LANGFUSE_CREDS = {"langfuse_public_key": "pk", "langfuse_secret_key": "sk"}
+
+
+def test_langfuse_dynamic_endpoint_follows_the_key_host():
+    """A key that names its own Langfuse host must move the export destination.
+
+    Without this the preset swapped the key's credentials in and left the exporter
+    aimed at the process-wide env host, so a tenant's spans were POSTed to the
+    operator's Langfuse signed with keys that host does not know.
+    """
+    assert (
+        dynamic_otlp_endpoint(
+            "langfuse_otel", {**LANGFUSE_CREDS, "langfuse_host": "http://team-b-langfuse:3100"}
+        )
+        == "http://team-b-langfuse:3100/api/public/otel"
+    )
+
+
+def test_langfuse_dynamic_endpoint_normalizes_a_bare_host():
+    assert (
+        dynamic_otlp_endpoint("langfuse_otel", {**LANGFUSE_CREDS, "langfuse_host": "langfuse.internal/"})
+        == "https://langfuse.internal/api/public/otel"
+    )
+
+
+def test_langfuse_dynamic_endpoint_is_none_without_a_host():
+    # No host on the key means "do not move the destination": the preset's
+    # env-resolved endpoint stands, matching V1's fallback to the env host.
+    assert dynamic_otlp_endpoint("langfuse_otel", LANGFUSE_CREDS) is None
+
+
+def test_langfuse_host_stamped_onto_owned_exporter_only():
+    """A Langfuse key's host must never repoint a co-configured exporter owned by
+    a different backend."""
+    cache = _cache(
+        "langfuse_otel",
+        exporters=[
+            ExporterSpec(
+                kind="otlp_http",
+                endpoint="http://self-hosted-collector:4318",
+                headers="x=base-collector",
+                owner=None,
+            ),
+            ExporterSpec(
+                kind="otlp_http",
+                endpoint="http://env-host:3100/api/public/otel",
+                headers="Authorization=Basic env",
+                owner="langfuse_otel",
+            ),
+        ],
+    )
+    new_cfg = cache._routed_config(
+        {"Authorization": "Basic team-b"}, {}, "http://team-b-langfuse:3100/api/public/otel"
+    )
+    by_owner = {e.owner: (e.endpoint, e.headers) for e in new_cfg.exporters}
+    assert by_owner["langfuse_otel"] == (
+        "http://team-b-langfuse:3100/api/public/otel",
+        "Authorization=Basic team-b",
+    )
+    assert by_owner[None] == ("http://self-hosted-collector:4318", "x=base-collector")