mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-14 23:21:35 +00:00
fix(opentelemetry): end litellm_request span after its child spans
This commit is contained in:
Devin AI
parent
2f03789927
commit
ef5402a915
2 changed files with 162 additions and 33 deletions
|
|
@ -1150,41 +1150,49 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
|
|||
# OR if USE_OTEL_LITELLM_REQUEST_SPAN is explicitly enabled
|
||||
should_create_primary_span = parent_span is None or get_secret_bool("USE_OTEL_LITELLM_REQUEST_SPAN")
|
||||
|
||||
if should_create_primary_span:
|
||||
# Create a new litellm_request span
|
||||
span = self._start_primary_span(kwargs, response_obj, start_time, end_time, ctx)
|
||||
# Raw-request sub-span (if enabled) - child of litellm_request span
|
||||
self._maybe_log_raw_request(kwargs, response_obj, start_time, end_time, span)
|
||||
# Do NOT duplicate attributes onto the parent proxy-request span.
|
||||
# The child litellm_request span already carries all attributes;
|
||||
# copying them to the parent doubles storage and complicates
|
||||
# search (Issue #4).
|
||||
else:
|
||||
# Do not create primary span (keep hierarchy shallow when parent exists)
|
||||
from opentelemetry.trace import Status, StatusCode
|
||||
span = None
|
||||
try:
|
||||
if should_create_primary_span:
|
||||
# Create a new litellm_request span. It is left open here and
|
||||
# ended in the finally block below so its on_end lifecycle
|
||||
# callback fires after every child span, keeping processors
|
||||
# that track the active parent (e.g. Langfuse) correctly
|
||||
# nested (Issue #33511).
|
||||
span = self._start_primary_span(kwargs, response_obj, start_time, ctx)
|
||||
# Raw-request sub-span (if enabled) - child of litellm_request span
|
||||
self._maybe_log_raw_request(kwargs, response_obj, start_time, end_time, span)
|
||||
# Do NOT duplicate attributes onto the parent proxy-request span.
|
||||
# The child litellm_request span already carries all attributes;
|
||||
# copying them to the parent doubles storage and complicates
|
||||
# search (Issue #4).
|
||||
else:
|
||||
# Do not create primary span (keep hierarchy shallow when parent exists)
|
||||
from opentelemetry.trace import Status, StatusCode
|
||||
|
||||
span = None
|
||||
# Only set attributes if the span is still recording (not closed)
|
||||
# Note: parent_span is guaranteed to be not None here
|
||||
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)
|
||||
# Only set attributes if the span is still recording (not closed)
|
||||
# Note: parent_span is guaranteed to be not None here
|
||||
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)
|
||||
|
||||
# 3. Guardrail span — ensure guardrails are always parented to an
|
||||
# existing span so they never become orphaned root spans (Issue #5).
|
||||
guardrail_ctx = self._resolve_guardrail_context(span=span, parent_span=parent_span, fallback_ctx=ctx)
|
||||
self._create_guardrail_span(kwargs=kwargs, context=guardrail_ctx)
|
||||
# 3. Guardrail span — ensure guardrails are always parented to an
|
||||
# existing span so they never become orphaned root spans (Issue #5).
|
||||
guardrail_ctx = self._resolve_guardrail_context(span=span, parent_span=parent_span, fallback_ctx=ctx)
|
||||
self._create_guardrail_span(kwargs=kwargs, context=guardrail_ctx)
|
||||
|
||||
# 4. Metrics & cost recording
|
||||
self._record_metrics(kwargs, response_obj, start_time, end_time)
|
||||
# 4. Metrics & cost recording
|
||||
self._record_metrics(kwargs, response_obj, start_time, end_time)
|
||||
|
||||
# 5. Semantic logs.
|
||||
if self.config.enable_events:
|
||||
log_span = span if span is not None else parent_span
|
||||
if log_span is not None:
|
||||
self._emit_semantic_logs(kwargs, response_obj, log_span)
|
||||
# 5. Semantic logs.
|
||||
if self.config.enable_events:
|
||||
log_span = span if span is not None else parent_span
|
||||
if log_span is not None:
|
||||
self._emit_semantic_logs(kwargs, response_obj, log_span)
|
||||
finally:
|
||||
if span is not None:
|
||||
span.end(end_time=self._to_ns(end_time))
|
||||
|
||||
# 6. Do NOT end parent span - it should be managed by its creator
|
||||
# External spans (from Langfuse, user code, HTTP headers, global context) must not be closed by LiteLLM
|
||||
|
|
@ -1212,7 +1220,6 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
|
|||
kwargs,
|
||||
response_obj,
|
||||
start_time,
|
||||
end_time,
|
||||
context,
|
||||
):
|
||||
from opentelemetry.trace import Status, StatusCode
|
||||
|
|
@ -1230,7 +1237,6 @@ class OpenTelemetry(OTELGenAISemconvMixin, CustomLogger):
|
|||
|
||||
span.set_status(Status(StatusCode.OK))
|
||||
self.set_attributes(span, kwargs, response_obj)
|
||||
span.end(end_time=self._to_ns(end_time))
|
||||
return span
|
||||
|
||||
def _maybe_log_raw_request(self, kwargs, response_obj, start_time, end_time, parent_span):
|
||||
|
|
|
|||
|
|
@ -3613,6 +3613,129 @@ class TestGuardrailSpanParenting(unittest.TestCase):
|
|||
)
|
||||
|
||||
|
||||
class TestOpenTelemetrySpanLifecycleOrdering(unittest.TestCase):
|
||||
"""Issue #33511: the litellm_request span must stay open until every
|
||||
child span (raw_gen_ai_request, guardrail) has been created and ended.
|
||||
|
||||
Even though exported start/end timestamps are explicit and look
|
||||
correctly nested, processors that track the active parent (e.g.
|
||||
Langfuse) rely on the on_start / on_end callback ORDER. If
|
||||
litellm_request.on_end fires before raw_gen_ai_request.on_start, the
|
||||
child is treated as a new root."""
|
||||
|
||||
def _build_kwargs_and_response(self, with_guardrail: bool):
|
||||
guardrail_info = {
|
||||
"guardrail_name": "pii_filter",
|
||||
"guardrail_mode": "pre_call",
|
||||
"guardrail_response": "ok",
|
||||
"start_time": time.time(),
|
||||
"end_time": time.time() + 0.1,
|
||||
}
|
||||
std_log = {
|
||||
"id": "test-lifecycle-id",
|
||||
"call_type": "completion",
|
||||
"metadata": {},
|
||||
"hidden_params": {},
|
||||
}
|
||||
if with_guardrail:
|
||||
std_log["guardrail_information"] = [guardrail_info]
|
||||
|
||||
kwargs = {
|
||||
"model": "gpt-4",
|
||||
"messages": [{"role": "user", "content": "Hello"}],
|
||||
"optional_params": {},
|
||||
"litellm_params": {"custom_llm_provider": "openai", "metadata": {}},
|
||||
"standard_logging_object": std_log,
|
||||
}
|
||||
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,
|
||||
},
|
||||
}
|
||||
return kwargs, response_obj
|
||||
|
||||
def _run_and_record(self, with_guardrail: bool):
|
||||
from opentelemetry.sdk.trace import SpanProcessor
|
||||
|
||||
events = []
|
||||
|
||||
class RecordingSpanProcessor(SpanProcessor):
|
||||
def on_start(self, span, parent_context=None):
|
||||
events.append((span.name, "start"))
|
||||
|
||||
def on_end(self, span):
|
||||
events.append((span.name, "end"))
|
||||
|
||||
tracer_provider = TracerProvider()
|
||||
tracer_provider.add_span_processor(RecordingSpanProcessor())
|
||||
|
||||
otel = OpenTelemetry(tracer_provider=tracer_provider)
|
||||
otel.tracer = tracer_provider.get_tracer(__name__)
|
||||
otel.message_logging = True
|
||||
|
||||
kwargs, response_obj = self._build_kwargs_and_response(with_guardrail)
|
||||
|
||||
start = datetime.utcnow()
|
||||
end = start + timedelta(seconds=1)
|
||||
otel._handle_success(kwargs, response_obj, start, end)
|
||||
return events
|
||||
|
||||
def test_raw_request_child_starts_and_ends_within_parent(self):
|
||||
events = self._run_and_record(with_guardrail=False)
|
||||
|
||||
self.assertIn(("litellm_request", "start"), events)
|
||||
self.assertIn(("raw_gen_ai_request", "start"), events)
|
||||
self.assertIn(("raw_gen_ai_request", "end"), events)
|
||||
self.assertIn(("litellm_request", "end"), events)
|
||||
|
||||
parent_start = events.index(("litellm_request", "start"))
|
||||
parent_end = events.index(("litellm_request", "end"))
|
||||
child_start = events.index(("raw_gen_ai_request", "start"))
|
||||
child_end = events.index(("raw_gen_ai_request", "end"))
|
||||
|
||||
self.assertLess(
|
||||
parent_start,
|
||||
child_start,
|
||||
"litellm_request must start before its raw_gen_ai_request child",
|
||||
)
|
||||
self.assertLess(
|
||||
child_end,
|
||||
parent_end,
|
||||
"litellm_request must end AFTER its raw_gen_ai_request child ends "
|
||||
"(Issue #33511)",
|
||||
)
|
||||
|
||||
def test_all_children_end_before_parent(self):
|
||||
events = self._run_and_record(with_guardrail=True)
|
||||
|
||||
parent_end = events.index(("litellm_request", "end"))
|
||||
child_names = ("raw_gen_ai_request", "guardrail")
|
||||
child_end_indices = [
|
||||
idx
|
||||
for idx, (name, phase) in enumerate(events)
|
||||
if phase == "end" and name in child_names
|
||||
]
|
||||
|
||||
self.assertTrue(child_end_indices, "Expected child spans to be recorded")
|
||||
for idx in child_end_indices:
|
||||
self.assertLess(
|
||||
idx,
|
||||
parent_end,
|
||||
"Every child span must end before litellm_request ends "
|
||||
"(Issue #33511)",
|
||||
)
|
||||
|
||||
|
||||
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."""
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue