From b717f24f742c0eb34fd1a15275989054c82598d8 Mon Sep 17 00:00:00 2001 From: Pablo Deputter Date: Fri, 18 Sep 2026 16:44:48 +0200 Subject: [PATCH 1/5] fix(boto3): harden StreamingBody span finalization --- .../integrations/boto3/_instrumentation.py | 124 +++++++++++++---- tests/integrations/boto3/test_client.py | 128 ++++++++++++++++++ 2 files changed, 222 insertions(+), 30 deletions(-) diff --git a/sentry_sdk/integrations/boto3/_instrumentation.py b/sentry_sdk/integrations/boto3/_instrumentation.py index 803b9233e8..0276c85418 100644 --- a/sentry_sdk/integrations/boto3/_instrumentation.py +++ b/sentry_sdk/integrations/boto3/_instrumentation.py @@ -1,10 +1,10 @@ from typing import TYPE_CHECKING import sentry_sdk -from sentry_sdk.consts import OP, SPANDATA +from sentry_sdk.consts import OP, SPANDATA, SPANSTATUS from sentry_sdk.integrations import DidNotEnable from sentry_sdk.integrations.boto3.consts import ORIGIN -from sentry_sdk.traces import StreamedSpan +from sentry_sdk.traces import NoOpStreamedSpan, StreamedSpan from sentry_sdk.tracing import BAGGAGE_HEADER_NAME, Span from sentry_sdk.tracing_utils import ( add_http_breadcrumb, @@ -164,26 +164,39 @@ def _replace_header(request: "AWSRequest", key: str, value: str) -> None: ) -def _sentry_after_call( - context: "Dict[str, Any]", parsed: "Dict[str, Any]", **kwargs: "Any" +def _finish_span( + span: "Union[Span, StreamedSpan]", + error: "Optional[BaseException]" = None, ) -> None: - span: "Optional[Union[Span, StreamedSpan]]" = context.pop("_sentrysdk_span", None) + with capture_internal_exceptions(): + if not isinstance(span, StreamedSpan): + if error is not None: + span.set_status(SPANSTATUS.INTERNAL_ERROR) + span.finish() + return - # Span could be absent if the integration is disabled. - if span is None: - return + if error is None: + span.end() + else: + span.__exit__(type(error), error, error.__traceback__) - span.__exit__(None, None, None) + +def _instrument_streaming_body( + span: "Union[Span, StreamedSpan]", parsed: "Dict[str, Any]" +) -> bool: + if isinstance(span, NoOpStreamedSpan): + return False body = parsed.get("Body") if not isinstance(body, StreamingBody): - return + return False streaming_span: "Union[Span, StreamedSpan]" if isinstance(span, StreamedSpan): streaming_span = sentry_sdk.traces.start_span( name=span.name, parent_span=span, + active=False, attributes={ "sentry.op": OP.HTTP_CLIENT_STREAM, "sentry.origin": ORIGIN, @@ -198,35 +211,86 @@ def _sentry_after_call( orig_read = body.read orig_close = body.close + raw_stream = body._raw_stream # type: ignore[attr-defined] + orig_raw_close = raw_stream.close + finished = False + + def finish(error: "Optional[BaseException]" = None) -> None: + nonlocal finished + if finished: + return + + finished = True + _finish_span(streaming_span, error) + + def content_length_reached() -> bool: + content_length = getattr(body, "_content_length", None) + amount_read = getattr(body, "_amount_read", None) + return ( + content_length is not None + and amount_read is not None + and amount_read >= int(content_length) + ) def sentry_streaming_body_read(*args: "Any", **kwargs: "Any") -> bytes: try: ret = orig_read(*args, **kwargs) - if ret: - return ret - - if isinstance(streaming_span, StreamedSpan): - streaming_span.end() - else: - streaming_span.finish() + with capture_internal_exceptions(): + amount = args[0] if args else kwargs.get("amt") + if ( + amount is None + or amount < 0 + or (amount > 0 and not ret) + or content_length_reached() + ): + finish() return ret - except Exception: - if isinstance(streaming_span, StreamedSpan): - streaming_span.end() - else: - streaming_span.finish() + except BaseException as error: + finish(error) raise - body.read = sentry_streaming_body_read # type: ignore - def sentry_streaming_body_close(*args: "Any", **kwargs: "Any") -> None: - if isinstance(streaming_span, StreamedSpan): - streaming_span.end() - else: - streaming_span.finish() - orig_close(*args, **kwargs) + try: + orig_close(*args, **kwargs) + finish() + except BaseException as error: + finish(error) + raise + + def sentry_raw_stream_close(*args: "Any", **kwargs: "Any") -> None: + try: + orig_raw_close(*args, **kwargs) + finish() + except BaseException as error: + finish(error) + raise + + try: + # StreamingBody.__exit__ closes `_raw_stream` directly, bypassing + # StreamingBody.close(), so both levels need to be instrumented. + raw_stream.close = sentry_raw_stream_close + body.read = sentry_streaming_body_read # type: ignore + body.close = sentry_streaming_body_close # type: ignore + except Exception: + finish() + raise + + return True + + +def _sentry_after_call( + context: "Dict[str, Any]", parsed: "Dict[str, Any]", **kwargs: "Any" +) -> None: + span: "Optional[Union[Span, StreamedSpan]]" = context.pop("_sentrysdk_span", None) + + # Span could be absent if the integration is disabled. + if span is None: + return - body.close = sentry_streaming_body_close # type: ignore + span.__exit__(None, None, None) + + with capture_internal_exceptions(): + _instrument_streaming_body(span, parsed) def _sentry_after_call_error( diff --git a/tests/integrations/boto3/test_client.py b/tests/integrations/boto3/test_client.py index db8f1b9263..de1d44de49 100644 --- a/tests/integrations/boto3/test_client.py +++ b/tests/integrations/boto3/test_client.py @@ -1,6 +1,134 @@ +import boto3 +import pytest +from botocore.awsrequest import AWSResponse +from botocore.config import Config + +import sentry_sdk +from sentry_sdk.consts import OP from sentry_sdk.integrations.boto3 import Boto3Integration +from tests.integrations.boto3.aws_mock import Body + +session = boto3.Session( # type: ignore[attr-defined] + aws_access_key_id="-", + aws_secret_access_key="-", + region_name="eu-north-1", +) def test_public_api(): assert Boto3Integration.__module__ == "sentry_sdk.integrations.boto3" assert Boto3Integration.identifier == "boto3" + + +@pytest.fixture +def client_factory(sentry_init, monkeypatch, span_streaming): + sentry_init( + traces_sample_rate=1.0, + integrations=[Boto3Integration()], + trace_lifecycle="stream" if span_streaming else "static", + # avoid SDK's machine hostname being used as server name. + server_name="", + ) + # remove retry delay to speed up tests + monkeypatch.setattr("botocore.endpoint.time.sleep", lambda delay: None) + + def make_client(service_name="s3", attempt_count=1, **client_kwargs): + return session.client( + service_name, + config=Config( + # `total_max_attempts` includes the initial request. + retries={"total_max_attempts": attempt_count, "mode": "standard"} + ), + **client_kwargs, + ) + + return make_client + + +def _capture_boto3_spans_by_op(invoke_client_method, capture_items, span_streaming): + items = capture_items() + + if span_streaming: + with sentry_sdk.traces.start_span(name="parent"): # type: ignore[attr-defined] + invoke_client_method() + + sentry_sdk.flush() + spans = [ + item.payload + for item in items + if item.type == "span" + and item.payload["attributes"].get("sentry.origin") + == Boto3Integration.origin + ] + else: + with sentry_sdk.start_transaction(): + invoke_client_method() + + transaction = next(item.payload for item in items if item.type == "transaction") + spans = [ + span + for span in transaction["spans"] + if span["origin"] == Boto3Integration.origin + ] + + spans_by_op = {} + for span in spans: + op = ( + span["attributes"].get("sentry.op") if span_streaming else span["op"] + ) + spans_by_op.setdefault(op, []).append(span) + return spans_by_op + + +def _assert_span_finished(span, span_streaming): + finished_timestamp = "end_timestamp" if span_streaming else "timestamp" + assert span[finished_timestamp] is not None + + +def _assert_one_failed_span(spans, span_streaming): + assert len(spans) == 1 + assert spans[0]["status"] in ("error", "internal_error") + _assert_span_finished(spans[0], span_streaming) + + +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_streaming_body_read_failure_finishes_stream_span( + capture_items, + client_factory, + span_streaming, +): + client = client_factory() + original_exception = OSError("stream read failed") + + class _FailingBody(Body): + def __init__(self, exception): + super().__init__(b"") + self._exception = exception + + def read(self, *args, **kwargs): + raise self._exception + + def respond(request, **kwargs): + return AWSResponse( + request.url, + 200, + {"content-length": "1"}, + _FailingBody(original_exception), + ) + + client.meta.events.register("before-send", respond) + + def invoke_client_method_and_read_body(): + body = client.get_object(Bucket="bucket", Key="foo")["Body"] + with pytest.raises(OSError) as exc_info: + body.read() + assert exc_info.value is original_exception + + spans_by_op = _capture_boto3_spans_by_op( + invoke_client_method_and_read_body, capture_items, span_streaming + ) + client_spans = spans_by_op.get(OP.HTTP_CLIENT, []) + stream_spans = spans_by_op.get(OP.HTTP_CLIENT_STREAM, []) + + assert len(client_spans) == 1 + _assert_one_failed_span(stream_spans, span_streaming) From 24b5d53aaa393f66dd47a0cf62a84d4e59f3a1b0 Mon Sep 17 00:00:00 2001 From: Pablo Deputter Date: Fri, 18 Sep 2026 17:13:42 +0200 Subject: [PATCH 2/5] lint --- tests/integrations/boto3/test_client.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/tests/integrations/boto3/test_client.py b/tests/integrations/boto3/test_client.py index de1d44de49..df08b19dd5 100644 --- a/tests/integrations/boto3/test_client.py +++ b/tests/integrations/boto3/test_client.py @@ -73,9 +73,7 @@ def _capture_boto3_spans_by_op(invoke_client_method, capture_items, span_streami spans_by_op = {} for span in spans: - op = ( - span["attributes"].get("sentry.op") if span_streaming else span["op"] - ) + op = span["attributes"].get("sentry.op") if span_streaming else span["op"] spans_by_op.setdefault(op, []).append(span) return spans_by_op From cfbc99f385eaf4c93dea35ec45f75102c00b44b4 Mon Sep 17 00:00:00 2001 From: Pablo Deputter Date: Mon, 21 Sep 2026 10:46:59 +0200 Subject: [PATCH 3/5] ref(boto3): renaming vars --- .../integrations/boto3/_instrumentation.py | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/sentry_sdk/integrations/boto3/_instrumentation.py b/sentry_sdk/integrations/boto3/_instrumentation.py index 0276c85418..46ae073e35 100644 --- a/sentry_sdk/integrations/boto3/_instrumentation.py +++ b/sentry_sdk/integrations/boto3/_instrumentation.py @@ -215,7 +215,7 @@ def _instrument_streaming_body( orig_raw_close = raw_stream.close finished = False - def finish(error: "Optional[BaseException]" = None) -> None: + def finish_span(error: "Optional[BaseException]" = None) -> None: nonlocal finished if finished: return @@ -234,35 +234,35 @@ def content_length_reached() -> bool: def sentry_streaming_body_read(*args: "Any", **kwargs: "Any") -> bytes: try: - ret = orig_read(*args, **kwargs) + read_return_value = orig_read(*args, **kwargs) with capture_internal_exceptions(): - amount = args[0] if args else kwargs.get("amt") + amount_of_bytes_requested = args[0] if args else kwargs.get("amt") if ( - amount is None - or amount < 0 - or (amount > 0 and not ret) + amount_of_bytes_requested is None + or amount_of_bytes_requested < 0 + or (amount_of_bytes_requested > 0 and not read_return_value) or content_length_reached() ): - finish() - return ret + finish_span() + return read_return_value except BaseException as error: - finish(error) + finish_span(error) raise def sentry_streaming_body_close(*args: "Any", **kwargs: "Any") -> None: try: orig_close(*args, **kwargs) - finish() + finish_span() except BaseException as error: - finish(error) + finish_span(error) raise def sentry_raw_stream_close(*args: "Any", **kwargs: "Any") -> None: try: orig_raw_close(*args, **kwargs) - finish() + finish_span() except BaseException as error: - finish(error) + finish_span(error) raise try: @@ -272,7 +272,7 @@ def sentry_raw_stream_close(*args: "Any", **kwargs: "Any") -> None: body.read = sentry_streaming_body_read # type: ignore body.close = sentry_streaming_body_close # type: ignore except Exception: - finish() + finish_span() raise return True From 9c7e0247b7f81d88db7f3dd131f63ac0d639204e Mon Sep 17 00:00:00 2001 From: Pablo Deputter Date: Mon, 21 Sep 2026 11:06:44 +0200 Subject: [PATCH 4/5] ref(boto3): Add specific comment on why we initialize with --- sentry_sdk/integrations/boto3/_instrumentation.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/sentry_sdk/integrations/boto3/_instrumentation.py b/sentry_sdk/integrations/boto3/_instrumentation.py index 46ae073e35..a30ebad643 100644 --- a/sentry_sdk/integrations/boto3/_instrumentation.py +++ b/sentry_sdk/integrations/boto3/_instrumentation.py @@ -195,7 +195,12 @@ def _instrument_streaming_body( if isinstance(span, StreamedSpan): streaming_span = sentry_sdk.traces.start_span( name=span.name, + # `parent_span` is set explicitly to the boto span. parent_span=span, + # avoid making the streaming span the current span on the scope since the application might + # keep `StreamingBody` open before reading it. Otherwise: 1. when the streamingspan ends it + # could restore the parent span on the scope, breaking the parent-child relation of newly + # created spans; 2. newly created spans would be attached to the streaming span. active=False, attributes={ "sentry.op": OP.HTTP_CLIENT_STREAM, From 65dfa5745ba191a06b61a904282ed3b57f524182 Mon Sep 17 00:00:00 2001 From: Pablo Deputter Date: Mon, 21 Sep 2026 18:07:05 +0200 Subject: [PATCH 5/5] fix(boto3): ensure span finishes correctly when reading streaming body --- sentry_sdk/integrations/boto3/_instrumentation.py | 8 +++++++- tests/integrations/boto3/test_client.py | 2 ++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/sentry_sdk/integrations/boto3/_instrumentation.py b/sentry_sdk/integrations/boto3/_instrumentation.py index a30ebad643..5dc90ce5d7 100644 --- a/sentry_sdk/integrations/boto3/_instrumentation.py +++ b/sentry_sdk/integrations/boto3/_instrumentation.py @@ -219,6 +219,7 @@ def _instrument_streaming_body( raw_stream = body._raw_stream # type: ignore[attr-defined] orig_raw_close = raw_stream.close finished = False + read_in_progress = False def finish_span(error: "Optional[BaseException]" = None) -> None: nonlocal finished @@ -238,6 +239,8 @@ def content_length_reached() -> bool: ) def sentry_streaming_body_read(*args: "Any", **kwargs: "Any") -> bytes: + nonlocal read_in_progress + read_in_progress = True try: read_return_value = orig_read(*args, **kwargs) with capture_internal_exceptions(): @@ -253,6 +256,8 @@ def sentry_streaming_body_read(*args: "Any", **kwargs: "Any") -> bytes: except BaseException as error: finish_span(error) raise + finally: + read_in_progress = False def sentry_streaming_body_close(*args: "Any", **kwargs: "Any") -> None: try: @@ -265,7 +270,8 @@ def sentry_streaming_body_close(*args: "Any", **kwargs: "Any") -> None: def sentry_raw_stream_close(*args: "Any", **kwargs: "Any") -> None: try: orig_raw_close(*args, **kwargs) - finish_span() + if not read_in_progress: + finish_span() except BaseException as error: finish_span(error) raise diff --git a/tests/integrations/boto3/test_client.py b/tests/integrations/boto3/test_client.py index df08b19dd5..6c81ba9313 100644 --- a/tests/integrations/boto3/test_client.py +++ b/tests/integrations/boto3/test_client.py @@ -104,6 +104,8 @@ def __init__(self, exception): self._exception = exception def read(self, *args, **kwargs): + # urllib3 closes the response before propagating some read failures. + self.close() raise self._exception def respond(request, **kwargs):