mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-11 03:38:38 +00:00
Merge pull request #21309 from BerriAI/litellm_fix_langfuse_otel_trace_v2
Litellm fix langfuse otel trace v2
This commit is contained in:
Harshit Jain
GitHub
commit
6134984008
3 changed files with 5954 additions and 5727 deletions
|
|
@ -40,9 +40,7 @@ if TYPE_CHECKING:
|
|||
Context = Union[_Context, Any]
|
||||
SpanExporter = Union[_SpanExporter, Any]
|
||||
UserAPIKeyAuth = Union[_UserAPIKeyAuth, Any]
|
||||
ManagementEndpointLoggingPayload = Union[
|
||||
_ManagementEndpointLoggingPayload, Any
|
||||
]
|
||||
ManagementEndpointLoggingPayload = Union[_ManagementEndpointLoggingPayload, Any]
|
||||
else:
|
||||
Span = Any
|
||||
Tracer = Any
|
||||
|
|
@ -76,7 +74,11 @@ class OpenTelemetryConfig:
|
|||
# automatically infer "otlp_http" to send traces to the endpoint.
|
||||
# This fixes an issue where UI-configured OTEL settings would default
|
||||
# to console output instead of sending traces to the configured endpoint.
|
||||
if self.endpoint and isinstance(self.exporter, str) and self.exporter == "console":
|
||||
if (
|
||||
self.endpoint
|
||||
and isinstance(self.exporter, str)
|
||||
and self.exporter == "console"
|
||||
):
|
||||
self.exporter = "otlp_http"
|
||||
|
||||
if not self.service_name:
|
||||
|
|
@ -104,16 +106,12 @@ class OpenTelemetryConfig:
|
|||
exporter = os.getenv(
|
||||
"OTEL_EXPORTER_OTLP_PROTOCOL", os.getenv("OTEL_EXPORTER", "console")
|
||||
)
|
||||
endpoint = os.getenv(
|
||||
"OTEL_EXPORTER_OTLP_ENDPOINT", os.getenv("OTEL_ENDPOINT")
|
||||
)
|
||||
endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", os.getenv("OTEL_ENDPOINT"))
|
||||
headers = os.getenv(
|
||||
"OTEL_EXPORTER_OTLP_HEADERS", os.getenv("OTEL_HEADERS")
|
||||
) # example: OTEL_HEADERS=x-honeycomb-team=B85YgLm96***"
|
||||
enable_metrics: bool = (
|
||||
os.getenv(
|
||||
"LITELLM_OTEL_INTEGRATION_ENABLE_METRICS", "false"
|
||||
).lower()
|
||||
os.getenv("LITELLM_OTEL_INTEGRATION_ENABLE_METRICS", "false").lower()
|
||||
== "true"
|
||||
)
|
||||
enable_events: bool = (
|
||||
|
|
@ -121,9 +119,7 @@ class OpenTelemetryConfig:
|
|||
== "true"
|
||||
)
|
||||
service_name = os.getenv("OTEL_SERVICE_NAME", "litellm")
|
||||
deployment_environment = os.getenv(
|
||||
"OTEL_ENVIRONMENT_NAME", "production"
|
||||
)
|
||||
deployment_environment = os.getenv("OTEL_ENVIRONMENT_NAME", "production")
|
||||
model_id = os.getenv("OTEL_MODEL_ID", service_name)
|
||||
|
||||
if exporter == "in_memory":
|
||||
|
|
@ -172,9 +168,7 @@ class OpenTelemetry(CustomLogger):
|
|||
logging.getLogger(__name__)
|
||||
|
||||
# Enable OpenTelemetry logging
|
||||
otel_exporter_logger = logging.getLogger(
|
||||
"opentelemetry.sdk.trace.export"
|
||||
)
|
||||
otel_exporter_logger = logging.getLogger("opentelemetry.sdk.trace.export")
|
||||
otel_exporter_logger.setLevel(logging.DEBUG)
|
||||
|
||||
# init CustomLogger params
|
||||
|
|
@ -229,6 +223,7 @@ class OpenTelemetry(CustomLogger):
|
|||
sdk_provider_class,
|
||||
create_new_provider_fn,
|
||||
set_provider_fn,
|
||||
skip_set_global: bool = False,
|
||||
):
|
||||
"""
|
||||
Generic helper to get or create an OpenTelemetry provider (Tracer, Meter, or Logger).
|
||||
|
|
@ -240,6 +235,7 @@ class OpenTelemetry(CustomLogger):
|
|||
sdk_provider_class: The SDK provider class to check for (e.g., TracerProvider from SDK)
|
||||
create_new_provider_fn: Function to create a new provider instance
|
||||
set_provider_fn: Function to set the provider globally
|
||||
skip_set_global: If True, don't set the provider globally (for dynamic-only providers)
|
||||
|
||||
Returns:
|
||||
The provider to use (either existing, new, or explicitly provided)
|
||||
|
|
@ -270,11 +266,15 @@ class OpenTelemetry(CustomLogger):
|
|||
# Don't call set_provider to preserve existing context
|
||||
else:
|
||||
# Default proxy provider or unknown type, create our own
|
||||
verbose_logger.debug(
|
||||
"OpenTelemetry: Creating new %s", provider_name
|
||||
)
|
||||
verbose_logger.debug("OpenTelemetry: Creating new %s", provider_name)
|
||||
provider = create_new_provider_fn()
|
||||
set_provider_fn(provider)
|
||||
if not skip_set_global:
|
||||
set_provider_fn(provider)
|
||||
else:
|
||||
verbose_logger.info(
|
||||
"OpenTelemetry: Created %s but NOT setting it globally (will use dynamic providers per-request)",
|
||||
provider_name,
|
||||
)
|
||||
except Exception as e:
|
||||
# Fallback: create a new provider if something goes wrong
|
||||
verbose_logger.debug(
|
||||
|
|
@ -283,7 +283,8 @@ class OpenTelemetry(CustomLogger):
|
|||
str(e),
|
||||
)
|
||||
provider = create_new_provider_fn()
|
||||
set_provider_fn(provider)
|
||||
if not skip_set_global:
|
||||
set_provider_fn(provider)
|
||||
|
||||
return provider
|
||||
|
||||
|
|
@ -293,12 +294,15 @@ class OpenTelemetry(CustomLogger):
|
|||
from opentelemetry.trace import SpanKind
|
||||
|
||||
def create_tracer_provider():
|
||||
provider = TracerProvider(
|
||||
resource=self._get_litellm_resource(self.config)
|
||||
)
|
||||
provider = TracerProvider(resource=self._get_litellm_resource(self.config))
|
||||
provider.add_span_processor(self._get_span_processor())
|
||||
return provider
|
||||
|
||||
# CRITICAL FIX: For Langfuse OTEL, skip setting global provider to prevent interference
|
||||
skip_global = (
|
||||
hasattr(self, "callback_name") and self.callback_name == "langfuse_otel"
|
||||
)
|
||||
|
||||
tracer_provider = self._get_or_create_provider(
|
||||
provider=tracer_provider,
|
||||
provider_name="TracerProvider",
|
||||
|
|
@ -306,6 +310,7 @@ class OpenTelemetry(CustomLogger):
|
|||
sdk_provider_class=TracerProvider,
|
||||
create_new_provider_fn=create_tracer_provider,
|
||||
set_provider_fn=trace.set_tracer_provider,
|
||||
skip_set_global=skip_global,
|
||||
)
|
||||
|
||||
# Grab our tracer from the TracerProvider (not from global context)
|
||||
|
|
@ -409,14 +414,10 @@ class OpenTelemetry(CustomLogger):
|
|||
def log_failure_event(self, kwargs, response_obj, start_time, end_time):
|
||||
self._handle_failure(kwargs, response_obj, start_time, end_time)
|
||||
|
||||
async def async_log_success_event(
|
||||
self, kwargs, response_obj, start_time, end_time
|
||||
):
|
||||
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
|
||||
self._handle_success(kwargs, response_obj, start_time, end_time)
|
||||
|
||||
async def async_log_failure_event(
|
||||
self, kwargs, response_obj, start_time, end_time
|
||||
):
|
||||
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
|
||||
self._handle_failure(kwargs, response_obj, start_time, end_time)
|
||||
|
||||
async def async_service_success_hook(
|
||||
|
|
@ -613,14 +614,37 @@ class OpenTelemetry(CustomLogger):
|
|||
|
||||
if dynamic_headers is not None:
|
||||
# Create spans using a temporary tracer with dynamic headers
|
||||
tracer_to_use = self._get_tracer_with_dynamic_headers(
|
||||
dynamic_headers
|
||||
)
|
||||
tracer_to_use = self._get_tracer_with_dynamic_headers(dynamic_headers)
|
||||
verbose_logger.debug(
|
||||
"Using dynamic headers for this request: %s", dynamic_headers
|
||||
"[OTEL DEBUG] Using DYNAMIC tracer with headers: %s", dynamic_headers
|
||||
)
|
||||
else:
|
||||
tracer_to_use = self.tracer
|
||||
# For langfuse_otel without dynamic headers, create a provider with env var credentials
|
||||
if hasattr(self, "callback_name") and self.callback_name == "langfuse_otel":
|
||||
# Use the headers from config (which were set from env vars during init)
|
||||
env_var_headers = (
|
||||
self._get_headers_dictionary(self.OTEL_HEADERS)
|
||||
if self.OTEL_HEADERS
|
||||
else {}
|
||||
)
|
||||
if env_var_headers:
|
||||
tracer_to_use = self._get_tracer_with_dynamic_headers(
|
||||
env_var_headers
|
||||
)
|
||||
verbose_logger.debug(
|
||||
"[OTEL DEBUG] Using env var credentials for langfuse_otel (master key request)"
|
||||
)
|
||||
else:
|
||||
# No env vars set, use global tracer (will be NoOp)
|
||||
tracer_to_use = self.tracer
|
||||
verbose_logger.debug(
|
||||
"[OTEL DEBUG] No credentials available for langfuse_otel"
|
||||
)
|
||||
else:
|
||||
tracer_to_use = self.tracer
|
||||
verbose_logger.debug(
|
||||
"[OTEL DEBUG] Using GLOBAL tracer (no dynamic headers)"
|
||||
)
|
||||
|
||||
return tracer_to_use
|
||||
|
||||
|
|
@ -651,9 +675,7 @@ class OpenTelemetry(CustomLogger):
|
|||
)
|
||||
|
||||
# Create a temporary tracer provider with dynamic headers
|
||||
temp_provider = TracerProvider(
|
||||
resource=self._get_litellm_resource(self.config)
|
||||
)
|
||||
temp_provider = TracerProvider(resource=self._get_litellm_resource(self.config))
|
||||
temp_provider.add_span_processor(
|
||||
self._get_span_processor(dynamic_headers=dynamic_headers)
|
||||
)
|
||||
|
|
@ -688,6 +710,15 @@ class OpenTelemetry(CustomLogger):
|
|||
)
|
||||
ctx, parent_span = self._get_span_context(kwargs)
|
||||
|
||||
# CRITICAL FIX: For langfuse_otel, ALWAYS create primary spans
|
||||
# Don't use parent spans from other providers as they cause trace corruption
|
||||
is_langfuse_otel = (
|
||||
hasattr(self, "callback_name") and self.callback_name == "langfuse_otel"
|
||||
)
|
||||
if is_langfuse_otel:
|
||||
parent_span = None # Ignore parent spans from other providers
|
||||
ctx = None
|
||||
|
||||
# Decide whether to create a primary span
|
||||
# Always create if no parent span exists (backward compatibility)
|
||||
# OR if USE_OTEL_LITELLM_REQUEST_SPAN is explicitly enabled
|
||||
|
|
@ -707,6 +738,7 @@ class OpenTelemetry(CustomLogger):
|
|||
# Ensure proxy-request parent span is annotated with the actual operation kind
|
||||
if (
|
||||
parent_span is not None
|
||||
and hasattr(parent_span, "name")
|
||||
and parent_span.name == LITELLM_PROXY_REQUEST_SPAN_NAME
|
||||
):
|
||||
self.set_attributes(parent_span, kwargs, response_obj)
|
||||
|
|
@ -717,8 +749,9 @@ class OpenTelemetry(CustomLogger):
|
|||
span = None
|
||||
# Only set attributes if the span is still recording (not closed)
|
||||
# Note: parent_span is guaranteed to be not None here
|
||||
parent_span.set_status(Status(StatusCode.OK))
|
||||
self.set_attributes(parent_span, kwargs, response_obj)
|
||||
if hasattr(parent_span, "set_status"):
|
||||
parent_span.set_status(Status(StatusCode.OK))
|
||||
self.set_attributes(parent_span, kwargs, response_obj)
|
||||
# Raw-request as direct child of parent_span
|
||||
self._maybe_log_raw_request(
|
||||
kwargs, response_obj, start_time, end_time, parent_span
|
||||
|
|
@ -741,6 +774,7 @@ class OpenTelemetry(CustomLogger):
|
|||
# However, proxy-created spans should be closed here
|
||||
if (
|
||||
parent_span is not None
|
||||
and hasattr(parent_span, "name")
|
||||
and parent_span.name == LITELLM_PROXY_REQUEST_SPAN_NAME
|
||||
):
|
||||
parent_span.end(end_time=self._to_ns(end_time))
|
||||
|
|
@ -784,9 +818,7 @@ class OpenTelemetry(CustomLogger):
|
|||
metadata = litellm_params.get("metadata") or {}
|
||||
generation_name = metadata.get("generation_name")
|
||||
|
||||
raw_span_name = (
|
||||
generation_name if generation_name else RAW_REQUEST_SPAN_NAME
|
||||
)
|
||||
raw_span_name = generation_name if generation_name else RAW_REQUEST_SPAN_NAME
|
||||
|
||||
otel_tracer: Tracer = self.get_tracer_to_use_for_request(kwargs)
|
||||
raw_span = otel_tracer.start_span(
|
||||
|
|
@ -811,9 +843,7 @@ class OpenTelemetry(CustomLogger):
|
|||
}
|
||||
|
||||
std_log = kwargs.get("standard_logging_object")
|
||||
md = getattr(std_log, "metadata", None) or (std_log or {}).get(
|
||||
"metadata", {}
|
||||
)
|
||||
md = getattr(std_log, "metadata", None) or (std_log or {}).get("metadata", {})
|
||||
for key in [
|
||||
"user_api_key_hash",
|
||||
"user_api_key_alias",
|
||||
|
|
@ -835,9 +865,9 @@ class OpenTelemetry(CustomLogger):
|
|||
common_attrs[f"metadata.{key}"] = str(md[key])
|
||||
|
||||
# get hidden params
|
||||
hidden_params = getattr(std_log, "hidden_params", None) or (
|
||||
std_log or {}
|
||||
).get("hidden_params", {})
|
||||
hidden_params = getattr(std_log, "hidden_params", None) or (std_log or {}).get(
|
||||
"hidden_params", {}
|
||||
)
|
||||
if hidden_params:
|
||||
common_attrs["hidden_params"] = safe_dumps(hidden_params)
|
||||
|
||||
|
|
@ -890,9 +920,7 @@ class OpenTelemetry(CustomLogger):
|
|||
except ValueError:
|
||||
return None
|
||||
|
||||
def _record_time_to_first_token_metric(
|
||||
self, kwargs: dict, common_attrs: dict
|
||||
):
|
||||
def _record_time_to_first_token_metric(self, kwargs: dict, common_attrs: dict):
|
||||
"""Record Time to First Token (TTFT) metric for streaming requests."""
|
||||
optional_params = kwargs.get("optional_params", {})
|
||||
is_streaming = optional_params.get("stream", False)
|
||||
|
|
@ -905,10 +933,7 @@ class OpenTelemetry(CustomLogger):
|
|||
api_call_start_time = kwargs.get("api_call_start_time", None)
|
||||
completion_start_time = kwargs.get("completion_start_time", None)
|
||||
|
||||
if (
|
||||
api_call_start_time is not None
|
||||
and completion_start_time is not None
|
||||
):
|
||||
if api_call_start_time is not None and completion_start_time is not None:
|
||||
# Convert to timestamps if needed (handles datetime, float, and string)
|
||||
api_call_start_ts = self._to_timestamp(api_call_start_time)
|
||||
completion_start_ts = self._to_timestamp(completion_start_time)
|
||||
|
|
@ -916,9 +941,7 @@ class OpenTelemetry(CustomLogger):
|
|||
if api_call_start_ts is None or completion_start_ts is None:
|
||||
return # Skip recording if conversion failed
|
||||
|
||||
time_to_first_token_seconds = (
|
||||
completion_start_ts - api_call_start_ts
|
||||
)
|
||||
time_to_first_token_seconds = completion_start_ts - api_call_start_ts
|
||||
self._time_to_first_token_histogram.record(
|
||||
time_to_first_token_seconds, attributes=common_attrs
|
||||
)
|
||||
|
|
@ -988,9 +1011,7 @@ class OpenTelemetry(CustomLogger):
|
|||
generation_time_seconds = duration_s
|
||||
|
||||
if generation_time_seconds > 0:
|
||||
time_per_output_token_seconds = (
|
||||
generation_time_seconds / completion_tokens
|
||||
)
|
||||
time_per_output_token_seconds = generation_time_seconds / completion_tokens
|
||||
self._time_per_output_token_histogram.record(
|
||||
time_per_output_token_seconds, attributes=common_attrs
|
||||
)
|
||||
|
|
@ -1052,6 +1073,7 @@ class OpenTelemetry(CustomLogger):
|
|||
# TODO: Refactor to use the proper OTEL Logs API instead of directly creating SDK LogRecords
|
||||
|
||||
from opentelemetry._logs import SeverityNumber, get_logger
|
||||
|
||||
try:
|
||||
from opentelemetry.sdk._logs import ( # type: ignore[attr-defined] # OTEL < 1.39.0
|
||||
LogRecord as SdkLogRecord,
|
||||
|
|
@ -1188,9 +1210,7 @@ class OpenTelemetry(CustomLogger):
|
|||
value=guardrail_information.get("guardrail_mode"),
|
||||
)
|
||||
|
||||
masked_entity_count = guardrail_information.get(
|
||||
"masked_entity_count"
|
||||
)
|
||||
masked_entity_count = guardrail_information.get("masked_entity_count")
|
||||
if masked_entity_count is not None:
|
||||
guardrail_span.set_attribute(
|
||||
"masked_entity_count", safe_dumps(masked_entity_count)
|
||||
|
|
@ -1214,12 +1234,20 @@ class OpenTelemetry(CustomLogger):
|
|||
)
|
||||
_parent_context, parent_otel_span = self._get_span_context(kwargs)
|
||||
|
||||
# CRITICAL FIX: For langfuse_otel, ALWAYS create primary spans
|
||||
# Don't use parent spans from other providers as they cause trace corruption
|
||||
is_langfuse_otel = (
|
||||
hasattr(self, "callback_name") and self.callback_name == "langfuse_otel"
|
||||
)
|
||||
if is_langfuse_otel:
|
||||
parent_otel_span = None # Ignore parent spans from other providers
|
||||
_parent_context = None
|
||||
|
||||
# Decide whether to create a primary span
|
||||
# Always create if no parent span exists (backward compatibility)
|
||||
# OR if USE_OTEL_LITELLM_REQUEST_SPAN is explicitly enabled
|
||||
should_create_primary_span = (
|
||||
parent_otel_span is None
|
||||
or get_secret_bool("USE_OTEL_LITELLM_REQUEST_SPAN")
|
||||
should_create_primary_span = parent_otel_span is None or get_secret_bool(
|
||||
"USE_OTEL_LITELLM_REQUEST_SPAN"
|
||||
)
|
||||
|
||||
if should_create_primary_span:
|
||||
|
|
@ -1245,9 +1273,7 @@ class OpenTelemetry(CustomLogger):
|
|||
if parent_otel_span.is_recording():
|
||||
parent_otel_span.set_status(Status(StatusCode.ERROR))
|
||||
self.set_attributes(parent_otel_span, kwargs, response_obj)
|
||||
self._record_exception_on_span(
|
||||
span=parent_otel_span, kwargs=kwargs
|
||||
)
|
||||
self._record_exception_on_span(span=parent_otel_span, kwargs=kwargs)
|
||||
|
||||
# Create span for guardrail information
|
||||
self._create_guardrail_span(kwargs=kwargs, context=_parent_context)
|
||||
|
|
@ -1257,6 +1283,7 @@ class OpenTelemetry(CustomLogger):
|
|||
# However, proxy-created spans should be closed here
|
||||
if (
|
||||
parent_otel_span is not None
|
||||
and hasattr(parent_otel_span, "name")
|
||||
and parent_otel_span.name == LITELLM_PROXY_REQUEST_SPAN_NAME
|
||||
):
|
||||
parent_otel_span.end(end_time=self._to_ns(end_time))
|
||||
|
|
@ -1282,17 +1309,15 @@ class OpenTelemetry(CustomLogger):
|
|||
span.record_exception(exception)
|
||||
|
||||
# Get StandardLoggingPayload for structured error information
|
||||
standard_logging_payload: Optional[StandardLoggingPayload] = (
|
||||
kwargs.get("standard_logging_object")
|
||||
standard_logging_payload: Optional[StandardLoggingPayload] = kwargs.get(
|
||||
"standard_logging_object"
|
||||
)
|
||||
|
||||
if standard_logging_payload is None:
|
||||
return
|
||||
|
||||
# Extract error_information from StandardLoggingPayload
|
||||
error_information = standard_logging_payload.get(
|
||||
"error_information"
|
||||
)
|
||||
error_information = standard_logging_payload.get("error_information")
|
||||
|
||||
if error_information is None:
|
||||
# Fallback to error_str if error_information is not available
|
||||
|
|
@ -1382,9 +1407,7 @@ class OpenTelemetry(CustomLogger):
|
|||
)
|
||||
pass
|
||||
|
||||
def cast_as_primitive_value_type(
|
||||
self, value
|
||||
) -> Union[str, bool, int, float]:
|
||||
def cast_as_primitive_value_type(self, value) -> Union[str, bool, int, float]:
|
||||
"""
|
||||
Casts the value to a primitive OTEL type if it is not already a primitive type.
|
||||
|
||||
|
|
@ -1454,8 +1477,8 @@ class OpenTelemetry(CustomLogger):
|
|||
|
||||
optional_params = kwargs.get("optional_params", {})
|
||||
litellm_params = kwargs.get("litellm_params", {}) or {}
|
||||
standard_logging_payload: Optional[StandardLoggingPayload] = (
|
||||
kwargs.get("standard_logging_object")
|
||||
standard_logging_payload: Optional[StandardLoggingPayload] = kwargs.get(
|
||||
"standard_logging_object"
|
||||
)
|
||||
if standard_logging_payload is None:
|
||||
raise ValueError("standard_logging_object not found in kwargs")
|
||||
|
|
@ -1482,8 +1505,8 @@ class OpenTelemetry(CustomLogger):
|
|||
value=safe_dumps(hidden_params),
|
||||
)
|
||||
# Cost breakdown tracking
|
||||
cost_breakdown: Optional[CostBreakdown] = (
|
||||
standard_logging_payload.get("cost_breakdown")
|
||||
cost_breakdown: Optional[CostBreakdown] = standard_logging_payload.get(
|
||||
"cost_breakdown"
|
||||
)
|
||||
if cost_breakdown:
|
||||
for key, value in cost_breakdown.items():
|
||||
|
|
@ -1696,9 +1719,7 @@ class OpenTelemetry(CustomLogger):
|
|||
"OpenTelemetry logging error in set_attributes %s", str(e)
|
||||
)
|
||||
|
||||
def _cast_as_primitive_value_type(
|
||||
self, value
|
||||
) -> Union[str, bool, int, float]:
|
||||
def _cast_as_primitive_value_type(self, value) -> Union[str, bool, int, float]:
|
||||
"""
|
||||
Casts the value to a primitive OTEL type if it is not already a primitive type.
|
||||
|
||||
|
|
@ -1776,11 +1797,9 @@ class OpenTelemetry(CustomLogger):
|
|||
message = choice.get("message") or {}
|
||||
finish_reason = choice.get("finish_reason")
|
||||
|
||||
transformed_msg = (
|
||||
self._transform_messages_to_otel_semantic_conventions(
|
||||
[message]
|
||||
)[0]
|
||||
)
|
||||
transformed_msg = self._transform_messages_to_otel_semantic_conventions(
|
||||
[message]
|
||||
)[0]
|
||||
if finish_reason:
|
||||
transformed_msg["finish_reason"] = finish_reason
|
||||
|
||||
|
|
@ -1792,9 +1811,7 @@ class OpenTelemetry(CustomLogger):
|
|||
self.set_attributes(span, kwargs, response_obj)
|
||||
kwargs.get("optional_params", {})
|
||||
litellm_params = kwargs.get("litellm_params", {}) or {}
|
||||
custom_llm_provider = litellm_params.get(
|
||||
"custom_llm_provider", "Unknown"
|
||||
)
|
||||
custom_llm_provider = litellm_params.get("custom_llm_provider", "Unknown")
|
||||
|
||||
_raw_response = kwargs.get("original_response")
|
||||
_additional_args = kwargs.get("additional_args", {}) or {}
|
||||
|
|
@ -1882,9 +1899,7 @@ class OpenTelemetry(CustomLogger):
|
|||
)
|
||||
|
||||
litellm_params = kwargs.get("litellm_params", {}) or {}
|
||||
proxy_server_request = (
|
||||
litellm_params.get("proxy_server_request", {}) or {}
|
||||
)
|
||||
proxy_server_request = litellm_params.get("proxy_server_request", {}) or {}
|
||||
headers = proxy_server_request.get("headers", {}) or {}
|
||||
traceparent = headers.get("traceparent", None)
|
||||
_metadata = litellm_params.get("metadata", {}) or {}
|
||||
|
|
@ -1951,6 +1966,19 @@ class OpenTelemetry(CustomLogger):
|
|||
headers=dynamic_headers or self.OTEL_HEADERS
|
||||
)
|
||||
|
||||
if dynamic_headers:
|
||||
verbose_logger.debug(
|
||||
"[OTEL DEBUG] Creating span processor with DYNAMIC headers: %s",
|
||||
{
|
||||
k: v[:20] + "..." if len(str(v)) > 20 else v
|
||||
for k, v in _split_otel_headers.items()
|
||||
},
|
||||
)
|
||||
else:
|
||||
verbose_logger.debug(
|
||||
"[OTEL DEBUG] Creating span processor with GLOBAL headers"
|
||||
)
|
||||
|
||||
if hasattr(
|
||||
self.OTEL_EXPORTER, "export"
|
||||
): # Check if it has the export method that SpanExporter requires
|
||||
|
|
@ -2034,14 +2062,10 @@ class OpenTelemetry(CustomLogger):
|
|||
self.OTEL_HEADERS,
|
||||
)
|
||||
|
||||
_split_otel_headers = OpenTelemetry._get_headers_dictionary(
|
||||
self.OTEL_HEADERS
|
||||
)
|
||||
_split_otel_headers = OpenTelemetry._get_headers_dictionary(self.OTEL_HEADERS)
|
||||
|
||||
# Normalize endpoint for logs - ensure it points to /v1/logs instead of /v1/traces
|
||||
normalized_endpoint = self._normalize_otel_endpoint(
|
||||
self.OTEL_ENDPOINT, "logs"
|
||||
)
|
||||
normalized_endpoint = self._normalize_otel_endpoint(self.OTEL_ENDPOINT, "logs")
|
||||
|
||||
verbose_logger.debug(
|
||||
"OpenTelemetry: Log endpoint normalized from %s to %s",
|
||||
|
|
@ -2129,18 +2153,14 @@ class OpenTelemetry(CustomLogger):
|
|||
self.OTEL_HEADERS,
|
||||
)
|
||||
|
||||
_split_otel_headers = OpenTelemetry._get_headers_dictionary(
|
||||
self.OTEL_HEADERS
|
||||
)
|
||||
_split_otel_headers = OpenTelemetry._get_headers_dictionary(self.OTEL_HEADERS)
|
||||
normalized_endpoint = self._normalize_otel_endpoint(
|
||||
self.OTEL_ENDPOINT, "metrics"
|
||||
)
|
||||
|
||||
if self.OTEL_EXPORTER == "console":
|
||||
exporter = ConsoleMetricExporter()
|
||||
return PeriodicExportingMetricReader(
|
||||
exporter, export_interval_millis=5000
|
||||
)
|
||||
return PeriodicExportingMetricReader(exporter, export_interval_millis=5000)
|
||||
|
||||
elif (
|
||||
self.OTEL_EXPORTER == "otlp_http"
|
||||
|
|
@ -2156,9 +2176,7 @@ class OpenTelemetry(CustomLogger):
|
|||
headers=_split_otel_headers,
|
||||
preferred_temporality={Histogram: AggregationTemporality.DELTA},
|
||||
)
|
||||
return PeriodicExportingMetricReader(
|
||||
exporter, export_interval_millis=5000
|
||||
)
|
||||
return PeriodicExportingMetricReader(exporter, export_interval_millis=5000)
|
||||
|
||||
elif self.OTEL_EXPORTER == "otlp_grpc" or self.OTEL_EXPORTER == "grpc":
|
||||
try:
|
||||
|
|
@ -2176,9 +2194,7 @@ class OpenTelemetry(CustomLogger):
|
|||
headers=_split_otel_headers,
|
||||
preferred_temporality={Histogram: AggregationTemporality.DELTA},
|
||||
)
|
||||
return PeriodicExportingMetricReader(
|
||||
exporter, export_interval_millis=5000
|
||||
)
|
||||
return PeriodicExportingMetricReader(exporter, export_interval_millis=5000)
|
||||
|
||||
else:
|
||||
verbose_logger.warning(
|
||||
|
|
@ -2186,9 +2202,7 @@ class OpenTelemetry(CustomLogger):
|
|||
self.OTEL_EXPORTER,
|
||||
)
|
||||
exporter = ConsoleMetricExporter()
|
||||
return PeriodicExportingMetricReader(
|
||||
exporter, export_interval_millis=5000
|
||||
)
|
||||
return PeriodicExportingMetricReader(exporter, export_interval_millis=5000)
|
||||
|
||||
def _normalize_otel_endpoint(
|
||||
self, endpoint: Optional[str], signal_type: str
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -41,7 +41,9 @@ class TestOpenTelemetryGuardrails(unittest.TestCase):
|
|||
}
|
||||
|
||||
# Create a kwargs dict with standard_logging_object containing guardrail information
|
||||
kwargs = {"standard_logging_object": {"guardrail_information": [ guardrail_info ]}}
|
||||
kwargs = {
|
||||
"standard_logging_object": {"guardrail_information": [guardrail_info]}
|
||||
}
|
||||
|
||||
# Call the method
|
||||
otel._create_guardrail_span(kwargs=kwargs, context=None)
|
||||
|
|
@ -195,11 +197,13 @@ class TestOpenTelemetryProviderInitialization(unittest.TestCase):
|
|||
|
||||
# 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"
|
||||
)
|
||||
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)
|
||||
@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.
|
||||
|
|
@ -221,11 +225,13 @@ class TestOpenTelemetryProviderInitialization(unittest.TestCase):
|
|||
|
||||
# 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"
|
||||
)
|
||||
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)
|
||||
@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.
|
||||
|
|
@ -247,9 +253,9 @@ class TestOpenTelemetryProviderInitialization(unittest.TestCase):
|
|||
|
||||
# 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"
|
||||
)
|
||||
assert (
|
||||
current_provider is existing_provider
|
||||
), "Existing LoggerProvider should be respected and not overridden"
|
||||
|
||||
|
||||
class TestOpenTelemetry(unittest.TestCase):
|
||||
|
|
@ -287,7 +293,9 @@ class TestOpenTelemetry(unittest.TestCase):
|
|||
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")
|
||||
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
|
||||
|
|
@ -365,7 +373,9 @@ class TestOpenTelemetry(unittest.TestCase):
|
|||
}
|
||||
|
||||
# Create a kwargs dict with standard_logging_object containing guardrail information
|
||||
kwargs = {"standard_logging_object": {"guardrail_information": [ guardrail_info ]}}
|
||||
kwargs = {
|
||||
"standard_logging_object": {"guardrail_information": [guardrail_info]}
|
||||
}
|
||||
|
||||
# Call the method
|
||||
otel._create_guardrail_span(kwargs=kwargs, context=None)
|
||||
|
|
@ -723,7 +733,6 @@ class TestOpenTelemetry(unittest.TestCase):
|
|||
# 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)
|
||||
|
|
@ -790,7 +799,9 @@ class TestOpenTelemetry(unittest.TestCase):
|
|||
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)
|
||||
@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()
|
||||
|
|
@ -1458,9 +1469,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
"""Set up common test fixtures"""
|
||||
self.span_exporter = InMemorySpanExporter()
|
||||
self.tracer_provider = TracerProvider()
|
||||
self.tracer_provider.add_span_processor(
|
||||
SimpleSpanProcessor(self.span_exporter)
|
||||
)
|
||||
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
|
||||
|
|
@ -1514,7 +1523,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
|
||||
self.assertTrue(
|
||||
parent_span.is_recording(),
|
||||
"External span should be recording before completion calls"
|
||||
"External span should be recording before completion calls",
|
||||
)
|
||||
|
||||
# First completion call
|
||||
|
|
@ -1525,7 +1534,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
# Verify parent span is still recording
|
||||
self.assertTrue(
|
||||
parent_span.is_recording(),
|
||||
"External span should still be recording after first completion"
|
||||
"External span should still be recording after first completion",
|
||||
)
|
||||
|
||||
# Second completion call
|
||||
|
|
@ -1536,7 +1545,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
# Verify parent span is still recording
|
||||
self.assertTrue(
|
||||
parent_span.is_recording(),
|
||||
"External span should still be recording after second completion"
|
||||
"External span should still be recording after second completion",
|
||||
)
|
||||
|
||||
# After exiting context, verify spans
|
||||
|
|
@ -1547,23 +1556,25 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
self.assertEqual(
|
||||
span.context.trace_id,
|
||||
parent_trace_id,
|
||||
f"Span {span.name} should have same trace_id as parent"
|
||||
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")
|
||||
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"
|
||||
"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"
|
||||
"Parent span should have model attribute from LiteLLM",
|
||||
)
|
||||
|
||||
# Should have raw_gen_ai_request spans (if message_logging is on)
|
||||
|
|
@ -1575,7 +1586,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
self.assertEqual(
|
||||
len(litellm_spans),
|
||||
0,
|
||||
"Should NOT have litellm_request spans when USE_OTEL_LITELLM_REQUEST_SPAN=false"
|
||||
"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
|
||||
|
|
@ -1583,7 +1594,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
self.assertEqual(
|
||||
raw_span.parent.span_id if raw_span.parent else None,
|
||||
parent_span_id,
|
||||
f"raw_gen_ai_request should be direct child of external_parent_span"
|
||||
"raw_gen_ai_request should be direct child of external_parent_span",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "true"}, clear=False)
|
||||
|
|
@ -1619,7 +1630,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
# Verify parent span is still recording
|
||||
self.assertTrue(
|
||||
parent_span.is_recording(),
|
||||
"External span should still be recording after first completion"
|
||||
"External span should still be recording after first completion",
|
||||
)
|
||||
|
||||
# Second completion call
|
||||
|
|
@ -1630,7 +1641,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
# Verify parent span is still recording
|
||||
self.assertTrue(
|
||||
parent_span.is_recording(),
|
||||
"External span should still be recording after second completion"
|
||||
"External span should still be recording after second completion",
|
||||
)
|
||||
|
||||
# After exiting context, verify spans
|
||||
|
|
@ -1641,7 +1652,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
self.assertEqual(
|
||||
span.context.trace_id,
|
||||
parent_trace_id,
|
||||
f"Span {span.name} should have same trace_id as parent"
|
||||
f"Span {span.name} should have same trace_id as parent",
|
||||
)
|
||||
|
||||
# Should have litellm_request spans (USE_OTEL_LITELLM_REQUEST_SPAN=true)
|
||||
|
|
@ -1649,7 +1660,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
self.assertEqual(
|
||||
len(litellm_spans),
|
||||
2,
|
||||
"Should have 2 litellm_request spans when USE_OTEL_LITELLM_REQUEST_SPAN=true"
|
||||
"Should have 2 litellm_request spans when USE_OTEL_LITELLM_REQUEST_SPAN=true",
|
||||
)
|
||||
|
||||
# Verify litellm_request spans are children of external span
|
||||
|
|
@ -1657,7 +1668,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
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"
|
||||
"litellm_request should be child of external_parent_span",
|
||||
)
|
||||
|
||||
# Verify raw_gen_ai_request spans (if present) are children of litellm_request
|
||||
|
|
@ -1668,7 +1679,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
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"
|
||||
"raw_gen_ai_request should be child of litellm_request",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
|
||||
|
|
@ -1706,7 +1717,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
# 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}"
|
||||
f"External span should still be recording after completion #{i+1}",
|
||||
)
|
||||
|
||||
# Verify all spans have the same trace_id
|
||||
|
|
@ -1715,19 +1726,21 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
self.assertEqual(
|
||||
span.context.trace_id,
|
||||
parent_trace_id,
|
||||
f"All spans should belong to the same trace"
|
||||
"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")
|
||||
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"
|
||||
"Parent span should have model attribute from LiteLLM",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
|
||||
|
|
@ -1759,7 +1772,9 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
|
||||
# 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")
|
||||
self.assertEqual(
|
||||
current_span, parent_span, "Span should be in global context"
|
||||
)
|
||||
|
||||
# Make completion call
|
||||
start_time = datetime.utcnow()
|
||||
|
|
@ -1769,7 +1784,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
# Verify parent span is still recording
|
||||
self.assertTrue(
|
||||
parent_span.is_recording(),
|
||||
"External span from global context should not be closed"
|
||||
"External span from global context should not be closed",
|
||||
)
|
||||
|
||||
# Verify trace structure
|
||||
|
|
@ -1778,7 +1793,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
self.assertEqual(
|
||||
span.context.trace_id,
|
||||
parent_trace_id,
|
||||
"All spans should have the same trace_id"
|
||||
"All spans should have the same trace_id",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
|
||||
|
|
@ -1793,7 +1808,9 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
"""
|
||||
# Initialize OpenTelemetry
|
||||
otel = OpenTelemetry(tracer_provider=self.tracer_provider)
|
||||
otel.message_logging = True # Enable message logging to get raw_gen_ai_request spans
|
||||
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()
|
||||
|
|
@ -1824,7 +1841,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
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"
|
||||
"raw_gen_ai_request should be child of external_parent_span",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, {"USE_OTEL_LITELLM_REQUEST_SPAN": "false"}, clear=False)
|
||||
|
|
@ -1864,7 +1881,7 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
# Verify parent span is still recording
|
||||
self.assertTrue(
|
||||
parent_span.is_recording(),
|
||||
"External span should still be recording even after failure"
|
||||
"External span should still be recording even after failure",
|
||||
)
|
||||
|
||||
# Verify trace structure
|
||||
|
|
@ -1875,40 +1892,42 @@ class TestOpenTelemetryExternalSpan(unittest.TestCase):
|
|||
self.assertEqual(
|
||||
span.context.trace_id,
|
||||
parent_trace_id,
|
||||
"All spans should have the same trace_id even on failure"
|
||||
"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")
|
||||
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"
|
||||
"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
|
||||
|
||||
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 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"}]}]
|
||||
"""
|
||||
|
|
@ -1943,14 +1962,19 @@ class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
|
|||
|
||||
# Find the call that set gen_ai.input.messages
|
||||
input_messages_calls = [
|
||||
call for call in mock_span.set_attribute.call_args_list
|
||||
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")
|
||||
|
||||
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)
|
||||
|
|
@ -1962,7 +1986,7 @@ class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
|
|||
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"}]
|
||||
"""
|
||||
|
|
@ -1997,14 +2021,19 @@ class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
|
|||
|
||||
# Find the call that set gen_ai.output.messages
|
||||
output_messages_calls = [
|
||||
call for call in mock_span.set_attribute.call_args_list
|
||||
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")
|
||||
|
||||
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)
|
||||
|
|
@ -2039,7 +2068,11 @@ class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
|
|||
"id": "test-response-id",
|
||||
"model": "gpt-4",
|
||||
"choices": [],
|
||||
"usage": {"prompt_tokens": 100, "completion_tokens": 50, "total_tokens": 150},
|
||||
"usage": {
|
||||
"prompt_tokens": 100,
|
||||
"completion_tokens": 50,
|
||||
"total_tokens": 150,
|
||||
},
|
||||
}
|
||||
|
||||
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
|
||||
|
|
@ -2052,7 +2085,7 @@ class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
|
|||
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()
|
||||
|
|
@ -2074,7 +2107,10 @@ class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
|
|||
"id": "test-response-id",
|
||||
"model": "gpt-4",
|
||||
"choices": [
|
||||
{"finish_reason": "stop", "message": {"role": "assistant", "content": "Hi"}},
|
||||
{
|
||||
"finish_reason": "stop",
|
||||
"message": {"role": "assistant", "content": "Hi"},
|
||||
},
|
||||
],
|
||||
"usage": {"prompt_tokens": 10, "completion_tokens": 20, "total_tokens": 30},
|
||||
}
|
||||
|
|
@ -2083,13 +2119,18 @@ class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
|
|||
|
||||
# Find the call that set gen_ai.response.finish_reasons
|
||||
finish_reasons_calls = [
|
||||
call for call in mock_span.set_attribute.call_args_list
|
||||
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")
|
||||
|
||||
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"])
|
||||
|
|
@ -2123,3 +2164,175 @@ class TestOpenTelemetrySemanticConventions138(unittest.TestCase):
|
|||
otel.set_attributes(span=mock_span, kwargs=kwargs, response_obj=response_obj)
|
||||
|
||||
mock_span.set_attribute.assert_any_call("gen_ai.operation.name", "chat")
|
||||
|
||||
def test_handle_failure_langfuse_otel_nulls_parent_span(self):
|
||||
"""
|
||||
For langfuse_otel, _handle_failure should ignore parent spans from other providers
|
||||
and create a root-level error span (symmetric with _handle_success).
|
||||
"""
|
||||
span_exporter = InMemorySpanExporter()
|
||||
tracer_provider = TracerProvider()
|
||||
tracer_provider.add_span_processor(SimpleSpanProcessor(span_exporter))
|
||||
|
||||
otel = OpenTelemetry(
|
||||
callback_name="langfuse_otel",
|
||||
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("other_provider_span")
|
||||
|
||||
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": other_span},
|
||||
},
|
||||
"standard_logging_object": {
|
||||
"id": "test-id",
|
||||
"call_type": "completion",
|
||||
"metadata": {},
|
||||
},
|
||||
"exception": Exception("test error"),
|
||||
}
|
||||
|
||||
otel._handle_failure(kwargs, None, start, end)
|
||||
|
||||
other_span.end()
|
||||
|
||||
spans = span_exporter.get_finished_spans()
|
||||
failure_spans = [s for s in spans if s.name != "other_provider_span"]
|
||||
|
||||
self.assertTrue(failure_spans, "Expected at least one failure span")
|
||||
for span in failure_spans:
|
||||
self.assertIsNone(
|
||||
span.parent,
|
||||
f"langfuse_otel failure span should be a root span, but has parent: {span.parent}",
|
||||
)
|
||||
|
||||
def test_handle_failure_non_langfuse_preserves_parent_span(self):
|
||||
"""
|
||||
For non-langfuse_otel callbacks, _handle_failure should still use parent spans normally.
|
||||
"""
|
||||
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.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_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"}):
|
||||
otel._handle_failure(kwargs, None, start, end)
|
||||
|
||||
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 failure span")
|
||||
for span in child_spans:
|
||||
self.assertIsNotNone(
|
||||
span.parent,
|
||||
"Non-langfuse_otel failure span should have a parent",
|
||||
)
|
||||
|
||||
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")
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue