diff --git a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py index a7d6c11a0e..6267e2c2b3 100644 --- a/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py +++ b/packages/opentelemetry-instrumentation-openai/opentelemetry/instrumentation/openai/shared/chat_wrappers.py @@ -739,9 +739,7 @@ def __next__(self): self._process_complete_response() else: # Handle cleanup for other exceptions during stream iteration - self._ensure_cleanup() - if self._span and self._span.is_recording(): - self._span.set_status(Status(StatusCode.ERROR, str(e))) + self._ensure_cleanup(e) raise else: self._process_item(chunk) @@ -755,9 +753,7 @@ async def __anext__(self): self._process_complete_response() else: # Handle cleanup for other exceptions during stream iteration - self._ensure_cleanup() - if self._span and self._span.is_recording(): - self._span.set_status(Status(StatusCode.ERROR, str(e))) + self._ensure_cleanup(e) raise else: self._process_item(chunk) @@ -834,8 +830,12 @@ def _process_complete_response(self): self._cleanup_completed = True @dont_throw - def _ensure_cleanup(self): - """Thread-safe cleanup method that handles different cleanup scenarios""" + def _ensure_cleanup(self, error=None): + """Thread-safe cleanup method that handles different cleanup scenarios + + Cleanup is what ends the span, so it decides the span's final status: + a stream that failed is an ERROR, the way a non-streaming call fails. + """ with self._cleanup_lock: if self._cleanup_completed: logger.debug("ChatStream cleanup already completed, skipping") @@ -849,7 +849,15 @@ def _ensure_cleanup(self): # Set span status and close it if self._span and self._span.is_recording(): - self._span.set_status(Status(StatusCode.OK)) + if error is None: + self._span.set_status(Status(StatusCode.OK)) + else: + self._span.set_attribute(ERROR_TYPE, type(error).__name__) + if should_send_prompts(): + self._span.record_exception(error) + self._span.set_status(Status(StatusCode.ERROR, str(error))) + else: + self._span.set_status(Status(StatusCode.ERROR)) self._span.end() logger.debug("ChatStream span closed successfully") diff --git a/packages/opentelemetry-instrumentation-openai/tests/traces/test_streaming_errors.py b/packages/opentelemetry-instrumentation-openai/tests/traces/test_streaming_errors.py new file mode 100644 index 0000000000..7c2cb992b2 --- /dev/null +++ b/packages/opentelemetry-instrumentation-openai/tests/traces/test_streaming_errors.py @@ -0,0 +1,123 @@ +"""A stream that fails mid-iteration is a failed call, not a successful one. + +The OpenAI SDK re-raises the error out of the iterator, so the caller sees the +failure; the span has to agree. ChatStream ends its span from the cleanup path +whenever iteration stops, and that path marks the span OK -- on an error that +must not be the last word. +""" + +import time + +import pytest +from opentelemetry.instrumentation.openai.shared.chat_wrappers import ChatStream +from opentelemetry.instrumentation.openai.utils import TRACELOOP_TRACE_CONTENT +from opentelemetry.trace import StatusCode + + +_ERROR_MESSAGE = "connection reset mid-stream: private prompt content" + + +def _first_chunk(): + return { + "id": "chatcmpl-broken", + "model": "gpt-4", + "choices": [ + {"index": 0, "delta": {"content": "partial"}, "finish_reason": None} + ], + } + + +class _BrokenStream: + """Yields one chunk, then fails the way a dropped connection does.""" + + def __init__(self, error): + self._delivered = False + self._error = error + + def __next__(self): + if not self._delivered: + self._delivered = True + return _first_chunk() + raise self._error + + def __aiter__(self): + return self + + async def __anext__(self): + return self.__next__() + + +def _broken_chat_stream(tracer, error): + span = tracer.start_span("openai.chat") + return ChatStream( + span, + _BrokenStream(error), + None, + None, + None, + None, + None, + None, + time.time(), + {"model": "gpt-4"}, + ) + + +def _assert_failure_recorded(span_exporter, content_enabled): + spans = [s for s in span_exporter.get_finished_spans() if s.name == "openai.chat"] + assert len(spans) == 1, f"expected one chat span, got {[s.name for s in spans]}" + span = spans[0] + + assert span.status.status_code is StatusCode.ERROR, ( + "a stream that raised was exported as " + f"{span.status.status_code}: the failed call reads as a success" + ) + assert span.attributes.get("error.type") == "RuntimeError" + exception_events = [event for event in span.events if event.name == "exception"] + if content_enabled: + assert len(exception_events) == 1, "the failure left no exception event" + assert exception_events[0].attributes["exception.message"] == _ERROR_MESSAGE + assert span.status.description == _ERROR_MESSAGE + else: + assert not exception_events, "exception details bypassed content tracing" + assert span.status.description is None + assert _ERROR_MESSAGE not in span.to_json() + assert "partial" not in span.to_json() + + +@pytest.fixture(params=[None, "true", "false"], ids=["default", "enabled", "disabled"]) +def content_enabled(request, monkeypatch): + if request.param is None: + monkeypatch.delenv(TRACELOOP_TRACE_CONTENT, raising=False) + else: + monkeypatch.setenv(TRACELOOP_TRACE_CONTENT, request.param) + return request.param != "false" + + +def test_sync_stream_failure_is_recorded_as_error( + span_exporter, tracer_provider, content_enabled +): + error = RuntimeError(_ERROR_MESSAGE) + stream = _broken_chat_stream(tracer_provider.get_tracer(__name__), error) + + with pytest.raises(RuntimeError) as exc_info: + for _ in stream: + pass + + assert exc_info.value is error + _assert_failure_recorded(span_exporter, content_enabled) + + +@pytest.mark.asyncio +async def test_async_stream_failure_is_recorded_as_error( + span_exporter, tracer_provider, content_enabled +): + error = RuntimeError(_ERROR_MESSAGE) + stream = _broken_chat_stream(tracer_provider.get_tracer(__name__), error) + + with pytest.raises(RuntimeError) as exc_info: + async for _ in stream: + pass + + assert exc_info.value is error + _assert_failure_recorded(span_exporter, content_enabled)