scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36
litellm/tests/test_litellm/integrations/test_opentelemetry.py
devin-ai-integration[bot] 4bf40c4e8d
fix(logging): stop billing and logging response reads as LLM calls (#36890) 
* fix(logging): stop billing and logging response reads as LLM calls

Retrieving, deleting or cancelling a stored response, and vector store management calls, run through the same logging lifecycle as inference. A retrieved response replays the usage of the call that created it, so every read priced it again and wrote a second spend log row for the same tokens. Non-inference calls now cost 0, report no usage, log no placeholder chat message, and get a litellm.responses_management operation name instead of reading as chat.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(responses): keep billing background response jobs after the poll

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(logging): use an empty list for read-call messages

A tuple matches no branch in the loggers that walk this value, so lunary's
parse_messages falls through to clean_message and raises AttributeError on the
success hook. An empty list reads as no messages everywhere: it satisfies the
isinstance(list) checks in newrelic, mlflow and datadog, iterates zero times in
traceloop and helicone, and is what StandardLoggingPayload.messages is typed to
hold. None would be type-legal too but is not iterable, so it trades one crash
for another in mlflow and traceloop.

* fix(otel): stop the legacy emitter reporting replayed tokens on response reads

The zeroing so far lands in the standard logging payload, which the legacy
OpenTelemetry emitter does not read for usage: it takes prompt, completion and
total tokens straight off the response object, so a retrieval span still carried
the token counts of the call that produced the response, and the token usage
histogram still recorded them. That emitter is the default, so the spend row said
zero while the trace said otherwise. The background cost poller keeps its counts,
the same exemption the pricing path already makes.

* fix(logging): keep billing a background response when its retrieval is read

A response created with background=true comes back queued and carries no usage, so
its create bills nothing. The retrieval that first sees the finished job is the only
place that job's tokens are ever visible, and pricing every read at zero therefore
loses the spend outright rather than deduplicating it. On a proxy without the
enterprise cost poller a background job ended up costing $0 end to end.

is_unbilled_non_inference_call now takes the response it is deciding about and treats
a background response the same way it already treats the poller's own read, which is
the same exemption seen from the other side. The legacy OpenTelemetry emitter's time
per output token metric picks up the read gate it was missing, so it stops dividing a
read's latency by the replayed completion token count.

* test(proxy): pass the read response to the non-inference predicate

The poller test called is_unbilled_non_inference_call with the pre-background signature, so it broke when the predicate gained the response it classifies. It now hands the predicate a foreground read, and asserts that the same read is free without the origin stamp, so the stamp is what the test proves.

* fix(otel): stop the v2 metrics recorder reporting replayed tokens on response reads

The v2 span builder sources usage from the standard logging payload, so the
earlier fix already zeroes it there. The metrics recorder reads response_obj
directly, so a responses-management read still recorded the original
generation's tokens into gen_ai.client.token.usage and divided generation time
by them for gen_ai.server.time_per_output_token.

The read still records operation and response duration, under the
litellm.responses_management operation, so it stays observable.

* fix(proxy): keep the response-cost headers on calls priced at zero

Pricing responses reads and vector-store management routes at zero dropped the whole
x-litellm-response-cost family off those replies. The header build reads a falsy zero as
a cost this response never recorded and filters it out, and a call that returns before
pricing stores no cost breakdown for the component headers to read, so a client parsing
the cost off a read got a KeyError where it had previously been handed a number.

Those calls now advertise the family at zero. Retrieving a background response, and the
cost poller's read of one, still report their real cost.

The params-taking form of the predicate moves from opentelemetry into
internal_call_metadata so the proxy header build and the OTEL recorders share one copy.

* fix(proxy): report a zero cost split only under a zero cost total

The component headers were filled from call-type membership alone, while the
total they sit beside keeps its real value when the read priced normally, so a
breakdown that had not landed by the time headers were built could advertise a
real total next to an all-zero split. The split is now reported as zero only
when the total agrees with it, and is otherwise left absent.

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Yucheng Zhu <yucheng@berri.ai>
2026-08-26 18:34:17 -07:00

6439 lines
259 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import asyncio
import concurrent.futures
import gc
import json
import os
import sys
import threading
import time
import unittest
import weakref
from datetime import datetime, timedelta, timezone
from types import MappingProxyType
from parameterized import parameterized
from unittest.mock import MagicMock, patch
# Adds the grandparent directory to sys.path to allow importing project modules
from opentelemetry import trace
from opentelemetry.sdk._logs import LoggerProvider as OTLoggerProvider
from opentelemetry.sdk._logs.export import InMemoryLogExporter, SimpleLogRecordProcessor
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import InMemoryMetricReader
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.integrations import opentelemetry as otel_module
from litellm.integrations.opentelemetry import (
OpenTelemetry,
OpenTelemetryConfig,
OTELMetricAttributeFilter,
OTELSemconvCategory,
_normalize_team_metadata_keys,
)
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.types.services import ServiceLoggerPayload, ServiceTypes
class TestOpenTelemetryGuardrails(unittest.TestCase):
@patch("litellm.integrations.opentelemetry.datetime")
def test_create_guardrail_span_with_valid_info(self, mock_datetime):
# Setup
otel = OpenTelemetry()
otel.tracer = MagicMock()
mock_span = MagicMock()
otel.tracer.start_span.return_value = mock_span
# Create guardrail information
guardrail_info = {
"guardrail_name": "test_guardrail",
"guardrail_mode": "input",
"masked_entity_count": {"CREDIT_CARD": 2},
"guardrail_response": "filtered_content",
"start_time": 1609459200.0,
"end_time": 1609459201.0,
}
# Create a kwargs dict with standard_logging_object containing guardrail information
kwargs = {
"standard_logging_object": {"guardrail_information": [guardrail_info]}
}
# Call the method
otel._create_guardrail_span(kwargs=kwargs, context=None)
# Assertions
otel.tracer.start_span.assert_called_once()
# print all calls to mock_span.set_attribute
print("Calls to mock_span.set_attribute:")
for call in mock_span.set_attribute.call_args_list:
print(call)
# Check that the span has the correct attributes set
mock_span.set_attribute.assert_any_call("guardrail_name", "test_guardrail")
mock_span.set_attribute.assert_any_call("guardrail_mode", "input")
mock_span.set_attribute.assert_any_call(
"guardrail_response", safe_dumps("filtered_content")
)
mock_span.set_attribute.assert_any_call(
"masked_entity_count", safe_dumps({"CREDIT_CARD": 2})
)
# Verify that the span was ended
mock_span.end.assert_called_once()
def test_create_guardrail_span_with_no_info(self):
# Setup
otel = OpenTelemetry()
otel.tracer = MagicMock()
# Test with no guardrail information
kwargs = {"standard_logging_object": {}}
otel._create_guardrail_span(kwargs=kwargs, context=None)
# Verify that start_span was never called
otel.tracer.start_span.assert_not_called()
@patch("litellm.integrations.opentelemetry.datetime")
def test_guardrail_response_dict_is_json_serialized(self, mock_datetime):
"""Dict guardrail_response (e.g. OpenAI moderation result) must reach
the span as a JSON string so downstream pipelines can parse it for
metric extraction — this is the bug the PR fixes."""
otel = OpenTelemetry()
otel.tracer = MagicMock()
mock_span = MagicMock()
otel.tracer.start_span.return_value = mock_span
moderation_payload = {
"id": "modr-7740",
"model": "omni-moderation-latest",
"results": [{"categories": {"harassment": False}}],
}
guardrail_info = {
"guardrail_name": "test_guardrail",
"guardrail_mode": "input",
"guardrail_response": moderation_payload,
"start_time": 1609459200.0,
"end_time": 1609459201.0,
}
kwargs = {
"standard_logging_object": {"guardrail_information": [guardrail_info]}
}
otel._create_guardrail_span(kwargs=kwargs, context=None)
mock_span.set_attribute.assert_any_call(
"guardrail_response", safe_dumps(moderation_payload)
)
@patch("litellm.integrations.opentelemetry.datetime")
def test_guardrail_response_none_is_skipped(self, mock_datetime):
"""When guardrail_response is None, the attribute must not be set —
guards against round-tripping ``"null"`` into traces."""
otel = OpenTelemetry()
otel.tracer = MagicMock()
mock_span = MagicMock()
otel.tracer.start_span.return_value = mock_span
guardrail_info = {
"guardrail_name": "test_guardrail",
"guardrail_mode": "input",
"guardrail_response": None,
"start_time": 1609459200.0,
"end_time": 1609459201.0,
}
kwargs = {
"standard_logging_object": {"guardrail_information": [guardrail_info]}
}
otel._create_guardrail_span(kwargs=kwargs, context=None)
attribute_keys = [
call.args[0] for call in mock_span.set_attribute.call_args_list
]
self.assertNotIn("guardrail_response", attribute_keys)
class TestOpenTelemetryTeamAttributesOnChildSpans(unittest.TestCase):
"""team_id / team_alias must land on every child span of a
litellm_request trace, not only the root litellm_request span."""
def _slo_metadata(self):
return {
"user_api_key_team_id": "team-123",
"user_api_key_team_alias": "my-team",
}
@patch("litellm.integrations.opentelemetry.datetime")
def test_guardrail_span_has_team_attributes(self, mock_datetime):
otel = OpenTelemetry()
otel.tracer = MagicMock()
mock_span = MagicMock()
otel.tracer.start_span.return_value = mock_span
guardrail_info = {
"guardrail_name": "test_guardrail",
"guardrail_mode": "input",
"guardrail_response": "filtered_content",
"start_time": 1609459200.0,
"end_time": 1609459201.0,
}
kwargs = {
"standard_logging_object": {
"guardrail_information": [guardrail_info],
"metadata": self._slo_metadata(),
}
}
otel._create_guardrail_span(kwargs=kwargs, context=None)
mock_span.set_attribute.assert_any_call(
"metadata.user_api_key_team_id", "team-123"
)
mock_span.set_attribute.assert_any_call(
"metadata.user_api_key_team_alias", "my-team"
)
@patch.dict(os.environ, {"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": ""})
@patch("litellm.turn_off_message_logging", False)
def test_raw_request_span_has_team_attributes(self):
otel = OpenTelemetry()
otel.message_logging = True
mock_tracer = MagicMock()
mock_span = MagicMock()
mock_tracer.start_span.return_value = mock_span
otel.get_tracer_to_use_for_request = MagicMock(return_value=mock_tracer)
otel.set_raw_request_attributes = MagicMock()
otel._to_ns = MagicMock(return_value=1234567890)
kwargs = {
"litellm_params": {"metadata": {}},
"standard_logging_object": {"metadata": self._slo_metadata()},
}
otel._maybe_log_raw_request(
kwargs, {}, datetime.now(), datetime.now(), MagicMock()
)
mock_span.set_attribute.assert_any_call(
"metadata.user_api_key_team_id", "team-123"
)
mock_span.set_attribute.assert_any_call(
"metadata.user_api_key_team_alias", "my-team"
)
def test_helper_skips_when_team_values_missing(self):
otel = OpenTelemetry()
mock_span = MagicMock()
otel._set_team_attributes_on_span(span=mock_span, team_id=None, team_alias=None)
mock_span.set_attribute.assert_not_called()
def test_helper_skips_when_team_values_are_empty_strings(self):
"""A master-key / team-less request carries user_api_key_team_id=''
in metadata. Propagating '' to every span is noise that makes
traces look mis-instrumented; treat empty as absent."""
otel = OpenTelemetry()
mock_span = MagicMock()
otel._set_team_attributes_on_span(span=mock_span, team_id="", team_alias="")
mock_span.set_attribute.assert_not_called()
def test_helper_reads_metadata_from_kwargs(self):
otel = OpenTelemetry()
mock_span = MagicMock()
otel._set_team_attributes_from_kwargs(
mock_span,
{"standard_logging_object": {"metadata": self._slo_metadata()}},
)
mock_span.set_attribute.assert_any_call(
"metadata.user_api_key_team_id", "team-123"
)
mock_span.set_attribute.assert_any_call(
"metadata.user_api_key_team_alias", "my-team"
)
def test_helper_handles_missing_standard_logging_object(self):
otel = OpenTelemetry()
mock_span = MagicMock()
otel._set_team_attributes_from_kwargs(mock_span, {})
mock_span.set_attribute.assert_not_called()
def test_failure_hook_exception_span_has_team_attributes(self):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer(__name__)
otel = OpenTelemetry()
otel.tracer = tracer
server_span = tracer.start_span("Received Proxy Server Request")
user_api_key_dict = MagicMock()
user_api_key_dict.parent_otel_span = server_span
user_api_key_dict.team_id = "team-123"
user_api_key_dict.team_alias = "my-team"
asyncio.run(
otel.async_post_call_failure_hook(
request_data={},
original_exception=ValueError("boom"),
user_api_key_dict=user_api_key_dict,
traceback_str="trace",
)
)
finished = {s.name: s for s in exporter.get_finished_spans()}
exception_span = finished["Failed Proxy Server Request"]
assert exception_span.attributes["metadata.user_api_key_team_id"] == "team-123"
assert (
exception_span.attributes["metadata.user_api_key_team_alias"] == "my-team"
)
class TestOpenTelemetryCostBreakdown(unittest.TestCase):
def test_cost_breakdown_emitted_to_otel_span(self):
"""
Test that cost breakdown from StandardLoggingPayload is emitted to OpenTelemetry span attributes.
"""
otel = OpenTelemetry()
mock_span = MagicMock()
cost_breakdown = {
"input_cost": 0.001,
"output_cost": 0.002,
"total_cost": 0.003,
"tool_usage_cost": 0.0001,
"original_cost": 0.004,
"discount_percent": 0.25,
"discount_amount": 0.001,
}
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
"cost_breakdown": cost_breakdown,
},
}
response_obj = {
"id": "test-response-id",
"model": "gpt-4",
"choices": [],
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
mock_span.set_attribute.assert_any_call("gen_ai.cost.input_cost", 0.001)
mock_span.set_attribute.assert_any_call("gen_ai.cost.output_cost", 0.002)
mock_span.set_attribute.assert_any_call("gen_ai.cost.total_cost", 0.003)
mock_span.set_attribute.assert_any_call("gen_ai.cost.tool_usage_cost", 0.0001)
mock_span.set_attribute.assert_any_call("gen_ai.cost.original_cost", 0.004)
mock_span.set_attribute.assert_any_call("gen_ai.cost.discount_percent", 0.25)
mock_span.set_attribute.assert_any_call("gen_ai.cost.discount_amount", 0.001)
def test_cost_breakdown_with_partial_fields(self):
"""
Test that cost breakdown works correctly when only some fields are present.
"""
otel = OpenTelemetry()
mock_span = MagicMock()
cost_breakdown = {
"input_cost": 0.001,
"output_cost": 0.002,
"total_cost": 0.003,
}
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
"cost_breakdown": cost_breakdown,
},
}
response_obj = {
"id": "test-response-id",
"model": "gpt-4",
"choices": [],
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
mock_span.set_attribute.assert_any_call("gen_ai.cost.input_cost", 0.001)
mock_span.set_attribute.assert_any_call("gen_ai.cost.output_cost", 0.002)
mock_span.set_attribute.assert_any_call("gen_ai.cost.total_cost", 0.003)
call_args_list = [call[0] for call in mock_span.set_attribute.call_args_list]
assert ("gen_ai.cost.tool_usage_cost", 0.0001) not in call_args_list
assert ("gen_ai.cost.original_cost", 0.004) not in call_args_list
class TestOpenTelemetryProviderInitialization(unittest.TestCase):
"""Test suite for verifying provider initialization respects existing providers"""
def test_init_tracing_respects_existing_tracer_provider(self):
"""
Unit test: _init_tracing() should respect existing TracerProvider.
When a TracerProvider already exists (e.g., set by Langfuse SDK),
LiteLLM should use it instead of creating a new one.
"""
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
# Setup: Create and set an existing TracerProvider
tracer_provider = TracerProvider()
trace.set_tracer_provider(tracer_provider)
existing_provider = trace.get_tracer_provider()
# Act: Initialize OpenTelemetry integration (should detect existing provider)
otel_integration = OpenTelemetry()
# Assert: The existing provider should still be active
current_provider = trace.get_tracer_provider()
assert (
current_provider is existing_provider
), "Existing TracerProvider should be respected and not overridden"
@patch.dict(
os.environ, {"LITELLM_OTEL_INTEGRATION_ENABLE_METRICS": "true"}, clear=True
)
def test_init_metrics_creates_instruments_under_their_published_names(self):
"""
The v1 engine's instrument names are a public contract.
Every name here is what a backend queries: four are GenAI semantic
conventions and gen_ai.usage.cost is the name backends query for spend.
A rename is breaking for anyone charting them, so it has to be a
deliberate edit to the shared Metric constants and to this list, never
a silent drift between the v1 and v2 engines.
"""
from opentelemetry import metrics
metrics.set_meter_provider(MeterProvider(metric_readers=[InMemoryMetricReader()]))
otel_integration = OpenTelemetry(config=OpenTelemetryConfig.from_env())
assert {
otel_integration._operation_duration_histogram.name,
otel_integration._token_usage_histogram.name,
otel_integration._cost_histogram.name,
otel_integration._time_to_first_token_histogram.name,
otel_integration._time_per_output_token_histogram.name,
otel_integration._response_duration_histogram.name,
} == {
"gen_ai.client.operation.duration",
"gen_ai.client.token.usage",
"gen_ai.usage.cost",
"gen_ai.server.time_to_first_token",
"gen_ai.server.time_per_output_token",
"gen_ai.client.response.duration",
}
@patch.dict(
os.environ, {"LITELLM_OTEL_INTEGRATION_ENABLE_METRICS": "true"}, clear=True
)
def test_init_metrics_respects_existing_meter_provider(self):
"""
Unit test: _init_metrics() should respect existing MeterProvider.
When a MeterProvider already exists (e.g., set by Langfuse SDK),
LiteLLM should use it instead of creating a new one.
"""
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider
# Create and set an existing MeterProvider
meter_provider = MeterProvider()
metrics.set_meter_provider(meter_provider)
existing_provider = metrics.get_meter_provider()
# Act: Initialize OpenTelemetry integration (should detect existing provider)
config = OpenTelemetryConfig.from_env()
otel_integration = OpenTelemetry(config=config)
# Assert: The existing provider should still be active
current_provider = metrics.get_meter_provider()
assert (
current_provider is existing_provider
), "Existing MeterProvider should be respected and not overridden"
@patch.dict(
os.environ, {"LITELLM_OTEL_INTEGRATION_ENABLE_EVENTS": "true"}, clear=True
)
def test_init_logs_respects_existing_logger_provider(self):
"""
Unit test: _init_logs() should respect existing LoggerProvider.
When a LoggerProvider already exists (e.g., set by Langfuse SDK),
LiteLLM should use it instead of creating a new one.
"""
from opentelemetry._logs import get_logger_provider, set_logger_provider
from opentelemetry.sdk._logs import LoggerProvider as OTLoggerProvider
# Create and set an existing LoggerProvider
logger_provider = OTLoggerProvider()
set_logger_provider(logger_provider)
existing_provider = get_logger_provider()
# Act: Initialize OpenTelemetry integration (should detect existing provider)
config = OpenTelemetryConfig.from_env()
otel_integration = OpenTelemetry(config=config)
# Assert: The existing provider should still be active
current_provider = get_logger_provider()
assert (
current_provider is existing_provider
), "Existing LoggerProvider should be respected and not overridden"
class TestOpenTelemetryDualHandlerIsolation(unittest.TestCase):
"""Two OpenTelemetry handlers coexisting via skip_set_global=True
must each get their own provider for every signal (tracer/meter/logger)."""
@staticmethod
def _wire_span_processor(exporter):
"""Context manager: while active, the next OpenTelemetry instance
wires its TracerProvider to `exporter`."""
return patch.object(
OpenTelemetry,
"_get_span_processor",
lambda self, dynamic_headers=None: SimpleSpanProcessor(exporter),
)
def test_skip_set_global_creates_isolated_tracer_provider(self):
from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider
fake_existing = SDKTracerProvider()
own_exporter = InMemorySpanExporter()
cfg = OpenTelemetryConfig(
exporter="console", service_name="iso-test", skip_set_global=True
)
with (
patch.object(trace, "get_tracer_provider", return_value=fake_existing),
patch.object(trace, "set_tracer_provider") as mock_set,
self._wire_span_processor(own_exporter),
):
handler = OpenTelemetry(config=cfg)
self.assertIsNot(handler._tracer_provider, fake_existing)
mock_set.assert_not_called()
handler.tracer.start_span("isolation_check").end()
handler._tracer_provider.force_flush(2000)
self.assertEqual(
[s.name for s in own_exporter.get_finished_spans()],
["isolation_check"],
)
def test_skip_set_global_via_callback_name_back_compat(self):
from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider
fake_existing = SDKTracerProvider()
cfg = OpenTelemetryConfig(exporter="console", service_name="lf-back-compat")
with (
patch.object(trace, "get_tracer_provider", return_value=fake_existing),
patch.object(trace, "set_tracer_provider"),
self._wire_span_processor(InMemorySpanExporter()),
):
handler = OpenTelemetry(config=cfg, callback_name="langfuse_otel")
self.assertIsNot(handler._tracer_provider, fake_existing)
def test_default_behavior_reuses_existing_sdk_tracer_provider(self):
from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider
fake_existing = SDKTracerProvider()
with patch.object(trace, "get_tracer_provider", return_value=fake_existing):
handler = OpenTelemetry(config=OpenTelemetryConfig(service_name="shared"))
self.assertIs(handler._tracer_provider, fake_existing)
def test_skip_set_global_creates_isolated_meter_provider(self):
from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider as SDKMeterProvider
fake_existing = SDKMeterProvider()
cfg = OpenTelemetryConfig(
exporter="console",
service_name="meter-iso-test",
enable_metrics=True,
skip_set_global=True,
)
with (
patch.object(metrics, "get_meter_provider", return_value=fake_existing),
patch.object(metrics, "set_meter_provider") as mock_set,
self._wire_span_processor(InMemorySpanExporter()),
):
handler = OpenTelemetry(config=cfg)
self.assertIsNot(handler._meter_provider, fake_existing)
mock_set.assert_not_called()
def test_skip_set_global_creates_isolated_logger_provider(self):
from opentelemetry import _logs
from opentelemetry.sdk._logs import LoggerProvider as SDKLoggerProvider
fake_existing = SDKLoggerProvider()
cfg = OpenTelemetryConfig(
exporter="console",
service_name="logger-iso-test",
enable_events=True,
skip_set_global=True,
)
with (
patch.object(_logs, "get_logger_provider", return_value=fake_existing),
patch.object(_logs, "set_logger_provider") as mock_set,
self._wire_span_processor(InMemorySpanExporter()),
):
handler = OpenTelemetry(config=cfg)
self.assertIsNot(handler._logger_provider, fake_existing)
mock_set.assert_not_called()
def test_emitted_logs_route_to_isolated_logger_provider(self):
# End-to-end: emitted logs land in the handler's private LoggerProvider,
# not the global one. Guards against get_logger() bypassing self._logger_provider.
from opentelemetry import _logs
from opentelemetry.sdk._logs import LoggerProvider as SDKLoggerProvider
global_exporter = InMemoryLogExporter()
fake_existing = SDKLoggerProvider()
fake_existing.add_log_record_processor(
SimpleLogRecordProcessor(global_exporter)
)
private_exporter = InMemoryLogExporter()
cfg = OpenTelemetryConfig(
exporter="console",
service_name="logger-emit-test",
enable_events=True,
skip_set_global=True,
)
with (
patch.object(_logs, "get_logger_provider", return_value=fake_existing),
patch.object(_logs, "set_logger_provider"),
patch.object(
OpenTelemetry, "_get_log_exporter", return_value=private_exporter
),
self._wire_span_processor(InMemorySpanExporter()),
):
handler = OpenTelemetry(config=cfg)
span = handler.tracer.start_span("emit-test")
handler._emit_semantic_logs(
kwargs={"messages": [{"role": "user", "content": "hi"}]},
response_obj={"choices": []},
span=span,
)
span.end()
handler._logger_provider.force_flush(2000)
self.assertGreater(len(private_exporter.get_finished_logs()), 0)
self.assertEqual(len(global_exporter.get_finished_logs()), 0)
def test_two_handlers_each_receive_their_own_spans(self):
# Handler A gets explicit injection (production-ish: claims the global).
exporter_a = InMemorySpanExporter()
provider_a = TracerProvider()
provider_a.add_span_processor(SimpleSpanProcessor(exporter_a))
handler_a = OpenTelemetry(
config=OpenTelemetryConfig(service_name="handler-a"),
tracer_provider=provider_a,
)
# Handler B comes along with the global appearing to be A's provider.
exporter_b = InMemorySpanExporter()
cfg_b = OpenTelemetryConfig(
exporter="console", service_name="handler-b", skip_set_global=True
)
with (
patch.object(trace, "get_tracer_provider", return_value=provider_a),
patch.object(trace, "set_tracer_provider"),
self._wire_span_processor(exporter_b),
):
handler_b = OpenTelemetry(config=cfg_b)
self.assertIsNot(handler_a._tracer_provider, handler_b._tracer_provider)
handler_a.tracer.start_span("from_handler_a").end()
handler_b.tracer.start_span("from_handler_b").end()
provider_a.force_flush(2000)
handler_b._tracer_provider.force_flush(2000)
self.assertEqual(
sorted(s.name for s in exporter_a.get_finished_spans()),
["from_handler_a"],
)
self.assertEqual(
sorted(s.name for s in exporter_b.get_finished_spans()),
["from_handler_b"],
)
class TestOpenTelemetryCaptureMessageContent(unittest.TestCase):
"""OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT and the
OpenTelemetryConfig.capture_message_content programmatic override
drive what the handler captures in spans vs events."""
@staticmethod
def _make(env=None, config_value=None, message_logging=True):
env_dict = (
{"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": env}
if env is not None
else {"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": ""}
)
with patch.dict(os.environ, env_dict):
handler = OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console", capture_message_content=config_value
)
)
handler.message_logging = message_logging
return handler, handler._resolve_capture_mode()
def test_no_explicit_setting_falls_back_to_message_logging_true(self):
_, mode = self._make()
self.assertEqual(mode, "SPAN_AND_EVENT")
def test_no_explicit_setting_falls_back_to_message_logging_false(self):
_, mode = self._make(message_logging=False)
self.assertEqual(mode, "NO_CONTENT")
def test_env_var_no_content(self):
_, mode = self._make(env="NO_CONTENT")
self.assertEqual(mode, "NO_CONTENT")
def test_env_var_span_only(self):
_, mode = self._make(env="SPAN_ONLY")
self.assertEqual(mode, "SPAN_ONLY")
def test_env_var_event_only(self):
_, mode = self._make(env="EVENT_ONLY")
self.assertEqual(mode, "EVENT_ONLY")
def test_env_var_span_and_event(self):
_, mode = self._make(env="SPAN_AND_EVENT")
self.assertEqual(mode, "SPAN_AND_EVENT")
def test_env_var_legacy_true_maps_to_event_only(self):
_, mode = self._make(env="true")
self.assertEqual(mode, "EVENT_ONLY")
def test_env_var_legacy_false_maps_to_no_content(self):
for env in ("false", "0"):
with self.subTest(env=env):
_, mode = self._make(env=env)
self.assertEqual(mode, "NO_CONTENT")
def test_env_var_unknown_value_falls_through_to_legacy(self):
_, mode = self._make(env="garbage", message_logging=True)
self.assertEqual(mode, "SPAN_AND_EVENT")
def test_config_field_overrides_env(self):
_, mode = self._make(env="EVENT_ONLY", config_value="SPAN_ONLY")
self.assertEqual(mode, "SPAN_ONLY")
def test_turn_off_message_logging_forces_no_content(self):
with patch("litellm.turn_off_message_logging", True):
_, mode = self._make(env="SPAN_AND_EVENT", message_logging=True)
self.assertEqual(mode, "NO_CONTENT")
def test_capture_in_span_and_event_predicates(self):
cases = {
"NO_CONTENT": (False, False),
"SPAN_ONLY": (True, False),
"EVENT_ONLY": (False, True),
"SPAN_AND_EVENT": (True, True),
}
for mode, (in_span, in_event) in cases.items():
handler, _ = self._make(env=mode)
self.assertEqual(handler._capture_in_span(), in_span, msg=mode)
self.assertEqual(handler._capture_in_event(), in_event, msg=mode)
def test_two_handlers_can_have_different_modes(self):
# FIL's stated requirement: one handler strips content, the other keeps it.
with patch.dict(
os.environ, {"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": ""}
):
stripped = OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console", capture_message_content="NO_CONTENT"
)
)
kept = OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console", capture_message_content="SPAN_AND_EVENT"
)
)
self.assertEqual(stripped._resolve_capture_mode(), "NO_CONTENT")
self.assertEqual(kept._resolve_capture_mode(), "SPAN_AND_EVENT")
self.assertFalse(stripped._capture_in_span())
self.assertFalse(stripped._capture_in_event())
self.assertTrue(kept._capture_in_span())
self.assertTrue(kept._capture_in_event())
class TestOpenTelemetrySemconvStability(unittest.TestCase):
"""OTEL_SEMCONV_STABILITY_OPT_IN=gen_ai_latest_experimental opts into
semconv-conformant span shape (name, kind, no raw_gen_ai_request child)."""
@staticmethod
def _make(env=None, config_value=None):
env_value = env if env is not None else ""
with patch.dict(os.environ, {"OTEL_SEMCONV_STABILITY_OPT_IN": env_value}):
return OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console",
semconv_stability_opt_in=config_value or set(),
)
)
def test_default_unset_keeps_legacy_span_name(self):
h = self._make()
self.assertFalse(h._gen_ai_semconv_latest_experimental)
kwargs = {"model": "gpt-4", "call_type": "acompletion"}
self.assertEqual(h._get_span_name(kwargs), "litellm_request")
def test_opt_in_emits_semconv_span_name(self):
h = self._make(env="gen_ai_latest_experimental")
self.assertTrue(h._gen_ai_semconv_latest_experimental)
kwargs = {"model": "gpt-4", "call_type": "acompletion"}
self.assertEqual(h._get_span_name(kwargs), "chat gpt-4")
def test_opt_in_supports_comma_separated_categories(self):
h = self._make(env="other_category,gen_ai_latest_experimental")
self.assertTrue(h._gen_ai_semconv_latest_experimental)
def test_opt_in_ignores_unrelated_category(self):
h = self._make(env="some_other_category")
self.assertFalse(h._gen_ai_semconv_latest_experimental)
def test_config_field_enables_without_env(self):
h = self._make(
env="", config_value={OTELSemconvCategory.GEN_AI_LATEST_EXPERIMENTAL}
)
self.assertTrue(h._gen_ai_semconv_latest_experimental)
def test_config_field_unions_with_env(self):
h = self._make(
env="gen_ai_latest_experimental",
config_value={OTELSemconvCategory.GEN_AI_LATEST_EXPERIMENTAL},
)
self.assertTrue(h._gen_ai_semconv_latest_experimental)
def test_operation_name_for_embeddings(self):
h = self._make(env="gen_ai_latest_experimental")
kwargs = {
"model": "text-embedding-3-small",
"call_type": "aembedding",
}
self.assertEqual(h._get_span_name(kwargs), "embeddings text-embedding-3-small")
def test_operation_name_for_text_completion(self):
h = self._make(env="gen_ai_latest_experimental")
kwargs = {"model": "babbage-002", "call_type": "atext_completion"}
self.assertEqual(h._get_span_name(kwargs), "text_completion babbage-002")
def test_operation_name_defaults_to_chat(self):
h = self._make(env="gen_ai_latest_experimental")
kwargs = {"model": "claude-sonnet-4-5", "call_type": "unknown"}
self.assertEqual(h._get_span_name(kwargs), "chat claude-sonnet-4-5")
def test_generation_name_metadata_overrides_semconv_name(self):
h = self._make(env="gen_ai_latest_experimental")
kwargs = {
"model": "gpt-4",
"call_type": "acompletion",
"litellm_params": {"metadata": {"generation_name": "user-named-span"}},
}
self.assertEqual(h._get_span_name(kwargs), "user-named-span")
def test_opt_in_skips_raw_gen_ai_request_span(self):
h = self._make(env="gen_ai_latest_experimental")
h._maybe_log_raw_request = OpenTelemetry._maybe_log_raw_request.__get__(h)
h.tracer = MagicMock()
h.set_raw_request_attributes = MagicMock()
kwargs = {"litellm_params": {"metadata": {}}}
h._maybe_log_raw_request(kwargs, {}, None, None, MagicMock())
h.tracer.start_span.assert_not_called()
def test_semconv_request_attributes_emit_when_present(self):
h = self._make(env="gen_ai_latest_experimental")
span = MagicMock()
optional_params = {
"frequency_penalty": 0.5,
"presence_penalty": 0.2,
"top_k": 40,
"seed": 42,
"stop": ["\n\n"],
"stream": True,
"n": 3,
}
h._set_semconv_request_attributes(span, optional_params)
calls = {
c.args[0] if c.args else c.kwargs.get("key"): c
for c in span.set_attribute.call_args_list
}
self.assertIn("gen_ai.request.frequency_penalty", calls)
self.assertIn("gen_ai.request.presence_penalty", calls)
self.assertIn("gen_ai.request.top_k", calls)
self.assertIn("gen_ai.request.seed", calls)
self.assertIn("gen_ai.request.stop_sequences", calls)
self.assertIn("gen_ai.request.stream", calls)
self.assertIn("gen_ai.request.choice.count", calls)
def test_semconv_request_choice_count_omitted_when_one(self):
h = self._make(env="gen_ai_latest_experimental")
span = MagicMock()
h._set_semconv_request_attributes(span, {"n": 1})
keys = {c.args[0] for c in span.set_attribute.call_args_list if c.args}
self.assertNotIn("gen_ai.request.choice.count", keys)
def test_semconv_request_choice_count_omitted_for_invalid_n(self):
# n must be a valid count (int > 1); 0/negative/non-int are suppressed.
h = self._make(env="gen_ai_latest_experimental")
for bad_n in (0, -1, "2", 2.0):
span = MagicMock()
h._set_semconv_request_attributes(span, {"n": bad_n})
keys = {c.args[0] for c in span.set_attribute.call_args_list if c.args}
self.assertNotIn(
"gen_ai.request.choice.count", keys, f"n={bad_n!r} should be omitted"
)
def _stream_calls(self, span):
return [
c
for c in span.set_attribute.call_args_list
if c.args and c.args[0] == "gen_ai.request.stream"
]
def test_semconv_request_stream_emitted_as_bool_when_streaming(self):
# Conditionally required per spec: present (as bool True) only when streaming.
h = self._make(env="gen_ai_latest_experimental")
span = MagicMock()
h._set_semconv_request_attributes(span, {"stream": True})
stream_calls = self._stream_calls(span)
self.assertEqual(len(stream_calls), 1)
self.assertIs(stream_calls[0].args[1], True)
def test_semconv_request_stream_omitted_when_not_streaming(self):
h = self._make(env="gen_ai_latest_experimental")
span = MagicMock()
h._set_semconv_request_attributes(span, {"stream": False})
self.assertEqual(self._stream_calls(span), [])
def test_semconv_request_stop_sequences_normalizes_string_to_list(self):
# Spec types gen_ai.request.stop_sequences as string[]; a scalar stop
# is wrapped, and the value is a real list (not a JSON-encoded string).
h = self._make(env="gen_ai_latest_experimental")
span = MagicMock()
h._set_semconv_request_attributes(span, {"stop": "STOP_TOKEN"})
stop_calls = [
c
for c in span.set_attribute.call_args_list
if c.args and c.args[0] == "gen_ai.request.stop_sequences"
]
self.assertEqual(len(stop_calls), 1)
self.assertEqual(stop_calls[0].args[1], ["STOP_TOKEN"])
def test_semconv_cache_token_attributes(self):
h = self._make(env="gen_ai_latest_experimental")
span = MagicMock()
std_log = {
"metadata": {
"usage_object": {
"cache_creation_input_tokens": 12,
"cache_read_input_tokens": 34,
}
}
}
h._set_semconv_cache_token_attributes(span, std_log)
keys = {
c.args[0]: c.args[1] for c in span.set_attribute.call_args_list if c.args
}
self.assertEqual(keys.get("gen_ai.usage.cache_creation.input_tokens"), 12)
self.assertEqual(keys.get("gen_ai.usage.cache_read.input_tokens"), 34)
def test_semconv_cache_token_attributes_handles_none_metadata(self):
# standard_logging_payload["metadata"] = None should not crash.
h = self._make(env="gen_ai_latest_experimental")
span = MagicMock()
h._set_semconv_cache_token_attributes(span, {"metadata": None})
span.set_attribute.assert_not_called()
def test_semconv_cache_token_attributes_omitted_when_zero(self):
h = self._make(env="gen_ai_latest_experimental")
span = MagicMock()
std_log = {
"metadata": {
"usage_object": {
"cache_creation_input_tokens": 0,
"cache_read_input_tokens": 0,
}
}
}
h._set_semconv_cache_token_attributes(span, std_log)
keys = {c.args[0] for c in span.set_attribute.call_args_list if c.args}
self.assertNotIn("gen_ai.usage.cache_creation.input_tokens", keys)
self.assertNotIn("gen_ai.usage.cache_read.input_tokens", keys)
def _set_attributes_keys(self, h):
"""Run set_attributes with a minimal chat payload; return {key: value}."""
span = MagicMock()
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "hi"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
}
response_obj = {"id": "r", "model": "gpt-4", "choices": []}
h.set_attributes(span=span, kwargs=kwargs, response_obj=response_obj)
return {
c.args[0]: c.args[1] for c in span.set_attribute.call_args_list if c.args
}
def test_semconv_mode_emits_provider_name_not_system(self):
# Latest-experimental semconv replaced gen_ai.system with
# gen_ai.provider.name; only the conformant key is emitted.
keys = self._set_attributes_keys(self._make(env="gen_ai_latest_experimental"))
self.assertEqual(keys.get("gen_ai.provider.name"), "openai")
self.assertNotIn("gen_ai.system", keys)
def test_legacy_mode_emits_system_not_provider_name(self):
keys = self._set_attributes_keys(self._make())
self.assertEqual(keys.get("gen_ai.system"), "openai")
self.assertNotIn("gen_ai.provider.name", keys)
def test_opt_in_emits_consolidated_inference_details_event(self):
from opentelemetry import _logs
from opentelemetry._logs._internal import ProxyLoggerProvider
log_exporter = InMemoryLogExporter()
# Make _init_logs see a non-SDK global (the proxy default) so it
# falls into the create_new branch and consults _get_log_exporter,
# which we patch to return our in-memory exporter.
with (
patch.dict(
os.environ,
{"OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental"},
),
patch.object(
_logs, "get_logger_provider", return_value=ProxyLoggerProvider()
),
patch.object(_logs, "set_logger_provider"),
patch.object(OpenTelemetry, "_get_log_exporter", return_value=log_exporter),
):
h = OpenTelemetry(
config=OpenTelemetryConfig(exporter="console", enable_events=True)
)
h.message_logging = True
kwargs = {
"model": "gpt-4",
"call_type": "acompletion",
"messages": [{"role": "user", "content": "hi"}],
"litellm_params": {"custom_llm_provider": "openai"},
}
response_obj = {
"choices": [
{
"message": {"role": "assistant", "content": "hello"},
"finish_reason": "stop",
}
]
}
span = h.tracer.start_span("test")
h._emit_semantic_logs(kwargs, response_obj, span)
span.end()
h._logger_provider.force_flush(2000)
records = [r.log_record for r in log_exporter.get_finished_logs()]
# Exactly ONE inference details event, not the legacy per-message/choice pair.
self.assertEqual(len(records), 1)
attrs = dict(records[0].attributes or {})
self.assertEqual(
attrs["event_name"], "gen_ai.client.inference.operation.details"
)
self.assertEqual(attrs["gen_ai.provider.name"], "openai")
self.assertEqual(attrs["gen_ai.operation.name"], "chat")
self.assertIn("gen_ai.input.messages", attrs)
self.assertIn("gen_ai.output.messages", attrs)
def test_opt_in_inference_details_respects_content_kill_switch(self):
from opentelemetry import _logs
from opentelemetry._logs._internal import ProxyLoggerProvider
log_exporter = InMemoryLogExporter()
with (
patch.dict(
os.environ,
{"OTEL_SEMCONV_STABILITY_OPT_IN": "gen_ai_latest_experimental"},
),
patch("litellm.turn_off_message_logging", True),
patch.object(
_logs, "get_logger_provider", return_value=ProxyLoggerProvider()
),
patch.object(_logs, "set_logger_provider"),
patch.object(OpenTelemetry, "_get_log_exporter", return_value=log_exporter),
):
h = OpenTelemetry(
config=OpenTelemetryConfig(exporter="console", enable_events=True)
)
h.message_logging = True
kwargs = {
"model": "gpt-4",
"call_type": "acompletion",
"messages": [{"role": "user", "content": "private prompt"}],
"litellm_params": {"custom_llm_provider": "openai"},
}
response_obj = {
"choices": [
{
"message": {
"role": "assistant",
"content": "private completion",
},
"finish_reason": "stop",
}
]
}
span = h.tracer.start_span("test")
h._emit_semantic_logs(kwargs, response_obj, span)
span.end()
h._logger_provider.force_flush(2000)
records = [r.log_record for r in log_exporter.get_finished_logs()]
self.assertEqual(len(records), 1)
attrs = dict(records[0].attributes or {})
self.assertNotIn("gen_ai.input.messages", attrs)
self.assertNotIn("gen_ai.output.messages", attrs)
class TestOpenTelemetry(unittest.TestCase):
POLL_INTERVAL = 0.05
POLL_TIMEOUT = 2.0
MODEL = "arn:aws:bedrock:us-west-2:1234567890123:inference-profile/us.anthropic.claude-sonnet-4-5-20250929-v1:0"
HERE = os.path.dirname(__file__)
@patch.dict(os.environ, {}, clear=True)
def test_open_telemetry_config_manual_defaults(self):
"""Manual OpenTelemetryConfig creation should populate default identifiers."""
config = OpenTelemetryConfig(exporter="console", endpoint="http://collector")
self.assertEqual(config.service_name, "litellm")
self.assertEqual(config.deployment_environment, "production")
self.assertEqual(config.model_id, "litellm")
@patch.dict(os.environ, {}, clear=True)
def test_open_telemetry_config_custom_service_name(self):
"""Model ID should inherit provided service name when not explicitly set."""
config = OpenTelemetryConfig(service_name="custom-service", exporter="console")
self.assertEqual(config.service_name, "custom-service")
self.assertEqual(config.deployment_environment, "production")
self.assertEqual(config.model_id, "custom-service")
@patch.dict(os.environ, {}, clear=True)
def test_open_telemetry_config_auto_infer_otlp_http_when_endpoint_set(self):
"""When endpoint is set but exporter is default 'console', auto-infer 'otlp_http'.
This fixes an issue where UI-configured OTEL settings would default to console
output instead of sending traces to the configured endpoint.
See: https://github.com/BerriAI/litellm/issues/XXXX
"""
# When endpoint is specified without explicit exporter, should auto-infer otlp_http
config = OpenTelemetryConfig(endpoint="https://otel-collector.example.com:443")
self.assertEqual(config.exporter, "otlp_http")
# When exporter is explicitly set to something other than console, should not override
config_grpc = OpenTelemetryConfig(
exporter="grpc", endpoint="https://otel-collector.example.com:443"
)
self.assertEqual(config_grpc.exporter, "grpc")
# When no endpoint is set, should keep console as default
config_no_endpoint = OpenTelemetryConfig()
self.assertEqual(config_no_endpoint.exporter, "console")
def wait_for_spans(self, exporter: InMemorySpanExporter, prefix: str):
"""Poll until we see at least one span with an attribute key starting with `prefix`."""
deadline = time.time() + self.POLL_TIMEOUT
while time.time() < deadline:
spans = exporter.get_finished_spans()
matches = [
s
for s in spans
if s.attributes and any(str(k).startswith(prefix) for k in s.attributes)
]
if matches:
return matches
time.sleep(self.POLL_INTERVAL)
return []
def wait_for_metric(self, reader: InMemoryMetricReader, name: str):
"""Poll until we see a metric with the given name."""
deadline = time.time() + self.POLL_TIMEOUT
while time.time() < deadline:
data = reader.get_metrics_data()
# guard against None or missing attribute
if not data or not hasattr(data, "resource_metrics"):
time.sleep(self.POLL_INTERVAL)
continue
for rm in data.resource_metrics:
for sm in rm.scope_metrics:
for m in sm.metrics:
if m.name == name:
return m
time.sleep(self.POLL_INTERVAL)
return None
def wait_for_log(self, reader: InMemoryLogExporter, name: str):
"""Poll until we see a log with the given name."""
deadline = time.time() + self.POLL_TIMEOUT
while time.time() < deadline:
logs = reader.get_finished_logs()
if not logs:
time.sleep(self.POLL_INTERVAL)
continue
matches = [
log
for log in logs
# if log.attributes and any(str(k).startswith(prefix) for k in log.attributes)
]
if matches:
return matches
time.sleep(self.POLL_INTERVAL)
return []
@patch("litellm.integrations.opentelemetry.datetime")
def test_create_guardrail_span_with_valid_info(self, mock_datetime):
# Setup
otel = OpenTelemetry()
otel.tracer = MagicMock()
mock_span = MagicMock()
otel.tracer.start_span.return_value = mock_span
# Create guardrail information
guardrail_info = {
"guardrail_name": "test_guardrail",
"guardrail_mode": "input",
"masked_entity_count": {"CREDIT_CARD": 2},
"guardrail_response": "filtered_content",
"start_time": 1609459200.0,
"end_time": 1609459201.0,
}
# Create a kwargs dict with standard_logging_object containing guardrail information
kwargs = {
"standard_logging_object": {"guardrail_information": [guardrail_info]}
}
# Call the method
otel._create_guardrail_span(kwargs=kwargs, context=None)
# Assertions
otel.tracer.start_span.assert_called_once()
# print all calls to mock_span.set_attribute
print("Calls to mock_span.set_attribute:")
for call in mock_span.set_attribute.call_args_list:
print(call)
# Check that the span has the correct attributes set
mock_span.set_attribute.assert_any_call("guardrail_name", "test_guardrail")
mock_span.set_attribute.assert_any_call("guardrail_mode", "input")
mock_span.set_attribute.assert_any_call(
"guardrail_response", safe_dumps("filtered_content")
)
mock_span.set_attribute.assert_any_call(
"masked_entity_count", safe_dumps({"CREDIT_CARD": 2})
)
# Verify that the span was ended
mock_span.end.assert_called_once()
def test_create_guardrail_span_with_no_info(self):
# Setup
otel = OpenTelemetry()
otel.tracer = MagicMock()
# Test with no guardrail information
kwargs = {"standard_logging_object": {}}
otel._create_guardrail_span(kwargs=kwargs, context=None)
# Verify that start_span was never called
otel.tracer.start_span.assert_not_called()
def test_get_tracer_to_use_for_request_with_dynamic_headers(self):
"""Test that get_tracer_to_use_for_request returns a dynamic tracer when dynamic headers are present."""
# Setup
otel = OpenTelemetry()
otel.tracer = MagicMock()
# Mock the dynamic header extraction and tracer creation
with (
patch.object(
otel, "_get_dynamic_otel_headers_from_kwargs"
) as mock_get_headers,
patch.object(otel, "_get_tracer_with_dynamic_headers") as mock_get_tracer,
):
# Test case 1: With dynamic headers
mock_get_headers.return_value = {
"arize-space-id": "test-space",
"api_key": "test-key",
}
mock_dynamic_tracer = MagicMock()
mock_get_tracer.return_value = mock_dynamic_tracer
kwargs = {
"standard_callback_dynamic_params": {"arize_space_key": "test-space"}
}
result = otel.get_tracer_to_use_for_request(kwargs)
# 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"}
)
self.assertEqual(result, mock_dynamic_tracer)
def test_get_tracer_to_use_for_request_without_dynamic_headers(self):
"""Test that get_tracer_to_use_for_request returns the default tracer when no dynamic headers are present."""
# Setup
otel = OpenTelemetry()
otel.tracer = MagicMock()
# Mock the dynamic header extraction to return None
with patch.object(
otel, "_get_dynamic_otel_headers_from_kwargs"
) as mock_get_headers:
mock_get_headers.return_value = None
kwargs = {}
result = otel.get_tracer_to_use_for_request(kwargs)
# Assertions
mock_get_headers.assert_called_once_with(kwargs)
self.assertEqual(result, otel.tracer)
def test_get_dynamic_otel_headers_from_kwargs(self):
"""Test that _get_dynamic_otel_headers_from_kwargs correctly extracts dynamic headers from kwargs."""
# Setup
otel = OpenTelemetry()
# Mock the construct_dynamic_otel_headers method
with patch.object(otel, "construct_dynamic_otel_headers") as mock_construct:
# Test case 1: With standard_callback_dynamic_params
mock_construct.return_value = {
"arize-space-id": "test-space",
"api_key": "test-key",
}
standard_params = {
"arize_space_key": "test-space",
"arize_api_key": "test-key",
}
kwargs = {"standard_callback_dynamic_params": standard_params}
result = otel._get_dynamic_otel_headers_from_kwargs(kwargs)
# Assertions
mock_construct.assert_called_once_with(
standard_callback_dynamic_params=standard_params
)
self.assertEqual(
result, {"arize-space-id": "test-space", "api_key": "test-key"}
)
# Test case 2: Without standard_callback_dynamic_params
kwargs_empty = {}
result_empty = otel._get_dynamic_otel_headers_from_kwargs(kwargs_empty)
# Should return None when no dynamic params
self.assertIsNone(result_empty)
# Test case 3: With empty construct result
mock_construct.return_value = {}
result_empty_construct = otel._get_dynamic_otel_headers_from_kwargs(kwargs)
# Should return None when construct returns empty dict
self.assertIsNone(result_empty_construct)
@patch("opentelemetry.sdk.trace.TracerProvider")
@patch("opentelemetry.sdk.resources.Resource")
def test_get_tracer_with_dynamic_headers(self, mock_resource, mock_tracer_provider):
"""Test that _get_tracer_with_dynamic_headers creates a temporary tracer with dynamic headers."""
# Setup
otel = OpenTelemetry()
# Mock the span processor creation
with patch.object(otel, "_get_span_processor") as mock_get_span_processor:
mock_span_processor = MagicMock()
mock_get_span_processor.return_value = mock_span_processor
# Mock the tracer provider and its methods
mock_provider_instance = MagicMock()
mock_tracer_provider.return_value = mock_provider_instance
mock_tracer = MagicMock()
mock_provider_instance.get_tracer.return_value = mock_tracer
# Mock the resource
mock_resource_instance = MagicMock()
mock_resource.return_value = mock_resource_instance
# Test
dynamic_headers = {"arize-space-id": "test-space", "api_key": "test-key"}
result = otel._get_tracer_with_dynamic_headers(dynamic_headers)
# Assertions
mock_get_span_processor.assert_called_once_with(
dynamic_headers=dynamic_headers
)
mock_provider_instance.add_span_processor.assert_called_once_with(
mock_span_processor
)
mock_provider_instance.get_tracer.assert_called_once_with("litellm")
self.assertEqual(result, mock_tracer)
@patch.dict(os.environ, {}, clear=True)
@patch("opentelemetry.sdk.resources.Resource.create")
@patch("opentelemetry.sdk.resources.OTELResourceDetector")
def test_get_litellm_resource_with_defaults(
self, mock_detector_cls, mock_resource_create
):
"""Test _get_litellm_resource with default values when no environment variables are set."""
# Mock the Resource.create method
mock_base_resource = MagicMock()
mock_resource_create.return_value = mock_base_resource
# Mock the OTELResourceDetector
mock_detector = MagicMock()
mock_detector_cls.return_value = mock_detector
mock_env_resource = MagicMock()
mock_detector.detect.return_value = mock_env_resource
# Mock the merged resource
mock_merged_resource = MagicMock()
mock_base_resource.merge.return_value = mock_merged_resource
config = OpenTelemetryConfig()
result = OpenTelemetry._get_litellm_resource(config)
# Verify Resource.create was called with correct default attributes
expected_attributes = {
"service.name": "litellm",
"deployment.environment": "production",
"model_id": "litellm",
}
mock_resource_create.assert_called_once_with(expected_attributes)
mock_detector.detect.assert_called_once()
mock_base_resource.merge.assert_called_once_with(mock_env_resource)
self.assertEqual(result, mock_merged_resource)
@patch.dict(
os.environ,
{
"OTEL_SERVICE_NAME": "test-service",
"OTEL_ENVIRONMENT_NAME": "staging",
"OTEL_MODEL_ID": "test-model",
},
clear=True,
)
@patch("opentelemetry.sdk.resources.Resource.create")
@patch("opentelemetry.sdk.resources.OTELResourceDetector")
def test_get_litellm_resource_with_litellm_env_vars(
self, mock_detector_cls, mock_resource_create
):
"""Test _get_litellm_resource with LiteLLM-specific environment variables."""
# Mock the Resource.create method
mock_base_resource = MagicMock()
mock_resource_create.return_value = mock_base_resource
# Mock the OTELResourceDetector
mock_detector = MagicMock()
mock_detector_cls.return_value = mock_detector
mock_env_resource = MagicMock()
mock_detector.detect.return_value = mock_env_resource
# Mock the merged resource
mock_merged_resource = MagicMock()
mock_base_resource.merge.return_value = mock_merged_resource
config = OpenTelemetryConfig.from_env()
result = OpenTelemetry._get_litellm_resource(config)
# Verify Resource.create was called with environment variable values
expected_attributes = {
"service.name": "test-service",
"deployment.environment": "staging",
"model_id": "test-model",
}
mock_resource_create.assert_called_once_with(expected_attributes)
mock_detector.detect.assert_called_once()
mock_base_resource.merge.assert_called_once_with(mock_env_resource)
self.assertEqual(result, mock_merged_resource)
@patch.dict(
os.environ,
{
"OTEL_RESOURCE_ATTRIBUTES": "service.name=otel-service,deployment.environment=production,custom.attr=value",
"OTEL_SERVICE_NAME": "should-be-overridden",
},
clear=True,
)
@patch("opentelemetry.sdk.resources.Resource.create")
@patch("opentelemetry.sdk.resources.OTELResourceDetector")
def test_get_litellm_resource_with_otel_resource_attributes(
self, mock_detector_cls, mock_resource_create
):
"""Test _get_litellm_resource with OTEL_RESOURCE_ATTRIBUTES environment variable."""
# Mock the Resource.create method to simulate the actual behavior
# In reality, Resource.create() would parse OTEL_RESOURCE_ATTRIBUTES and merge it
mock_base_resource = MagicMock()
mock_resource_create.return_value = mock_base_resource
# Mock the OTELResourceDetector
mock_detector = MagicMock()
mock_detector_cls.return_value = mock_detector
mock_env_resource = MagicMock()
mock_detector.detect.return_value = mock_env_resource
# Mock the merged resource
mock_merged_resource = MagicMock()
mock_base_resource.merge.return_value = mock_merged_resource
config = OpenTelemetryConfig.from_env()
result = OpenTelemetry._get_litellm_resource(config)
# Verify Resource.create was called with the base attributes
# The actual OTEL_RESOURCE_ATTRIBUTES parsing is handled by OpenTelemetry SDK
expected_attributes = {
"service.name": "should-be-overridden",
"deployment.environment": "production",
"model_id": "should-be-overridden",
}
mock_resource_create.assert_called_once_with(expected_attributes)
mock_detector.detect.assert_called_once()
mock_base_resource.merge.assert_called_once_with(mock_env_resource)
self.assertEqual(result, mock_merged_resource)
@patch.dict(os.environ, {}, clear=True)
def test_get_litellm_resource_integration_with_real_resource(self):
"""Integration test to verify _get_litellm_resource works with actual OpenTelemetry Resource."""
config = OpenTelemetryConfig()
result = OpenTelemetry._get_litellm_resource(config)
# Verify the result is a Resource instance
from opentelemetry.sdk.resources import Resource
self.assertIsInstance(result, Resource)
# Verify the resource has the expected default attributes
attributes = result.attributes
self.assertEqual(attributes.get("service.name"), "litellm")
self.assertEqual(attributes.get("deployment.environment"), "production")
self.assertEqual(attributes.get("model_id"), "litellm")
@patch.dict(
os.environ,
{
"OTEL_RESOURCE_ATTRIBUTES": "service.name=from-env,custom.attribute=test-value,deployment.environment=test-env"
},
clear=True,
)
def test_get_litellm_resource_real_otel_resource_attributes(self):
"""Integration test to verify OTEL_RESOURCE_ATTRIBUTES is properly handled."""
config = OpenTelemetryConfig.from_env()
result = OpenTelemetry._get_litellm_resource(config)
print("RESULT", result)
# Verify the result is a Resource instance
from opentelemetry.sdk.resources import Resource
self.assertIsInstance(result, Resource)
# Verify that OTEL_RESOURCE_ATTRIBUTES values override the defaults
attributes = result.attributes
self.assertEqual(attributes.get("service.name"), "from-env")
self.assertEqual(attributes.get("deployment.environment"), "test-env")
self.assertEqual(attributes.get("custom.attribute"), "test-value")
# model_id should still be set from the base attributes since it wasn't in OTEL_RESOURCE_ATTRIBUTES
self.assertEqual(attributes.get("model_id"), "litellm")
@patch.dict(
os.environ,
{
"OTEL_SERVICE_NAME": "litellm-service",
"OTEL_RESOURCE_ATTRIBUTES": "service.name=otel-override,extra.attr=extra-value",
},
clear=True,
)
def test_get_litellm_resource_precedence(self):
"""Test that OTEL_SERVICE_NAME takes precedence over OTEL_RESOURCE_ATTRIBUTES according to OpenTelemetry spec."""
config = OpenTelemetryConfig.from_env()
result = OpenTelemetry._get_litellm_resource(config)
# Verify the result is a Resource instance
from opentelemetry.sdk.resources import Resource
self.assertIsInstance(result, Resource)
# According to OpenTelemetry spec, OTEL_SERVICE_NAME takes precedence over service.name in OTEL_RESOURCE_ATTRIBUTES
attributes = result.attributes
self.assertEqual(attributes.get("service.name"), "litellm-service")
# But other attributes from OTEL_RESOURCE_ATTRIBUTES should still be present
self.assertEqual(attributes.get("extra.attr"), "extra-value")
def test_handle_success_spans_only(self):
# make sure neither events nor metrics is on
os.environ.pop("LITELLM_OTEL_INTEGRATION_ENABLE_EVENTS", None)
os.environ.pop("LITELLM_OTEL_INTEGRATION_ENABLE_METRICS", None)
# ─── build in‐memory OTEL providers/exporters ─────────────────────────────
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
# no logs / no metrics
log_exporter = InMemoryLogExporter()
logger_provider = OTLoggerProvider()
logger_provider.add_log_record_processor(SimpleLogRecordProcessor(log_exporter))
metric_reader = InMemoryMetricReader()
meter_provider = MeterProvider(metric_readers=[metric_reader])
# ─── instantiate our OpenTelemetry logger with test providers ───────────
otel = OpenTelemetry(
tracer_provider=tracer_provider,
meter_provider=meter_provider,
logger_provider=logger_provider, # pass even if events disabled (safe)
)
# bind our tracer to the test tracer provider (global registration is a no-op after the first time)
otel.tracer = tracer_provider.get_tracer(__name__)
# ─── minimal input / output for a chat call ──────────────────────────────
start = datetime.utcnow()
end = start + timedelta(seconds=1)
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_kwargs.json")
) as f:
kwargs = json.load(f)
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_response.json")
) as f:
response_obj = json.load(f)
# ─── exercise the hook ───────────────────────────────────────────────────
otel._handle_success(kwargs, response_obj, start, end)
# ─── assert spans only ───────────────────────────────────────────────────
spans = span_exporter.get_finished_spans()
self.assertTrue(spans, "Expected at least one span")
# must have the top‐level litellm_request span
# self.assertIn(
# LITELLM_REQUEST_SPAN_NAME,
# [s.name for s in spans],
# "litellm_request span missing",
# )
# model attribute should be on that span
found = any(
s.attributes and s.attributes.get("gen_ai.request.model") == self.MODEL
for s in spans
)
self.assertTrue(found, "expected gen_ai.request.model on span attributes")
# no metrics recorded
self.assertIsNone(
self.wait_for_metric(metric_reader, "gen_ai.client.operation.duration"),
"Did not expect any metrics",
)
# no logs emitted
logs = log_exporter.get_finished_logs()
self.assertFalse(logs, "Did not expect any logs")
@patch.dict(
os.environ, {"LITELLM_OTEL_INTEGRATION_ENABLE_METRICS": "true"}, clear=True
)
def test_handle_success_spans_and_metrics(self):
# ─── build in‐memory OTEL providers/exporters ─────────────────────────────
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
log_exporter = InMemoryLogExporter()
logger_provider = OTLoggerProvider()
logger_provider.add_log_record_processor(SimpleLogRecordProcessor(log_exporter))
metric_reader = InMemoryMetricReader()
meter_provider = MeterProvider(metric_readers=[metric_reader])
# ─── instantiate our OpenTelemetry logger with test providers ───────────
otel = OpenTelemetry(
tracer_provider=tracer_provider,
meter_provider=meter_provider,
logger_provider=logger_provider, # needed if events were enabled
)
otel.tracer = tracer_provider.get_tracer(__name__)
# ─── minimal input / output for a chat call ──────────────────────────────
start = datetime.utcnow()
end = start + timedelta(seconds=1)
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_kwargs.json")
) as f:
kwargs = json.load(f)
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_response.json")
) as f:
response_obj = json.load(f)
# ─── exercise the hook ───────────────────────────────────────────────────
otel._handle_success(kwargs, response_obj, start, end)
# ─── assert spans ────────────────────────────────────────────────────────
spans = span_exporter.get_finished_spans()
self.assertTrue(spans, "Expected at least one span")
# ─── assert metrics ──────────────────────────────────────────────────────
duration_metric = self.wait_for_metric(
metric_reader, "gen_ai.client.operation.duration"
)
self.assertIsNotNone(duration_metric, "duration histogram was not recorded")
# model attribute should be present on a data point
found_dp = False
if (
duration_metric
and hasattr(duration_metric, "data")
and hasattr(duration_metric.data, "data_points")
):
found_dp = any(
dp.attributes.get("gen_ai.request.model") == self.MODEL
for dp in duration_metric.data.data_points
)
self.assertTrue(
found_dp, "expected gen_ai.request.model attribute on a data point"
)
# ─── no events when only metrics enabled ─────────────────────────────────
logs = log_exporter.get_finished_logs()
self.assertFalse(logs, "Did not expect any logs")
def test_get_span_name_with_generation_name(self):
"""Test _get_span_name returns generation_name when present"""
otel = OpenTelemetry()
kwargs = {"litellm_params": {"metadata": {"generation_name": "custom_span"}}}
result = otel._get_span_name(kwargs)
self.assertEqual(result, "custom_span")
def test_get_span_name_without_generation_name(self):
"""Test _get_span_name returns default when generation_name missing"""
from litellm.integrations.opentelemetry import LITELLM_REQUEST_SPAN_NAME
otel = OpenTelemetry()
kwargs = {"litellm_params": {"metadata": {}}}
result = otel._get_span_name(kwargs)
self.assertEqual(result, LITELLM_REQUEST_SPAN_NAME)
@patch.dict(os.environ, {"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT": ""})
@patch("litellm.turn_off_message_logging", False)
def test_maybe_log_raw_request_creates_span(self):
"""Test _maybe_log_raw_request creates span when logging enabled"""
from litellm.integrations.opentelemetry import RAW_REQUEST_SPAN_NAME
otel = OpenTelemetry()
otel.message_logging = True
mock_tracer = MagicMock()
mock_span = MagicMock()
mock_tracer.start_span.return_value = mock_span
otel.get_tracer_to_use_for_request = MagicMock(return_value=mock_tracer)
otel.set_raw_request_attributes = MagicMock()
otel._to_ns = MagicMock(return_value=1234567890)
kwargs = {"litellm_params": {"metadata": {}}}
otel._maybe_log_raw_request(
kwargs, {}, datetime.now(), datetime.now(), MagicMock()
)
mock_tracer.start_span.assert_called_once()
self.assertEqual(
mock_tracer.start_span.call_args[1]["name"], RAW_REQUEST_SPAN_NAME
)
@patch("litellm.turn_off_message_logging", True)
def test_maybe_log_raw_request_skips_when_logging_disabled(self):
"""Test _maybe_log_raw_request skips when logging disabled"""
otel = OpenTelemetry()
mock_tracer = MagicMock()
otel.get_tracer_to_use_for_request = MagicMock(return_value=mock_tracer)
kwargs = {"litellm_params": {"metadata": {}}}
otel._maybe_log_raw_request(
kwargs, {}, datetime.now(), datetime.now(), MagicMock()
)
mock_tracer.start_span.assert_not_called()
class TestOpenTelemetryToNs(unittest.TestCase):
"""``_to_ns`` converts a span boundary to epoch nanoseconds. Service spans now
feed it real float/datetime windows, and a missing boundary arrives as
``None`` — all three shapes must convert without raising the ``AttributeError``
a bare ``dt.timestamp()`` would on a float or ``None``."""
def setUp(self):
self.otel = OpenTelemetry()
def test_datetime_converts_to_epoch_ns(self):
dt = datetime(2026, 5, 26, 12, 0, 0, tzinfo=timezone.utc)
self.assertEqual(self.otel._to_ns(dt), int(dt.timestamp() * 1e9))
def test_float_epoch_seconds_scaled_to_ns(self):
self.assertEqual(self.otel._to_ns(1700.5), 1_700_500_000_000)
def test_int_epoch_seconds_scaled_to_ns(self):
self.assertEqual(self.otel._to_ns(1700), 1_700_000_000_000)
@patch("litellm.integrations.opentelemetry.datetime")
def test_none_falls_back_to_current_time(self, mock_datetime):
mock_datetime.now.return_value.timestamp.return_value = 1700.0
self.assertEqual(self.otel._to_ns(None), 1_700_000_000_000)
class TestOpenTelemetryHeaderSplitting(unittest.TestCase):
"""Test suite for _get_headers_dictionary method"""
def test_split_multiple_headers_comma_separated(self):
"""Test splitting multiple headers separated by commas"""
otel = OpenTelemetry()
headers = "api-key=key,other-config-value=value"
result = otel._get_headers_dictionary(headers)
self.assertEqual(result, {"api-key": "key", "other-config-value": "value"})
def test_split_headers_with_equals_in_values(self):
"""Test splitting headers where values contain equals signs (split only on first '=')"""
otel = OpenTelemetry()
headers = "api-key=value1=part2,config=setting=enabled"
result = otel._get_headers_dictionary(headers)
self.assertEqual(
result, {"api-key": "value1=part2", "config": "setting=enabled"}
)
def test_accepts_any_mapping_not_only_dict(self):
"""The parameter is typed Mapping, so a non-dict Mapping must not silently drop
every header and leave the exporter unauthenticated."""
otel = OpenTelemetry()
headers = MappingProxyType({"authorization": "Basic abc"})
self.assertEqual(otel._get_headers_dictionary(headers), {"authorization": "Basic abc"})
def test_returns_a_copy_so_the_exporter_never_aliases_the_caller(self):
"""The result is handed to a long-lived exporter, so it must not be the caller's
own dict."""
otel = OpenTelemetry()
headers = {"authorization": "Basic abc"}
result = otel._get_headers_dictionary(headers)
self.assertIsNot(result, headers)
self.assertEqual(result, headers)
class TestOpenTelemetryEndpointNormalization(unittest.TestCase):
"""Test suite for the unified _normalize_otel_endpoint method"""
def test_normalize_traces_endpoint_from_logs_path(self):
"""Test normalizing endpoint with /v1/logs to /v1/traces"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/v1/logs", "traces"
)
self.assertEqual(result, "http://collector:4318/v1/traces")
def test_normalize_traces_endpoint_from_metrics_path(self):
"""Test normalizing endpoint with /v1/metrics to /v1/traces"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/v1/metrics", "traces"
)
self.assertEqual(result, "http://collector:4318/v1/traces")
def test_normalize_traces_endpoint_from_base_url(self):
"""Test adding /v1/traces to base URL"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint("http://collector:4318", "traces")
self.assertEqual(result, "http://collector:4318/v1/traces")
def test_normalize_traces_endpoint_from_v1_path(self):
"""Test adding traces to /v1 path"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint("http://collector:4318/v1", "traces")
self.assertEqual(result, "http://collector:4318/v1/traces")
def test_normalize_traces_endpoint_already_correct(self):
"""Test endpoint already ending with /v1/traces remains unchanged"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/v1/traces", "traces"
)
self.assertEqual(result, "http://collector:4318/v1/traces")
def test_normalize_metrics_endpoint_from_traces_path(self):
"""Test normalizing endpoint with /v1/traces to /v1/metrics"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/v1/traces", "metrics"
)
self.assertEqual(result, "http://collector:4318/v1/metrics")
def test_normalize_metrics_endpoint_from_logs_path(self):
"""Test normalizing endpoint with /v1/logs to /v1/metrics"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/v1/logs", "metrics"
)
self.assertEqual(result, "http://collector:4318/v1/metrics")
def test_normalize_metrics_endpoint_from_base_url(self):
"""Test adding /v1/metrics to base URL"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint("http://collector:4318", "metrics")
self.assertEqual(result, "http://collector:4318/v1/metrics")
def test_normalize_metrics_endpoint_already_correct(self):
"""Test endpoint already ending with /v1/metrics remains unchanged"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/v1/metrics", "metrics"
)
self.assertEqual(result, "http://collector:4318/v1/metrics")
def test_normalize_logs_endpoint_from_traces_path(self):
"""Test normalizing endpoint with /v1/traces to /v1/logs"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/v1/traces", "logs"
)
self.assertEqual(result, "http://collector:4318/v1/logs")
def test_normalize_logs_endpoint_from_metrics_path(self):
"""Test normalizing endpoint with /v1/metrics to /v1/logs"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/v1/metrics", "logs"
)
self.assertEqual(result, "http://collector:4318/v1/logs")
def test_normalize_logs_endpoint_from_base_url(self):
"""Test adding /v1/logs to base URL"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint("http://collector:4318", "logs")
self.assertEqual(result, "http://collector:4318/v1/logs")
def test_normalize_logs_endpoint_already_correct(self):
"""Test endpoint already ending with /v1/logs remains unchanged"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint("http://collector:4318/v1/logs", "logs")
self.assertEqual(result, "http://collector:4318/v1/logs")
def test_normalize_endpoint_with_trailing_slash(self):
"""Test that trailing slashes are properly handled"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint("http://collector:4318/", "traces")
self.assertEqual(result, "http://collector:4318/v1/traces")
@parameterized.expand(
[
(
"https://ingest.eu1.observability.splunkcloud.com/v2/trace/otlp",
"https://ingest.eu1.observability.splunkcloud.com/v2/trace/otlp",
),
(
"https://ingest.us0.observability.splunkcloud.com/v2/trace/otlp/",
"https://ingest.us0.observability.splunkcloud.com/v2/trace/otlp",
),
(
"https://ingest.eu0.signalfx.com/v2/trace/otlp",
"https://ingest.eu0.signalfx.com/v2/trace/otlp",
),
(
"https://example.com/prefix/v2/trace/otlp",
"https://example.com/prefix/v2/trace/otlp",
),
]
)
def test_normalize_traces_nonstandard_otlp_ingest_urls_unchanged(
self, input_url: str, expected: str
) -> None:
"""Splunk-style /v2/trace/otlp endpoints must not get /v1/traces appended."""
otel = OpenTelemetry()
self.assertEqual(
otel._normalize_otel_endpoint(input_url, "traces"),
expected,
)
def test_normalize_endpoint_none(self):
"""Test that None endpoint returns None"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(None, "traces")
self.assertIsNone(result)
def test_normalize_endpoint_empty_string(self):
"""Test that empty string returns empty string"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint("", "traces")
self.assertEqual(result, "")
def test_normalize_endpoint_invalid_signal_type(self):
"""Test that invalid signal type returns endpoint unchanged with warning"""
otel = OpenTelemetry()
endpoint = "http://collector:4318/v1/traces"
with patch("litellm._logging.verbose_logger.warning") as mock_warning:
result = otel._normalize_otel_endpoint(endpoint, "invalid")
# Should return endpoint unchanged
self.assertEqual(result, endpoint)
# Should log a warning
mock_warning.assert_called_once()
# Check the warning was called with the expected format string and parameters
call_args = mock_warning.call_args[0]
self.assertIn("Invalid signal_type", call_args[0])
self.assertEqual(call_args[1], "invalid") # signal_type parameter
self.assertEqual(
call_args[2], {"traces", "metrics", "logs"}
) # valid_signals parameter
def test_normalize_endpoint_https(self):
"""Test normalization works with https URLs"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"https://collector.example.com:4318", "logs"
)
self.assertEqual(result, "https://collector.example.com:4318/v1/logs")
def test_normalize_endpoint_with_path_prefix(self):
"""Test normalization works with URLs that have path prefixes"""
otel = OpenTelemetry()
result = otel._normalize_otel_endpoint(
"http://collector:4318/otel/v1/traces", "logs"
)
# Should replace the final /traces with /logs
self.assertEqual(result, "http://collector:4318/otel/v1/logs")
def test_normalize_endpoint_consistency_across_signals(self):
"""Test that normalization is consistent for all signal types from the same base"""
otel = OpenTelemetry()
base = "http://collector:4318"
traces_result = otel._normalize_otel_endpoint(base, "traces")
metrics_result = otel._normalize_otel_endpoint(base, "metrics")
logs_result = otel._normalize_otel_endpoint(base, "logs")
# All should have the same base with different signal paths
self.assertEqual(traces_result, "http://collector:4318/v1/traces")
self.assertEqual(metrics_result, "http://collector:4318/v1/metrics")
self.assertEqual(logs_result, "http://collector:4318/v1/logs")
def test_normalize_endpoint_signal_switching(self):
"""Test switching between different signal types on the same endpoint"""
otel = OpenTelemetry()
# Start with traces
endpoint = "http://collector:4318/v1/traces"
# Switch to metrics
metrics = otel._normalize_otel_endpoint(endpoint, "metrics")
self.assertEqual(metrics, "http://collector:4318/v1/metrics")
# Switch to logs
logs = otel._normalize_otel_endpoint(metrics, "logs")
self.assertEqual(logs, "http://collector:4318/v1/logs")
# Switch back to traces
traces = otel._normalize_otel_endpoint(logs, "traces")
self.assertEqual(traces, "http://collector:4318/v1/traces")
class TestOpenTelemetryProtocolSelection(unittest.TestCase):
"""Test suite for verifying correct exporter selection based on protocol"""
def test_get_span_processor_uses_http_exporter_for_otlp_http(self):
"""Test that otlp_http protocol uses OTLPSpanExporterHTTP"""
from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
OTLPSpanExporter as OTLPSpanExporterHTTP,
)
from opentelemetry.sdk.trace.export import BatchSpanProcessor
config = OpenTelemetryConfig(
exporter="otlp_http", endpoint="http://collector:4318"
)
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify it's a BatchSpanProcessor
self.assertIsInstance(processor, BatchSpanProcessor)
# Verify the exporter is the HTTP variant
self.assertIsInstance(processor.span_exporter, OTLPSpanExporterHTTP)
def test_get_span_processor_uses_grpc_exporter_for_otlp_grpc(self):
"""Test that otlp_grpc protocol uses OTLPSpanExporterGRPC"""
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
OTLPSpanExporter as OTLPSpanExporterGRPC,
)
from opentelemetry.sdk.trace.export import BatchSpanProcessor
config = OpenTelemetryConfig(
exporter="otlp_grpc", endpoint="http://collector:4317"
)
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify it's a BatchSpanProcessor
self.assertIsInstance(processor, BatchSpanProcessor)
# Verify the exporter is the gRPC variant
self.assertIsInstance(processor.span_exporter, OTLPSpanExporterGRPC)
def test_get_span_processor_uses_grpc_exporter_for_grpc_alias(self):
"""Test that 'grpc' protocol alias uses OTLPSpanExporterGRPC"""
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
OTLPSpanExporter as OTLPSpanExporterGRPC,
)
from opentelemetry.sdk.trace.export import BatchSpanProcessor
config = OpenTelemetryConfig(exporter="grpc", endpoint="http://collector:4317")
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify it's a BatchSpanProcessor
self.assertIsInstance(processor, BatchSpanProcessor)
# Verify the exporter is the gRPC variant
self.assertIsInstance(processor.span_exporter, OTLPSpanExporterGRPC)
def test_get_span_processor_uses_http_exporter_for_http_protobuf(self):
"""Test that http/protobuf protocol uses OTLPSpanExporterHTTP"""
from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
OTLPSpanExporter as OTLPSpanExporterHTTP,
)
from opentelemetry.sdk.trace.export import BatchSpanProcessor
config = OpenTelemetryConfig(
exporter="http/protobuf", endpoint="http://collector:4318"
)
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify it's a BatchSpanProcessor
self.assertIsInstance(processor, BatchSpanProcessor)
# Verify the exporter is the HTTP variant
self.assertIsInstance(processor.span_exporter, OTLPSpanExporterHTTP)
def test_get_span_processor_uses_console_exporter_for_console(self):
"""Test that console protocol uses ConsoleSpanExporter"""
from opentelemetry.sdk.trace.export import (
BatchSpanProcessor,
ConsoleSpanExporter,
)
config = OpenTelemetryConfig(exporter="console")
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify it's a BatchSpanProcessor
self.assertIsInstance(processor, BatchSpanProcessor)
# Verify the exporter is the console variant
self.assertIsInstance(processor.span_exporter, ConsoleSpanExporter)
def test_get_log_exporter_uses_http_exporter_for_otlp_http(self):
"""Test that otlp_http protocol uses HTTP OTLPLogExporter"""
from opentelemetry.exporter.otlp.proto.http._log_exporter import OTLPLogExporter
config = OpenTelemetryConfig(
exporter="otlp_http", endpoint="http://collector:4318", enable_events=True
)
otel = OpenTelemetry(config=config)
exporter = otel._get_log_exporter()
# Verify the exporter is the HTTP variant
self.assertIsInstance(exporter, OTLPLogExporter)
# Check that it's from the http module by checking the module name
self.assertIn("http", exporter.__class__.__module__)
def test_get_log_exporter_uses_grpc_exporter_for_otlp_grpc(self):
"""Test that otlp_grpc protocol uses gRPC OTLPLogExporter"""
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
config = OpenTelemetryConfig(
exporter="otlp_grpc", endpoint="http://collector:4317", enable_events=True
)
otel = OpenTelemetry(config=config)
exporter = otel._get_log_exporter()
# Verify the exporter is the gRPC variant
self.assertIsInstance(exporter, OTLPLogExporter)
# Check that it's from the grpc module by checking the module name
self.assertIn("grpc", exporter.__class__.__module__)
def test_get_log_exporter_uses_grpc_exporter_for_grpc_alias(self):
"""Test that 'grpc' protocol alias uses gRPC OTLPLogExporter"""
from opentelemetry.exporter.otlp.proto.grpc._log_exporter import OTLPLogExporter
config = OpenTelemetryConfig(
exporter="grpc", endpoint="http://collector:4317", enable_events=True
)
otel = OpenTelemetry(config=config)
exporter = otel._get_log_exporter()
# Verify the exporter is the gRPC variant
self.assertIsInstance(exporter, OTLPLogExporter)
# Check that it's from the grpc module by checking the module name
self.assertIn("grpc", exporter.__class__.__module__)
def test_get_log_exporter_uses_console_exporter_for_console(self):
"""Test that console protocol uses ConsoleLogExporter"""
from opentelemetry.sdk._logs.export import ConsoleLogExporter
config = OpenTelemetryConfig(exporter="console", enable_events=True)
otel = OpenTelemetry(config=config)
exporter = otel._get_log_exporter()
# Verify the exporter is the console variant
self.assertIsInstance(exporter, ConsoleLogExporter)
def test_get_log_exporter_defaults_to_console_for_unknown_protocol(self):
"""Test that unknown protocol defaults to ConsoleLogExporter with warning"""
from opentelemetry.sdk._logs.export import ConsoleLogExporter
config = OpenTelemetryConfig(exporter="unknown_protocol", enable_events=True)
otel = OpenTelemetry(config=config)
with patch("litellm._logging.verbose_logger.warning") as mock_warning:
exporter = otel._get_log_exporter()
# Verify the exporter defaults to console
self.assertIsInstance(exporter, ConsoleLogExporter)
# Verify a warning was logged
mock_warning.assert_called_once()
args = mock_warning.call_args[0]
self.assertIn("Unknown log exporter", args[0])
self.assertIn("unknown_protocol", args[1])
@patch.dict(
os.environ,
{
"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318",
},
clear=False,
)
def test_protocol_selection_from_environment_http(self):
"""Test that protocol selection works correctly from environment variables for HTTP"""
from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
OTLPSpanExporter as OTLPSpanExporterHTTP,
)
from opentelemetry.sdk.trace.export import BatchSpanProcessor
config = OpenTelemetryConfig.from_env()
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify the HTTP exporter is used
self.assertIsInstance(processor, BatchSpanProcessor)
self.assertIsInstance(processor.span_exporter, OTLPSpanExporterHTTP)
@patch.dict(
os.environ,
{
"OTEL_EXPORTER_OTLP_PROTOCOL": "grpc",
"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4317",
},
clear=False,
)
def test_protocol_selection_from_environment_grpc(self):
"""Test that protocol selection works correctly from environment variables for gRPC"""
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
OTLPSpanExporter as OTLPSpanExporterGRPC,
)
from opentelemetry.sdk.trace.export import BatchSpanProcessor
config = OpenTelemetryConfig.from_env()
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify the gRPC exporter is used
self.assertIsInstance(processor, BatchSpanProcessor)
self.assertIsInstance(processor.span_exporter, OTLPSpanExporterGRPC)
@patch.dict(
os.environ,
{
"OTEL_EXPORTER": "otlp_http",
"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4318",
},
clear=False,
)
def test_protocol_selection_from_otel_exporter_fallback_http(self):
"""OTEL_EXPORTER drives protocol when OTEL_EXPORTER_OTLP_PROTOCOL is unset."""
from opentelemetry.exporter.otlp.proto.http.trace_exporter import (
OTLPSpanExporter as OTLPSpanExporterHTTP,
)
from opentelemetry.sdk.trace.export import BatchSpanProcessor
popped_protocol = os.environ.pop("OTEL_EXPORTER_OTLP_PROTOCOL", None)
try:
config = OpenTelemetryConfig.from_env()
self.assertEqual(config.exporter, "otlp_http")
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
self.assertIsInstance(processor, BatchSpanProcessor)
self.assertIsInstance(processor.span_exporter, OTLPSpanExporterHTTP)
finally:
if popped_protocol is not None:
os.environ["OTEL_EXPORTER_OTLP_PROTOCOL"] = popped_protocol
@patch.dict(
os.environ,
{
"OTEL_EXPORTER": "otlp_grpc",
"OTEL_EXPORTER_OTLP_ENDPOINT": "http://collector:4317",
},
clear=False,
)
def test_protocol_selection_from_otel_exporter_fallback_grpc(self):
"""OTEL_EXPORTER drives protocol when OTEL_EXPORTER_OTLP_PROTOCOL is unset."""
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import (
OTLPSpanExporter as OTLPSpanExporterGRPC,
)
from opentelemetry.sdk.trace.export import BatchSpanProcessor
popped_protocol = os.environ.pop("OTEL_EXPORTER_OTLP_PROTOCOL", None)
try:
config = OpenTelemetryConfig.from_env()
self.assertEqual(config.exporter, "otlp_grpc")
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
self.assertIsInstance(processor, BatchSpanProcessor)
self.assertIsInstance(processor.span_exporter, OTLPSpanExporterGRPC)
finally:
if popped_protocol is not None:
os.environ["OTEL_EXPORTER_OTLP_PROTOCOL"] = popped_protocol
def test_http_exporter_endpoint_normalization_for_traces(self):
"""Test that HTTP trace exporter gets properly normalized endpoint"""
config = OpenTelemetryConfig(
exporter="otlp_http", endpoint="http://collector:4318"
)
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify the endpoint was normalized to include /v1/traces
# Access the private _endpoint attribute if available
if hasattr(processor.span_exporter, "_endpoint"):
self.assertEqual(processor.span_exporter._endpoint, "http://collector:4318/v1/traces") # type: ignore[attr-defined]
def test_grpc_exporter_endpoint_normalization_for_traces(self):
"""Test that gRPC trace exporter gets properly normalized endpoint"""
config = OpenTelemetryConfig(
exporter="otlp_grpc", endpoint="http://collector:4317"
)
otel = OpenTelemetry(config=config)
processor = otel._get_span_processor()
# Verify the endpoint was normalized to include /v1/traces
# Note: gRPC exporters strip the http:// prefix, so we check for the normalized path
if hasattr(processor.span_exporter, "_endpoint"):
# gRPC exporter strips http:// prefix
self.assertIn("collector:4317", processor.span_exporter._endpoint) # type: ignore[attr-defined]
# The endpoint should have been normalized with /v1/traces before being passed to gRPC exporter
# We verify this by checking the normalization function was called correctly
normalized = otel._normalize_otel_endpoint(
"http://collector:4317", "traces"
)
self.assertEqual(normalized, "http://collector:4317/v1/traces")
def test_http_log_exporter_endpoint_normalization_for_logs(self):
"""Test that HTTP log exporter gets properly normalized endpoint"""
config = OpenTelemetryConfig(
exporter="otlp_http",
endpoint="http://collector:4318/v1/traces",
enable_events=True,
)
otel = OpenTelemetry(config=config)
exporter = otel._get_log_exporter()
# Verify the endpoint was normalized to /v1/logs (not /v1/traces)
# Access the private _endpoint attribute if available
if hasattr(exporter, "_endpoint"):
self.assertEqual(exporter._endpoint, "http://collector:4318/v1/logs") # type: ignore[attr-defined]
def test_grpc_log_exporter_endpoint_normalization_for_logs(self):
"""Test that gRPC log exporter gets properly normalized endpoint"""
config = OpenTelemetryConfig(
exporter="otlp_grpc",
endpoint="http://collector:4317/v1/traces",
enable_events=True,
)
otel = OpenTelemetry(config=config)
exporter = otel._get_log_exporter()
# Verify the endpoint was normalized to /v1/logs (not /v1/traces)
# Note: gRPC exporters strip the http:// prefix, so we check for the normalized path
if hasattr(exporter, "_endpoint"):
# gRPC exporter strips http:// prefix
self.assertIn("collector:4317", exporter._endpoint) # type: ignore[attr-defined]
# The endpoint should have been normalized with /v1/logs before being passed to gRPC exporter
# We verify this by checking the normalization function was called correctly
normalized = otel._normalize_otel_endpoint(
"http://collector:4317/v1/traces", "logs"
)
self.assertEqual(normalized, "http://collector:4317/v1/logs")
def test_get_metric_reader_uses_http_exporter_for_http_protobuf(self):
"""Test that http/protobuf protocol uses OTLPMetricExporterHTTP"""
from opentelemetry.exporter.otlp.proto.http.metric_exporter import (
OTLPMetricExporter,
)
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
config = OpenTelemetryConfig(
exporter="http/protobuf", endpoint="http://collector:4318"
)
otel = OpenTelemetry(config=config)
reader = otel._get_metric_reader()
self.assertIsInstance(reader, PeriodicExportingMetricReader)
self.assertIsInstance(reader._exporter, OTLPMetricExporter)
class TestOpenTelemetryExternalSpan(unittest.TestCase):
"""
Test suite for external span handling in OpenTelemetry integration.
These tests verify that LiteLLM correctly handles spans created outside
of LiteLLM (e.g., by Langfuse SDK, user application code, or global context)
without closing them prematurely.
Background:
- External spans can come from: Langfuse SDK, user code, HTTP traceparent headers, global context
- LiteLLM should NEVER close spans it did not create
- Bug: LiteLLM was reusing and closing external spans in _start_primary_span
"""
HERE = os.path.dirname(__file__)
def setUp(self):
"""Set up common test fixtures"""
self.span_exporter = InMemorySpanExporter()
self.tracer_provider = TracerProvider()
self.tracer_provider.add_span_processor(SimpleSpanProcessor(self.span_exporter))
# Don't set global tracer provider - instead, get tracers directly from our provider
# This avoids "Overriding of current TracerProvider is not allowed" warnings
# Clear any existing spans
self.span_exporter.clear()
def _create_test_kwargs_and_response(self):
"""Load test data from JSON files"""
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_kwargs.json")
) as f:
kwargs = json.load(f)
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_response.json")
) as f:
response_obj = json.load(f)
return kwargs, response_obj
def _get_spans_by_name(self, name):
"""Get all spans with the given name"""
spans = self.span_exporter.get_finished_spans()
return [s for s in spans if s.name == name]
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
def test_external_span_not_closed_with_use_otel_litellm_request_span_false(self):
"""
Test that external spans are not closed when USE_OTEL_LITELLM_REQUEST_SPAN=false (default).
Expected behavior:
- External span remains open (is_recording = True)
- raw_gen_ai_request spans are direct children of external span (shallow hierarchy)
- No litellm_request span is created
- Multiple completions work correctly
"""
# Initialize OpenTelemetry
otel = OpenTelemetry(tracer_provider=self.tracer_provider)
# Load test data
kwargs, response_obj = self._create_test_kwargs_and_response()
# Create external parent span using our test TracerProvider
tracer = self.tracer_provider.get_tracer(__name__)
with tracer.start_as_current_span("external_parent_span") as parent_span:
parent_ctx = parent_span.get_span_context()
parent_trace_id = parent_ctx.trace_id
parent_span_id = parent_ctx.span_id
self.assertTrue(
parent_span.is_recording(),
"External span should be recording before completion calls",
)
# First completion call
start_time = datetime.utcnow()
end_time = start_time + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj, start_time, end_time)
# Verify parent span is still recording
self.assertTrue(
parent_span.is_recording(),
"External span should still be recording after first completion",
)
# Second completion call
start_time2 = end_time
end_time2 = start_time2 + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj, start_time2, end_time2)
# Verify parent span is still recording
self.assertTrue(
parent_span.is_recording(),
"External span should still be recording after second completion",
)
# After exiting context, verify spans
spans = self.span_exporter.get_finished_spans()
# All spans should have the same trace_id
for span in spans:
self.assertEqual(
span.context.trace_id,
parent_trace_id,
f"Span {span.name} should have same trace_id as parent",
)
# Should have external_parent_span
parent_spans = self._get_spans_by_name("external_parent_span")
self.assertEqual(
len(parent_spans), 1, "Should have exactly one external_parent_span"
)
# Verify LiteLLM set attributes on external parent span
parent_span_finished = parent_spans[0]
self.assertIsNotNone(
parent_span_finished.attributes,
"Parent span should have attributes set by LiteLLM",
)
self.assertIn(
"gen_ai.request.model",
parent_span_finished.attributes,
"Parent span should have model attribute from LiteLLM",
)
# Should have raw_gen_ai_request spans (if message_logging is on)
raw_spans = self._get_spans_by_name("raw_gen_ai_request")
# Note: May be 0 if message_logging is off, or 2 if on
# Should NOT have litellm_request spans (USE_OTEL_LITELLM_REQUEST_SPAN=false)
litellm_spans = self._get_spans_by_name("litellm_request")
self.assertEqual(
len(litellm_spans),
0,
"Should NOT have litellm_request spans when USE_OTEL_LITELLM_REQUEST_SPAN=false",
)
# Verify raw_gen_ai_request spans are direct children of external span
for raw_span in raw_spans:
self.assertEqual(
raw_span.parent.span_id if raw_span.parent else None,
parent_span_id,
"raw_gen_ai_request should be direct child of external_parent_span",
)
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "true"}, clear=False)
def test_external_span_not_closed_with_use_otel_litellm_request_span_true(self):
"""
Test that external spans are not closed when USE_OTEL_LITELLM_REQUEST_SPAN=true.
Expected behavior:
- External span remains open (is_recording = True)
- litellm_request spans are created as children of external span
- raw_gen_ai_request spans are children of litellm_request spans
- Correct hierarchy: external_parent → litellm_request → raw_gen_ai_request
"""
import copy
# Initialize OpenTelemetry
otel = OpenTelemetry(tracer_provider=self.tracer_provider)
kwargs1, response_obj = self._create_test_kwargs_and_response()
kwargs2 = copy.deepcopy(kwargs1)
# Create external parent span using our test TracerProvider
tracer = self.tracer_provider.get_tracer(__name__)
with tracer.start_as_current_span("external_parent_span") as parent_span:
parent_ctx = parent_span.get_span_context()
parent_trace_id = parent_ctx.trace_id
parent_span_id = parent_ctx.span_id
# First completion call
start_time = datetime.utcnow()
end_time = start_time + timedelta(seconds=1)
otel._handle_success(kwargs1, response_obj, start_time, end_time)
# Verify parent span is still recording
self.assertTrue(
parent_span.is_recording(),
"External span should still be recording after first completion",
)
# Second completion call
start_time2 = end_time
end_time2 = start_time2 + timedelta(seconds=1)
otel._handle_success(kwargs2, response_obj, start_time2, end_time2)
# Verify parent span is still recording
self.assertTrue(
parent_span.is_recording(),
"External span should still be recording after second completion",
)
# After exiting context, verify spans
spans = self.span_exporter.get_finished_spans()
# All spans should have the same trace_id
for span in spans:
self.assertEqual(
span.context.trace_id,
parent_trace_id,
f"Span {span.name} should have same trace_id as parent",
)
# Should have litellm_request spans (USE_OTEL_LITELLM_REQUEST_SPAN=true)
litellm_spans = self._get_spans_by_name("litellm_request")
self.assertEqual(
len(litellm_spans),
2,
"Should have 2 litellm_request spans when USE_OTEL_LITELLM_REQUEST_SPAN=true",
)
# Verify litellm_request spans are children of external span
for litellm_span in litellm_spans:
self.assertEqual(
litellm_span.parent.span_id if litellm_span.parent else None,
parent_span_id,
"litellm_request should be child of external_parent_span",
)
# Verify raw_gen_ai_request spans (if present) are children of litellm_request
raw_spans = self._get_spans_by_name("raw_gen_ai_request")
if raw_spans:
litellm_span_ids = {s.context.span_id for s in litellm_spans}
for raw_span in raw_spans:
self.assertIn(
raw_span.parent.span_id if raw_span.parent else None,
litellm_span_ids,
"raw_gen_ai_request should be child of litellm_request",
)
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
def test_external_span_with_multiple_completions(self):
"""
Test that multiple completion calls work correctly within external span context.
Expected behavior:
- Both completion calls succeed
- All spans belong to the same trace
- External span remains open throughout
- No errors or warnings about "ended span"
"""
# Initialize OpenTelemetry
otel = OpenTelemetry(tracer_provider=self.tracer_provider)
# Load test data
kwargs, response_obj = self._create_test_kwargs_and_response()
# Create external parent span using our test TracerProvider
tracer = self.tracer_provider.get_tracer(__name__)
with tracer.start_as_current_span("external_parent_span") as parent_span:
parent_ctx = parent_span.get_span_context()
parent_trace_id = parent_ctx.trace_id
# Make multiple completion calls
for i in range(3):
start_time = datetime.utcnow()
end_time = start_time + timedelta(seconds=1)
# This should not raise any exceptions
otel._handle_success(kwargs, response_obj, start_time, end_time)
# Verify parent span is still recording after each call
self.assertTrue(
parent_span.is_recording(),
f"External span should still be recording after completion #{i + 1}",
)
# Verify all spans have the same trace_id
spans = self.span_exporter.get_finished_spans()
for span in spans:
self.assertEqual(
span.context.trace_id,
parent_trace_id,
"All spans should belong to the same trace",
)
# Should have the external parent span
parent_spans = self._get_spans_by_name("external_parent_span")
self.assertEqual(
len(parent_spans), 1, "Should have exactly one external_parent_span"
)
# Verify LiteLLM set attributes on external parent span
parent_span_finished = parent_spans[0]
self.assertIn(
"gen_ai.request.model",
parent_span_finished.attributes,
"Parent span should have model attribute from LiteLLM",
)
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
def test_external_span_from_global_context(self):
"""
Test external span detection from global context (Priority 3 in _get_span_context).
This simulates the case where a span is set in the global context
(e.g., by user code or Langfuse SDK) and LiteLLM detects it via
trace.get_current_span().
Expected behavior:
- LiteLLM detects the span from global context
- External span is not closed
- Correct parent-child relationship
"""
# Initialize OpenTelemetry
otel = OpenTelemetry(tracer_provider=self.tracer_provider)
# Load test data
kwargs, response_obj = self._create_test_kwargs_and_response()
# Create external parent span and set it as current using our test TracerProvider
tracer = self.tracer_provider.get_tracer(__name__)
with tracer.start_as_current_span("external_global_span") as parent_span:
parent_ctx = parent_span.get_span_context()
parent_trace_id = parent_ctx.trace_id
# Verify the span is in global context
current_span = trace.get_current_span()
self.assertEqual(
current_span, parent_span, "Span should be in global context"
)
# Make completion call
start_time = datetime.utcnow()
end_time = start_time + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj, start_time, end_time)
# Verify parent span is still recording
self.assertTrue(
parent_span.is_recording(),
"External span from global context should not be closed",
)
# Verify trace structure
spans = self.span_exporter.get_finished_spans()
for span in spans:
self.assertEqual(
span.context.trace_id,
parent_trace_id,
"All spans should have the same trace_id",
)
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
def test_external_span_hierarchy_preserved(self):
"""
Test that span hierarchy is correctly preserved with external parent.
Expected behavior:
- Parent span IDs are correct
- Trace structure matches expected hierarchy
- Span names are correct
"""
# Initialize OpenTelemetry
otel = OpenTelemetry(tracer_provider=self.tracer_provider)
otel.message_logging = (
True # Enable message logging to get raw_gen_ai_request spans
)
# Load test data
kwargs, response_obj = self._create_test_kwargs_and_response()
# Create external parent span using our test TracerProvider
tracer = self.tracer_provider.get_tracer(__name__)
with tracer.start_as_current_span("external_parent_span") as parent_span:
parent_span_id = parent_span.get_span_context().span_id
# Make completion call
start_time = datetime.utcnow()
end_time = start_time + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj, start_time, end_time)
# Verify hierarchy
spans = self.span_exporter.get_finished_spans()
# Get spans by name
parent_spans = self._get_spans_by_name("external_parent_span")
raw_spans = self._get_spans_by_name("raw_gen_ai_request")
self.assertEqual(len(parent_spans), 1, "Should have one parent span")
# Verify parent-child relationship
if raw_spans: # If message_logging is on
for raw_span in raw_spans:
self.assertEqual(
raw_span.parent.span_id if raw_span.parent else None,
parent_span_id,
"raw_gen_ai_request should be child of external_parent_span",
)
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
def test_external_span_not_ended_on_failure(self):
"""
Test that external spans are not closed even on failure.
Expected behavior:
- When _handle_failure is called with external span context
- External span remains open (is_recording = True)
- Error span is created correctly
- External span status is NOT changed by LiteLLM
"""
# Initialize OpenTelemetry
otel = OpenTelemetry(tracer_provider=self.tracer_provider)
# Load test data
kwargs, response_obj = self._create_test_kwargs_and_response()
# Create external parent span using our test TracerProvider
tracer = self.tracer_provider.get_tracer(__name__)
with tracer.start_as_current_span("external_parent_span") as parent_span:
parent_ctx = parent_span.get_span_context()
parent_trace_id = parent_ctx.trace_id
# Simulate failure
start_time = datetime.utcnow()
end_time = start_time + timedelta(seconds=1)
# Create error response object
error_response = {"error": "Test error"}
# Call _handle_failure
otel._handle_failure(kwargs, error_response, start_time, end_time)
# Verify parent span is still recording
self.assertTrue(
parent_span.is_recording(),
"External span should still be recording even after failure",
)
# Verify trace structure
spans = self.span_exporter.get_finished_spans()
# All spans should have the same trace_id
for span in spans:
self.assertEqual(
span.context.trace_id,
parent_trace_id,
"All spans should have the same trace_id even on failure",
)
# Should have external_parent_span
parent_spans = self._get_spans_by_name("external_parent_span")
self.assertEqual(
len(parent_spans), 1, "Should have exactly one external_parent_span"
)
# Verify LiteLLM set attributes on external parent span even on failure
parent_span_finished = parent_spans[0]
self.assertIn(
"gen_ai.request.model",
parent_span_finished.attributes,
"Parent span should have model attribute from LiteLLM even on failure",
)
class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
"""
Test suite for OpenTelemetry 1.38 Semantic Conventions compliance.
These tests verify that LiteLLM emits span attributes following the
OpenTelemetry GenAI semantic conventions v1.38, including:
- gen_ai.input.messages (JSON string with parts array)
- gen_ai.output.messages (JSON string with parts array)
- gen_ai.usage.input_tokens / output_tokens (new naming)
- gen_ai.response.finish_reasons (JSON array)
See: https://github.com/BerriAI/litellm/issues/17794
"""
def setUp(self):
# Insulate from a shell-set OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT
# so these tests exercise the legacy default path (message_logging=True).
self._prev = os.environ.pop(
"OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT", None
)
def tearDown(self):
if self._prev is not None:
os.environ["OTEL_INSTRUMENTATION_GENAI_CAPTURE_MESSAGE_CONTENT"] = (
self._prev
)
def test_input_messages_uses_parts_structure(self):
"""
Test that gen_ai.input.messages uses the OTEL 1.38 parts array structure.
Expected format:
[{"role": "user", "parts": [{"type": "text", "content": "Hello"}]}]
"""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello world"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
}
response_obj = {
"id": "test-response-id",
"model": "gpt-4",
"choices": [
{
"finish_reason": "stop",
"message": {"role": "assistant", "content": "Hi there!"},
}
],
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
# Find the call that set gen_ai.input.messages
input_messages_calls = [
call
for call in mock_span.set_attribute.call_args_list
if call[0][0] == "gen_ai.input.messages"
]
self.assertEqual(
len(input_messages_calls),
1,
"Should have exactly one gen_ai.input.messages attribute",
)
input_messages_value = input_messages_calls[0][0][1]
parsed = json.loads(input_messages_value)
# Verify structure
self.assertIsInstance(parsed, list)
self.assertEqual(len(parsed), 1)
self.assertEqual(parsed[0]["role"], "user")
self.assertIn("parts", parsed[0])
self.assertEqual(parsed[0]["parts"][0]["type"], "text")
self.assertEqual(parsed[0]["parts"][0]["content"], "Hello world")
def test_output_messages_uses_parts_structure(self):
"""
Test that gen_ai.output.messages uses the OTEL 1.38 parts array structure.
Expected format:
[{"role": "assistant", "parts": [{"type": "text", "content": "Hi!"}], "finish_reason": "stop"}]
"""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
}
response_obj = {
"id": "test-response-id",
"model": "gpt-4",
"choices": [
{
"finish_reason": "stop",
"message": {"role": "assistant", "content": "Hello back!"},
}
],
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
# Find the call that set gen_ai.output.messages
output_messages_calls = [
call
for call in mock_span.set_attribute.call_args_list
if call[0][0] == "gen_ai.output.messages"
]
self.assertEqual(
len(output_messages_calls),
1,
"Should have exactly one gen_ai.output.messages attribute",
)
output_messages_value = output_messages_calls[0][0][1]
parsed = json.loads(output_messages_value)
# Verify structure
self.assertIsInstance(parsed, list)
self.assertEqual(len(parsed), 1)
self.assertEqual(parsed[0]["role"], "assistant")
self.assertIn("parts", parsed[0])
self.assertEqual(parsed[0]["parts"][0]["type"], "text")
self.assertEqual(parsed[0]["parts"][0]["content"], "Hello back!")
self.assertEqual(parsed[0]["finish_reason"], "stop")
def test_usage_tokens_use_new_naming_convention(self):
"""
Test that token usage uses the OTEL 1.38 naming convention:
- gen_ai.usage.input_tokens (not prompt_tokens)
- gen_ai.usage.output_tokens (not completion_tokens)
"""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
}
response_obj = {
"id": "test-response-id",
"model": "gpt-4",
"choices": [],
"usage": {
"prompt_tokens": 100,
"completion_tokens": 50,
"total_tokens": 150,
},
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
# Verify new naming convention is used
mock_span.set_attribute.assert_any_call("gen_ai.usage.input_tokens", 100)
mock_span.set_attribute.assert_any_call("gen_ai.usage.output_tokens", 50)
mock_span.set_attribute.assert_any_call("gen_ai.usage.total_tokens", 150)
def test_finish_reasons_is_json_array(self):
"""
Test that gen_ai.response.finish_reasons is a proper JSON array.
Expected: '["stop"]' (not "['stop']")
"""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
}
response_obj = {
"id": "test-response-id",
"model": "gpt-4",
"choices": [
{
"finish_reason": "stop",
"message": {"role": "assistant", "content": "Hi"},
},
],
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
# Find the call that set gen_ai.response.finish_reasons
finish_reasons_calls = [
call
for call in mock_span.set_attribute.call_args_list
if call[0][0] == "gen_ai.response.finish_reasons"
]
self.assertEqual(
len(finish_reasons_calls),
1,
"Should have exactly one gen_ai.response.finish_reasons attribute",
)
finish_reasons_value = finish_reasons_calls[0][0][1]
# Verify it's valid JSON (not Python repr)
parsed = json.loads(finish_reasons_value)
self.assertEqual(parsed, ["stop"])
def test_operation_name_is_chat_for_completion(self):
"""
Test that gen_ai.operation.name is 'chat' for completion calls.
"""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
}
response_obj = {
"id": "test-response-id",
"model": "gpt-4",
"choices": [],
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
mock_span.set_attribute.assert_any_call("gen_ai.operation.name", "chat")
@parameterized.expand([("_handle_success",), ("_handle_failure",)])
def test_handle_success_failure_nulls_parent_span_if_ignore_context_propagation(
self, handle_method: str
):
"""
If ignore_context_propagation is True, _handle_success should ignore any parent span
and create a root-level span. This could be useful for langfuse_otel where
_handle_success may ignore parent spans from other providers and create a root-level
span (symmetric with _handle_failure).
"""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(
config=OpenTelemetryConfig(ignore_context_propagation=True),
tracer_provider=tracer_provider,
)
otel.tracer = tracer_provider.get_tracer("litellm")
other_tracer = tracer_provider.get_tracer("other_provider")
other_span = other_tracer.start_span("parent_span")
start = datetime.now(timezone.utc)
end = start + timedelta(seconds=1)
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {
"custom_llm_provider": "openai",
"metadata": {"litellm_parent_otel_span": other_span},
},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
"exception": Exception("test error"),
}
with patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "true"}):
if handle_method == "_handle_success":
otel._handle_success(kwargs, None, start, end)
elif handle_method == "_handle_failure":
otel._handle_failure(kwargs, None, start, end)
else:
self.fail(f"Invalid handle_method: {handle_method}")
other_span.end()
spans = span_exporter.get_finished_spans()
child_spans = [s for s in spans if s.name != "parent_span"]
child_span_ids = {s.context.span_id for s in child_spans if s.context}
self.assertTrue(child_spans, "Expected at least one child span")
for span in child_spans:
assert (
span.parent is None or span.parent.span_id in child_span_ids
), f"if ignore_context_propagation is True, span should not have parent from other providers, but got parent: {span.parent}"
@parameterized.expand([("_handle_success",), ("_handle_failure",)])
def test_handle_success_failure_default_preserves_parent_span(
self, handle_method: str
):
"""
For default otel callbacks, _handle_success should use parent spans normally.
(symmetric with _handle_failure)
"""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer("litellm")
parent_span = otel.tracer.start_span("parent_span")
start = datetime.now(timezone.utc)
end = start + timedelta(seconds=1)
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {
"custom_llm_provider": "openai",
"metadata": {"litellm_parent_otel_span": parent_span},
},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
"exception": Exception("test error"),
}
with patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "true"}):
if handle_method == "_handle_success":
otel._handle_success(kwargs, None, start, end)
elif handle_method == "_handle_failure":
otel._handle_failure(kwargs, None, start, end)
else:
self.fail(f"Invalid handle_method: {handle_method}")
parent_span.end()
spans = span_exporter.get_finished_spans()
child_spans = [s for s in spans if s.name != "parent_span"]
self.assertTrue(child_spans, "Expected at least one child span")
for span in child_spans:
assert (
span.parent is not None
), f"By default parent span should be preserved, but got None parent for span: {span.name}"
@parameterized.expand([("_handle_success",), ("_handle_failure",)])
def test_handle_success_failure_with_context_propagation_preserves_parent_span(
self, handle_method: str
):
"""
For otel callbacks with context propagation enabled, _handle_success should
use parent spans normally. (symmetric with _handle_failure)
"""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(
config=OpenTelemetryConfig(ignore_context_propagation=False),
tracer_provider=tracer_provider,
)
otel.tracer = tracer_provider.get_tracer("litellm")
parent_span = otel.tracer.start_span("parent_span")
start = datetime.now(timezone.utc)
end = start + timedelta(seconds=1)
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {
"custom_llm_provider": "openai",
"metadata": {"litellm_parent_otel_span": parent_span},
},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
"exception": Exception("test error"),
}
with patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "true"}):
if handle_method == "_handle_success":
otel._handle_success(kwargs, None, start, end)
elif handle_method == "_handle_failure":
otel._handle_failure(kwargs, None, start, end)
else:
self.fail(f"Invalid handle_method: {handle_method}")
parent_span.end()
spans = span_exporter.get_finished_spans()
child_spans = [s for s in spans if s.name != "parent_span"]
self.assertTrue(child_spans, "Expected at least one child span")
for span in child_spans:
assert (
span.parent is not None
), f"If ignore_context_propagation is False, parent span should be preserved, but got None parent for span: {span.name}"
def test_handle_failure_hasattr_guard_on_parent_name(self):
"""
_handle_failure should not raise AttributeError when parent_otel_span
lacks a 'name' attribute (e.g., NonRecordingSpan).
"""
otel = OpenTelemetry()
otel.tracer = MagicMock()
mock_span = MagicMock()
otel.tracer.start_span.return_value = mock_span
parent_without_name = MagicMock()
del parent_without_name.name
start = datetime.utcnow()
end = start + timedelta(seconds=1)
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {
"custom_llm_provider": "openai",
"metadata": {"litellm_parent_otel_span": parent_without_name},
},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
}
try:
otel._handle_failure(kwargs, None, start, end)
except AttributeError as e:
self.fail(
f"_handle_failure raised AttributeError on parent span without 'name': {e}"
)
def test_handle_failure_creates_error_span(self):
"""
_handle_failure should create a span with ERROR status.
"""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer("litellm")
start = datetime.utcnow()
end = start + timedelta(seconds=1)
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
"exception": Exception("test error"),
}
otel._handle_failure(kwargs, None, start, end)
spans = span_exporter.get_finished_spans()
self.assertTrue(spans, "Expected at least one span")
from opentelemetry.trace import StatusCode
error_spans = [s for s in spans if s.status.status_code == StatusCode.ERROR]
self.assertTrue(error_spans, "Expected at least one span with ERROR status")
class TestRawSpanAttributeIsolation(unittest.TestCase):
"""Issue #3: raw_gen_ai_request span should only contain provider-specific
llm.{provider}.* attributes, not the duplicated gen_ai.* / metadata.* attrs."""
@patch("litellm.turn_off_message_logging", False)
def test_raw_span_does_not_duplicate_parent_attributes(self):
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.message_logging = True
mock_tracer = tracer_provider.get_tracer(__name__)
otel.get_tracer_to_use_for_request = MagicMock(return_value=mock_tracer)
raw_span = mock_tracer.start_span("raw_gen_ai_request")
kwargs = {
"litellm_params": {"custom_llm_provider": "vertex_ai"},
"optional_params": {"temperature": 0.7},
"original_response": '{"predictions": [1,2,3]}',
"additional_args": {
"complete_input_dict": {"instances": [{"content": "hello"}]}
},
"standard_logging_object": {
"id": "test-id",
"call_type": "embedding",
"metadata": {"user_api_key_hash": "abc123"},
"hidden_params": {},
},
}
response_obj = {"model": "text-embedding-004", "usage": {"total_tokens": 5}}
otel.set_raw_request_attributes(raw_span, kwargs, response_obj)
raw_span.end()
spans = span_exporter.get_finished_spans()
raw = [s for s in spans if s.name == "raw_gen_ai_request"][0]
attr_keys = set(raw.attributes.keys()) if raw.attributes else set()
# Provider-specific attributes SHOULD be present
self.assertTrue(
any(k.startswith("llm.vertex_ai.") for k in attr_keys),
f"Expected llm.vertex_ai.* attributes, got: {attr_keys}",
)
# Standard gen_ai / metadata attributes should NOT be present
self.assertFalse(
any(k.startswith("gen_ai.") for k in attr_keys),
f"raw span should not contain gen_ai.* attributes, got: {attr_keys}",
)
self.assertFalse(
any(k.startswith("metadata.") for k in attr_keys),
f"raw span should not contain metadata.* attributes, got: {attr_keys}",
)
class TestNoParentSpanDuplication(unittest.TestCase):
"""Issue #4: When litellm_request child span exists, the parent
litellm_proxy_request span should NOT get set_attributes() called."""
HERE = os.path.dirname(__file__)
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "true"}, clear=False)
def test_parent_proxy_span_not_duplicated(self):
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_kwargs.json")
) as f:
kwargs = json.load(f)
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_response.json")
) as f:
response_obj = json.load(f)
# Simulate proxy flow: create a parent proxy span
tracer = tracer_provider.get_tracer(__name__)
from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME
parent_span = tracer.start_span(name=LITELLM_PROXY_REQUEST_SPAN_NAME)
# Inject parent span into kwargs so _get_span_context finds it
kwargs["litellm_params"]["metadata"]["litellm_parent_otel_span"] = parent_span
start = datetime.utcnow()
end = start + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj, start, end)
spans = span_exporter.get_finished_spans()
proxy_spans = [s for s in spans if s.name == LITELLM_PROXY_REQUEST_SPAN_NAME]
self.assertEqual(len(proxy_spans), 1, "Should have exactly one proxy span")
proxy_attrs = proxy_spans[0].attributes or {}
# The parent proxy span should NOT have gen_ai.request.model set
self.assertNotIn(
"gen_ai.request.model",
proxy_attrs,
"Parent proxy span should NOT duplicate gen_ai.request.model (Issue #4)",
)
class TestGuardrailSpanParenting(unittest.TestCase):
"""Issue #5: Guardrail spans must not be orphaned — they should always
be children of the litellm_request span (or parent span)."""
def test_guardrail_span_is_child_of_litellm_request(self):
"""When no parent proxy span exists, guardrail spans should be
children of the litellm_request span, not orphaned root spans."""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
guardrail_info = {
"guardrail_name": "pii_filter",
"guardrail_mode": "pre_call",
"guardrail_response": "ok",
"start_time": time.time(),
"end_time": time.time() + 0.1,
}
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai", "metadata": {}},
"standard_logging_object": {
"id": "test-guardrail-id",
"call_type": "completion",
"metadata": {},
"hidden_params": {},
"guardrail_information": [guardrail_info],
},
}
response_obj = {
"id": "chatcmpl-test",
"choices": [
{
"finish_reason": "stop",
"index": 0,
"message": {"content": "Hi!", "role": "assistant"},
}
],
"usage": {
"prompt_tokens": 5,
"completion_tokens": 2,
"total_tokens": 7,
},
}
start = datetime.utcnow()
end = start + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj, start, end)
spans = span_exporter.get_finished_spans()
guardrail_spans = [s for s in spans if s.name == "guardrail"]
litellm_spans = [s for s in spans if s.name == "litellm_request"]
self.assertTrue(guardrail_spans, "Expected at least one guardrail span")
self.assertTrue(litellm_spans, "Expected a litellm_request span")
litellm_span = litellm_spans[0]
for gs in guardrail_spans:
# All spans should share the same trace_id (not orphaned)
self.assertEqual(
gs.context.trace_id,
litellm_span.context.trace_id,
"Guardrail span should share trace_id with litellm_request (not orphaned)",
)
# Guardrail should be a child of the litellm_request span
self.assertIsNotNone(
gs.parent,
"Guardrail span should have a parent (not be a root span)",
)
self.assertEqual(
gs.parent.span_id,
litellm_span.context.span_id,
"Guardrail span should be a child of litellm_request",
)
def test_guardrail_span_parented_on_failure(self):
"""Guardrail spans should also be properly parented in the failure path."""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
guardrail_info = {
"guardrail_name": "content_filter",
"guardrail_mode": "pre_call",
"guardrail_response": "blocked",
"start_time": time.time(),
"end_time": time.time() + 0.05,
}
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai", "metadata": {}},
"standard_logging_object": {
"id": "test-fail-id",
"call_type": "completion",
"metadata": {},
"hidden_params": {},
"guardrail_information": [guardrail_info],
},
"exception": Exception("test error"),
}
start = datetime.utcnow()
end = start + timedelta(seconds=1)
otel._handle_failure(kwargs, None, start, end)
spans = span_exporter.get_finished_spans()
guardrail_spans = [s for s in spans if s.name == "guardrail"]
self.assertTrue(guardrail_spans, "Expected at least one guardrail span")
for gs in guardrail_spans:
self.assertIsNotNone(
gs.parent,
"Guardrail span should have a parent on failure path too",
)
class TestResponseIdFallback(unittest.TestCase):
"""Issue #8: gen_ai.response.id should be set for embeddings and image gen
using standard_logging_payload['id'] as fallback."""
def test_response_id_from_response_obj(self):
"""When response_obj has an id, it should be used."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "gpt-4",
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "litellm-call-id-123",
"call_type": "completion",
"metadata": {},
},
}
response_obj = {
"id": "chatcmpl-provider-id-456",
"choices": [
{
"finish_reason": "stop",
"index": 0,
"message": {"content": "Hi", "role": "assistant"},
}
],
"usage": {
"prompt_tokens": 5,
"completion_tokens": 2,
"total_tokens": 7,
},
}
otel.set_attributes(mock_span, kwargs, response_obj)
# Should use provider response ID, not litellm call ID
mock_span.set_attribute.assert_any_call(
"gen_ai.response.id", "chatcmpl-provider-id-456"
)
def test_response_id_fallback_for_embeddings(self):
"""When response_obj has no id (embeddings), fallback to
standard_logging_payload['id']."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "text-embedding-ada-002",
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "litellm-embed-call-789",
"call_type": "embedding",
"metadata": {},
},
}
# Embedding response has no "id" field
response_obj = {
"object": "list",
"data": [{"embedding": [0.1, 0.2], "index": 0}],
"model": "text-embedding-ada-002",
"usage": {"prompt_tokens": 5, "total_tokens": 5},
}
otel.set_attributes(mock_span, kwargs, response_obj)
# Should fallback to litellm call ID
mock_span.set_attribute.assert_any_call(
"gen_ai.response.id", "litellm-embed-call-789"
)
def test_response_id_fallback_for_image_gen(self):
"""When response_obj has no id (image gen), fallback to
standard_logging_payload['id']."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "dall-e-3",
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "litellm-img-call-101",
"call_type": "image_generation",
"metadata": {},
},
}
# Image response has no "id" field
response_obj = {
"created": 1234567890,
"data": [{"url": "https://example.com/img.png"}],
}
otel.set_attributes(mock_span, kwargs, response_obj)
# Should fallback to litellm call ID
mock_span.set_attribute.assert_any_call(
"gen_ai.response.id", "litellm-img-call-101"
)
def test_litellm_call_id_emitted_as_span_attribute(self):
"""litellm.call_id must be set on the span from standard_logging_payload."""
otel = OpenTelemetry()
mock_span = MagicMock()
call_id = "my-litellm-call-uuid-456"
kwargs = {
"model": "gpt-4o",
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "chatcmpl-provider-id",
"litellm_call_id": call_id,
"call_type": "completion",
"metadata": {},
},
}
response_obj = {"id": "chatcmpl-provider-id", "model": "gpt-4o"}
otel.set_attributes(mock_span, kwargs, response_obj)
mock_span.set_attribute.assert_any_call("litellm.call_id", call_id)
class TestOpenTelemetryResponsesAPI(unittest.TestCase):
"""
Tests for Responses API (/v1/responses) OTel span attributes.
The Responses API uses ``output`` (list of output items) instead of
``choices``, ``instructions`` instead of ``system_instructions``, and
``status`` instead of per-choice ``finish_reason``.
See: https://github.com/BerriAI/litellm/issues/25840
"""
def _base_kwargs(self, **overrides):
"""Return minimal kwargs for set_attributes with Responses API defaults."""
kwargs = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "What is 2+2?"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "resp_abc123",
"call_type": "responses",
"metadata": {},
},
}
kwargs.update(overrides)
return kwargs
def _responses_api_response_obj(self, text="The answer is 4.", status="completed"):
"""Return a dict mimicking ResponsesAPIResponse with a message output."""
return {
"id": "resp_abc123",
"model": "gpt-4o",
"status": status,
"output": [
{
"type": "message",
"role": "assistant",
"content": [
{
"type": "output_text",
"text": text,
}
],
}
],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 20,
"total_tokens": 30,
},
}
def _get_attr(self, mock_span, attr_name):
"""Extract the value set for a specific attribute name, or None."""
calls = [
call
for call in mock_span.set_attribute.call_args_list
if call[0][0] == attr_name
]
if not calls:
return None
return calls[0][0][1]
# ------------------------------------------------------------------
# gen_ai.output.messages
# ------------------------------------------------------------------
def test_output_messages_populated_for_responses_api(self):
"""gen_ai.output.messages must be set when response has output items."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = self._base_kwargs()
response_obj = self._responses_api_response_obj(text="The answer is 4.")
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
raw = self._get_attr(mock_span, "gen_ai.output.messages")
self.assertIsNotNone(raw, "gen_ai.output.messages should be set")
parsed = json.loads(raw)
self.assertIsInstance(parsed, list)
self.assertEqual(len(parsed), 1)
self.assertEqual(parsed[0]["role"], "assistant")
self.assertIn("parts", parsed[0])
self.assertEqual(parsed[0]["parts"][0]["type"], "text")
self.assertEqual(parsed[0]["parts"][0]["content"], "The answer is 4.")
def test_output_messages_with_multiple_content_items(self):
"""Multiple output_text items in a single message should all appear as parts."""
otel = OpenTelemetry()
mock_span = MagicMock()
response_obj = {
"id": "resp_multi",
"model": "gpt-4o",
"status": "completed",
"output": [
{
"type": "message",
"role": "assistant",
"content": [
{"type": "output_text", "text": "First paragraph."},
{"type": "output_text", "text": "Second paragraph."},
],
}
],
}
otel.set_attributes(
span=mock_span, kwargs=self._base_kwargs(), response_obj=response_obj
)
raw = self._get_attr(mock_span, "gen_ai.output.messages")
parsed = json.loads(raw)
self.assertEqual(len(parsed[0]["parts"]), 2)
self.assertEqual(parsed[0]["parts"][0]["content"], "First paragraph.")
self.assertEqual(parsed[0]["parts"][1]["content"], "Second paragraph.")
def test_output_messages_with_function_call(self):
"""function_call output items should appear as tool_call parts."""
otel = OpenTelemetry()
mock_span = MagicMock()
response_obj = {
"id": "resp_fc",
"model": "gpt-4o",
"status": "completed",
"output": [
{
"type": "function_call",
"name": "get_weather",
"call_id": "call_abc",
"arguments": '{"location": "SF"}',
}
],
}
otel.set_attributes(
span=mock_span, kwargs=self._base_kwargs(), response_obj=response_obj
)
raw = self._get_attr(mock_span, "gen_ai.output.messages")
parsed = json.loads(raw)
self.assertEqual(len(parsed), 1)
self.assertEqual(parsed[0]["role"], "assistant")
self.assertEqual(parsed[0]["parts"][0]["type"], "tool_call")
self.assertEqual(parsed[0]["parts"][0]["name"], "get_weather")
self.assertEqual(parsed[0]["parts"][0]["arguments"], '{"location": "SF"}')
self.assertEqual(parsed[0]["parts"][0]["id"], "call_abc")
def test_output_messages_mixed_message_and_function_call(self):
"""Mixed output with both message and function_call items."""
otel = OpenTelemetry()
mock_span = MagicMock()
response_obj = {
"id": "resp_mixed",
"model": "gpt-4o",
"status": "completed",
"output": [
{
"type": "message",
"role": "assistant",
"content": [
{"type": "output_text", "text": "Let me check the weather."},
],
},
{
"type": "function_call",
"name": "get_weather",
"call_id": "call_xyz",
"arguments": "{}",
},
],
}
otel.set_attributes(
span=mock_span, kwargs=self._base_kwargs(), response_obj=response_obj
)
raw = self._get_attr(mock_span, "gen_ai.output.messages")
parsed = json.loads(raw)
self.assertEqual(len(parsed), 2)
self.assertEqual(parsed[0]["role"], "assistant")
self.assertEqual(parsed[0]["parts"][0]["content"], "Let me check the weather.")
self.assertEqual(parsed[1]["parts"][0]["type"], "tool_call")
def test_output_messages_empty_text_skipped(self):
"""Output items with empty text should not produce parts."""
otel = OpenTelemetry()
mock_span = MagicMock()
response_obj = {
"id": "resp_empty",
"model": "gpt-4o",
"status": "completed",
"output": [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": ""}],
}
],
}
otel.set_attributes(
span=mock_span, kwargs=self._base_kwargs(), response_obj=response_obj
)
# No output messages should be set since the text is empty
raw = self._get_attr(mock_span, "gen_ai.output.messages")
self.assertIsNone(
raw, "Empty output text should not produce gen_ai.output.messages"
)
def test_choices_still_work(self):
"""Existing choices-based responses must still work (no regression)."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
},
}
response_obj = {
"id": "chatcmpl-123",
"model": "gpt-4",
"choices": [
{
"finish_reason": "stop",
"message": {"role": "assistant", "content": "Hi there!"},
}
],
"usage": {"prompt_tokens": 5, "completion_tokens": 10, "total_tokens": 15},
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
raw = self._get_attr(mock_span, "gen_ai.output.messages")
parsed = json.loads(raw)
self.assertEqual(parsed[0]["parts"][0]["content"], "Hi there!")
self.assertEqual(parsed[0]["finish_reason"], "stop")
# ------------------------------------------------------------------
# gen_ai.response.finish_reasons
# ------------------------------------------------------------------
def test_finish_reasons_from_status(self):
"""gen_ai.response.finish_reasons should use ResponsesAPIResponse.status."""
otel = OpenTelemetry()
mock_span = MagicMock()
otel.set_attributes(
span=mock_span,
kwargs=self._base_kwargs(),
response_obj=self._responses_api_response_obj(status="completed"),
)
raw = self._get_attr(mock_span, "gen_ai.response.finish_reasons")
self.assertIsNotNone(raw)
parsed = json.loads(raw)
self.assertEqual(parsed, ["completed"])
def test_finish_reasons_incomplete_status(self):
"""Non-completed status values should still be captured."""
otel = OpenTelemetry()
mock_span = MagicMock()
otel.set_attributes(
span=mock_span,
kwargs=self._base_kwargs(),
response_obj=self._responses_api_response_obj(status="incomplete"),
)
raw = self._get_attr(mock_span, "gen_ai.response.finish_reasons")
parsed = json.loads(raw)
self.assertEqual(parsed, ["incomplete"])
# ------------------------------------------------------------------
# gen_ai.system_instructions
# ------------------------------------------------------------------
def test_system_instructions_from_instructions_kwarg(self):
"""Responses API passes system prompt as kwargs['instructions']."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = self._base_kwargs(instructions="You are a math tutor.")
response_obj = self._responses_api_response_obj()
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
value = self._get_attr(mock_span, "gen_ai.system_instructions")
self.assertEqual(value, "You are a math tutor.")
def test_system_instructions_from_system_kwarg(self):
"""Anthropic Messages API passes system prompt as kwargs['system']."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = self._base_kwargs(system="You are a helpful assistant.")
response_obj = self._responses_api_response_obj()
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
value = self._get_attr(mock_span, "gen_ai.system_instructions")
self.assertEqual(value, "You are a helpful assistant.")
def test_system_instructions_from_system_instructions_kwarg(self):
"""Vertex AI Gemini path uses kwargs['system_instructions'] (existing behavior)."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = self._base_kwargs(
system_instructions=[{"role": "system", "content": "Be concise."}]
)
response_obj = self._responses_api_response_obj()
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
raw = self._get_attr(mock_span, "gen_ai.system_instructions")
self.assertIsNotNone(raw)
parsed = json.loads(raw)
self.assertEqual(parsed[0]["role"], "system")
self.assertIn("parts", parsed[0])
def test_system_instructions_precedence(self):
"""system_instructions takes precedence over instructions and system."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = self._base_kwargs(
system_instructions="From Gemini",
instructions="From Responses API",
system="From Anthropic",
)
response_obj = self._responses_api_response_obj()
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
# system_instructions (string) should win — it's checked first
value = self._get_attr(mock_span, "gen_ai.system_instructions")
self.assertEqual(value, "From Gemini")
def test_no_system_instructions_when_absent(self):
"""No gen_ai.system_instructions attr when none of the kwargs are set."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = self._base_kwargs()
response_obj = self._responses_api_response_obj()
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
value = self._get_attr(mock_span, "gen_ai.system_instructions")
self.assertIsNone(value)
class TestTransformResponsesAPIOutput(unittest.TestCase):
"""
Unit tests for _transform_responses_api_output_to_otel.
"""
def test_message_with_output_text(self):
otel = OpenTelemetry()
output = [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello!"}],
}
]
result = otel._transform_responses_api_output_to_otel(output)
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["role"], "assistant")
self.assertEqual(result[0]["parts"], [{"type": "text", "content": "Hello!"}])
def test_function_call_item(self):
otel = OpenTelemetry()
output = [
{
"type": "function_call",
"name": "search",
"call_id": "call_1",
"arguments": '{"q": "test"}',
}
]
result = otel._transform_responses_api_output_to_otel(output)
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["role"], "assistant")
self.assertEqual(result[0]["parts"][0]["type"], "tool_call")
self.assertEqual(result[0]["parts"][0]["name"], "search")
self.assertEqual(result[0]["parts"][0]["id"], "call_1")
def test_function_call_without_call_id(self):
otel = OpenTelemetry()
output = [
{
"type": "function_call",
"name": "search",
"arguments": "{}",
}
]
result = otel._transform_responses_api_output_to_otel(output)
self.assertNotIn("id", result[0]["parts"][0])
def test_unknown_type_ignored(self):
otel = OpenTelemetry()
output = [{"type": "reasoning", "content": "thinking..."}]
result = otel._transform_responses_api_output_to_otel(output)
self.assertEqual(result, [])
def test_non_dict_items_ignored(self):
otel = OpenTelemetry()
output = ["not a dict", 42, None]
result = otel._transform_responses_api_output_to_otel(output)
self.assertEqual(result, [])
def test_empty_output(self):
otel = OpenTelemetry()
result = otel._transform_responses_api_output_to_otel([])
self.assertEqual(result, [])
def test_message_with_empty_text_skipped(self):
otel = OpenTelemetry()
output = [
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": ""}],
}
]
result = otel._transform_responses_api_output_to_otel(output)
self.assertEqual(result, [])
def test_message_default_role(self):
"""Messages without explicit role should default to assistant."""
otel = OpenTelemetry()
output = [
{
"type": "message",
"content": [{"type": "output_text", "text": "Hi"}],
}
]
result = otel._transform_responses_api_output_to_otel(output)
self.assertEqual(result[0]["role"], "assistant")
def test_pydantic_like_objects_accepted(self):
"""Items with .get() but not isinstance(dict) should be accepted."""
class FakeOutputItem:
"""Mimics BaseLiteLLMOpenAIResponseObject duck-typing."""
def __init__(self, data):
self._data = data
def get(self, key, default=None):
return self._data.get(key, default)
class FakeContent:
def __init__(self, data):
self._data = data
def get(self, key, default=None):
return self._data.get(key, default)
otel = OpenTelemetry()
output = [
FakeOutputItem(
{
"type": "message",
"role": "assistant",
"content": [
FakeContent({"type": "output_text", "text": "Pydantic works!"}),
],
}
)
]
result = otel._transform_responses_api_output_to_otel(output)
self.assertEqual(len(result), 1)
self.assertEqual(result[0]["parts"][0]["content"], "Pydantic works!")
class TestSystemInstructionsPrecedence(unittest.TestCase):
"""Tests for the is-not-None precedence in system_instructions coalescing."""
def _get_attr(self, mock_span, attr_name):
calls = [
call
for call in mock_span.set_attribute.call_args_list
if call[0][0] == attr_name
]
if not calls:
return None
return calls[0][0][1]
def _base_kwargs(self, **overrides):
kwargs = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "Hi"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "test-id",
"call_type": "responses",
"metadata": {},
},
}
kwargs.update(overrides)
return kwargs
def test_empty_list_system_instructions_does_not_fallthrough(self):
"""An empty list for system_instructions should NOT fall through to instructions."""
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = self._base_kwargs(
system_instructions=[],
instructions="Should not be used",
)
response_obj = {"id": "r1", "model": "gpt-4o"}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
# system_instructions is [] (falsy but not None), so it wins.
# Since it's an empty list, no attribute should be set (nothing to transform).
value = self._get_attr(mock_span, "gen_ai.system_instructions")
# The empty list is truthy for `is not None` but produces empty
# transformed output — the attribute should NOT contain "Should not be used".
if value is not None:
self.assertNotIn("Should not be used", str(value))
class TestResponsesAPIToolCallSpanAttributes(unittest.TestCase):
"""Tests for per-tool-call span attributes on Responses API function_call items."""
def _base_kwargs(self):
return {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "What is the weather?"}],
"optional_params": {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": {
"id": "resp_tc",
"call_type": "responses",
"metadata": {},
},
}
def test_per_tool_call_attributes_emitted(self):
"""function_call output items should produce per-tool-call span attributes."""
otel = OpenTelemetry()
mock_span = MagicMock()
response_obj = {
"id": "resp_tc",
"model": "gpt-4o",
"status": "completed",
"output": [
{
"type": "function_call",
"name": "get_weather",
"call_id": "call_abc",
"arguments": '{"location": "SF"}',
}
],
}
otel.set_attributes(
span=mock_span, kwargs=self._base_kwargs(), response_obj=response_obj
)
# Verify per-tool-call attributes were set (same format as choices branch)
attr_names = [call[0][0] for call in mock_span.set_attribute.call_args_list]
tool_call_attrs = [a for a in attr_names if "function_call" in a]
self.assertTrue(
len(tool_call_attrs) > 0, "Per-tool-call span attributes should be emitted"
)
# Verify the name attribute specifically
mock_span.set_attribute.assert_any_call(
"gen_ai.completion.0.function_call.name", "get_weather"
)
mock_span.set_attribute.assert_any_call(
"gen_ai.completion.0.function_call.arguments", '{"location": "SF"}'
)
def test_multiple_tool_calls_indexed(self):
"""Multiple function_call items should be indexed correctly."""
otel = OpenTelemetry()
mock_span = MagicMock()
response_obj = {
"id": "resp_tc2",
"model": "gpt-4o",
"status": "completed",
"output": [
{
"type": "function_call",
"name": "get_weather",
"call_id": "call_1",
"arguments": "{}",
},
{
"type": "function_call",
"name": "get_time",
"call_id": "call_2",
"arguments": "{}",
},
],
}
otel.set_attributes(
span=mock_span, kwargs=self._base_kwargs(), response_obj=response_obj
)
mock_span.set_attribute.assert_any_call(
"gen_ai.completion.0.function_call.name", "get_weather"
)
mock_span.set_attribute.assert_any_call(
"gen_ai.completion.1.function_call.name", "get_time"
)
class TestOpenTelemetryProxyParentSpanChildEmission(unittest.TestCase):
"""When metadata includes litellm_parent_otel_span (the proxy
span), the primary litellm_request span must still be created as a child
so the trace hierarchy is complete."""
def _build_kwargs(self, parent_span):
return {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {
"custom_llm_provider": "openai",
"metadata": {"litellm_parent_otel_span": parent_span},
},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
"hidden_params": {},
},
}
def test_get_span_context_returns_none_parent_for_metadata_span(self):
"""_get_span_context Priority 1 must return (ctx, None) — never the
parent span object — so callers always create litellm_request as a
child of ctx."""
tracer_provider = TracerProvider()
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
parent_span = otel.tracer.start_span("some_external_parent")
kwargs = self._build_kwargs(parent_span)
ctx, returned_parent = otel._get_span_context(kwargs)
self.assertIsNotNone(ctx, "ctx should carry the parent for child spans")
self.assertIsNone(
returned_parent,
"parent_span return slot must be None so callers create litellm_request",
)
parent_span.end()
def test_litellm_request_emitted_as_child_of_proxy_parent_span(self):
"""End-to-end: proxy span in metadata should yield exactly one
litellm_request span parented to it, with no extra root span."""
from litellm.integrations.opentelemetry import (
LITELLM_PROXY_REQUEST_SPAN_NAME,
LITELLM_REQUEST_SPAN_NAME,
)
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
proxy_span = otel.tracer.start_span(LITELLM_PROXY_REQUEST_SPAN_NAME)
kwargs = self._build_kwargs(proxy_span)
start = datetime.utcnow()
end = start + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj=None, start_time=start, end_time=end)
spans = span_exporter.get_finished_spans()
litellm_spans = [s for s in spans if s.name == LITELLM_REQUEST_SPAN_NAME]
proxy_spans = [s for s in spans if s.name == LITELLM_PROXY_REQUEST_SPAN_NAME]
self.assertEqual(
len(litellm_spans), 1, "Exactly one litellm_request span must be emitted"
)
self.assertEqual(
len(proxy_spans), 1, "Proxy span should be closed exactly once"
)
litellm_span = litellm_spans[0]
self.assertIsNotNone(
litellm_span.parent, "litellm_request must have a parent (not root)"
)
self.assertEqual(
litellm_span.parent.span_id,
proxy_spans[0].context.span_id,
"litellm_request must be a child of the proxy span",
)
def test_end_proxy_span_from_kwargs_closes_recording_proxy_span(self):
from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
proxy_span = otel.tracer.start_span(LITELLM_PROXY_REQUEST_SPAN_NAME)
self.assertTrue(proxy_span.is_recording())
kwargs = {
"litellm_params": {
"metadata": {"litellm_parent_otel_span": proxy_span},
}
}
otel._end_proxy_span_from_kwargs(kwargs, end_time=datetime.utcnow())
self.assertFalse(
proxy_span.is_recording(), "Proxy span should be closed by helper"
)
def test_end_proxy_span_from_kwargs_does_not_close_external_span(self):
"""Spans not named LITELLM_PROXY_REQUEST_SPAN_NAME must not be closed —
they may belong to external owners (Langfuse SDK, user code, etc.)."""
tracer_provider = TracerProvider()
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
external = otel.tracer.start_span("external_caller_span")
kwargs = {
"litellm_params": {
"metadata": {"litellm_parent_otel_span": external},
}
}
otel._end_proxy_span_from_kwargs(kwargs, end_time=datetime.utcnow())
self.assertTrue(
external.is_recording(),
"External (non-proxy) parent span must not be closed by LiteLLM",
)
external.end()
class TestOpenTelemetryProxyLoggerFirstRegisteredWins(unittest.TestCase):
"""open_telemetry_logger ownership must not be silently
overwritten by later handlers. First-registered wins."""
def _install_fake_proxy_server(self):
"""Install a stub ``litellm.proxy.proxy_server`` so the test does
not depend on optional proxy dependencies (websockets, etc.).
Returns (fake_module, cleanup_fn)."""
import importlib
import types
proxy_pkg_name = "litellm.proxy"
proxy_server_name = "litellm.proxy.proxy_server"
previous_pkg = sys.modules.get(proxy_pkg_name)
previous_mod = sys.modules.get(proxy_server_name)
# Ensure litellm.proxy package object exists
if previous_pkg is None:
try:
pkg = importlib.import_module(proxy_pkg_name)
except Exception:
pkg = types.ModuleType(proxy_pkg_name)
sys.modules[proxy_pkg_name] = pkg
else:
pkg = previous_pkg
fake = types.ModuleType(proxy_server_name)
fake.open_telemetry_logger = None
sys.modules[proxy_server_name] = fake
setattr(pkg, "proxy_server", fake)
def cleanup():
if previous_mod is not None:
sys.modules[proxy_server_name] = previous_mod
setattr(pkg, "proxy_server", previous_mod)
else:
sys.modules.pop(proxy_server_name, None)
if hasattr(pkg, "proxy_server"):
try:
delattr(pkg, "proxy_server")
except AttributeError:
pass
if previous_pkg is None and proxy_pkg_name in sys.modules:
if sys.modules[proxy_pkg_name] is pkg:
# Leave it in place — removing it would break later imports
pass
return fake, cleanup
def test_first_registered_handler_keeps_ownership(self):
fake_proxy_server, cleanup = self._install_fake_proxy_server()
try:
first = OpenTelemetry()
self.assertIs(
fake_proxy_server.open_telemetry_logger,
first,
"First registered handler must own the proxy logger slot",
)
second = OpenTelemetry()
self.assertIs(
fake_proxy_server.open_telemetry_logger,
first,
"Second handler must NOT overwrite the first-registered logger",
)
self.assertIsNot(
fake_proxy_server.open_telemetry_logger,
second,
"Proxy logger must remain pointed at the first handler",
)
finally:
cleanup()
def test_assignment_happens_when_slot_is_unset(self):
fake_proxy_server, cleanup = self._install_fake_proxy_server()
try:
handler = OpenTelemetry()
self.assertIs(fake_proxy_server.open_telemetry_logger, handler)
finally:
cleanup()
def test_existing_non_none_logger_is_preserved(self):
"""If ``proxy_server.open_telemetry_logger`` is already set to any
non-None value, a new handler must not overwrite it — even if the
existing value is not an OpenTelemetry instance."""
fake_proxy_server, cleanup = self._install_fake_proxy_server()
try:
sentinel = object()
fake_proxy_server.open_telemetry_logger = sentinel
OpenTelemetry()
self.assertIs(
fake_proxy_server.open_telemetry_logger,
sentinel,
"Existing non-None logger must not be overwritten",
)
finally:
cleanup()
class TestOpenTelemetrySpanDedupe(unittest.TestCase):
"""``_emit_once`` is a per-request, per-handler idempotency guard that
prevents duplicate span emission across two distinct dual-fire patterns:
1. Handler-level: streaming triggers both sync and async success/failure
callbacks for one request — the second call would otherwise produce a
duplicate ``litellm_request`` span.
2. Payload-driven entry-level: ``_create_guardrail_span`` is invoked
from three lifecycle points (post-call hook, success, failure) and
re-reads a mutating list — the same logical guardrail invocation
would otherwise be emitted up to three times.
"""
def _build_kwargs(self, *, exception: bool = False):
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": {},
"litellm_params": {
"custom_llm_provider": "openai",
"metadata": {},
},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
"hidden_params": {},
},
}
if exception:
kwargs["exception"] = Exception("test error")
return kwargs
def test_emit_once_first_call_returns_true_then_false(self):
otel = OpenTelemetry()
kwargs = self._build_kwargs()
self.assertTrue(otel._emit_once(kwargs, "success"))
self.assertFalse(
otel._emit_once(kwargs, "success"),
"Repeat call for same handler+scope+kwargs must be deduped",
)
def test_emit_once_distinct_scopes_dont_collide(self):
"""Different scopes on the same handler+kwargs must each emit once."""
otel = OpenTelemetry()
kwargs = self._build_kwargs()
self.assertTrue(otel._emit_once(kwargs, "success"))
self.assertTrue(
otel._emit_once(kwargs, "failure"),
"Failure scope must be independent of success scope",
)
self.assertTrue(
otel._emit_once(kwargs, "guardrail", "block-code", 1.0, "pre_call"),
"Guardrail entry scope must be independent of success/failure scopes",
)
self.assertFalse(otel._emit_once(kwargs, "success"))
self.assertFalse(otel._emit_once(kwargs, "failure"))
self.assertFalse(
otel._emit_once(kwargs, "guardrail", "block-code", 1.0, "pre_call")
)
def test_emit_once_separate_handlers_each_emit(self):
"""Two distinct handler instances must each emit exactly once for the
same scope."""
otel_a = OpenTelemetry()
otel_b = OpenTelemetry()
kwargs = self._build_kwargs()
self.assertTrue(otel_a._emit_once(kwargs, "success"))
self.assertTrue(
otel_b._emit_once(kwargs, "success"),
"Different handler instance must not share the first handler's marker",
)
self.assertFalse(otel_a._emit_once(kwargs, "success"))
self.assertFalse(otel_b._emit_once(kwargs, "success"))
def test_emit_once_handles_missing_metadata(self):
otel = OpenTelemetry()
kwargs = {"litellm_params": {}}
self.assertTrue(otel._emit_once(kwargs, "success"))
self.assertFalse(otel._emit_once(kwargs, "success"))
def test_emit_once_handles_missing_litellm_params(self):
otel = OpenTelemetry()
kwargs = {}
self.assertTrue(otel._emit_once(kwargs, "success"))
self.assertFalse(otel._emit_once(kwargs, "success"))
def test_emit_once_accepts_list_valued_scope_part(self):
"""Regression for LIT-3428 / LIT-3764: a list-valued ``guardrail_mode``
(the shape Presidio expands to with ``output_parse_pii: true``) must
not raise ``TypeError: unhashable type: 'list'`` when building the
dedupe key. Pre-fix, this call crashed inside ``dict.get``."""
otel = OpenTelemetry()
kwargs = self._build_kwargs()
self.assertTrue(
otel._emit_once(kwargs, "guardrail", "pii", 1.0, ["pre_call", "post_call"])
)
self.assertFalse(
otel._emit_once(kwargs, "guardrail", "pii", 1.0, ["pre_call", "post_call"]),
"Same list scope must dedupe to False on the second call",
)
def test_emit_once_distinct_list_scopes_dont_collide(self):
"""Two different list-valued scopes on the same handler/kwargs must
each emit exactly once. Catches a regression where every list collapses
to the same key (e.g. ``str(list)`` collisions on near-identical input)."""
otel = OpenTelemetry()
kwargs = self._build_kwargs()
self.assertTrue(otel._emit_once(kwargs, "guardrail", "pii", 1.0, ["pre_call"]))
self.assertTrue(
otel._emit_once(kwargs, "guardrail", "pii", 1.0, ["pre_call", "post_call"]),
"Distinct list scopes must produce distinct dedupe keys",
)
self.assertFalse(otel._emit_once(kwargs, "guardrail", "pii", 1.0, ["pre_call"]))
self.assertFalse(
otel._emit_once(kwargs, "guardrail", "pii", 1.0, ["pre_call", "post_call"])
)
def test_emit_once_accepts_dict_and_set_scope_parts(self):
"""``guardrail_mode`` can also arrive as a ``GuardrailMode`` TypedDict
(i.e. a plain dict at runtime). Sets are not produced today but flow
through the same normalization. Both must hash without raising."""
otel = OpenTelemetry()
kwargs = self._build_kwargs()
self.assertTrue(
otel._emit_once(kwargs, "guardrail", "pii", 1.0, {"tags": ["pre", "post"]})
)
self.assertFalse(
otel._emit_once(kwargs, "guardrail", "pii", 1.0, {"tags": ["pre", "post"]})
)
self.assertTrue(otel._emit_once(kwargs, "guardrail", "pii", 1.0, {"a", "b"}))
def test_emit_once_handles_self_referential_scope_without_recursion_error(self):
"""``_freeze_for_dedupe`` caps recursion at ``_FREEZE_MAX_DEPTH`` and
falls back to ``repr`` past the cap, so a self-referential container
in scope must not crash ``_emit_once``. ``guardrail_mode`` cannot
construct such input today, but the cap is the bound that justifies
recursion on the logging hot path."""
otel = OpenTelemetry()
kwargs = self._build_kwargs()
cyclic: list = []
cyclic.append(cyclic)
self.assertTrue(otel._emit_once(kwargs, "guardrail", "pii", 1.0, cyclic))
self.assertFalse(otel._emit_once(kwargs, "guardrail", "pii", 1.0, cyclic))
def test_create_guardrail_span_does_not_raise_on_list_mode(self):
"""End-to-end regression for LIT-3428: ``_create_guardrail_span``
must produce exactly one span (not raise ``TypeError``) when the
guardrail entry's ``guardrail_mode`` is a list."""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
kwargs = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "Hello"}],
"litellm_params": {"custom_llm_provider": "openai", "metadata": {}},
"standard_logging_object": {
"id": "test-id",
"call_type": "completion",
"metadata": {},
"hidden_params": {},
"guardrail_information": [
{
"guardrail_name": "presidio-pii",
"guardrail_mode": ["pre_call", "post_call"],
"guardrail_response": "ok",
"start_time": 1.0,
"end_time": 2.0,
}
],
},
}
otel._create_guardrail_span(kwargs=kwargs, context=None)
otel._create_guardrail_span(kwargs=kwargs, context=None)
guardrail_spans = [
s for s in span_exporter.get_finished_spans() if s.name == "guardrail"
]
self.assertEqual(
len(guardrail_spans),
1,
"List-valued guardrail_mode must emit exactly one guardrail span "
"across repeated lifecycle entrypoints",
)
def test_handle_success_emits_single_litellm_request_span_on_double_call(self):
"""Sync + async callback paths firing for the same kwargs must
result in exactly one litellm_request span."""
from litellm.integrations.opentelemetry import LITELLM_REQUEST_SPAN_NAME
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
kwargs = self._build_kwargs()
start = datetime.utcnow()
end = start + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj=None, start_time=start, end_time=end)
otel._handle_success(kwargs, response_obj=None, start_time=start, end_time=end)
spans = span_exporter.get_finished_spans()
litellm_spans = [s for s in spans if s.name == LITELLM_REQUEST_SPAN_NAME]
self.assertEqual(
len(litellm_spans),
1,
f"Exactly one litellm_request span expected, got {len(litellm_spans)}",
)
def test_handle_success_dedupe_skip_still_closes_proxy_span(self):
"""When the success path is short-circuited as a duplicate, the
proxy span must still be closed so traces don't leak."""
from litellm.integrations.opentelemetry import LITELLM_PROXY_REQUEST_SPAN_NAME
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
proxy_span = otel.tracer.start_span(LITELLM_PROXY_REQUEST_SPAN_NAME)
kwargs = self._build_kwargs()
kwargs["litellm_params"]["metadata"]["litellm_parent_otel_span"] = proxy_span
otel._emit_once(kwargs, "success") # pre-mark to force dedupe-skip branch
self.assertTrue(proxy_span.is_recording())
start = datetime.utcnow()
end = start + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj=None, start_time=start, end_time=end)
self.assertFalse(
proxy_span.is_recording(),
"Dedupe-skip path must still close the proxy span via _end_proxy_span_from_kwargs",
)
def test_handle_failure_emits_single_error_span_on_double_call(self):
"""Sync + async failure callback paths firing for the same kwargs
must result in exactly one ERROR litellm_request span."""
from opentelemetry.trace import StatusCode
from litellm.integrations.opentelemetry import LITELLM_REQUEST_SPAN_NAME
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
kwargs = self._build_kwargs(exception=True)
start = datetime.utcnow()
end = start + timedelta(seconds=1)
otel._handle_failure(kwargs, response_obj=None, start_time=start, end_time=end)
otel._handle_failure(kwargs, response_obj=None, start_time=start, end_time=end)
spans = span_exporter.get_finished_spans()
litellm_spans = [s for s in spans if s.name == LITELLM_REQUEST_SPAN_NAME]
self.assertEqual(
len(litellm_spans),
1,
f"Exactly one litellm_request ERROR span expected, got {len(litellm_spans)}",
)
self.assertEqual(litellm_spans[0].status.status_code, StatusCode.ERROR)
def test_create_guardrail_span_dedupes_across_lifecycle_entrypoints(self):
"""``_create_guardrail_span`` is called from post-call-success hook,
``_handle_success``, and ``_handle_failure``. A single guardrail
invocation (identified by ``(name, start_time, mode)``) must produce
exactly one span per handler even when the underlying entry is
mutated between calls (e.g. proxy enriches ``guardrail_response``)."""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
kwargs = self._build_kwargs()
guardrail_entry = {
"guardrail_name": "block-code",
"guardrail_mode": "pre_call",
"guardrail_response": "allow",
"start_time": 1.0,
"end_time": 2.0,
}
kwargs["standard_logging_object"]["guardrail_information"] = [guardrail_entry]
otel._create_guardrail_span(kwargs=kwargs, context=None)
# Mutate the entry between calls — proxy enriches the response.
guardrail_entry["guardrail_response"] = [
{"type": "code_block", "action_taken": "block"}
]
guardrail_entry["end_time"] = 3.0
otel._create_guardrail_span(kwargs=kwargs, context=None)
otel._create_guardrail_span(kwargs=kwargs, context=None)
guardrail_spans = [
s for s in span_exporter.get_finished_spans() if s.name == "guardrail"
]
self.assertEqual(
len(guardrail_spans),
1,
f"Exactly one guardrail span expected per logical invocation, got {len(guardrail_spans)}",
)
def test_create_guardrail_span_emits_distinct_entries(self):
"""Two real guardrail invocations (different ``start_time``) must
each emit a span — entry-level dedupe must not collapse them."""
span_exporter = InMemorySpanExporter()
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
otel = OpenTelemetry(tracer_provider=tracer_provider)
otel.tracer = tracer_provider.get_tracer(__name__)
kwargs = self._build_kwargs()
kwargs["standard_logging_object"]["guardrail_information"] = [
{
"guardrail_name": "block-code",
"guardrail_mode": "pre_call",
"guardrail_response": "allow",
"start_time": 1.0,
"end_time": 2.0,
},
{
"guardrail_name": "block-code",
"guardrail_mode": "post_call",
"guardrail_response": "allow",
"start_time": 5.0,
"end_time": 6.0,
},
]
otel._create_guardrail_span(kwargs=kwargs, context=None)
otel._create_guardrail_span(kwargs=kwargs, context=None)
guardrail_spans = [
s for s in span_exporter.get_finished_spans() if s.name == "guardrail"
]
self.assertEqual(
len(guardrail_spans),
2,
f"Two distinct guardrail invocations expected, got {len(guardrail_spans)}",
)
class TestOpenTelemetryHttpStatusCodeAttribute(unittest.TestCase):
"""PR 1: the failure recorder also exposes the HTTP status under the
OTel-standard ``http.response.status_code`` (as an int), while keeping the
legacy ``error.code`` for back-compat and leaving span status untouched.
"""
def _record(self, error_information):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer(__name__)
otel = OpenTelemetry()
span = tracer.start_span("Received Proxy Server Request")
kwargs = {
"exception": ValueError("boom"),
"standard_logging_object": {"error_information": error_information},
}
otel._record_exception_on_span(span=span, kwargs=kwargs)
span.end()
finished = exporter.get_finished_spans()
assert len(finished) == 1
return finished[0]
def test_401_sets_int_status_code_and_error_type(self):
span = self._record({"error_code": "401", "error_class": "AuthenticationError"})
assert span.attributes["http.response.status_code"] == 401
assert isinstance(span.attributes["http.response.status_code"], int)
assert span.attributes["error.type"] == "AuthenticationError"
def test_429_terminal(self):
span = self._record({"error_code": "429"})
assert span.attributes["http.response.status_code"] == 429
def test_500_sets_status_code_and_records_exception_event(self):
span = self._record({"error_code": "500"})
assert span.attributes["http.response.status_code"] == 500
assert any(e.name == "exception" for e in span.events)
def test_legacy_error_code_still_present_no_regression(self):
span = self._record({"error_code": "401"})
assert span.attributes["error.code"] == "401"
def test_non_numeric_error_code_omits_status_code(self):
span = self._record({"error_code": "ContextWindowExceeded"})
assert "http.response.status_code" not in span.attributes
# legacy attribute still set so existing dashboards don't regress
assert span.attributes["error.code"] == "ContextWindowExceeded"
def test_empty_error_code_omits_status_code(self):
span = self._record({"error_code": ""})
assert "http.response.status_code" not in span.attributes
def test_recorder_does_not_touch_span_status(self):
span = self._record({"error_code": "401"})
assert span.status.status_code == trace.StatusCode.UNSET
class TestOpenTelemetryFailureHookStampsServerSpan(unittest.TestCase):
"""Error attributes must land on the SERVER span dashboards query.
``_handle_failure`` records on the litellm_request child span, so
``async_post_call_failure_hook`` — which holds the SERVER span via
``user_api_key_dict.parent_otel_span`` — is where it gets stamped.
"""
def _run_hook(self, exception):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer(__name__)
otel = OpenTelemetry()
otel.tracer = tracer
server_span = tracer.start_span("Received Proxy Server Request")
user_api_key_dict = MagicMock()
user_api_key_dict.parent_otel_span = server_span
asyncio.run(
otel.async_post_call_failure_hook(
request_data={},
original_exception=exception,
user_api_key_dict=user_api_key_dict,
traceback_str="trace",
)
)
finished = {s.name: s for s in exporter.get_finished_spans()}
assert "Received Proxy Server Request" in finished
return finished["Received Proxy Server Request"]
def test_server_span_gets_int_status_code_and_error_type(self):
class _Boom(Exception):
status_code = 500
span = self._run_hook(_Boom("upstream blew up"))
assert span.attributes["http.response.status_code"] == 500
assert isinstance(span.attributes["http.response.status_code"], int)
assert span.attributes["error.type"] == "_Boom"
assert span.attributes["error.code"] == "500" # legacy, string
assert span.status.status_code == trace.StatusCode.ERROR
def test_non_numeric_code_omits_status_code_no_crash(self):
class _Boom(Exception):
code = "ContextWindowExceeded"
span = self._run_hook(_Boom("bad"))
assert "http.response.status_code" not in span.attributes
assert span.attributes["error.code"] == "ContextWindowExceeded"
def test_no_parent_span_is_noop(self):
otel = OpenTelemetry()
otel.tracer = MagicMock()
user_api_key_dict = MagicMock()
user_api_key_dict.parent_otel_span = None
# Must not raise when there is no SERVER span (e.g. pre-auth 401).
asyncio.run(
otel.async_post_call_failure_hook(
request_data={},
original_exception=ValueError("x"),
user_api_key_dict=user_api_key_dict,
traceback_str=None,
)
)
class TestOpenTelemetrySetProxyRequestRouteAttributes(unittest.TestCase):
"""http.route (template) + url.path (literal) must land on the SERVER
span. The logging handlers write the litellm_request child span, so
this is set from the auth path on the freshly-created SERVER span.
"""
def _set(self, **kwargs):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer(__name__)
otel = OpenTelemetry()
span = tracer.start_span("Received Proxy Server Request")
otel.set_proxy_request_route_attributes(span, **kwargs)
span.end()
return exporter.get_finished_spans()[0]
def test_sets_named_template_and_literal(self):
span = self._set(
url_path="/v1/threads/abc123/runs",
http_route="/v1/threads/{thread_id}/runs",
)
# Exact OTel-standard names — NOT metadata.* (naming regression guard).
assert span.attributes["url.path"] == "/v1/threads/abc123/runs"
assert span.attributes["http.route"] == "/v1/threads/{thread_id}/runs"
assert span.attributes["http.route"] != span.attributes["url.path"]
assert "metadata.http_route" not in span.attributes
def test_flat_route_template_equals_literal(self):
span = self._set(
url_path="/v1/chat/completions",
http_route="/v1/chat/completions",
)
assert span.attributes["http.route"] == "/v1/chat/completions"
assert span.attributes["url.path"] == "/v1/chat/completions"
def test_missing_http_route_omits_only_that_attribute(self):
span = self._set(url_path="/v1/chat/completions", http_route=None)
assert span.attributes["url.path"] == "/v1/chat/completions"
assert "http.route" not in span.attributes
def test_missing_both_sets_nothing(self):
span = self._set(url_path=None, http_route=None)
assert "url.path" not in span.attributes
assert "http.route" not in span.attributes
def test_none_span_is_noop(self):
otel = OpenTelemetry()
# Mirrors the Langfuse-override path (create span returns None).
otel.set_proxy_request_route_attributes(None, url_path="/x", http_route="/x")
class TestOpenTelemetrySetResponseStatusCodeAttribute(unittest.TestCase):
"""http.response.status_code must land on the SERVER span on the
success path too (failure path sets it in _record_exception_on_span).
Without this the attribute is failure-only, so error-ratio /
status-breakdown dashboards have no 2xx bucket.
"""
def _set(self, status_code):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer(__name__)
otel = OpenTelemetry()
span = tracer.start_span("Received Proxy Server Request")
otel.set_response_status_code_attribute(span, status_code)
span.end()
return exporter.get_finished_spans()[0]
def test_success_sets_int_200(self):
span = self._set(200)
# Exact OTel-standard name, stored as int (regression guard).
assert span.attributes["http.response.status_code"] == 200
assert isinstance(span.attributes["http.response.status_code"], int)
def test_none_status_code_omits_attribute(self):
span = self._set(None)
assert "http.response.status_code" not in span.attributes
def test_none_span_is_noop(self):
otel = OpenTelemetry()
# Mirrors the Langfuse-override path (create span returns None).
otel.set_response_status_code_attribute(None, 200)
class TestOpenTelemetryPreprocessingDuration(unittest.TestCase):
"""litellm.preprocessing.duration_ms (proxy-receive -> first provider
handoff) on the SERVER span. Read from container metadata so the
success (model_call_details) and failure (request_data) paths work
uniformly. Excludes retries via the set-once first_api_call_start_time.
"""
def _span(self):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer(__name__)
return tracer.start_span("Received Proxy Server Request"), exporter
def _attr(self, span, exporter):
span.end()
return exporter.get_finished_spans()[0].attributes
def test_success_shape_model_call_details(self):
# success path: first_api_call_start_time top-level,
# received-at under litellm_params.metadata
received = datetime(2026, 1, 1, 0, 0, 0)
handoff = datetime(2026, 1, 1, 0, 0, 0, 250000) # +250ms
otel = OpenTelemetry()
span, exp = self._span()
otel.set_preprocessing_duration_attribute(
span,
{
"first_api_call_start_time": handoff,
"litellm_params": {"metadata": {"litellm_received_at": received}},
},
)
attrs = self._attr(span, exp)
self.assertAlmostEqual(
attrs["litellm.preprocessing.duration_ms"], 250.0, places=1
)
def test_failure_shape_request_data(self):
# failure path: request_data with first_api_call_start_time lifted
# to the TOP LEVEL by the proxy (off the logging object, before it
# is popped) and received-at riding the metadata variable. The
# user metadata sub-dict is never used for the handoff anchor.
received = datetime(2026, 1, 1, 0, 0, 0)
handoff = datetime(2026, 1, 1, 0, 0, 0, 30000) # +30ms
otel = OpenTelemetry()
span, exp = self._span()
otel.set_preprocessing_duration_attribute(
span,
{
"first_api_call_start_time": handoff,
"metadata": {"litellm_received_at": received},
},
)
attrs = self._attr(span, exp)
self.assertAlmostEqual(
attrs["litellm.preprocessing.duration_ms"], 30.0, places=1
)
def test_missing_received_at_omits(self):
otel = OpenTelemetry()
span, exp = self._span()
otel.set_preprocessing_duration_attribute(
span, {"first_api_call_start_time": datetime(2026, 1, 1)}
)
assert "litellm.preprocessing.duration_ms" not in self._attr(span, exp)
def test_missing_handoff_omits(self):
otel = OpenTelemetry()
span, exp = self._span()
otel.set_preprocessing_duration_attribute(
span, {"metadata": {"litellm_received_at": datetime(2026, 1, 1)}}
)
assert "litellm.preprocessing.duration_ms" not in self._attr(span, exp)
def test_negative_duration_omitted(self):
# clock skew: handoff before receive -> omit, not a negative value
otel = OpenTelemetry()
span, exp = self._span()
otel.set_preprocessing_duration_attribute(
span,
{
"first_api_call_start_time": datetime(2026, 1, 1, 0, 0, 0),
"metadata": {"litellm_received_at": datetime(2026, 1, 1, 0, 0, 5)},
},
)
assert "litellm.preprocessing.duration_ms" not in self._attr(span, exp)
def test_none_span_is_noop(self):
OpenTelemetry().set_preprocessing_duration_attribute(
None, {"first_api_call_start_time": datetime(2026, 1, 1)}
)
def test_non_dict_container_is_noop(self):
otel = OpenTelemetry()
span, exp = self._span()
otel.set_preprocessing_duration_attribute(span, None)
assert "litellm.preprocessing.duration_ms" not in self._attr(span, exp)
class TestGetSpanContextLitellmMetadataFallback(unittest.TestCase):
"""
Tests for _get_span_context() falling back to litellm_metadata.
On /v1/messages (Anthropic Messages API) and other LITELLM_METADATA_ROUTES,
litellm_parent_otel_span is stored in litellm_params["litellm_metadata"]
instead of litellm_params["metadata"]. _get_span_context() must check
both locations.
Fixes: https://github.com/BerriAI/litellm/issues/27934
"""
def test_span_context_from_metadata(self):
"""Parent span is found when stored in litellm_params['metadata'] (OpenAI path)."""
otel = OpenTelemetry()
mock_span = MagicMock()
mock_span.get_span_context.return_value = MagicMock(is_valid=True)
kwargs = {
"litellm_params": {
"metadata": {"litellm_parent_otel_span": mock_span},
}
}
ctx, detected_span = otel._get_span_context(kwargs)
self.assertIsNotNone(ctx)
# Should NOT fall through to "no parent context" path
self.assertIsNone(detected_span)
def test_span_context_from_litellm_metadata_fallback(self):
"""Parent span is found when stored in litellm_params['litellm_metadata'] (Anthropic path)."""
otel = OpenTelemetry()
mock_span = MagicMock()
mock_span.get_span_context.return_value = MagicMock(is_valid=True)
kwargs = {
"litellm_params": {
"metadata": {
"user_id": "test-user"
}, # Anthropic native metadata, no span
"litellm_metadata": {"litellm_parent_otel_span": mock_span},
}
}
ctx, detected_span = otel._get_span_context(kwargs)
self.assertIsNotNone(ctx)
self.assertIsNone(detected_span)
def test_span_context_metadata_takes_priority(self):
"""When both metadata and litellm_metadata have the span, metadata wins."""
otel = OpenTelemetry()
span_from_metadata = MagicMock(name="span_from_metadata")
span_from_metadata.get_span_context.return_value = MagicMock(is_valid=True)
span_from_litellm_metadata = MagicMock(name="span_from_litellm_metadata")
span_from_litellm_metadata.get_span_context.return_value = MagicMock(
is_valid=True
)
kwargs = {
"litellm_params": {
"metadata": {"litellm_parent_otel_span": span_from_metadata},
"litellm_metadata": {
"litellm_parent_otel_span": span_from_litellm_metadata
},
}
}
ctx, detected_span = otel._get_span_context(kwargs)
self.assertIsNotNone(ctx)
self.assertIsNone(detected_span)
# metadata span is found first, so get_span_context on the
# litellm_metadata span should never be called — proving
# metadata takes priority over litellm_metadata.
span_from_litellm_metadata.get_span_context.assert_not_called()
def test_span_context_no_parent_when_neither_has_span(self):
"""When neither metadata nor litellm_metadata has a span, returns (None, None)."""
otel = OpenTelemetry()
kwargs = {
"litellm_params": {
"metadata": {"user_id": "test-user"},
"litellm_metadata": {"some_key": "some_value"},
}
}
ctx, detected_span = otel._get_span_context(kwargs)
# No parent span in either metadata dict and no active span in test
# context, so both should be None.
self.assertIsNone(ctx)
self.assertIsNone(detected_span)
class TestEndProxySpanLitellmMetadataFallback(unittest.TestCase):
"""
Tests for _end_proxy_span_from_kwargs() falling back to litellm_metadata.
Fixes: https://github.com/BerriAI/litellm/issues/27934
"""
def test_end_proxy_span_from_metadata(self):
"""Proxy span is found and ended from litellm_params['metadata']."""
otel = OpenTelemetry()
mock_span = MagicMock()
mock_span.name = "Received Proxy Server Request"
mock_span.is_recording.return_value = True
kwargs = {
"litellm_params": {
"metadata": {"litellm_parent_otel_span": mock_span},
}
}
otel._end_proxy_span_from_kwargs(kwargs, end_time=datetime.now())
mock_span.end.assert_called_once()
def test_end_proxy_span_from_litellm_metadata(self):
"""Proxy span is found and ended from litellm_params['litellm_metadata'] (fallback)."""
otel = OpenTelemetry()
mock_span = MagicMock()
mock_span.name = "Received Proxy Server Request"
mock_span.is_recording.return_value = True
kwargs = {
"litellm_params": {
"metadata": {"user_id": "test-user"}, # No span here
"litellm_metadata": {"litellm_parent_otel_span": mock_span},
}
}
otel._end_proxy_span_from_kwargs(kwargs, end_time=datetime.now())
mock_span.end.assert_called_once()
class TestOpenTelemetryInferenceIdentityAttributes(unittest.TestCase):
"""team_metadata, http.route, and both model names (the user-facing
model_group alias and the dispatched provider model) must land on the
inference span via set_attributes."""
def _span(self):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
tracer = provider.get_tracer(__name__)
return tracer.start_span("litellm_request"), exporter
def _attr(self, span, exporter):
span.end()
return exporter.get_finished_spans()[0].attributes
def _kwargs(self):
return {
"model": "gpt-4o",
"optional_params": {},
"litellm_params": {
"custom_llm_provider": "azure",
"metadata": {
"user_api_key_team_metadata": {
"tier": "gold",
"cost_center": "42",
}
},
},
"standard_logging_object": {
"metadata": {
"user_api_key_request_route": "/v1/chat/completions",
"user_api_key_team_id": "team-1",
},
"call_type": "completion",
"model_group": "gpt-4o",
"model": "azure/my-deployment",
"hidden_params": {"litellm_model_name": "azure/my-deployment"},
"id": "req-1",
"litellm_call_id": "call-1",
},
}
def _otel_with_team_metadata_keys(self, keys):
return OpenTelemetry(
config=OpenTelemetryConfig(baggage_team_metadata_keys=keys)
)
def test_all_identity_attributes_stamped(self):
otel = self._otel_with_team_metadata_keys(["tier", "cost_center"])
span, exp = self._span()
otel.set_attributes(span, self._kwargs(), {"model": "azure/gpt-4o"})
attrs = self._attr(span, exp)
assert attrs["http.route"] == "/v1/chat/completions"
assert json.loads(attrs["litellm.team.metadata"]) == {
"tier": "gold",
"cost_center": "42",
}
assert attrs["litellm.model_group"] == "gpt-4o"
assert attrs["litellm.provider.model"] == "azure/my-deployment"
def test_team_metadata_defaults_to_none_stamped(self):
"""With no allowlist configured (the default), a team's metadata must
never be stamped, even when present on the request."""
otel = OpenTelemetry()
span, exp = self._span()
otel.set_attributes(span, self._kwargs(), {"model": "azure/gpt-4o"})
assert "litellm.team.metadata" not in self._attr(span, exp)
def test_only_allowlisted_team_metadata_keys_stamped(self):
"""Sub-keys outside the allowlist are excluded from the stamped value."""
otel = self._otel_with_team_metadata_keys(["tier"])
span, exp = self._span()
otel.set_attributes(span, self._kwargs(), {"model": "azure/gpt-4o"})
assert json.loads(self._attr(span, exp)["litellm.team.metadata"]) == {
"tier": "gold"
}
def test_team_metadata_allowlist_from_config_yaml_kwarg(self):
"""callback_settings.otel.baggage_team_metadata_keys arrives as a kwarg
and must drive the allowlist."""
otel = OpenTelemetry(baggage_team_metadata_keys=["cost_center"])
span, exp = self._span()
otel.set_attributes(span, self._kwargs(), {"model": "azure/gpt-4o"})
assert json.loads(self._attr(span, exp)["litellm.team.metadata"]) == {
"cost_center": "42"
}
def test_provider_model_falls_back_to_payload_model(self):
"""Without hidden_params.litellm_model_name the dispatched model is
the payload model (the SDK path, where no router renaming happened)."""
otel = OpenTelemetry()
kwargs = self._kwargs()
kwargs["standard_logging_object"]["hidden_params"] = {}
span, exp = self._span()
otel.set_attributes(span, kwargs, {"model": "azure/gpt-4o"})
assert self._attr(span, exp)["litellm.provider.model"] == "azure/my-deployment"
def test_empty_team_metadata_is_dropped(self):
"""An empty team_metadata dict must not stamp a useless '{}'."""
otel = OpenTelemetry()
kwargs = self._kwargs()
kwargs["litellm_params"]["metadata"]["user_api_key_team_metadata"] = {}
span, exp = self._span()
otel.set_attributes(span, kwargs, {"model": "azure/gpt-4o"})
assert "litellm.team.metadata" not in self._attr(span, exp)
def test_missing_route_is_dropped(self):
"""An SDK request has no route; http.route must be absent, not empty."""
otel = OpenTelemetry()
kwargs = self._kwargs()
del kwargs["standard_logging_object"]["metadata"]["user_api_key_request_route"]
span, exp = self._span()
otel.set_attributes(span, kwargs, {"model": "azure/gpt-4o"})
assert "http.route" not in self._attr(span, exp)
def test_team_metadata_json_helper(self):
keys = ["a", "b"]
assert OpenTelemetry._team_metadata_json(None, keys) is None
assert OpenTelemetry._team_metadata_json("not-a-dict", keys) is None
assert OpenTelemetry._team_metadata_json({}, keys) is None
# empty allowlist -> nothing stamped, even with data present
assert OpenTelemetry._team_metadata_json({"a": 1}, []) is None
# no allowlisted key present -> dropped, not a useless "{}"
assert OpenTelemetry._team_metadata_json({"c": 1}, keys) is None
# only allowlisted sub-keys survive
assert json.loads(
OpenTelemetry._team_metadata_json({"a": 1, "c": 2}, keys)
) == {"a": 1}
class TestOpenTelemetryTeamMetadataKeysConfig(unittest.TestCase):
def test_normalize_from_csv_string(self):
# comma-separated env var: strip whitespace and drop empties
assert _normalize_team_metadata_keys("tier, cost_center , ,") == [
"tier",
"cost_center",
]
def test_normalize_from_list(self):
assert _normalize_team_metadata_keys(["tier", " cost_center ", ""]) == [
"tier",
"cost_center",
]
def test_normalize_none(self):
assert _normalize_team_metadata_keys(None) == []
def test_config_reads_csv_env_var(self):
with patch.dict(
"os.environ",
{"LITELLM_OTEL_BAGGAGE_TEAM_METADATA_KEYS": "tier, cost_center"},
):
assert OpenTelemetryConfig().baggage_team_metadata_keys == [
"tier",
"cost_center",
]
def test_explicit_keys_win_over_env_var(self):
with patch.dict(
"os.environ",
{"LITELLM_OTEL_BAGGAGE_TEAM_METADATA_KEYS": "from_env"},
):
cfg = OpenTelemetryConfig(baggage_team_metadata_keys=["from_arg"])
assert cfg.baggage_team_metadata_keys == ["from_arg"]
class TestOpenTelemetryMetricAttributeFiltering(unittest.TestCase):
"""LIT-3600: include/exclude control over which attributes are stamped on
emitted metrics, to cap metric cardinality. These drive the real
_handle_success -> _record_metrics path through an in-memory reader and
read attributes straight off the recorded data points, so they fail if the
filtering feature is reverted and pass only when it works end to end."""
HERE = os.path.dirname(__file__)
POLL_INTERVAL = 0.05
POLL_TIMEOUT = 2.0
DURATION_METRIC = "gen_ai.client.operation.duration"
TOKEN_METRIC = "gen_ai.client.token.usage"
# High-cardinality attributes the captured fixture emits by default. Each is
# a member of VALID_METRIC_ATTRIBUTE_NAMES and is present on the recorded
# metric when no filter is configured (verified by the backward-compat test).
HIGH_CARDINALITY_KEYS = (
"hidden_params",
"metadata.user_api_key_hash",
"metadata.requester_ip_address",
"metadata.requester_metadata",
"metadata.applied_guardrails",
)
RETAINED_LOW_CARDINALITY_KEY = "gen_ai.request.model"
def _load_fixtures(self):
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_kwargs.json")
) as f:
kwargs = json.load(f)
with open(
os.path.join(self.HERE, "open_telemetry", "data", "captured_response.json")
) as f:
response_obj = json.load(f)
return kwargs, response_obj
def _record(self, attributes):
"""Run a real success hook with metrics enabled and return the reader."""
metric_reader = InMemoryMetricReader()
meter_provider = MeterProvider(metric_readers=[metric_reader])
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(InMemorySpanExporter()))
otel = OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console", enable_metrics=True, attributes=attributes
),
tracer_provider=tracer_provider,
meter_provider=meter_provider,
)
otel.tracer = tracer_provider.get_tracer(__name__)
kwargs, response_obj = self._load_fixtures()
start = datetime.utcnow()
end = start + timedelta(seconds=1)
otel._handle_success(kwargs, response_obj, start, end)
return metric_reader
def _keysets(self, reader, metric_name):
"""Attribute-key sets, one per recorded data point of `metric_name`."""
deadline = time.time() + self.POLL_TIMEOUT
while time.time() < deadline:
data = reader.get_metrics_data()
if data and hasattr(data, "resource_metrics"):
for rm in data.resource_metrics:
for sm in rm.scope_metrics:
for m in sm.metrics:
if m.name == metric_name:
return [
set(dp.attributes.keys())
for dp in m.data.data_points
]
time.sleep(self.POLL_INTERVAL)
return None
def test_exclude_list_strips_high_cardinality_keys_across_metrics(self):
"""The bug: high-cardinality metadata/hidden_params explode metric
cardinality. With exclude_list set, none of them reach any data point,
while the retained low-cardinality model attribute survives. Asserted
on both the duration and token-usage histograms."""
reader = self._record(
OTELMetricAttributeFilter(exclude_list=list(self.HIGH_CARDINALITY_KEYS))
)
excluded = set(self.HIGH_CARDINALITY_KEYS)
for metric_name in (self.DURATION_METRIC, self.TOKEN_METRIC):
keysets = self._keysets(reader, metric_name)
self.assertTrue(keysets, f"{metric_name} was not recorded")
for keys in keysets:
self.assertTrue(
excluded.isdisjoint(keys),
f"{metric_name} leaked excluded keys: {excluded & keys}",
)
self.assertIn(self.RETAINED_LOW_CARDINALITY_KEY, keys)
def test_include_list_allows_only_listed_attributes(self):
"""An allowlist caps emitted attributes to exactly the listed set.
gen_ai.token.type is a structural discriminator added to the token
histogram after filtering, so it is the only key permitted beyond the
allowlist, and only on that metric."""
include = ["gen_ai.request.model", "gen_ai.system"]
reader = self._record(OTELMetricAttributeFilter(include_list=include))
allowed = set(include)
duration_keysets = self._keysets(reader, self.DURATION_METRIC)
self.assertTrue(duration_keysets, "duration metric was not recorded")
for keys in duration_keysets:
self.assertEqual(keys, allowed)
token_keysets = self._keysets(reader, self.TOKEN_METRIC)
self.assertTrue(token_keysets, "token-usage metric was not recorded")
for keys in token_keysets:
self.assertEqual(keys - {"gen_ai.token.type"}, allowed)
def test_no_filter_preserves_high_cardinality_keys(self):
"""Backward compatibility: with no attributes config, every
high-cardinality key the fixture carries is still stamped on the
metric, so existing customers who rely on them are unaffected."""
reader = self._record(None)
expected = set(self.HIGH_CARDINALITY_KEYS)
for metric_name in (self.DURATION_METRIC, self.TOKEN_METRIC):
keysets = self._keysets(reader, metric_name)
self.assertTrue(keysets, f"{metric_name} was not recorded")
for keys in keysets:
self.assertTrue(
expected.issubset(keys),
f"{metric_name} dropped {expected - keys} by default",
)
self.assertIn(self.RETAINED_LOW_CARDINALITY_KEY, keys)
def test_proxy_callback_settings_attributes_applied_without_kwarg(self):
"""Regression for the proxy path: the OpenTelemetry logger is constructed
before the proxy populates litellm.callback_settings['otel']['attributes'],
and without the attributes kwarg, so the filter must be resolved at record
time rather than at __init__. Otherwise metrics ship at full cardinality
(the bug the live proxy surfaced; constructing with the kwarg, or with
callback_settings already set, hid it)."""
previous = litellm.callback_settings
litellm.callback_settings = {} # not yet populated when the logger is built
try:
metric_reader = InMemoryMetricReader()
meter_provider = MeterProvider(metric_readers=[metric_reader])
tracer_provider = TracerProvider()
tracer_provider.add_span_processor(
SimpleSpanProcessor(InMemorySpanExporter())
)
otel = OpenTelemetry(
config=OpenTelemetryConfig(exporter="console", enable_metrics=True),
tracer_provider=tracer_provider,
meter_provider=meter_provider,
)
otel.tracer = tracer_provider.get_tracer(__name__)
# The proxy sets this only after the logger already exists.
litellm.callback_settings = {
"otel": {
"attributes": {"exclude_list": list(self.HIGH_CARDINALITY_KEYS)}
}
}
kwargs, response_obj = self._load_fixtures()
start = datetime.utcnow()
otel._handle_success(
kwargs, response_obj, start, start + timedelta(seconds=1)
)
finally:
litellm.callback_settings = previous
excluded = set(self.HIGH_CARDINALITY_KEYS)
for metric_name in (self.DURATION_METRIC, self.TOKEN_METRIC):
keysets = self._keysets(metric_reader, metric_name)
self.assertTrue(keysets, f"{metric_name} was not recorded")
for keys in keysets:
self.assertTrue(
excluded.isdisjoint(keys),
f"{metric_name} leaked {excluded & keys} via callback_settings",
)
self.assertIn(self.RETAINED_LOW_CARDINALITY_KEY, keys)
def test_callback_settings_validation_failure_is_not_sticky(self):
"""On the lazy callback_settings path a validation failure must not cache
the bad config. Once the operator corrects
callback_settings['otel']['attributes'], the next record resolves the
fixed filter instead of re-raising the stale error until a restart."""
previous = litellm.callback_settings
litellm.callback_settings = {
"otel": {
"attributes": {
"include_list": ["gen_ai.system"],
"exclude_list": ["hidden_params"],
}
}
}
try:
otel = OpenTelemetry(config=OpenTelemetryConfig(exporter="console"))
attrs = {"gen_ai.system": "openai", "hidden_params": "{}"}
with self.assertRaises(ValueError):
otel._filter_metric_attributes(attrs)
litellm.callback_settings = {
"otel": {"attributes": {"exclude_list": ["hidden_params"]}}
}
filtered = otel._filter_metric_attributes(attrs)
finally:
litellm.callback_settings = previous
self.assertEqual(filtered, {"gen_ai.system": "openai"})
def test_include_and_exclude_together_raise_value_error(self):
with self.assertRaises(ValueError):
OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console",
attributes=OTELMetricAttributeFilter(
include_list=["gen_ai.system"],
exclude_list=["hidden_params"],
),
)
)
def test_unknown_include_name_raises_value_error(self):
with self.assertRaises(ValueError):
OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console",
attributes=OTELMetricAttributeFilter(
include_list=["not.a.real.attribute"]
),
)
)
def test_unknown_exclude_name_raises_value_error(self):
with self.assertRaises(ValueError):
OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console",
attributes=OTELMetricAttributeFilter(
exclude_list=["metadata.does_not_exist"]
),
)
)
def test_dict_attributes_kwarg_path_validates(self):
"""The YAML/kwargs entry point (a plain dict) flows through
_build_metric_attribute_filter and hits the same validation."""
with self.assertRaises(ValueError):
OpenTelemetry(
attributes={
"include_list": ["gen_ai.system"],
"exclude_list": ["hidden_params"],
}
)
def test_no_filter_returns_attrs_object_unchanged(self):
"""The no-config path is a hot-path no-op: it returns the same dict
object, so default emission pays zero copy cost. Locking identity makes
a future refactor that always copies/filters trip here."""
otel = OpenTelemetry(config=OpenTelemetryConfig(exporter="console"))
attrs = {"gen_ai.request.model": "m", "hidden_params": "{}"}
self.assertIs(otel._filter_metric_attributes(attrs), attrs)
def test_token_type_discriminator_rejected_from_either_list(self):
"""gen_ai.token.type is a structural discriminator stamped onto the
input/output token series after filtering; it cannot be filtered without
collapsing the two series into one. Listing it in include_list or
exclude_list is rejected loudly at startup rather than silently ignored,
so an operator gets an error instead of a no-op."""
for attributes in (
OTELMetricAttributeFilter(exclude_list=["gen_ai.token.type"]),
OTELMetricAttributeFilter(include_list=["gen_ai.token.type"]),
):
with self.assertRaises(ValueError):
OpenTelemetry(
config=OpenTelemetryConfig(
exporter="console", attributes=attributes
)
)
class TestOTELServiceTierAttributes(unittest.TestCase):
"""The tier a request asked for and the tier the provider served must land on
the litellm_request span, so tier usage is segmentable in traces."""
REQUEST_KEY = "gen_ai.openai.request.service_tier"
RESPONSE_KEY = "gen_ai.openai.response.service_tier"
def _span_attributes(self, standard_logging_object, response_obj):
otel = OpenTelemetry()
mock_span = MagicMock()
kwargs = {
"model": "gpt-5-mini",
"messages": [{"role": "user", "content": "Hello"}],
"optional_params": standard_logging_object.get("model_parameters") or {},
"litellm_params": {"custom_llm_provider": "openai"},
"standard_logging_object": standard_logging_object,
}
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
return {call[0][0]: call[0][1] for call in mock_span.set_attribute.call_args_list}
def test_served_tier_from_response_and_requested_tier_are_stamped(self):
response_obj = {
"id": "chatcmpl-1",
"model": "gpt-5-mini",
"service_tier": "scale",
"usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
}
attributes = self._span_attributes(
{
"id": "test-id",
"call_type": "completion",
"metadata": {},
"model_parameters": {"service_tier": "auto"},
"response": response_obj,
},
response_obj,
)
self.assertEqual(attributes[self.RESPONSE_KEY], "scale")
self.assertEqual(attributes[self.REQUEST_KEY], "auto")
def test_served_tier_read_from_usage_object(self):
"""Anthropic reports the served tier on the usage object, not the top level."""
response_obj = {
"id": "chatcmpl-2",
"model": "claude-sonnet-4-5",
"usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
}
attributes = self._span_attributes(
{
"id": "test-id",
"call_type": "completion",
"metadata": {"usage_object": {"service_tier": "priority"}},
"model_parameters": {},
"response": response_obj,
},
response_obj,
)
self.assertEqual(attributes[self.RESPONSE_KEY], "priority")
self.assertNotIn(self.REQUEST_KEY, attributes)
def test_no_tier_anywhere_stamps_nothing(self):
response_obj = {
"id": "chatcmpl-3",
"model": "gpt-5-mini",
"usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
}
attributes = self._span_attributes(
{
"id": "test-id",
"call_type": "completion",
"metadata": {},
"model_parameters": {},
"response": response_obj,
},
response_obj,
)
self.assertNotIn(self.RESPONSE_KEY, attributes)
self.assertNotIn(self.REQUEST_KEY, attributes)
def test_unknown_requested_tier_is_not_stamped(self):
"""The requested tier is caller-controlled, so an unrecognized value is
dropped rather than written verbatim onto the span."""
response_obj = {
"id": "chatcmpl-4",
"model": "gpt-5-mini",
"usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
}
attributes = self._span_attributes(
{
"id": "test-id",
"call_type": "completion",
"metadata": {},
"model_parameters": {"service_tier": "Z" * 5000},
"response": response_obj,
},
response_obj,
)
self.assertNotIn(self.REQUEST_KEY, attributes)
def test_served_tier_is_stamped_even_when_unrecognized(self):
"""The served tier comes from the provider, not the caller, so a tier a
provider adds later is still stamped."""
response_obj = {
"id": "chatcmpl-5",
"model": "gpt-5-mini",
"service_tier": "tier-added-by-provider-later",
"usage": {"prompt_tokens": 1, "completion_tokens": 2, "total_tokens": 3},
}
attributes = self._span_attributes(
{
"id": "test-id",
"call_type": "completion",
"metadata": {},
"model_parameters": {},
"response": response_obj,
},
response_obj,
)
self.assertEqual(attributes[self.RESPONSE_KEY], "tier-added-by-provider-later")
class TestDynamicTracerProviderCache(unittest.TestCase):
"""Every credential-scoped TracerProvider that owns its exporter also owns a
BatchSpanProcessor worker thread that only stops on shutdown, so the cache holding them
must be bounded and must shut down whatever it drops (LIT-5437: threads accumulated
until pods were OOMKilled)."""
BSP_THREAD_NAME = "OtelBatchSpanProcessor"
def _logger(self, cap=3, exporter="console"):
logger = OpenTelemetry(
config=OpenTelemetryConfig(exporter=exporter, skip_set_global=True),
max_dynamic_tracer_providers=cap,
)
self.addCleanup(logger._tracer_provider.shutdown)
self.addCleanup(self._drain, logger)
return logger
def _drain(self, logger):
for entry in list(logger._tracer_provider_cache.values()):
entry.provider.shutdown()
logger._tracer_provider_cache.clear()
def _live_exporter_threads(self):
return [t for t in threading.enumerate() if t.name == self.BSP_THREAD_NAME]
def _wait_for_exporter_threads(self, expected, timeout=10.0):
"""Dropped providers are shut down off-thread, so poll instead of sleeping."""
deadline = time.time() + timeout
while time.time() < deadline:
count = len(self._live_exporter_threads())
if count <= expected:
return count
time.sleep(0.05)
return len(self._live_exporter_threads())
def test_distinct_credential_sets_stay_bounded(self):
"""One tenant per credential set must not mean one live thread per credential set."""
logger = self._logger(cap=3)
before = len(self._live_exporter_threads())
for i in range(25):
logger._get_tracer_with_dynamic_headers({"authorization": f"Basic tenant-{i}"})
self.assertEqual(len(logger._tracer_provider_cache), 3)
# Guards the thread-name constant: a rename upstream would make this read 0 and the
# bound assertion below would pass while measuring nothing.
self.assertGreaterEqual(len(self._live_exporter_threads()), 1)
self.assertLessEqual(self._wait_for_exporter_threads(before + 3) - before, 3)
def test_evicted_provider_is_shut_down(self):
"""An evicted provider is stopped, not silently dropped with its thread running."""
logger = self._logger(cap=3)
before = len(self._live_exporter_threads())
with patch.object(otel_module, "_shutdown_tracer_provider") as mock_shutdown:
logger._get_tracer_with_dynamic_headers({"authorization": "Basic evict-me"})
evicted = next(iter(logger._tracer_provider_cache.values()))
for i in range(3):
logger._get_tracer_with_dynamic_headers({"authorization": f"Basic keep-{i}"})
self.assertNotIn(evicted, logger._tracer_provider_cache.values())
self._wait_for_call(mock_shutdown)
mock_shutdown.assert_called_once_with(evicted.provider)
# The patch stopped the real shutdown, so stop the victim here; leaving its
# exporter thread alive would perturb the thread-census assertions elsewhere.
evicted.provider.shutdown()
still_cached = len(logger._tracer_provider_cache)
self.assertEqual(self._wait_for_exporter_threads(before + still_cached), before + still_cached)
def _wait_for_call(self, mock_fn, timeout=10.0):
"""The shutdown runs on a worker thread, so give it a moment to land."""
deadline = time.time() + timeout
while time.time() < deadline and not mock_fn.call_args_list:
time.sleep(0.05)
def test_concurrent_first_requests_build_one_provider(self):
"""Concurrent misses on one credential set race to build; only the winner may survive,
and the losers must be shut down rather than orphaned with their threads running."""
logger = self._logger(cap=3)
before = len(self._live_exporter_threads())
headers = {"authorization": "Basic same-tenant"}
barrier = threading.Barrier(16)
def _request_tracer(_):
barrier.wait()
return logger._get_tracer_with_dynamic_headers(headers)
with concurrent.futures.ThreadPoolExecutor(max_workers=16) as pool:
list(pool.map(_request_tracer, range(16)))
self.assertEqual(len(logger._tracer_provider_cache), 1)
self.assertEqual(self._wait_for_exporter_threads(before + 1) - before, 1)
def test_shared_exporter_instance_survives_dropped_providers(self):
"""A caller-supplied SpanExporter is shared with the logger's own provider, so a
dropped provider must not shut it down and silence the whole process."""
shared = InMemorySpanExporter()
logger = self._logger(cap=1, exporter=shared)
with logger.tracer.start_as_current_span("before"):
pass
for i in range(4):
logger._get_tracer_with_dynamic_headers({"authorization": f"Basic tenant-{i}"})
with logger.tracer.start_as_current_span("after"):
pass
self.assertEqual(
[span.name for span in shared.get_finished_spans()], ["before", "after"]
)
def test_mixed_ownership_cache_shuts_down_only_the_victims_that_own_their_exporter(self):
"""Both dynamic entry points share one cache, so it can hold providers of mixed
ownership. Whether an evicted provider may be shut down is a property of that
provider, not of the request that evicted it."""
shared = InMemorySpanExporter()
logger = self._logger(cap=1, exporter=shared)
with logger.tracer.start_as_current_span("before"):
pass
# Cached by the headers path, so its processor wraps the SHARED exporter.
logger._get_tracer_with_dynamic_headers({"authorization": "Basic shared-owner"})
# Evicted by the config path, which builds its OWN exporter from a named kind.
logger._get_tracer_with_dynamic_config(
OpenTelemetryConfig(exporter="console", skip_set_global=True)
)
with logger.tracer.start_as_current_span("after"):
pass
self.assertFalse(shared._stopped)
self.assertEqual(
[span.name for span in shared.get_finished_spans()], ["before", "after"]
)
def test_mixed_ownership_cache_still_reclaims_a_thread_owning_victim(self):
"""The other direction of the same defect: a victim that owns a real exporter
thread must still be shut down even when the evicting request does not."""
shared = InMemorySpanExporter()
logger = self._logger(cap=1, exporter=shared)
before = len(self._live_exporter_threads())
# Cached by the config path with a named kind, so it owns a BatchSpanProcessor thread.
logger._get_tracer_with_dynamic_config(
OpenTelemetryConfig(exporter="console", skip_set_global=True)
)
self.assertEqual(len(self._live_exporter_threads()) - before, 1)
# Evicted by the headers path, whose own exporter is the shared instance.
logger._get_tracer_with_dynamic_headers({"authorization": "Basic shared-owner"})
self.assertEqual(self._wait_for_exporter_threads(before) - before, 0)
def test_dropped_shared_exporter_provider_is_not_retained_by_an_exit_hook(self):
"""A provider we may never shut down must not register an interpreter-exit hook.
The hook holds a strong reference, so the provider would be pinned for the life of
the process (the very leak this fixes) and would stop the shared exporter at exit."""
shared = InMemorySpanExporter()
logger = self._logger(cap=1, exporter=shared)
logger._get_tracer_with_dynamic_headers({"authorization": "Basic a"})
entry = next(iter(logger._tracer_provider_cache.values()))
self.assertFalse(entry.owns_exporter)
victim = weakref.ref(entry.provider)
logger._get_tracer_with_dynamic_headers({"authorization": "Basic b"})
del entry
gc.collect()
self.assertIsNone(victim(), "evicted shared-exporter provider is still referenced")
def test_provider_that_owns_its_exporter_keeps_its_exit_flush(self):
"""The counterpart: a provider that owns a buffering processor must keep its exit
hook so its last batch still flushes when the process stops."""
logger = self._logger(cap=3)
logger._get_tracer_with_dynamic_headers({"authorization": "Basic owned"})
entry = next(iter(logger._tracer_provider_cache.values()))
self.assertTrue(entry.owns_exporter)
self.assertIsNotNone(entry.provider._atexit_handler)
def test_dynamic_providers_share_one_resource(self):
"""Building the Resource scans every installed distribution's entry points, and the
dynamic providers reach it from the async logging path, so one logger builds it once."""
logger = self._logger(cap=8)
for i in range(4):
logger._get_tracer_with_dynamic_headers({"authorization": f"Basic tenant-{i}"})
entries = list(logger._tracer_provider_cache.values())
self.assertEqual(len(entries), 4)
self.assertEqual(len({id(entry.provider.resource) for entry in entries}), 1)
self.assertIs(entries[0].provider.resource, logger._litellm_resource())
def test_resource_is_memoized_per_logger_not_shared(self):
"""Two loggers must not share a Resource; the second's service.name would be wrong."""
first = self._logger()
second = OpenTelemetry(
config=OpenTelemetryConfig(exporter="console", skip_set_global=True, service_name="svc-second")
)
self.addCleanup(second._tracer_provider.shutdown)
self.assertIsNot(first._litellm_resource(), second._litellm_resource())
self.assertEqual(second._litellm_resource().attributes.get("service.name"), "svc-second")
class TestOpenTelemetryDatabaseSemconvAttributes(unittest.TestCase):
"""A Postgres service span must name the PostgreSQL server it reached.
Without ``db.system`` and ``server.address``, the only host in the trace is
the loopback address of Prisma's local query engine, so the backend
attributes the wait to ``localhost`` and it cannot be correlated with the
database's own metrics.
"""
DSN = "postgresql://llmproxy:dbpassword9090@litellm-prod.abc123.us-east-1.rds.amazonaws.com:6432/litellm?schema=reporting"
REPLICA_DSN = "postgresql://reader:r3ad0nly@litellm-prod-ro.abc123.us-east-1.rds.amazonaws.com/litellm"
def _service_span(self, service, call_type, dsn, error=None, replica_dsn=None):
exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
otel = OpenTelemetry()
otel.tracer = provider.get_tracer(__name__)
parent = otel.tracer.start_span("Received Proxy Server Request")
payload = ServiceLoggerPayload(
is_error=error is not None,
error=error,
service=service,
duration=0.25,
call_type=call_type,
event_metadata=None,
)
hook = otel.async_service_failure_hook if error else otel.async_service_success_hook
kwargs = {"error": error} if error else {}
env = {k: v for k, v in (("DATABASE_URL", dsn), ("DATABASE_URL_READ_REPLICA", replica_dsn)) if v}
with patch.dict(os.environ, env, clear=False):
for absent in {"DATABASE_URL", "DATABASE_URL_READ_REPLICA"} - set(env):
os.environ.pop(absent, None)
asyncio.run(
hook(
payload=payload,
parent_otel_span=parent,
start_time=datetime.now(),
end_time=datetime.now(),
**kwargs,
)
)
parent.end()
return next(s for s in exporter.get_finished_spans() if s.name == service.value)
def test_postgres_span_names_the_database_server(self):
span = self._service_span(ServiceTypes.DB, "get_data", self.DSN)
self.assertEqual(span.attributes["db.system.name"], "postgresql")
self.assertEqual(span.attributes["db.operation.name"], "get_data")
self.assertEqual(
span.attributes["server.address"],
"litellm-prod.abc123.us-east-1.rds.amazonaws.com",
)
self.assertEqual(span.attributes["server.port"], 6432)
self.assertEqual(span.attributes["db.namespace"], "litellm|reporting")
def test_datastore_span_is_a_client_span_carrying_the_legacy_db_system(self):
"""Datadog types a span as a database call from CLIENT kind plus
``db.system``; an INTERNAL span is classified as custom work."""
span = self._service_span(ServiceTypes.DB, "get_data", self.DSN)
self.assertEqual(span.kind, trace.SpanKind.CLIENT)
self.assertEqual(span.attributes["db.system"], "postgresql")
def test_internal_service_span_stays_internal(self):
span = self._service_span(ServiceTypes.RESET_BUDGET_JOB, "reset_budget", self.DSN)
self.assertEqual(span.kind, trace.SpanKind.INTERNAL)
self.assertNotIn("db.system.name", span.attributes)
self.assertNotIn("server.address", span.attributes)
def test_existing_service_and_call_type_attributes_are_unchanged(self):
span = self._service_span(ServiceTypes.DB, "get_data", self.DSN)
self.assertEqual(span.attributes["service"], "postgres")
self.assertEqual(span.attributes["call_type"], "get_data")
def test_failed_postgres_span_also_names_the_database_server(self):
span = self._service_span(ServiceTypes.DB, "get_data", self.DSN, error="connection refused")
self.assertEqual(span.attributes["db.system.name"], "postgresql")
self.assertEqual(span.kind, trace.SpanKind.CLIENT)
self.assertEqual(
span.attributes["server.address"],
"litellm-prod.abc123.us-east-1.rds.amazonaws.com",
)
self.assertEqual(span.attributes["error"], "connection refused")
def test_no_credential_from_the_dsn_lands_on_the_span(self):
span = self._service_span(ServiceTypes.DB, "get_data", self.DSN)
exported = " ".join(str(value) for value in span.attributes.values())
self.assertIn("litellm-prod.abc123.us-east-1.rds.amazonaws.com", exported)
self.assertNotIn("dbpassword9090", exported)
self.assertNotIn("llmproxy", exported)
def test_redis_span_does_not_borrow_the_postgres_endpoint(self):
span = self._service_span(ServiceTypes.REDIS, "async_set_cache", self.DSN)
self.assertEqual(span.attributes["db.system.name"], "redis")
self.assertEqual(span.kind, trace.SpanKind.CLIENT)
self.assertNotIn("server.address", span.attributes)
def test_configured_read_replica_suppresses_the_endpoint(self):
span = self._service_span(ServiceTypes.DB, "get_data", self.DSN, replica_dsn=self.REPLICA_DSN)
self.assertEqual(span.attributes["db.system.name"], "postgresql")
self.assertNotIn("server.address", span.attributes)
self.assertNotIn("db.namespace", span.attributes)
def test_unset_database_url_leaves_the_span_without_endpoint_attributes(self):
span = self._service_span(ServiceTypes.DB, "get_data", None)
self.assertEqual(span.attributes["db.system.name"], "postgresql")
self.assertNotIn("server.address", span.attributes)
class TestOpenTelemetryNonInferenceUsage(unittest.TestCase):
"""Reading a stored response replays the usage of the call that created it, so emitting those
token counts again on the read's span reports the same tokens a second time. Regression tests
for LIT-5602, covering the legacy emitter that runs by default."""
USAGE = {"prompt_tokens": 4000, "completion_tokens": 2000, "total_tokens": 6000}
TOKEN_KEYS = frozenset({"gen_ai.usage.input_tokens", "gen_ai.usage.output_tokens", "gen_ai.usage.total_tokens"})
BACKGROUND_POLL = {"internal_call_origin": "background_response_cost_poll"}
RESPONSE_OBJ = {"id": "resp_lit5602", "model": "gpt-4o", "usage": USAGE}
BACKGROUND_RESPONSE_OBJ = {**RESPONSE_OBJ, "background": True}
def _kwargs(self, call_type, litellm_metadata=None):
return {
"model": "gpt-4o",
"call_type": call_type,
"optional_params": {},
"litellm_params": {
"custom_llm_provider": "openai",
"litellm_metadata": litellm_metadata or {},
},
"standard_logging_object": {"id": "lit5602", "call_type": call_type, "metadata": {}},
}
def _token_attributes_on_span(self, call_type, litellm_metadata=None, response_obj=None):
otel = OpenTelemetry()
mock_span = MagicMock()
otel.set_attributes(
span=mock_span,
kwargs=self._kwargs(call_type, litellm_metadata),
response_obj=response_obj or dict(self.RESPONSE_OBJ),
)
return {call[0][0] for call in mock_span.set_attribute.call_args_list if call[0][0] in self.TOKEN_KEYS}
def _token_histogram_calls(self, call_type, litellm_metadata=None, response_obj=None):
otel = OpenTelemetry()
otel._operation_duration_histogram = MagicMock()
otel._token_usage_histogram = MagicMock()
otel._cost_histogram = None
now = datetime.now()
otel._record_metrics(
self._kwargs(call_type, litellm_metadata), response_obj or dict(self.RESPONSE_OBJ), now, now
)
return otel._token_usage_histogram.record.call_count
def _time_per_output_token_calls(self, call_type, litellm_metadata=None, response_obj=None):
otel = OpenTelemetry()
otel._time_per_output_token_histogram = MagicMock()
now = datetime.now()
otel._record_time_per_output_token_metric(
self._kwargs(call_type, litellm_metadata), response_obj or dict(self.RESPONSE_OBJ), now, 1.0, {}
)
return otel._time_per_output_token_histogram.record.call_count
def test_inference_call_still_reports_its_tokens_on_the_span(self):
self.assertEqual(self._token_attributes_on_span("acompletion"), set(self.TOKEN_KEYS))
def test_response_read_does_not_report_the_retrieved_tokens_on_the_span(self):
self.assertEqual(self._token_attributes_on_span("aget_responses"), set())
def test_background_cost_poll_read_still_reports_its_tokens_on_the_span(self):
self.assertEqual(self._token_attributes_on_span("aget_responses", self.BACKGROUND_POLL), set(self.TOKEN_KEYS))
def test_inference_call_still_records_the_token_usage_histogram(self):
self.assertEqual(self._token_histogram_calls("acompletion"), 2)
def test_response_read_does_not_record_the_token_usage_histogram(self):
self.assertEqual(self._token_histogram_calls("aget_responses"), 0)
def test_background_cost_poll_read_still_records_the_token_usage_histogram(self):
self.assertEqual(self._token_histogram_calls("aget_responses", self.BACKGROUND_POLL), 2)
def test_background_response_read_still_reports_its_tokens_on_the_span(self):
self.assertEqual(
self._token_attributes_on_span("aget_responses", response_obj=self.BACKGROUND_RESPONSE_OBJ),
set(self.TOKEN_KEYS),
)
def test_background_response_read_still_records_the_token_usage_histogram(self):
self.assertEqual(self._token_histogram_calls("aget_responses", response_obj=self.BACKGROUND_RESPONSE_OBJ), 2)
def test_inference_call_still_records_time_per_output_token(self):
self.assertEqual(self._time_per_output_token_calls("acompletion"), 1)
def test_response_read_does_not_divide_its_latency_by_the_retrieved_token_count(self):
self.assertEqual(self._time_per_output_token_calls("aget_responses"), 0)
def test_background_response_read_still_records_time_per_output_token(self):
self.assertEqual(
self._time_per_output_token_calls("aget_responses", response_obj=self.BACKGROUND_RESPONSE_OBJ), 1
)
Reference in a new issue View git blame Copy permalink