From eddd81166044b16e6dfb74edea61db1fc2d6dc6e Mon Sep 17 00:00:00 2001 From: fei <204683769+feiiiiii5@users.noreply.github.com> Date: Thu, 1 Oct 2026 12:44:59 +0800 Subject: [PATCH 1/2] fix(openai): record a failed stream as an ERROR span A chat completion with stream=True that fails after the first chunk was exported with status OK. __next__/__anext__ called _ensure_cleanup(), which sets OK and ends the span, and the set_status(ERROR) that followed could never run because an ended span no longer records. The caller still received the exception; the trace said the call succeeded. Cleanup is what ends the span, so let it decide the final status. On error it now sets error.type, records the exception and sets ERROR, the same way the non-streaming paths in this file do. error=None keeps the previous OK for __del__ and __exit__. Adds tests for the sync and the async iterator. --- .../openai/shared/chat_wrappers.py | 23 +++-- .../tests/traces/test_streaming_errors.py | 96 +++++++++++++++++++ 2 files changed, 110 insertions(+), 9 deletions(-) create mode 100644 packages/opentelemetry-instrumentation-openai/tests/traces/test_streaming_errors.py 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 7c805d6995..f9f37514c7 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,12 @@ 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__) + self._span.record_exception(error) + self._span.set_status(Status(StatusCode.ERROR, str(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..d7823141ba --- /dev/null +++ b/packages/opentelemetry-instrumentation-openai/tests/traces/test_streaming_errors.py @@ -0,0 +1,96 @@ +"""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.trace import StatusCode + + +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): + self._delivered = False + + def __next__(self): + if not self._delivered: + self._delivered = True + return _first_chunk() + raise RuntimeError("connection reset mid-stream") + + def __aiter__(self): + return self + + async def __anext__(self): + return self.__next__() + + +def _broken_chat_stream(tracer): + span = tracer.start_span("openai.chat") + return ChatStream( + span, + _BrokenStream(), + None, + None, + None, + None, + None, + None, + time.time(), + {"model": "gpt-4"}, + ) + + +def _assert_failure_recorded(span_exporter): + 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" + assert any(event.name == "exception" for event in span.events), ( + "the failure left no exception event" + ) + + +def test_sync_stream_failure_is_recorded_as_error(span_exporter, tracer_provider): + stream = _broken_chat_stream(tracer_provider.get_tracer(__name__)) + + with pytest.raises(RuntimeError): + for _ in stream: + pass + + _assert_failure_recorded(span_exporter) + + +@pytest.mark.asyncio +async def test_async_stream_failure_is_recorded_as_error( + span_exporter, tracer_provider +): + stream = _broken_chat_stream(tracer_provider.get_tracer(__name__)) + + with pytest.raises(RuntimeError): + async for _ in stream: + pass + + _assert_failure_recorded(span_exporter) From d6d9b754cffee012d5b840837bf0d60d6c036dfd Mon Sep 17 00:00:00 2001 From: Chen Yufeiyang Date: Wed, 4 Nov 2026 17:53:11 +0800 Subject: [PATCH 2/2] fix(openai): respect content tracing for stream errors --- .../openai/shared/chat_wrappers.py | 7 ++- .../tests/traces/test_streaming_errors.py | 63 +++++++++++++------ 2 files changed, 50 insertions(+), 20 deletions(-) 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 b8a5ca1d1e..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 @@ -853,8 +853,11 @@ def _ensure_cleanup(self, error=None): self._span.set_status(Status(StatusCode.OK)) else: self._span.set_attribute(ERROR_TYPE, type(error).__name__) - self._span.record_exception(error) - self._span.set_status(Status(StatusCode.ERROR, str(error))) + 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 index d7823141ba..7c2cb992b2 100644 --- a/packages/opentelemetry-instrumentation-openai/tests/traces/test_streaming_errors.py +++ b/packages/opentelemetry-instrumentation-openai/tests/traces/test_streaming_errors.py @@ -10,9 +10,13 @@ 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", @@ -26,14 +30,15 @@ def _first_chunk(): class _BrokenStream: """Yields one chunk, then fails the way a dropped connection does.""" - def __init__(self): + def __init__(self, error): self._delivered = False + self._error = error def __next__(self): if not self._delivered: self._delivered = True return _first_chunk() - raise RuntimeError("connection reset mid-stream") + raise self._error def __aiter__(self): return self @@ -42,11 +47,11 @@ async def __anext__(self): return self.__next__() -def _broken_chat_stream(tracer): +def _broken_chat_stream(tracer, error): span = tracer.start_span("openai.chat") return ChatStream( span, - _BrokenStream(), + _BrokenStream(error), None, None, None, @@ -58,7 +63,7 @@ def _broken_chat_stream(tracer): ) -def _assert_failure_recorded(span_exporter): +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] @@ -68,29 +73,51 @@ def _assert_failure_recorded(span_exporter): f"{span.status.status_code}: the failed call reads as a success" ) assert span.attributes.get("error.type") == "RuntimeError" - assert any(event.name == "exception" for event in span.events), ( - "the failure left no exception event" - ) - - -def test_sync_stream_failure_is_recorded_as_error(span_exporter, tracer_provider): - stream = _broken_chat_stream(tracer_provider.get_tracer(__name__)) + 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): + with pytest.raises(RuntimeError) as exc_info: for _ in stream: pass - _assert_failure_recorded(span_exporter) + 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 + span_exporter, tracer_provider, content_enabled ): - stream = _broken_chat_stream(tracer_provider.get_tracer(__name__)) + error = RuntimeError(_ERROR_MESSAGE) + stream = _broken_chat_stream(tracer_provider.get_tracer(__name__), error) - with pytest.raises(RuntimeError): + with pytest.raises(RuntimeError) as exc_info: async for _ in stream: pass - _assert_failure_recorded(span_exporter) + assert exc_info.value is error + _assert_failure_recorded(span_exporter, content_enabled)