diff --git a/litellm/integrations/langfuse/langfuse_otel.py b/litellm/integrations/langfuse/langfuse_otel.py index fa978c04af2..f195cd97e3c 100644 --- a/litellm/integrations/langfuse/langfuse_otel.py +++ b/litellm/integrations/langfuse/langfuse_otel.py @@ -5,18 +5,13 @@ from collections.abc import Mapping, Sequence from datetime import datetime from typing import TYPE_CHECKING, Any, Final, Optional -from pydantic import TypeAdapter - from litellm._logging import verbose_logger -from litellm.integrations._types.open_inference import ErrorAttributes from litellm.integrations.arize import _utils from litellm.integrations.langfuse.langfuse_otel_attributes import ( LangfuseLLMObsOTELAttributes, serialize_langfuse_datetime, ) -from litellm.integrations.opentelemetry import HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE, OpenTelemetry, OpenTelemetryConfig -from litellm.integrations.otel.mappers.langfuse import TRACE_IDENTITY_FIELDS -from litellm.integrations.otel.model.metadata import IDENTITY_METADATA_FIELDS, exported_request_metadata +from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig from litellm.litellm_core_utils.safe_json_loads import safe_json_loads from litellm.types.integrations.langfuse_otel import ( LangfuseSpanAttributes, @@ -31,19 +26,6 @@ else: Span = Any -_MAPPING_ADAPTER: Final[TypeAdapter[Mapping[object, object]]] = TypeAdapter(Mapping[object, object]) -# The attributes ``OpenTelemetry._record_exception_on_span`` adds to a failed call's span after these. -_ERROR_ATTRIBUTES: Final = frozenset( - { - ErrorAttributes.ERROR_CODE, - HTTP_RESPONSE_STATUS_CODE_ATTRIBUTE, - ErrorAttributes.ERROR_TYPE, - ErrorAttributes.ERROR_MESSAGE, - ErrorAttributes.ERROR_LLM_PROVIDER, - ErrorAttributes.ERROR_STACK_TRACE, - } -) - LANGFUSE_CLOUD_EU_ENDPOINT: Final = "https://cloud.langfuse.com/api/public/otel" LANGFUSE_CLOUD_US_ENDPOINT: Final = "https://us.cloud.langfuse.com/api/public/otel" LANGFUSE_INGESTION_VERSION_HEADER: Final = "x-langfuse-ingestion-version" @@ -64,10 +46,7 @@ class LangfuseOtelLogger(OpenTelemetry): Sets OpenTelemetry span attributes for Langfuse observability. Uses the same attribute setting logic as Arize Phoenix for consistency. """ - from opentelemetry import trace - # The caller's active span, which LiteLLM writes onto instead of starting its own, gets no caller identity. - own_span: Final = span is not trace.get_current_span() _utils.set_attributes(span, kwargs, response_obj, LangfuseLLMObsOTELAttributes) span.set_attribute("langfuse.observation.type", "generation") @@ -75,10 +54,6 @@ class LangfuseOtelLogger(OpenTelemetry): # Set Langfuse specific attributes ######################################################### LangfuseOtelLogger._set_langfuse_specific_attributes(span=span, kwargs=kwargs, response_obj=response_obj) - if own_span: - LangfuseOtelLogger._set_request_metadata_attributes( - span=span, kwargs=_MAPPING_ADAPTER.validate_python(kwargs) - ) @staticmethod def _extract_langfuse_metadata(kwargs: Mapping[str, Any]) -> dict: @@ -152,46 +127,6 @@ class LangfuseOtelLogger(OpenTelemetry): value = str(value) safe_set_attribute(span, enum_attr.value, value) - @staticmethod - def _set_request_metadata_attributes( - span: Span, - kwargs: Mapping[object, object], - ) -> None: - """Set the caller identity into the room ``span``'s attribute limit leaves next to the error attributes - a failed call adds after it, so the identity never evicts or drops another attribute.""" - from litellm.integrations.otel.emitter import carried_keys, span_attribute_limit - - raw_standard_logging_object: Final = kwargs.get("standard_logging_object") - if not isinstance(raw_standard_logging_object, dict): - return - raw_metadata: Final = _MAPPING_ADAPTER.validate_python(raw_standard_logging_object).get("metadata") - if not isinstance(raw_metadata, dict): - return - metadata: Final = _MAPPING_ADAPTER.validate_python(raw_metadata) - observation_metadata: Final = exported_request_metadata( - {field: metadata.get(field) for field in IDENTITY_METADATA_FIELDS} - ) - trace_prefix: Final = LangfuseSpanAttributes.TRACE_METADATA.value - identity: Final = ( - *( - ((LangfuseSpanAttributes.OBSERVATION_METADATA.value, json.dumps(dict(observation_metadata))),) - if observation_metadata - else () - ), - *( - (f"{trace_prefix}.{field}", value) - for field in TRACE_IDENTITY_FIELDS - if (value := observation_metadata.get(field)) - ), - ) - limit: Final = span_attribute_limit(span) - carried: Final = carried_keys(span) - room: Final = ( - len(identity) if limit is None else max(limit - len(carried) - len(_ERROR_ATTRIBUTES - carried), 0) - ) - for key, value in identity[:room]: - _utils.safe_set_attribute(span, key, value) - @staticmethod def _set_observation_output(span: Span, response_obj): """Helper to set observation output attributes.""" diff --git a/litellm/types/integrations/langfuse_otel.py b/litellm/types/integrations/langfuse_otel.py index 33b708b2ac6..5647f28c757 100644 --- a/litellm/types/integrations/langfuse_otel.py +++ b/litellm/types/integrations/langfuse_otel.py @@ -30,7 +30,6 @@ class LangfuseSpanAttributes(str, Enum): OBSERVATION_INPUT = "langfuse.observation.input" OBSERVATION_OUTPUT = "langfuse.observation.output" OBSERVATION_COMPLETION_START_TIME = "langfuse.observation.completion_start_time" - OBSERVATION_METADATA = "langfuse.observation.metadata" # ---- Trace-level metadata ---- TRACE_USER_ID = "user.id" diff --git a/tests/integration/observability/test_langfuse_otel_metadata.py b/tests/integration/observability/test_langfuse_otel_metadata.py index ddd9bca229b..78b31d8ee4e 100644 --- a/tests/integration/observability/test_langfuse_otel_metadata.py +++ b/tests/integration/observability/test_langfuse_otel_metadata.py @@ -256,6 +256,7 @@ def _langfuse_rig( "LANGFUSE_PUBLIC_KEY": "pk-lf-integration", "LANGFUSE_SECRET_KEY": "sk-lf-integration", "LANGFUSE_HOST": collector.url, + "LITELLM_OTEL_V2": "1", **{ name: (collector.url if value == "__collector__" else value) for name, value in dict(overrides or {}).items() @@ -282,7 +283,7 @@ def _response_id(value: object) -> str | None: def _span_response_id(span: Mapping[str, object]) -> str | None: - return _response_id(span.get("llm.response.id")) + return _response_id(span.get("gen_ai.response.id")) def _watched_spans(collector: Wire) -> Callable[[], tuple[dict[str, object], ...]]: @@ -307,32 +308,6 @@ def _v2_llm_span(collector: Wire, response_id: str) -> dict[str, object]: return next(span for span in observed if matches(span)) -def _observation_span(collector: Wire, response_id: str) -> dict[str, object]: - def spans() -> tuple[dict[str, object], ...]: - return tuple( - attributes - for batch in collector.drain() - for attributes in _span_attributes(batch.body) - if _span_response_id(attributes) == _response_id(response_id) - ) - - observed: Final = eventually(spans, lambda values: len(values) == 1, seconds=25) - return observed[0] - - -def _marked_span(collector: Wire, marker: str, needle: str) -> dict[str, object]: - def spans() -> tuple[dict[str, object], ...]: - return tuple( - attributes - for batch in collector.drain() - for attributes in _span_attributes(batch.body) - if needle in json.dumps(attributes, default=str) and marker in json.dumps(attributes, default=str) - ) - - observed: Final = eventually(spans, lambda values: len(values) == 1, seconds=25) - return observed[0] - - def _scenario_identity( scenario: Scenario, provider: Wire, @@ -360,7 +335,6 @@ def _scenario_identity( def _assert_identity(attrs: Mapping[str, object], expected: Mapping[str, object]) -> None: - assert "metadata" in attrs, sorted(attrs) assert "langfuse.observation.metadata" in attrs, sorted(attrs) observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) assert {key: observation.get(key) for key in expected} == dict(expected), observation @@ -383,63 +357,6 @@ def _chat_identity_fields(identity: Mapping[str, str | None]) -> dict[str, objec } -def test_langfuse_otel_emits_request_metadata_under_langfuse_observation_and_trace_keys( - gateway: Gateway, tmp_path: Path -) -> None: - marker: Final = "lf-meta-" + uuid.uuid4().hex - with ( - _langfuse_rig(gateway, tmp_path, {}, marker) as rig, - rig.candidate.scenario() as scenario, - ): - key, identity = _scenario_identity(scenario, rig.provider) - response: Final = rig.candidate.request( - "POST", - "/v1/chat/completions", - { - "model": identity["model"], - "messages": [{"role": "user", "content": marker}], - "user": identity["end_user"], - "cache": {"no-cache": True}, - }, - key=key, - ) - assert response.status_code == 200, response.text - assert response.json()["choices"][0]["message"]["content"] == "langfuse-echo" - - attrs: Final = _observation_span(rig.collector, response.json()["id"]) - _assert_identity(attrs, _chat_identity_fields(identity)) - - -def test_langfuse_otel_metadata_redacts_user_api_key_fields_like_vanilla_langfuse( - gateway: Gateway, tmp_path: Path -) -> None: - marker: Final = "lf-redact-" + uuid.uuid4().hex - with ( - _langfuse_rig(gateway, tmp_path, {"redact_user_api_key_info": True}, marker) as rig, - rig.candidate.scenario() as scenario, - ): - key: Final = scenario.key( - key_alias="lf-redact-" + uuid.uuid4().hex, - metadata={"spend_logs_metadata": {"ticket": "LIT-8283"}}, - ) - model: Final = scenario.model(api_base=rig.provider.url + "/v1") - response: Final = rig.candidate.request( - "POST", - "/v1/chat/completions", - { - "model": model, - "messages": [{"role": "user", "content": marker}], - "cache": {"no-cache": True}, - }, - key=key, - ) - assert response.status_code == 200, response.text - - attrs: Final = _observation_span(rig.collector, response.json()["id"]) - assert "langfuse.observation.metadata" not in attrs, attrs["langfuse.observation.metadata"] - assert not [name for name in attrs if name.startswith("langfuse.trace.metadata.")], sorted(attrs) - - def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_stream(gateway: Gateway, tmp_path: Path) -> None: marker: Final = "lf-sdkstream-" + uuid.uuid4().hex with ( @@ -463,7 +380,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_stream(gateway: ) assert chunks, "stream produced no chunks" response_id: Final = chunks[-1].id - attrs: Final = _observation_span(rig.collector, response_id) + attrs: Final = _v2_llm_span(rig.collector, response_id) _assert_identity(attrs, _chat_identity_fields(identity)) @@ -485,7 +402,7 @@ async def test_langfuse_otel_metadata_reaches_langfuse_for_openai_sdk_async(gate user=identity["end_user"], extra_body={"cache": {"no-cache": True}}, ) - attrs: Final = _observation_span(rig.collector, response.id) + attrs: Final = _v2_llm_span(rig.collector, response.id) _assert_identity(attrs, _chat_identity_fields(identity)) @@ -508,7 +425,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_messages(gate messages=[{"role": "user", "content": marker}], extra_body={"litellm_metadata": {"tags": ["lf8283"]}, "metadata": {"user_id": identity["end_user"]}}, ) - attrs: Final = _observation_span(rig.collector, message.id) + attrs: Final = _v2_llm_span(rig.collector, message.id) _assert_identity(attrs, _chat_identity_fields(identity)) @@ -536,7 +453,7 @@ async def test_langfuse_otel_metadata_reaches_langfuse_for_anthropic_sdk_stream( ) as stream, ): final: Final = await stream.get_final_message() - attrs: Final = _observation_span(rig.collector, final.id) + attrs: Final = _v2_llm_span(rig.collector, final.id) _assert_identity(attrs, _chat_identity_fields(identity)) @@ -558,7 +475,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_responses_api(gateway: Gate user=identity["end_user"], extra_body={"cache": {"no-cache": True}}, ) - attrs: Final = _observation_span(rig.collector, response.id) + attrs: Final = _v2_llm_span(rig.collector, response.id) _assert_identity(attrs, _chat_identity_fields(identity)) @@ -589,32 +506,7 @@ def test_langfuse_otel_metadata_reaches_langfuse_for_responses_sse_stream(gatewa response_id: Final = next( event["response_id"] for event in events if event.get("type") == "response.output_text.delta" ) - attrs: Final = _observation_span(rig.collector, response_id) - _assert_identity(attrs, _chat_identity_fields(identity)) - - -def test_langfuse_otel_caller_trace_metadata_survives_flattened_identity(gateway: Gateway, tmp_path: Path) -> None: - marker: Final = "lf-deploy-" + uuid.uuid4().hex - with ( - _langfuse_rig(gateway, tmp_path, {}, marker) as rig, - rig.candidate.scenario() as scenario, - ): - key, identity = _scenario_identity(scenario, rig.provider) - response: Final = rig.candidate.request( - "POST", - "/v1/chat/completions", - { - "model": identity["model"], - "messages": [{"role": "user", "content": marker}], - "user": identity["end_user"], - "metadata": {"trace_metadata": {"deploy": "blue"}, "tags": ["lf8283"]}, - "cache": {"no-cache": True}, - }, - key=key, - ) - assert response.status_code == 200, response.text - attrs: Final = _observation_span(rig.collector, response.json()["id"]) - assert json.loads(str(attrs["langfuse.trace.metadata"])) == {"deploy": "blue"}, attrs + attrs: Final = _v2_llm_span(rig.collector, response_id) _assert_identity(attrs, _chat_identity_fields(identity)) @@ -637,65 +529,19 @@ def test_langfuse_otel_failure_span_carries_request_metadata(gateway: Gateway, t key=key, ) assert response.status_code >= 500, response.text - attrs: Final = _marked_span(rig.collector, marker, "langfuse.observation.") - _assert_identity(attrs, _chat_identity_fields(identity)) + def failed_calls(spans: Sequence[Mapping[str, object]]) -> tuple[Mapping[str, object], ...]: + return tuple( + span + for span in spans + if span.get("langfuse.trace.metadata.team_id") == identity["team_id"] and "error.type" in span + ) -def test_langfuse_otel_teamless_key_emits_no_team_identity(gateway: Gateway, tmp_path: Path) -> None: - marker: Final = "lf-noteam-" + uuid.uuid4().hex - with ( - _langfuse_rig(gateway, tmp_path, {}, marker) as rig, - rig.candidate.scenario() as scenario, - ): - key, identity = _scenario_identity(scenario, rig.provider, team=False) - response: Final = rig.candidate.request( - "POST", - "/v1/chat/completions", - { - "model": identity["model"], - "messages": [{"role": "user", "content": marker}], - "cache": {"no-cache": True}, - }, - key=key, + failed: Final = failed_calls( + eventually(_watched_spans(rig.collector), lambda spans: bool(failed_calls(spans)), seconds=25) ) - assert response.status_code == 200, response.text - attrs: Final = _observation_span(rig.collector, response.json()["id"]) - assert "langfuse.observation.metadata" in attrs, sorted(attrs) - observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) - for field in ("user_api_key_team_id", "user_api_key_team_alias", "user_api_key_end_user_id"): - assert field not in observation, observation - assert set(observation) <= _EXPORTED_FIELDS, observation - assert { - attribute: attrs.get(attribute) - for attribute in (f"langfuse.trace.metadata.{field}" for field in _IDENTITIES) - if attribute in attrs - } == {"langfuse.trace.metadata.user_api_key_alias": identity["key_alias"]}, attrs - - -def test_langfuse_otel_teamless_key_keeps_caller_trace_metadata(gateway: Gateway, tmp_path: Path) -> None: - marker: Final = "lf-noteam-meta-" + uuid.uuid4().hex - with ( - _langfuse_rig(gateway, tmp_path, {}, marker) as rig, - rig.candidate.scenario() as scenario, - ): - key, identity = _scenario_identity(scenario, rig.provider, team=False) - response: Final = rig.candidate.request( - "POST", - "/v1/chat/completions", - { - "model": identity["model"], - "messages": [{"role": "user", "content": marker}], - "metadata": {"trace_metadata": {"deploy": "blue"}}, - "cache": {"no-cache": True}, - }, - key=key, - ) - assert response.status_code == 200, response.text - attrs: Final = _observation_span(rig.collector, response.json()["id"]) - assert json.loads(str(attrs["langfuse.trace.metadata"])) == {"deploy": "blue"}, attrs - assert "langfuse.trace.metadata.user_api_key_team_id" not in attrs, attrs - assert "langfuse.trace.metadata.user_api_key_team_alias" not in attrs, attrs - assert attrs.get("langfuse.trace.metadata.user_api_key_alias") == identity["key_alias"], attrs + for attrs in failed: + _assert_identity(attrs, _chat_identity_fields(identity)) def test_langfuse_otel_oversized_caller_metadata_is_not_copied_into_the_langfuse_keys( @@ -727,11 +573,18 @@ def test_langfuse_otel_oversized_caller_metadata_is_not_copied_into_the_langfuse key=key, ) assert response.status_code == 200, response.text - attrs: Final = _observation_span(rig.collector, response.json()["id"]) + attrs: Final = _v2_llm_span(rig.collector, response.json()["id"]) _assert_identity(attrs, _chat_identity_fields(identity)) assert len(str(attrs["langfuse.observation.metadata"])) < 1000, len(str(attrs["langfuse.observation.metadata"])) - assert caller_note in str(attrs["metadata"]), "the bare metadata attribute keeps the caller metadata" - assert spend_logs_metadata["blob"] in str(attrs["metadata"]), "the bare metadata attribute keeps it" + langfuse_metadata: Final = json.dumps( + { + name: value + for name, value in attrs.items() + if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.") + } + ) + assert caller_note not in langfuse_metadata, len(langfuse_metadata) + assert spend_logs_metadata["blob"] not in langfuse_metadata, len(langfuse_metadata) def test_langfuse_otel_unauthenticated_request_leaks_no_identity(gateway: Gateway, tmp_path: Path) -> None: @@ -779,20 +632,13 @@ def test_langfuse_otel_unauthenticated_request_leaks_no_identity(gateway: Gatewa settled: Final = eventually( watch, lambda spans: sum(_span_response_id(s) == second_id for s in spans) == 1, seconds=25 ) - denied_spans: Final = tuple(span for span in settled if f"{marker}-denied" in json.dumps(span, default=str)) - assert len(denied_spans) == 1, [s.get("llm.response.id") for s in settled] - denied_span: Final = denied_spans[0] - assert not [name for name in denied_span if name.startswith("langfuse.trace.metadata.user_api_key")], ( - denied_span + carriers: Final = tuple( + _span_response_id(span) + for span in settled + if "langfuse.observation.metadata" in span + or any(name.startswith("langfuse.trace.metadata.user_api_key_") for name in span) ) - leaked: Final = json.loads(str(denied_span.get("langfuse.observation.metadata", "{}"))) - for field in ( - "user_api_key_alias", - "user_api_key_team_id", - "user_api_key_team_alias", - "user_api_key_end_user_id", - ): - assert field not in leaked, leaked + assert sorted(carriers, key=str) == sorted((first_id, second_id)), carriers def test_langfuse_otel_repeated_identical_requests_emit_one_span_each(gateway: Gateway, tmp_path: Path) -> None: @@ -903,10 +749,12 @@ def test_langfuse_otel_sink_outage_mid_burst_lands_every_response_id_once(gatewa watch, lambda spans: all(sum(_span_response_id(s) == rid for s in spans) == 1 for rid in response_ids), seconds=80, + return_last_on_timeout=True, ) - assert sorted(_span_response_id(span) for span in settled if _span_response_id(span) in response_ids) == sorted( - response_ids + landed: Final = sorted( + str(_span_response_id(span)) for span in settled if _span_response_id(span) in response_ids ) + assert landed == sorted(response_ids), sorted(set(response_ids) - set(landed)) assert len(rejected) >= 1, "sink outage never served a 503" @@ -1006,7 +854,6 @@ def test_langfuse_otel_v2_mapper_emits_request_metadata_and_identity(gateway: Ga tmp_path, {}, marker, - overrides={"LITELLM_OTEL_V2": "1"}, ) as rig, rig.candidate.scenario() as scenario, ): @@ -1053,7 +900,6 @@ def test_langfuse_otel_v2_mapper_redacts_user_api_key_fields(gateway: Gateway, t tmp_path, {"redact_user_api_key_info": True}, marker, - overrides={"LITELLM_OTEL_V2": "1"}, ) as rig, rig.candidate.scenario() as scenario, ): @@ -1092,7 +938,6 @@ def test_langfuse_otel_v2_mapper_teamless_key_emits_no_team_identity(gateway: Ga tmp_path, {}, marker, - overrides={"LITELLM_OTEL_V2": "1"}, ) as rig, rig.candidate.scenario() as scenario, ): @@ -1154,7 +999,6 @@ def test_otel_v2_scalar_requester_metadata_keeps_the_llm_span( {"callbacks": [callback]}, marker, overrides={ - "LITELLM_OTEL_V2": "1", **({"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": "__collector__"} if callback == "otel" else {}), }, ) as rig, @@ -1179,26 +1023,6 @@ def test_otel_v2_scalar_requester_metadata_keeps_the_llm_span( assert observation["user_api_key_alias"] == identity["key_alias"], observation -def test_langfuse_otel_scalar_requester_metadata_keeps_the_observation_input(gateway: Gateway, tmp_path: Path) -> None: - marker: Final = "lf-scalar-" + uuid.uuid4().hex - with ( - _langfuse_rig(gateway, tmp_path, {}, marker) as rig, - rig.candidate.scenario() as scenario, - ): - key, identity = _scenario_identity(scenario, rig.provider) - model: Final = scenario.model(model="anthropic/claude-sonnet-4-6", api_base=rig.provider.url) - response: Final = rig.candidate.request( - "POST", "/v1/messages", _scalar_requester_metadata_request("/v1/messages", model, marker), key=key - ) - assert response.status_code == 200, response.text - - attrs: Final = _observation_span(rig.collector, response.json()["id"]) - assert marker in str(attrs.get("langfuse.observation.input")), sorted(attrs) - observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) - assert "requester_metadata" not in observation, observation - assert observation["user_api_key_alias"] == identity["key_alias"], observation - - @pytest.mark.parametrize( "callbacks", [["langfuse_otel", "otel"], ["langfuse_otel"]], @@ -1216,7 +1040,7 @@ def test_otel_v2_collector_beside_langfuse_gets_no_langfuse_request_metadata( tmp_path, {"callbacks": callbacks}, marker, - overrides={"LITELLM_OTEL_V2": "1", "OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url}, + overrides={"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url}, ) as rig, rig.candidate.scenario() as scenario, ): @@ -1258,7 +1082,7 @@ def test_otel_v2_langfuse_gets_no_request_headers_or_caller_metadata( marker: Final = "lf-v2-probe-" + uuid.uuid4().hex probe: Final = "lf-probe-" + uuid.uuid4().hex with ( - _langfuse_rig(gateway, tmp_path, {}, marker, overrides={"LITELLM_OTEL_V2": "1"}) as rig, + _langfuse_rig(gateway, tmp_path, {}, marker) as rig, rig.candidate.scenario() as scenario, ): key, identity = _scenario_identity(scenario, rig.provider) @@ -1346,10 +1170,7 @@ def _keyword_router(router: str, keyword: str) -> Callable[[str], Sequence[Mappi return models -@pytest.mark.parametrize("otel_v2", [False, True], ids=["v1", "v2"]) -def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields( - otel_v2: bool, gateway: Gateway, tmp_path: Path -) -> None: +def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields(gateway: Gateway, tmp_path: Path) -> None: marker: Final = "lf-route-" + uuid.uuid4().hex router: Final = "lf-router-" + uuid.uuid4().hex[:8] keyword: Final = "acquire-globex-" + uuid.uuid4().hex[:8] @@ -1359,11 +1180,7 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields( tmp_path, {}, marker, - overrides=( - {"LITELLM_OTEL_V2": "1", "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "no_content"} - if otel_v2 - else None - ), + overrides={"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "no_content"}, model_list=_keyword_router(router, keyword), ) as rig, rig.candidate.scenario() as scenario, @@ -1381,11 +1198,7 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields( ) assert response.status_code == 200, response.text - attrs: Final = ( - _v2_llm_span(rig.collector, response.json()["id"]) - if otel_v2 - else _observation_span(rig.collector, response.json()["id"]) - ) + attrs: Final = _v2_llm_span(rig.collector, response.json()["id"]) observation: Final = json.loads(str(attrs["langfuse.observation.metadata"])) assert set(observation) <= _EXPORTED_FIELDS, observation langfuse_metadata: Final = { @@ -1394,8 +1207,7 @@ def test_langfuse_otel_metadata_omits_prompt_quoting_routing_fields( if name == "langfuse.observation.metadata" or name.startswith("langfuse.trace.metadata.") } assert keyword not in json.dumps(langfuse_metadata, default=str), langfuse_metadata - if otel_v2: - assert keyword not in json.dumps(attrs, default=str), attrs + assert keyword not in json.dumps(attrs, default=str), attrs def _span_records(body: bytes) -> tuple[tuple[dict[str, object], int], ...]: @@ -1451,59 +1263,6 @@ def _limit_sweep( return {size: record for size, record in observed.items() if record is not None} -def test_langfuse_otel_identity_never_evicts_or_drops_another_attribute_of_a_full_span( - gateway: Gateway, tmp_path: Path -) -> None: - marker: Final = "lf-limit-" + uuid.uuid4().hex - with ( - _langfuse_rig(gateway, tmp_path, {}, marker) as rig, - rig.candidate.scenario() as scenario, - ): - key, identity = _scenario_identity(scenario, rig.provider) - - def body(tool_count: int) -> dict[str, object]: - return { - "model": identity["model"], - "messages": [{"role": "user", "content": f"{marker}:{tool_count}:"}], - "user": identity["end_user"], - "tools": [ - { - "type": "function", - "function": { - "name": f"tool_{n}", - "description": f"looks up thing {n}", - "parameters": {"type": "object", "properties": {"q": {"type": "string"}}}, - }, - } - for n in range(tool_count) - ], - "cache": {"no-cache": True}, - } - - spans: Final = _limit_sweep( - rig, key, marker, {tool_count: body(tool_count) for tool_count in (0, *range(30, 41))} - ) - - reference, reference_dropped = spans[0] - identity_count: Final = len(_identity_attributes(reference)) - assert reference_dropped == 0, reference_dropped - assert identity_count == len(_chat_identity_fields(identity)) + 1, sorted(reference) - kept_by_size: Final[dict[int, tuple[int, int]]] = {} # mutable-ok: filled once per swept span below - for tool_count, (attrs, dropped) in spans.items(): - kept = len(_identity_attributes(attrs)) - kept_by_size[tool_count] = (kept, dropped) - if kept: - assert dropped == 0, (tool_count, kept, dropped) - expected = {"openinference.span.kind", "metadata", "llm.model_name", "llm.request.type"} - assert expected <= set(attrs), (tool_count, sorted(expected - set(attrs))) - tools = {f"llm.tools.{n}.name" for n in range(tool_count)} - assert tools <= set(attrs), (tool_count, sorted(tools - set(attrs))) - assert [size for size, (kept, dropped) in kept_by_size.items() if kept < identity_count and not dropped], ( - kept_by_size - ) - assert [size for size, (_, dropped) in kept_by_size.items() if dropped], kept_by_size - - def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: Gateway, tmp_path: Path) -> None: marker: Final = "lf-v2-limit-" + uuid.uuid4().hex with ( @@ -1513,7 +1272,6 @@ def test_otel_v2_langfuse_identity_sheds_before_any_message_attribute(gateway: G {}, marker, overrides={ - "LITELLM_OTEL_V2": "1", "MAPPER_NAMES": "genai,openinference", "OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": "span_only", }, @@ -1589,7 +1347,7 @@ def test_otel_v2_span_another_tracer_wrote_keeps_its_langfuse_metadata_keys_at_t tmp_path, {"callbacks": ["langfuse_otel", "otel", "lit8283_app_span.instance"]}, marker, - overrides={"LITELLM_OTEL_V2": "1", "OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url}, + overrides={"OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url}, ) as rig, rig.candidate.scenario() as scenario, ): @@ -1647,7 +1405,6 @@ def test_otel_v2_collector_keeps_the_dropped_counts_of_a_span_without_langfuse_r {"callbacks": ["langfuse_otel", "otel"]}, marker, overrides={ - "LITELLM_OTEL_V2": "1", "OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": generic.url, "OTEL_SPAN_EVENT_COUNT_LIMIT": "0", @@ -1693,7 +1450,6 @@ def test_otel_v2_langfuse_mapper_sends_no_langfuse_request_metadata_to_a_plain_c {"callbacks": ["otel"]}, marker, overrides={ - "LITELLM_OTEL_V2": "1", "MAPPER_NAMES": "genai,langfuse", "OTEL_EXPORTER": "otlp_http", "OTEL_ENDPOINT": "__collector__", diff --git a/tests/integration/sdk/test_langfuse_otel_caller_span_sdk.py b/tests/integration/sdk/test_langfuse_otel_caller_span_sdk.py deleted file mode 100644 index 2e0f59974bd..00000000000 --- a/tests/integration/sdk/test_langfuse_otel_caller_span_sdk.py +++ /dev/null @@ -1,113 +0,0 @@ -import asyncio -import json -import time -from collections.abc import Mapping -from typing import Final - -import pytest -from integration._support.wire import Reply, Request, Wire, wire_server -from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest -from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.export import SimpleSpanProcessor -from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter - -import litellm -from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER - -_IDENTITY: Final = { - "user_api_key_alias": "caller-span-alias", - "user_api_key_user_id": "caller-span-user", - "user_api_key_team_id": "caller-span-team", -} -_CHAT_REPLY: Final = Reply( - body=json.dumps( - { - "id": "chatcmpl-caller-span", - "object": "chat.completion", - "created": 1, - "model": "gpt-4o-mini", - "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], - "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2}, - } - ).encode() -) - - -def _is_identity(key: str) -> bool: - return key == "langfuse.observation.metadata" or key.startswith("langfuse.trace.metadata.user_api_key_") - - -def _exported(langfuse: Wire, seen: list[tuple[str, dict[str, object]]]) -> list[tuple[str, dict[str, object]]]: - for batch in langfuse.drain(): - request: ExportTraceServiceRequest = ExportTraceServiceRequest.FromString(batch.body) - seen.extend( - ( - span.name, - { - attribute.key: getattr(attribute.value, attribute.value.WhichOneof("value")) - for attribute in span.attributes - }, - ) - for resource in request.resource_spans - for scope in resource.scope_spans - for span in scope.spans - ) - return seen - - -async def _exported_until( - langfuse: Wire, seen: list[tuple[str, dict[str, object]]], name: str, prompt: str -) -> list[tuple[str, dict[str, object]]]: - """The spans the Langfuse sink got, once a ``name`` span for ``prompt`` is among them; polls without blocking the loop.""" - deadline: Final = time.monotonic() + 25 - while not any(span == name and prompt in json.dumps(attributes) for span, attributes in _exported(langfuse, seen)): - assert time.monotonic() < deadline, [span for span, _ in seen] - await asyncio.sleep(0.2) - return seen - - -async def _complete(upstream: Wire, prompt: str) -> None: - await litellm.acompletion( - model="openai/gpt-4o-mini", - api_base=upstream.url + "/v1", - api_key="synthetic-caller-span-key", - messages=[{"role": "user", "content": prompt}], - metadata=dict(_IDENTITY), - ) - await GLOBAL_LOGGING_WORKER.flush() - - -@pytest.mark.asyncio -async def test_langfuse_otel_keeps_the_caller_identity_off_the_callers_active_span( - monkeypatch: pytest.MonkeyPatch, -) -> None: - app_exporter: Final = InMemorySpanExporter() - app_provider: Final = TracerProvider() - app_provider.add_span_processor(SimpleSpanProcessor(app_exporter)) - - def chat(_request: Request) -> Reply: - return _CHAT_REPLY - - with wire_server(chat) as upstream, wire_server(lambda _request: Reply()) as langfuse: - monkeypatch.setenv("LANGFUSE_PUBLIC_KEY", "pk-lf-caller-span") - monkeypatch.setenv("LANGFUSE_SECRET_KEY", "sk-lf-caller-span") - monkeypatch.setenv("LANGFUSE_OTEL_HOST", langfuse.url) - monkeypatch.setenv("OTEL_BSP_SCHEDULE_DELAY", "100") - monkeypatch.setattr(litellm, "callbacks", ["langfuse_otel"]) - - seen: Final[list[tuple[str, dict[str, object]]]] = [] # mutable-ok: drain() consumes the sink queue - with app_provider.get_tracer("app").start_as_current_span("app.operation"): - await _complete(upstream, "inside the caller span") - await _exported_until(langfuse, seen, "raw_gen_ai_request", "inside the caller span") - await _complete(upstream, "no caller span") - exported: Final = await _exported_until(langfuse, seen, "litellm_request", "no caller span") - - (app_span,) = app_exporter.get_finished_spans() - app_attributes: Final[Mapping[str, object]] = app_span.attributes or {} - assert app_attributes["langfuse.observation.type"] == "generation", sorted(app_attributes) - assert not [key for key in app_attributes if _is_identity(key)], sorted(app_attributes) - (own_span,) = [attributes for name, attributes in exported if name == "litellm_request"] - assert json.loads(str(own_span["langfuse.observation.metadata"])) == _IDENTITY, own_span - assert {key: value for key, value in own_span.items() if key.startswith("langfuse.trace.metadata.")} == { - f"langfuse.trace.metadata.{field}": value for field, value in _IDENTITY.items() - }, sorted(own_span) diff --git a/tests/unit/integrations/test_langfuse_otel.py b/tests/unit/integrations/test_langfuse_otel.py index fee996a5d1d..9f536ab0dac 100644 --- a/tests/unit/integrations/test_langfuse_otel.py +++ b/tests/unit/integrations/test_langfuse_otel.py @@ -1,20 +1,13 @@ import base64 import json import os -from collections.abc import Mapping -from contextlib import nullcontext from datetime import datetime, timezone -from types import MappingProxyType from typing import Final from unittest.mock import MagicMock, patch import pytest -from opentelemetry import trace -from opentelemetry.sdk.trace import ReadableSpan, SpanLimits, TracerProvider -from opentelemetry.sdk.trace.export import SimpleSpanProcessor -from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.sdk.trace import TracerProvider -import litellm from litellm.integrations.langfuse.langfuse_otel import LangfuseOtelLogger from litellm.integrations.opentelemetry import OpenTelemetry, OpenTelemetryConfig from litellm.types.llms.openai import ResponsesAPIResponse @@ -519,313 +512,6 @@ class TestLangfuseOtelIntegration: assert isinstance(config, OpenTelemetryConfig) # Endpoint assertion removed as side effect is gone - def test_request_metadata_is_emitted_under_langfuse_observation_and_trace_metadata_keys(self) -> None: - request_metadata: Final = { - "user_api_key_hash": "hash123", - "user_api_key_alias": "prod-key", - "user_api_key_user_id": "user-1", - "user_api_key_end_user_id": "end-user-1", - "user_api_key_team_id": "team-1", - "user_api_key_team_alias": "team-a", - "user_api_key_org_id": "org-1", - "spend_logs_metadata": {"env": "prod"}, - "requester_ip_address": "10.0.0.1", - "requester_metadata": {"ticket": "LIT-8283"}, - } - kwargs: Final[dict[str, object]] = { - "litellm_params": {"metadata": {}}, - "standard_logging_object": {"metadata": request_metadata}, - } - - with ( - patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, - patch("litellm.integrations.arize._utils.set_attributes"), - ): - LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) - - actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} - - assert json.loads(actual["langfuse.observation.metadata"]) == { - "user_api_key_alias": "prod-key", - "user_api_key_user_id": "user-1", - "user_api_key_end_user_id": "end-user-1", - "user_api_key_team_id": "team-1", - "user_api_key_team_alias": "team-a", - "user_api_key_org_id": "org-1", - } - assert {key: value for key, value in actual.items() if key.startswith("langfuse.trace.metadata.")} == { - "langfuse.trace.metadata.user_api_key_alias": "prod-key", - "langfuse.trace.metadata.user_api_key_user_id": "user-1", - "langfuse.trace.metadata.user_api_key_end_user_id": "end-user-1", - "langfuse.trace.metadata.user_api_key_team_id": "team-1", - "langfuse.trace.metadata.user_api_key_team_alias": "team-a", - } - - def test_request_metadata_redaction_matches_vanilla_langfuse(self, monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(litellm, "redact_user_api_key_info", True) - request_metadata: Final = { - "user_api_key_hash": "hash123", - "user_api_key_alias": "prod-key", - "user_api_key_user_id": "user-1", - "user_api_key_end_user_id": "end-user-1", - "user_api_key_team_id": "team-1", - "user_api_key_team_alias": "team-a", - "spend_logs_metadata": {"env": "prod"}, - "requester_ip_address": "10.0.0.1", - "requester_metadata": {"ticket": "LIT-8283"}, - } - kwargs: Final[dict[str, object]] = { - "litellm_params": {"metadata": {}}, - "standard_logging_object": {"metadata": request_metadata}, - } - - with ( - patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, - patch("litellm.integrations.arize._utils.set_attributes"), - ): - LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) - - actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} - - assert "langfuse.observation.metadata" not in actual - assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")] - - def test_request_metadata_beyond_the_identity_fields_stays_off_the_langfuse_metadata_keys(self) -> None: - request_metadata: Final = { - "user_api_key_alias": "prod-key", - "requester_ip_address": "10.0.0.1", - "user_api_key_user_email": "caller-email-secret", - "user_api_key_auth_metadata": {"note": "auth-metadata-secret"}, - "user_agent": "user-agent-secret", - "headers": {"x-tenant-secret": "header-secret"}, - "requester_custom_headers": {"x-tenant-secret": "custom-header-secret"}, - "requester_metadata": {"note": "requester-metadata-secret"}, - "spend_logs_metadata": {"note": "spend-logs-secret"}, - "prompt_management_metadata": {"prompt_variables": {"input": "prompt-variables-secret"}}, - "mcp_tool_call_metadata": {"arguments": {"location": "mcp-arguments-secret"}}, - "vector_store_request_metadata": [{"query": "vector-query-secret"}], - "routing_decision": {"matched_keyword": "routing-prompt-secret"}, - } - kwargs: Final[dict[str, object]] = { - "litellm_params": {"metadata": {}}, - "standard_logging_object": {"metadata": request_metadata}, - } - - with ( - patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, - patch("litellm.integrations.arize._utils.set_attributes"), - ): - LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) - - actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} - langfuse_metadata: Final = { - key: value - for key, value in actual.items() - if key == "langfuse.observation.metadata" or key.startswith("langfuse.trace.metadata.") - } - - assert langfuse_metadata == { - "langfuse.observation.metadata": json.dumps({"user_api_key_alias": "prod-key"}), - "langfuse.trace.metadata.user_api_key_alias": "prod-key", - } - - def test_request_metadata_drops_empty_fields_and_skips_empty_trace_identities(self) -> None: - request_metadata: Final = { - "user_api_key_alias": "prod-key", - "user_api_key_user_id": "user-1", - "user_api_key_end_user_id": None, - "user_api_key_team_id": "", - "user_api_key_team_alias": None, - "team_id": None, - "team_alias": None, - "spend_logs_metadata": {"env": "prod"}, - } - kwargs: Final[dict[str, object]] = { - "litellm_params": {"metadata": {}}, - "standard_logging_object": {"metadata": request_metadata}, - } - - with ( - patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, - patch("litellm.integrations.arize._utils.set_attributes"), - ): - LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) - - actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} - - assert json.loads(actual["langfuse.observation.metadata"]) == { - "user_api_key_alias": "prod-key", - "user_api_key_user_id": "user-1", - } - assert {key: value for key, value in actual.items() if key.startswith("langfuse.trace.metadata.")} == { - "langfuse.trace.metadata.user_api_key_alias": "prod-key", - "langfuse.trace.metadata.user_api_key_user_id": "user-1", - } - - def test_request_metadata_keys_are_absent_without_identity_fields(self) -> None: - kwargs: Final[dict[str, object]] = { - "litellm_params": {"metadata": {}}, - "standard_logging_object": { - "metadata": {"spend_logs_metadata": {"env": "prod"}, "requester_ip_address": "10.0.0.1"} - }, - } - - with ( - patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, - patch("litellm.integrations.arize._utils.set_attributes"), - ): - LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) - - actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} - - assert "langfuse.observation.metadata" not in actual - assert not [key for key in actual if key.startswith("langfuse.trace.metadata.")] - - def test_request_metadata_keys_are_absent_without_standard_logging_object(self) -> None: - kwargs: Final = {"litellm_params": {"metadata": {"trace_metadata": {"k": "v"}}}} - - with ( - patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, - patch("litellm.integrations.arize._utils.set_attributes"), - ): - LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, None) - - actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} - - assert [key for key in actual if key.startswith("langfuse.trace.metadata")] == ["langfuse.trace.metadata"] - assert "langfuse.observation.metadata" not in actual - - @pytest.mark.parametrize( - "unexpected_shape", - [ - {"prompt_management_metadata": {1: "one", "prompt_id": "prompt-1"}}, - {"mcp_tool_call_metadata": {1: "one", "name": "lookup_weather"}}, - {"vector_store_request_metadata": [{1: "one", "vector_store_id": "vs-1"}]}, - {"requester_metadata": 1}, - {"requester_metadata": True}, - {"requester_metadata": 1.5}, - {1: "one"}, - {"user_api_key_user_id": 5}, - {"user_api_key_team_id": {"id": "team-1"}}, - ], - ) - def test_request_metadata_of_unexpected_shape_keeps_the_observation_output( - self, unexpected_shape: dict[object, object] - ) -> None: - kwargs: Final[dict[str, object]] = { - "litellm_params": {"metadata": {}}, - "standard_logging_object": {"metadata": {**unexpected_shape, "user_api_key_alias": "prod-key"}}, - } - response_obj: Final = {"choices": [{"message": {"role": "assistant", "content": "hello"}}]} - - with ( - patch("litellm.integrations.arize._utils.safe_set_attribute") as mock_safe_set_attribute, - patch("litellm.integrations.arize._utils.set_attributes"), - ): - LangfuseOtelLogger.set_langfuse_otel_attributes(MagicMock(), kwargs, response_obj) - - actual: Final = {call.args[1]: call.args[2] for call in mock_safe_set_attribute.call_args_list} - - assert json.loads(actual["langfuse.observation.metadata"]) == {"user_api_key_alias": "prod-key"} - assert json.loads(actual["langfuse.observation.output"]) == {"role": "assistant", "content": "hello"} - - # What ``OpenTelemetry._record_exception_on_span`` adds to a failed call's span after its attributes. - _FAILED_CALL_ATTRIBUTES: Final = MappingProxyType( - { - "error.code": "429", - "http.response.status_code": 429, - "error.type": "RateLimitError", - "error.message": "rate limited", - "error.llm_provider": "openai", - "error.stack_trace": "Traceback", - } - ) - - @staticmethod - def _exported_span( - kwargs: dict[str, object], - max_span_attributes: int | None, - carried: dict[str, str] | None = None, - added_after: Mapping[str, str | int] = MappingProxyType({}), - active: bool = False, - ) -> ReadableSpan: - exporter: Final = InMemorySpanExporter() - provider: Final = TracerProvider(span_limits=SpanLimits(max_span_attributes=max_span_attributes)) - provider.add_span_processor(SimpleSpanProcessor(exporter)) - span: Final = provider.get_tracer("langfuse-otel-test").start_span("litellm_request", attributes=carried) - response_obj: Final = {"id": "chatcmpl-1", "choices": [{"message": {"role": "assistant", "content": "hello"}}]} - with trace.use_span(span, end_on_exit=False) if active else nullcontext(): - LangfuseOtelLogger.set_langfuse_otel_attributes(span, kwargs, response_obj) - span.set_attributes(added_after) - span.end() - (finished,) = exporter.get_finished_spans() - return finished - - @staticmethod - def _tool_call_kwargs() -> dict[str, object]: - tools: Final = [ - { - "type": "function", - "function": {"name": f"tool_{n}", "description": f"does thing {n}", "parameters": {"type": "object"}}, - } - for n in range(3) - ] - messages: Final = [{"role": "user", "content": "hello"}] - return { - "model": "gpt-4o", - "messages": messages, - "optional_params": {"tools": tools}, - "litellm_params": {"metadata": {}}, - "standard_logging_object": { - "model": "gpt-4o", - "custom_llm_provider": "openai", - "call_type": "completion", - "messages": messages, - "model_parameters": {"tools": tools}, - "metadata": {"user_api_key_alias": "prod-key", "user_api_key_team_id": "team-1"}, - }, - } - - def test_request_metadata_never_evicts_or_drops_another_attribute_at_the_span_limit( - self, monkeypatch: pytest.MonkeyPatch - ) -> None: - kwargs: Final = self._tool_call_kwargs() - failed: Final = self._FAILED_CALL_ATTRIBUTES - monkeypatch.setattr(litellm, "redact_user_api_key_info", True) - plain: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET, added_after=failed).attributes or {}) - succeeded: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET).attributes or {}) - monkeypatch.setattr(litellm, "redact_user_api_key_info", False) - full: Final = dict(self._exported_span(kwargs, SpanLimits.UNSET, added_after=failed).attributes or {}) - identity_keys: Final = set(full) - set(plain) - assert identity_keys == { - "langfuse.observation.metadata", - "langfuse.trace.metadata.user_api_key_alias", - "langfuse.trace.metadata.user_api_key_team_id", - } - - kept_identity: list[int] = [] # mutable-ok: one count per limit - for limit in range(len(plain), len(full) + 1): - failed_span = self._exported_span(kwargs, limit, added_after=failed) - succeeded_span = self._exported_span(kwargs, limit - len(failed)) - for span, expected in ((failed_span, plain), (succeeded_span, succeeded)): - attributes = dict(span.attributes or {}) - assert span.dropped_attributes == 0 - assert {key: attributes.get(key) for key in expected} == expected - assert set(attributes) - set(expected) <= identity_keys - kept_identity.append(len(set(failed_span.attributes or {}) & identity_keys)) - assert kept_identity == list(range(len(identity_keys) + 1)) - - @pytest.mark.parametrize("carried", [None, {"caller.attribute": "kept"}], ids=["bare", "with_attributes"]) - def test_request_metadata_skips_the_callers_active_span(self, carried: dict[str, str] | None) -> None: - attributes: Final = dict( - self._exported_span(self._tool_call_kwargs(), SpanLimits.UNSET, carried, active=True).attributes or {} - ) - - assert {key: attributes[key] for key in carried or {}} == (carried or {}) - assert attributes["langfuse.observation.type"] == "generation" - assert "langfuse.observation.metadata" not in attributes - assert not [key for key in attributes if key.startswith("langfuse.trace.metadata.")] - class TestLangfuseOtelKeyDynamicConfig: """Key/team-scoped Langfuse credentials must define the full export target @@ -1405,38 +1091,3 @@ def test_langfuse_otel_does_not_start_proxy_request_span() -> None: finally: langfuse_provider.shutdown() generic_provider.shutdown() - - -def test_langfuse_otel_success_writes_identity_onto_its_own_span_and_never_onto_the_callers() -> None: - langfuse_exporter: Final = InMemorySpanExporter() - langfuse_provider: Final = TracerProvider() - langfuse_provider.add_span_processor(SimpleSpanProcessor(langfuse_exporter)) - app_exporter: Final = InMemorySpanExporter() - app_provider: Final = TracerProvider() - app_provider.add_span_processor(SimpleSpanProcessor(app_exporter)) - with patch.dict(os.environ, LANGFUSE_ENV_ONLY, clear=True): - logger: Final = LangfuseOtelLogger(callback_name="langfuse_otel", tracer_provider=langfuse_provider) - response_obj: Final = {"id": "chatcmpl-1", "choices": [{"message": {"role": "assistant", "content": "hello"}}]} - started_at: Final = datetime(2026, 1, 1, tzinfo=timezone.utc) - try: - with app_provider.get_tracer("app").start_as_current_span("app.operation"): - logger._handle_success( - TestLangfuseOtelIntegration._tool_call_kwargs(), response_obj, started_at, started_at - ) - logger._handle_success(TestLangfuseOtelIntegration._tool_call_kwargs(), response_obj, started_at, started_at) - - (app_span,) = app_exporter.get_finished_spans() - (own_span,) = [span for span in langfuse_exporter.get_finished_spans() if span.name == "litellm_request"] - app_attributes: Final = dict(app_span.attributes or {}) - own_attributes: Final = dict(own_span.attributes or {}) - assert app_attributes["langfuse.observation.type"] == "generation" - identity_prefixes: Final = ("langfuse.observation.metadata", "langfuse.trace.metadata.") - assert not [key for key in app_attributes if key.startswith(identity_prefixes)] - assert json.loads(str(own_attributes["langfuse.observation.metadata"])) == { - "user_api_key_alias": "prod-key", - "user_api_key_team_id": "team-1", - } - assert own_attributes["langfuse.trace.metadata.user_api_key_team_id"] == "team-1" - finally: - langfuse_provider.shutdown() - app_provider.shutdown()