diff --git a/sentry_sdk/consts.py b/sentry_sdk/consts.py index 9bad8aa7db..3b8a1a6855 100644 --- a/sentry_sdk/consts.py +++ b/sentry_sdk/consts.py @@ -390,6 +390,60 @@ class SPANDATA: Example: ["Token limit exceeded"] """ + AWS_EXTENDED_REQUEST_ID = "aws.extended_request_id" + """ + The AWS extended request ID as returned in the response headers. + Example: "wzHcyEWfmOGDIE5QOhTAqFDoDWP3y8IUvpNINCwL9N4TEHbUw0/gZJ+VZTmCNCWR7fezEN3eCiQ=" + """ + + AWS_REQUEST_ID = "aws.request_id" + """ + The AWS request ID as returned in the response headers. + Example: "79b9da39-b7ae-508a-a6bc-864b2829c622" + """ + + AWS_S3_BUCKET = "aws.s3.bucket" + """ + The S3 bucket name the request refers to. + Example: "ot-demo-test" + """ + + AWS_S3_COPY_SOURCE = "aws.s3.copy_source" + """ + The source object (in the form bucket/key) for the copy operation. + Example: "someFile.yml" + """ + + AWS_S3_DELETE = "aws.s3.delete" + """ + The delete request container that specifies the objects to be deleted. + Example: "Objects=[{Key=string,VersionId=string},{Key=string,VersionId=string}],Quiet=boolean" + """ + + AWS_S3_KEY = "aws.s3.key" + """ + The S3 object key the request refers to. Corresponds to the --key parameter of the S3 API operations. + Example: "someFile.yml" + """ + + AWS_S3_OBJECT_SIZE = "aws.s3.object_size" + """ + The size of the S3 object in bytes. + Example: 434234 + """ + + AWS_S3_PART_NUMBER = "aws.s3.part_number" + """ + The part number of the part being uploaded in a multipart-upload operation. This is a positive integer between 1 and 10,000. + Example: 3456 + """ + + AWS_S3_UPLOAD_ID = "aws.s3.upload_id" + """ + Upload ID that identifies the multipart upload. + Example: "dfRtDYWFbkRONycy.Yxwh66Yjlx.cph0gtNBtJ" + """ + CACHE_HIT = "cache.hit" """ A boolean indicating whether the requested data was found in the cache. @@ -414,6 +468,12 @@ class SPANDATA: Example: "10.1.2.80" """ + CLOUD_REGION = "cloud.region" + """ + The geographical region the resource is running. + Example: "us-east-1" + """ + CODE_FILEPATH = "code.filepath" """ .. deprecated:: @@ -541,6 +601,12 @@ class SPANDATA: Example: my_user """ + ERROR_TYPE = "error.type" + """ + Describes a class of error the operation ended with. + Example: "timeout" + """ + GEN_AI_AGENT_NAME = "gen_ai.agent.name" """ The name of the agent being used. @@ -880,6 +946,12 @@ class SPANDATA: Example: GET """ + HTTP_REQUEST_RESEND_COUNT = "http.request.resend_count" + """ + The ordinal number of request resending attempt (for any reason, including redirects). + Example: 2 + """ + HTTP_ROUTE = "http.route" """ The matched route, that is, the path template used to match the request. @@ -892,6 +964,12 @@ class SPANDATA: Example: ?foo=bar&bar=baz """ + HTTP_RESPONSE_BODY_SIZE = "http.response.body.size" + """ + The encoded body size of the response (in bytes). + Example: 123 + """ + HTTP_STATUS_CODE = "http.response.status_code" """ The HTTP status code as an integer. @@ -977,12 +1055,24 @@ class SPANDATA: Example: "com.example.ExampleService/exampleMethod" """ + RPC_SERVICE = "rpc.service" + """ + The full (logical) name of the service being called, including its package name, if applicable. + Example: "myService.BestService" + """ + RPC_RESPONSE_STATUS_CODE = "rpc.response.status_code" """ Status code of the RPC returned by the RPC server or generated by the client. Example: "DEADLINE_EXCEEDED" """ + RPC_SYSTEM_NAME = "rpc.system.name" + """ + A string identifying the remoting system. + Example: "aws-api" + """ + SERVER_ADDRESS = "server.address" """ Name of the database host. @@ -1172,6 +1262,24 @@ class SPANDATA: Used in inbound filters. """ + SENTRY_KIND = "sentry.kind" + """ + Used to clarify the relationship between parents and children, or to distinguish between spans, e.g. a `server` and `client` span with the same name. + Example: "client", "server", "producer", "consumer", "internal" + """ + + SENTRY_OP = "sentry.op" + """ + The operation of a span. + Example: "http.client" + """ + + SENTRY_ORIGIN = "sentry.origin" + """ + The origin of the instrumentation (e.g. span, log, etc.) + Example: "auto.http.otel.fastify" + """ + SENTRY_RELEASE = "sentry.release" """ The Sentry release. diff --git a/sentry_sdk/integrations/boto3.py b/sentry_sdk/integrations/boto3.py deleted file mode 100644 index 18d7accf6d..0000000000 --- a/sentry_sdk/integrations/boto3.py +++ /dev/null @@ -1,269 +0,0 @@ -from functools import partial -from typing import TYPE_CHECKING - -import sentry_sdk -from sentry_sdk.consts import OP, SPANDATA -from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version -from sentry_sdk.traces import StreamedSpan -from sentry_sdk.tracing import BAGGAGE_HEADER_NAME, Span -from sentry_sdk.tracing_utils import ( - add_http_breadcrumb, - add_sentry_baggage_to_headers, - get_url_attributes, - has_span_streaming_enabled, - should_propagate_trace, -) -from sentry_sdk.utils import ( - capture_internal_exceptions, - parse_url, - parse_version, -) - -if TYPE_CHECKING: - from typing import Any, Dict, Optional, Type, Union - - from botocore.model import ServiceId - - -try: - from botocore import __version__ as BOTOCORE_VERSION - from botocore.awsrequest import AWSRequest - from botocore.client import BaseClient - from botocore.response import StreamingBody -except ImportError: - raise DidNotEnable("botocore is not installed") - - -class Boto3Integration(Integration): - identifier = "boto3" - origin = f"auto.http.{identifier}" - - @staticmethod - def setup_once() -> None: - version = parse_version(BOTOCORE_VERSION) - _check_minimum_version(Boto3Integration, version, "botocore") - - orig_init = BaseClient.__init__ - - def sentry_patched_init( - self: "BaseClient", *args: "Any", **kwargs: "Any" - ) -> None: - orig_init(self, *args, **kwargs) - meta = self.meta - service_id = meta.service_model.service_id - meta.events.register( - "request-created", - partial(_sentry_request_created, service_id=service_id), - ) - # run after other `before-sign` handlers, allowing it to see and preserve existing baggage. - meta.events.register_last("before-sign", _sentry_before_sign) - meta.events.register("after-call", _sentry_after_call) - meta.events.register("after-call-error", _sentry_after_call_error) - - BaseClient.__init__ = sentry_patched_init # type: ignore - - -def _sentry_request_created( - service_id: "ServiceId", request: "AWSRequest", operation_name: str, **kwargs: "Any" -) -> None: - description = "aws.%s.%s" % (service_id.hyphenize(), operation_name) - - client = sentry_sdk.get_client() - if client.get_integration(Boto3Integration) is None: - return - - parsed_url = None - if request.url is not None: - with capture_internal_exceptions(): - parsed_url = parse_url(request.url, sanitize=False) - - breadcrumb: "dict[str, Any]" = {} - - is_span_streaming_enabled = has_span_streaming_enabled(client.options) - span: "Union[Span, StreamedSpan, None]" = None - if is_span_streaming_enabled: - url_attributes = get_url_attributes(client, parsed_url) - breadcrumb.update(url_attributes) - - if request.method is not None: - breadcrumb[SPANDATA.HTTP_REQUEST_METHOD] = request.method - - if sentry_sdk.traces.get_current_span() is not None: - span = sentry_sdk.traces.start_span( - name=description, - attributes={ - "sentry.op": OP.HTTP_CLIENT, - "sentry.origin": Boto3Integration.origin, - SPANDATA.RPC_METHOD: f"{service_id}/{operation_name}", - }, - ) - span.set_attributes(url_attributes) - - if request.method is not None: - span.set_attribute(SPANDATA.HTTP_REQUEST_METHOD, request.method) - else: - span = sentry_sdk.start_span( - op=OP.HTTP_CLIENT, - name=description, - origin=Boto3Integration.origin, - ) - - if parsed_url: - span.set_data("aws.request.url", parsed_url.url) - span.set_data(SPANDATA.HTTP_QUERY, parsed_url.query) - span.set_data(SPANDATA.HTTP_FRAGMENT, parsed_url.fragment) - breadcrumb.update( - { - "aws.request.url": parsed_url.url, - SPANDATA.HTTP_QUERY: parsed_url.query, - SPANDATA.HTTP_FRAGMENT: parsed_url.fragment, - } - ) - - span.set_tag("aws.service_id", service_id.hyphenize()) - span.set_tag("aws.operation_name", operation_name) - if request.method is not None: - span.set_data(SPANDATA.HTTP_METHOD, request.method) - breadcrumb[SPANDATA.HTTP_METHOD] = request.method - - # We do it in order for subsequent http calls/retries be - # attached to this span. - span.__enter__() - - add_http_breadcrumb(None, breadcrumb) - - if span is not None: - # request.context is an open-ended data-structure - # where we can add anything useful in request life cycle. - request.context["_sentrysdk_span"] = span - - -def _sentry_before_sign( - request: "AWSRequest", signature_version: "Any", **kwargs: "Any" -) -> None: - client = sentry_sdk.get_client() - if client.get_integration(Boto3Integration) is None: - return - - with capture_internal_exceptions(): - # presigned requests are executed later by another caller. Adding propagation - # headers here would make those headers part of the signature, requiring the caller to reproduce the same values. - if isinstance(signature_version, str) and signature_version.endswith( - ("-query", "-presign-post") - ): - return - - if request.url is None or not should_propagate_trace(client, request.url): - return - - def _replace_header(request: "AWSRequest", key: str, value: str) -> None: - """ - Botocore's `HTTPHeaders` inherits from `email.message.Message`, where: - headers["foo"] = "old" - headers["foo"] = "new" - produces two fields: {"foo": "old", "foo": "new"}. So delete existing - fields before assigning replacement. - """ - if key in request.headers: - del request.headers[key] - request.headers[key] = value - - # use span associated with this botocore request - span = request.context.get("_sentrysdk_span") - - headers = sentry_sdk.get_current_scope().iter_trace_propagation_headers( - span=span - ) - for header_name, header_value in headers: - if header_name != BAGGAGE_HEADER_NAME: - # normal headers (e.g. `sentry-trace`) are non-shared, so replace stale values - _replace_header(request, header_name, header_value) - continue - - # merge existing `baggage` values under single header - existing_values = request.headers.get_all(BAGGAGE_HEADER_NAME, []) - combined_baggage = { - BAGGAGE_HEADER_NAME: ",".join(str(value) for value in existing_values) - } - # preserve third-party baggage, replace stale `sentry-*` values - add_sentry_baggage_to_headers(combined_baggage, header_value) - _replace_header( - request, BAGGAGE_HEADER_NAME, combined_baggage[BAGGAGE_HEADER_NAME] - ) - - -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 - - span.__exit__(None, None, None) - - body = parsed.get("Body") - if not isinstance(body, StreamingBody): - return - - streaming_span: "Union[Span, StreamedSpan]" - if isinstance(span, StreamedSpan): - streaming_span = sentry_sdk.traces.start_span( - name=span.name, - parent_span=span, - attributes={ - "sentry.op": OP.HTTP_CLIENT_STREAM, - "sentry.origin": Boto3Integration.origin, - }, - ) - else: - streaming_span = span.start_child( - op=OP.HTTP_CLIENT_STREAM, - name=span.description, - origin=Boto3Integration.origin, - ) - - orig_read = body.read - orig_close = body.close - - 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() - return ret - except Exception: - if isinstance(streaming_span, StreamedSpan): - streaming_span.end() - else: - streaming_span.finish() - 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) - - body.close = sentry_streaming_body_close # type: ignore - - -def _sentry_after_call_error( - context: "Dict[str, Any]", exception: "Type[BaseException]", **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 - - span.__exit__(type(exception), exception, None) diff --git a/sentry_sdk/integrations/boto3/__init__.py b/sentry_sdk/integrations/boto3/__init__.py new file mode 100644 index 0000000000..d36b44d271 --- /dev/null +++ b/sentry_sdk/integrations/boto3/__init__.py @@ -0,0 +1,21 @@ +from sentry_sdk.integrations import DidNotEnable, Integration, _check_minimum_version +from sentry_sdk.integrations.boto3._client import _patch_botocore_client +from sentry_sdk.integrations.boto3.consts import IDENTIFIER, ORIGIN +from sentry_sdk.utils import parse_version + +try: + from botocore import __version__ as BOTOCORE_VERSION +except ImportError: + raise DidNotEnable("botocore is not installed") + + +class Boto3Integration(Integration): + identifier = IDENTIFIER + origin = ORIGIN + + @staticmethod + def setup_once() -> None: + version = parse_version(BOTOCORE_VERSION) + _check_minimum_version(Boto3Integration, version, "botocore") + + _patch_botocore_client() diff --git a/sentry_sdk/integrations/boto3/_client.py b/sentry_sdk/integrations/boto3/_client.py new file mode 100644 index 0000000000..32015c802e --- /dev/null +++ b/sentry_sdk/integrations/boto3/_client.py @@ -0,0 +1,155 @@ +from contextlib import contextmanager +from typing import TYPE_CHECKING + +import sentry_sdk +from sentry_sdk.integrations import DidNotEnable +from sentry_sdk.integrations.boto3._context import AwsCallContext +from sentry_sdk.integrations.boto3._instrumentation import ( + _finish_span, + _get_error_attributes, + _get_response_attributes, + _instrument_streaming_body, + _sentry_before_sign, + _sentry_request_created, + _set_span_attributes, + _start_client_span, +) +from sentry_sdk.integrations.boto3._services.registry import ( + _resolve_service, +) +from sentry_sdk.integrations.boto3.consts import IDENTIFIER +from sentry_sdk.traces import NoOpStreamedSpan, StreamedSpan +from sentry_sdk.utils import capture_internal_exceptions + +if TYPE_CHECKING: + from typing import Any, Dict, Iterator, Optional, Union + + from sentry_sdk._types import Attributes + from sentry_sdk.tracing import Span + +try: + from botocore.client import BaseClient + from botocore.exceptions import ClientError +except ImportError: + raise DidNotEnable("botocore not installed") + + +@contextmanager +def _activate_client_span( + span: "Union[Span, StreamedSpan]", +) -> "Iterator[Union[Span, StreamedSpan]]": + """ + Activate the client span temporarily during `_make_api_call()` without ending it. + + Botocore returns a `StreamingBody` before its bytes are consumed. Using the + context manager would finish it as soon as `_make_api_call()` returns, so + restore the caller's span here and let the `StreamingBody` wrapper finish + the client span when the body is consumed or closed. + + faulty: desired: + boto3 [_make_api_call] boto3 [_make_api_call------] + http [request] http [request] + stream [read] stream [read] + """ + if isinstance(span, NoOpStreamedSpan): + yield span + return + + scope = sentry_sdk.get_current_scope() + if not isinstance(span, StreamedSpan): + previous_span = scope.span + scope.span = span + try: + yield span + finally: + scope.span = previous_span + return + + previous_streamed_span = scope.streamed_span + scope.streamed_span = span + try: + yield span + finally: + scope.streamed_span = previous_streamed_span + + +def _patch_botocore_client() -> None: + orig_init = BaseClient.__init__ + orig_make_api_call = BaseClient._make_api_call # type: ignore + + def sentry_patched_init(self: "BaseClient", *args: "Any", **kwargs: "Any") -> None: + orig_init(self, *args, **kwargs) + with capture_internal_exceptions(): + self.meta.events.register("request-created", _sentry_request_created) + # run after other `before-sign` handlers so existing baggage is preserved. + self.meta.events.register_last("before-sign", _sentry_before_sign) + + def sentry_patched_make_api_call( + self: "BaseClient", operation_name: str, api_params: "Dict[str, Any]" + ) -> "Any": + """ + Track a single API call, including retries, serialization, and endpoint + resolution. For streaming responses, keep the span open until the + response body is consumed or closed. + https://github.com/boto/botocore/blob/358f8eec8c76201bb1a7a35644abcbc9036de7ed/botocore/client.py + https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans/#rpc-client-span + """ + client = sentry_sdk.get_client() + if client.get_integration(IDENTIFIER) is None: + return orig_make_api_call(self, operation_name, api_params) + + span: "Optional[Union[Span, StreamedSpan]]" = None + with capture_internal_exceptions(): + ctx = AwsCallContext(operation_name, api_params) + with capture_internal_exceptions(): + # add optional metadata to the context, e.g. service-name, region-name, etc. + ctx.add_metadata(self) + service_ext = _resolve_service(ctx.service_name) + span = _start_client_span(ctx, service_ext) + + if span is None: + return orig_make_api_call(self, operation_name, api_params) + + # activate without finishing; a streaming response may outlive the call. + span_ctx = _activate_client_span(span) + + attributes: "Attributes" = {} + try: + with span_ctx: + try: + parsed = orig_make_api_call(self, operation_name, api_params) + except BaseException as error: + if service_ext is not None and isinstance(error, ClientError): + with capture_internal_exceptions(): + attributes.update( + service_ext.get_response_attributes(ctx, error.response) + ) + with capture_internal_exceptions(): + attributes.update(_get_error_attributes(error)) + with capture_internal_exceptions(): + _set_span_attributes(span, attributes) + raise + if service_ext is not None: + with capture_internal_exceptions(): + attributes.update( + service_ext.get_response_attributes(ctx, parsed) + ) + with capture_internal_exceptions(): + attributes.update(_get_response_attributes(parsed)) + with capture_internal_exceptions(): + _set_span_attributes(span, attributes) + except BaseException as error: + _finish_span(span, error) + raise + + streaming_body_instrumented = False + with capture_internal_exceptions(): + streaming_body_instrumented = _instrument_streaming_body(span, parsed) + + # `StreamingBody`s finish their span when consumed or closed. + if not streaming_body_instrumented: + _finish_span(span) + return parsed + + BaseClient.__init__ = sentry_patched_init # type: ignore + BaseClient._make_api_call = sentry_patched_make_api_call # type: ignore diff --git a/sentry_sdk/integrations/boto3/_context.py b/sentry_sdk/integrations/boto3/_context.py new file mode 100644 index 0000000000..4ab95eb366 --- /dev/null +++ b/sentry_sdk/integrations/boto3/_context.py @@ -0,0 +1,49 @@ +from typing import TYPE_CHECKING + +from sentry_sdk.integrations import DidNotEnable +from sentry_sdk.utils import capture_internal_exceptions + +if TYPE_CHECKING: + from typing import Any, Dict, Optional + +try: + from botocore.client import BaseClient +except ImportError: + raise DidNotEnable("botocore not installed") + + +class AwsCallContext: + __slots__ = ( + "service_name", + "service_id", + "service_id_hyphenized", + "operation_name", + "region_name", + "endpoint_url", + "params", + ) + + def __init__(self, operation_name: str, params: "Dict[str, Any]") -> None: + self.operation_name: "str" = operation_name + self.params: "Dict[str, Any]" = dict(params) + self.service_name: "Optional[str]" = None + self.service_id: "Optional[str]" = None + self.service_id_hyphenized: "Optional[str]" = None + self.region_name: "Optional[str]" = None + self.endpoint_url: "Optional[str]" = None + + def add_metadata(self, client: "BaseClient") -> None: + with capture_internal_exceptions(): + service_model = client.meta.service_model + # botocore's internal identifier, e.g. `apigateway`. + self.service_name = service_model.service_name + service_id = service_model.service_id + # modeled AWS service identity used in span names, e.g. `API Gateway`. + self.service_id = str(service_id) + self.service_id_hyphenized = service_id.hyphenize() + + with capture_internal_exceptions(): + self.region_name = client.meta.region_name + + with capture_internal_exceptions(): + self.endpoint_url = client.meta.endpoint_url diff --git a/sentry_sdk/integrations/boto3/_instrumentation.py b/sentry_sdk/integrations/boto3/_instrumentation.py new file mode 100644 index 0000000000..3c180a025c --- /dev/null +++ b/sentry_sdk/integrations/boto3/_instrumentation.py @@ -0,0 +1,496 @@ +from typing import TYPE_CHECKING, cast +from urllib.parse import urlsplit + +import sentry_sdk +from sentry_sdk.consts import OP, SPANDATA, SPANSTATUS +from sentry_sdk.integrations import DidNotEnable +from sentry_sdk.integrations.boto3.consts import ( + AWS_RPC_SYSTEM_NAME, + DEFAULT_PORTS, + IDENTIFIER, + ORIGIN, +) +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, + add_sentry_baggage_to_headers, + get_url_attributes, + has_span_streaming_enabled, + should_propagate_trace, +) +from sentry_sdk.utils import ( + capture_internal_exceptions, + parse_url, +) + +if TYPE_CHECKING: + from typing import Any, Dict, Mapping, Optional, Union + + from sentry_sdk._types import Attributes + from sentry_sdk.integrations.boto3._context import AwsCallContext + from sentry_sdk.integrations.boto3._services.base import _ServiceExtension + +try: + from botocore.awsrequest import AWSRequest + from botocore.exceptions import ClientError + from botocore.response import StreamingBody +except ImportError: + raise DidNotEnable("botocore not installed") + + +def _set_span_attributes( + span: "Union[Span, StreamedSpan]", attributes: "Attributes" +) -> None: + """ + Will be removed in the next major version (3.0). This helper makes + it easier to migrate to `StreamedSpan` without having to remove + multiple conditional blocks intertwined with other logic. + """ + if isinstance(span, StreamedSpan): + span.set_attributes(attributes) + return + + for key, value in attributes.items(): + span.set_data(key, value) + + +def _get_server_attributes(endpoint_url: "Optional[str]") -> "Attributes": + if not endpoint_url: + return {} + + try: + parsed_url = urlsplit(endpoint_url) + if parsed_url.scheme not in DEFAULT_PORTS or not parsed_url.hostname: + return {} + + # `server.port` is only defined together with `server.address`. + # Infer the effective port when the configured HTTP(S) endpoint omits it. + # https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans/ + return { + SPANDATA.SERVER_ADDRESS: parsed_url.hostname, + SPANDATA.SERVER_PORT: parsed_url.port or DEFAULT_PORTS[parsed_url.scheme], + } + + except (TypeError, UnicodeError, ValueError): + # invalid client metadata must not prevent the AWS call from running. + return {} + + +def _get_client_attributes( + ctx: "AwsCallContext", +) -> "Attributes": + attributes: "Attributes" = {} + + # `rpc.service` is deprecated in OTel, but js still uses it. + if ctx.service_id is not None: + attributes[SPANDATA.RPC_SERVICE] = ctx.service_id + + if ctx.region_name: + attributes[SPANDATA.CLOUD_REGION] = ctx.region_name + + attributes.update(_get_server_attributes(ctx.endpoint_url)) + return attributes + + +def _get_response_attributes(response: "Mapping[str, Any]") -> "Attributes": + metadata = response.get("ResponseMetadata", {}) + attributes: "Attributes" = {} + + # botocore injects HTTP status into `ResponseMetadata` after parsing. + # https://github.com/boto/botocore/blob/358f8eec8c76201bb1a7a35644abcbc9036de7ed/botocore/parsers.py#L273-L284 + status_code = metadata.get("HTTPStatusCode") + if status_code is not None: + attributes[SPANDATA.HTTP_STATUS_CODE] = status_code + + retry_attempts = metadata.get("RetryAttempts", 0) + if retry_attempts > 0: + attributes[SPANDATA.HTTP_REQUEST_RESEND_COUNT] = retry_attempts + + headers = metadata.get("HTTPHeaders", {}) + + request_id = ( + metadata.get("RequestId") + or headers.get("x-amzn-requestid") + or headers.get("x-amzn-request-id") + or headers.get("x-amz-request-id") + ) + if request_id: + attributes[SPANDATA.AWS_REQUEST_ID] = request_id + + # S3's `HostId` is the extended request ID returned in `x-amz-id-2`. + # https://docs.aws.amazon.com/AmazonS3/latest/developerguide/get-request-ids.html + extended_request_id = metadata.get("HostId") or headers.get("x-amz-id-2") + if extended_request_id: + attributes[SPANDATA.AWS_EXTENDED_REQUEST_ID] = extended_request_id + + return attributes + + +def _get_error_type(exception: "BaseException") -> str: + if isinstance(exception, ClientError): + # `ClientError` wraps AWS service errors; `Error.Code` identifies the + # actual service error, e.g. `AccessDeniedException`. + # https://docs.aws.amazon.com/boto3/latest/guide/error-handling.html + error_code: "Optional[str]" = exception.response.get("Error", {}).get("Code") + if error_code: + return error_code + + # failures before a service response have no AWS error code. + # https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans/ + exception_type = type(exception) + exception_name = exception_type.__qualname__ + exception_module = exception_type.__module__ + if exception_module not in ("builtins", "__builtins__"): + return f"{exception_module}.{exception_name}" + return exception_name + + +def _get_error_attributes(exception: "BaseException") -> "Attributes": + attributes: "Attributes" = {} + if isinstance(exception, ClientError): + attributes.update(_get_response_attributes(exception.response)) + + attributes[SPANDATA.ERROR_TYPE] = _get_error_type(exception) + return attributes + + +def _start_client_span( + ctx: "AwsCallContext", + service_ext: "Optional[_ServiceExtension]" = None, +) -> "Optional[Union[Span, StreamedSpan]]": + + client = sentry_sdk.get_client() + + # use "unknown" if `service_id_hyphenized` is not set so span name can still be created. + # e.g. "aws.unknown.GetObject" + service_name = ctx.service_id_hyphenized or "unknown" + span_name = f"aws.{service_name}.{ctx.operation_name}" + attributes: "Attributes" = { + SPANDATA.RPC_METHOD: ctx.operation_name, + SPANDATA.RPC_SYSTEM_NAME: AWS_RPC_SYSTEM_NAME, + # all client call spans are by default "client" spans. + # https://opentelemetry.io/docs/specs/semconv/cloud-providers/aws-sdk/#aws-sdk-spans + SPANDATA.SENTRY_KIND: "client", + } + with capture_internal_exceptions(): + attributes.update(_get_client_attributes(ctx)) + + # `sentry.span_op` and `sentry.span_origin` are set to generic defaults; + # a service extension can override them with `get_span_op()` and `get_span_origin()`. + span_op = OP.HTTP_CLIENT + span_origin = ORIGIN + + if service_ext is not None: + with capture_internal_exceptions(): + service_op = service_ext.get_span_op(ctx) + if service_op is not None: + span_op = service_op + + with capture_internal_exceptions(): + service_origin = service_ext.get_span_origin(ctx) + if service_origin is not None: + span_origin = service_origin + + with capture_internal_exceptions(): + attributes.update(service_ext.get_request_attributes(ctx)) + + if has_span_streaming_enabled(client.options): + if sentry_sdk.traces.get_current_span() is None: + return None + + # `start_span()` evaluates `ignore_spans` against the initial attributes. + # https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans/#rpc-client-span + attributes.update( + { + SPANDATA.SENTRY_OP: span_op, + SPANDATA.SENTRY_ORIGIN: span_origin, + } + ) + return sentry_sdk.traces.start_span( + name=span_name, + attributes=attributes, + # `StreamingBody` responses outlive `_make_api_call()`. `_activate_client_span()` + # activates this span only while the call itself runs. + active=False, + ) + + span = sentry_sdk.start_span( + name=span_name, + op=span_op, + origin=span_origin, + ) + with capture_internal_exceptions(): + _set_span_attributes(span, attributes) + with capture_internal_exceptions(): + if ctx.service_id_hyphenized is not None: + span.set_tag("aws.service_id", ctx.service_id_hyphenized) + span.set_tag("aws.operation_name", ctx.operation_name) + return span + + +def _finish_span( + span: "Union[Span, StreamedSpan]", + error: "Optional[BaseException]" = None, +) -> None: + with capture_internal_exceptions(): + if not isinstance(span, StreamedSpan): + if error is not None: + span.set_status(SPANSTATUS.INTERNAL_ERROR) + span.finish() + return + + if error is None: + span.end() + else: + span.__exit__(type(error), error, error.__traceback__) + + +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 False + + streaming_span: "Union[Span, StreamedSpan]" + if isinstance(span, StreamedSpan): + streaming_span = sentry_sdk.traces.start_span( + name=span.name, + # keep the stream span under the client span after + # `_make_api_call()` returns. + parent_span=span, + # the body may outlive the api call, so keep it inactive. Otherwise it + # 1. could restore the already-finished client span when it ends; 2. make + # unrelated new spans attach to the stream span since it's the current span. + active=False, + attributes={ + SPANDATA.SENTRY_OP: OP.HTTP_CLIENT_STREAM, + SPANDATA.SENTRY_ORIGIN: ORIGIN, + }, + ) + else: + streaming_span = span.start_child( + op=OP.HTTP_CLIENT_STREAM, + name=span.description, + origin=ORIGIN, + ) + + finished = False + read_in_progress = False + + def finish_span(error: "Optional[BaseException]" = None) -> None: + nonlocal finished + if finished: + return + + finished = True + # finish the stream span before the client span, and only once across + # read and close. + if error is not None: + with capture_internal_exceptions(): + attributes = _get_error_attributes(error) + _set_span_attributes(streaming_span, attributes) + _set_span_attributes(span, attributes) + + _finish_span(streaming_span, error) + _finish_span(span, error) + + def content_length_reached() -> bool: + content_length = body._content_length # type: ignore[attr-defined] + return content_length is not None and body._amount_read >= int(content_length) # type: ignore[attr-defined] + + 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(): + amount_of_bytes_requested = args[0] if args else kwargs.get("amt") + # detect read-to-end, eof, or the known content length being consumed. + if ( + 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_span() + return read_return_value + except BaseException as error: + finish_span(error) + raise + finally: + read_in_progress = False + + def sentry_streaming_body_close(*args: "Any", **kwargs: "Any") -> None: + try: + orig_close(*args, **kwargs) + finish_span() + except BaseException as error: + finish_span(error) + raise + + def sentry_raw_stream_close(*args: "Any", **kwargs: "Any") -> None: + try: + orig_raw_close(*args, **kwargs) + if not read_in_progress: + finish_span() + except BaseException as error: + finish_span(error) + raise + + try: + orig_read = body.read + orig_close = body.close + raw_stream = body._raw_stream # type: ignore[attr-defined] + orig_raw_close = raw_stream.close + + 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_span() + raise + + return True + + +def _set_request_attributes( + span: "Union[Span, StreamedSpan]", + request: "AWSRequest", +) -> None: + client = sentry_sdk.get_client() + + parsed_url = None + with capture_internal_exceptions(): + parsed_url = parse_url(cast(str, request.url), sanitize=False) + + # overwrite server attributes when actual request URL is resolved. + _set_span_attributes(span, _get_server_attributes(request.url)) + + if isinstance(span, StreamedSpan): + span.set_attributes(get_url_attributes(client, parsed_url)) + span.set_attribute(SPANDATA.HTTP_REQUEST_METHOD, cast(str, request.method)) + return + + if parsed_url is not None: + span.set_data("aws.request.url", parsed_url.url) + span.set_data(SPANDATA.HTTP_QUERY, parsed_url.query) + span.set_data(SPANDATA.HTTP_FRAGMENT, parsed_url.fragment) + + span.set_data(SPANDATA.HTTP_METHOD, request.method) + + +def _add_request_breadcrumb(request: "AWSRequest") -> None: + client = sentry_sdk.get_client() + + parsed_url = None + with capture_internal_exceptions(): + parsed_url = parse_url(cast(str, request.url), sanitize=False) + + breadcrumb: "dict[str, Any]" = {} + + if has_span_streaming_enabled(client.options): + breadcrumb.update(get_url_attributes(client, parsed_url)) + breadcrumb[SPANDATA.HTTP_REQUEST_METHOD] = request.method + else: + if parsed_url is not None: + breadcrumb.update( + { + "aws.request.url": parsed_url.url, + SPANDATA.HTTP_QUERY: parsed_url.query, + SPANDATA.HTTP_FRAGMENT: parsed_url.fragment, + } + ) + + breadcrumb[SPANDATA.HTTP_METHOD] = request.method + + add_http_breadcrumb(None, breadcrumb) + + +def _sentry_request_created( + request: "AWSRequest", operation_name: str, **kwargs: "Any" +) -> None: + """ + Enrich a single `AWSRequest` attempt. Botocore creates a + fresh `AWSRequest` on every retry. + https://github.com/boto/botocore/blob/f9195c79ea2bf46350dd320d2a0bf3db7da0b460/botocore/endpoint.py#L178-L202 + """ + + client = sentry_sdk.get_client() + if client.get_integration(IDENTIFIER) is None: + return + + with capture_internal_exceptions(): + _add_request_breadcrumb(request) + + span = ( + sentry_sdk.traces.get_current_span() + if has_span_streaming_enabled(client.options) + else sentry_sdk.get_current_span() + ) + if span is None: + return + + # an ignored streamed span is not active; avoid enriching its parent. + if isinstance(span, StreamedSpan) and span.active: + return + + _set_request_attributes(span, request) + # each attempt has a fresh `request.context`; carry the active client span. + request.context["_sentrysdk_span"] = span + + +def _sentry_before_sign( + request: "AWSRequest", signature_version: "Any", **kwargs: "Any" +) -> None: + client = sentry_sdk.get_client() + if client.get_integration(IDENTIFIER) is None: + return + + with capture_internal_exceptions(): + # presigned requests are executed later by another caller. Adding propagation + # headers here would make those headers part of the signature, requiring the caller to reproduce the same values. + if isinstance(signature_version, str) and signature_version.endswith( + ("-query", "-presign-post") + ): + return + + if request.url is None or not should_propagate_trace(client, request.url): + return + + def _replace_header(request: "AWSRequest", key: str, value: str) -> None: + """ + Botocore's `HTTPHeaders` inherits from `email.message.Message`, where: + headers["foo"] = "old" + headers["foo"] = "new" + produces two fields: {"foo": "old", "foo": "new"}. So delete existing + fields before assigning replacement. + """ + if key in request.headers: + del request.headers[key] + request.headers[key] = value + + # use span associated with this botocore request + span = request.context.get("_sentrysdk_span") + headers = sentry_sdk.get_current_scope().iter_trace_propagation_headers( + span=span + ) + for header_name, header_value in headers: + if header_name != BAGGAGE_HEADER_NAME: + # normal headers (e.g. `sentry-trace`) are non-shared, so replace stale values + _replace_header(request, header_name, header_value) + continue + + # merge existing `baggage` values under single header + existing_values = request.headers.get_all(BAGGAGE_HEADER_NAME, []) + combined_baggage = { + BAGGAGE_HEADER_NAME: ",".join(str(value) for value in existing_values) + } + add_sentry_baggage_to_headers(combined_baggage, header_value) + _replace_header( + request, BAGGAGE_HEADER_NAME, combined_baggage[BAGGAGE_HEADER_NAME] + ) diff --git a/sentry_sdk/integrations/boto3/_services/__init__.py b/sentry_sdk/integrations/boto3/_services/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/sentry_sdk/integrations/boto3/_services/_attribute_extraction.py b/sentry_sdk/integrations/boto3/_services/_attribute_extraction.py new file mode 100644 index 0000000000..ff57dc5065 --- /dev/null +++ b/sentry_sdk/integrations/boto3/_services/_attribute_extraction.py @@ -0,0 +1,36 @@ +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from typing import Any, Callable, Optional, Sequence, Tuple + + from sentry_sdk._types import Attributes + + _Converter = Callable[[Any], Optional[Any]] + # e.g. ("Limit", "aws.dynamodb.limit", _as_integer) converts + # {"Limit": 10} into {"aws.dynamodb.limit": 10} using `_extract_attributes()` + _AttributeSpec = Tuple[str, str, _Converter] + + +def _as_integer(value: "Any") -> "Optional[int]": + if isinstance(value, int) and not isinstance(value, bool): + return value + return None + + +def _as_string(value: "Any") -> "Optional[str]": + return value if isinstance(value, str) and value else None + + +def _extract_attributes( + source: "Any", specs: "Sequence[_AttributeSpec]" +) -> "Attributes": + if not isinstance(source, dict): + return {} + + attributes = {} + for param, attribute, convert in specs: + value = convert(source.get(param)) + # an unexpected type results in that attribute being omitted. + if value is not None: + attributes[attribute] = value + return attributes diff --git a/sentry_sdk/integrations/boto3/_services/base.py b/sentry_sdk/integrations/boto3/_services/base.py new file mode 100644 index 0000000000..0d9cb1c24d --- /dev/null +++ b/sentry_sdk/integrations/boto3/_services/base.py @@ -0,0 +1,36 @@ +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from typing import Any, Optional + + from sentry_sdk._types import Attributes + from sentry_sdk.integrations.boto3._context import AwsCallContext + + +class _ServiceExtension: + """ + Optional hooks for adding service-specific behavior to AWS client + span; non-overridden methods keep the generic instrumentation. + Services without a registered extension in `_SERVICE_EXTENSIONS` continue + to use the generic instrumentation. + """ + + __slots__ = () + + def get_span_op(self, ctx: "AwsCallContext") -> "Optional[str]": + """Return an optional `sentry.op` override, or `None` to keep the default.""" + return None + + def get_span_origin(self, ctx: "AwsCallContext") -> "Optional[str]": + """Return an optional `sentry.origin` override, or `None` to keep the default.""" + return None + + def get_request_attributes(self, ctx: "AwsCallContext") -> "Attributes": + """Return request attributes to add before the AWS request is made.""" + return {} + + def get_response_attributes( + self, ctx: "AwsCallContext", response: "Any" + ) -> "Attributes": + """Return response attributes to add after the AWS request is made.""" + return {} diff --git a/sentry_sdk/integrations/boto3/_services/registry.py b/sentry_sdk/integrations/boto3/_services/registry.py new file mode 100644 index 0000000000..991da042d9 --- /dev/null +++ b/sentry_sdk/integrations/boto3/_services/registry.py @@ -0,0 +1,27 @@ +from typing import TYPE_CHECKING + +from sentry_sdk.integrations.boto3._services.s3 import _S3Extension + +if TYPE_CHECKING: + from typing import Dict, Optional + + from sentry_sdk.integrations.boto3._services.base import _ServiceExtension + + +# add a ServiceExtension here when one is implemented. for example: +# _SERVICE_EXTENSIONS = {"s3": _S3Extension()} +# when py 3.15 drops, we might want to take a look at using +# a lazy-loading approach using the new `lazy` keyword. +# e.g. {"s3": _S3Extension()} +_SERVICE_EXTENSIONS: "Dict[str, _ServiceExtension]" = { + "s3": _S3Extension(), +} + + +def _resolve_service( + service_name: "Optional[str]", +) -> "Optional[_ServiceExtension]": + """Return the extension for a service, or `None` for generic instrumentation.""" + if service_name is None: + return None + return _SERVICE_EXTENSIONS.get(service_name) diff --git a/sentry_sdk/integrations/boto3/_services/s3.py b/sentry_sdk/integrations/boto3/_services/s3.py new file mode 100644 index 0000000000..ed4f201d30 --- /dev/null +++ b/sentry_sdk/integrations/boto3/_services/s3.py @@ -0,0 +1,109 @@ +import json +from typing import TYPE_CHECKING + +from sentry_sdk.consts import SPANDATA +from sentry_sdk.integrations.boto3._services._attribute_extraction import ( + _as_integer, + _as_string, + _extract_attributes, +) +from sentry_sdk.integrations.boto3._services.base import _ServiceExtension + +if TYPE_CHECKING: + from typing import Any, Optional, Sequence + + from sentry_sdk._types import Attributes + from sentry_sdk.integrations.boto3._context import AwsCallContext + from sentry_sdk.integrations.boto3._services._attribute_extraction import ( + _AttributeSpec, + ) + +_RESPONSE_BODY_SIZE_OPERATIONS = frozenset( + ( + "GetObject", + "GetObjectAnnotation", + ) +) + +_RESPONSE_OBJECT_SIZE_FIELDS = { + "GetObjectAttributes": "ObjectSize", + "PutObject": "Size", +} + + +def _json_dict(value: "Any") -> "Optional[str]": + if not isinstance(value, dict): + return None + + try: + return json.dumps( + value, + allow_nan=False, + separators=(",", ":"), + sort_keys=True, + ) + except (TypeError, ValueError): + return None + + +_REQUEST_ATTRIBUTES: "Sequence[_AttributeSpec]" = ( + # s3-specific attributes defined by OTel SemConv. Specified as a tuple of + # (param_name, attribute_name, converter_func). the `converter_func` is + # used to 1. validate the value (otherwise omitted) and 2. convert it to + # the appropriate type. + # https://opentelemetry.io/docs/specs/semconv/object-stores/s3/ + ("Bucket", SPANDATA.AWS_S3_BUCKET, _as_string), + ("CopySource", SPANDATA.AWS_S3_COPY_SOURCE, _as_string), + ("Delete", SPANDATA.AWS_S3_DELETE, _json_dict), + ("Key", SPANDATA.AWS_S3_KEY, _as_string), + ("PartNumber", SPANDATA.AWS_S3_PART_NUMBER, _as_integer), + ("UploadId", SPANDATA.AWS_S3_UPLOAD_ID, _as_string), +) + + +class _S3Extension(_ServiceExtension): + __slots__ = () + + def get_request_attributes(self, ctx: "AwsCallContext") -> "Attributes": + attributes: "Attributes" = _extract_attributes(ctx.params, _REQUEST_ATTRIBUTES) + + if ctx.operation_name == "CompleteMultipartUpload": + object_size = _as_integer(ctx.params.get("MpuObjectSize")) + if object_size is not None and object_size >= 0: + attributes[SPANDATA.AWS_S3_OBJECT_SIZE] = object_size + + return attributes + + def get_response_attributes( + self, ctx: "AwsCallContext", response: "Any" + ) -> "Attributes": + if not isinstance(response, dict): + return {} + + attributes: "Attributes" = {} + operation_name = ctx.operation_name + + if operation_name in _RESPONSE_BODY_SIZE_OPERATIONS: + # `ContentLength` is the size of the HTTP body returned, which may be a range. + content_length = _as_integer(response.get("ContentLength")) + if content_length is not None and content_length >= 0: + attributes[SPANDATA.HTTP_RESPONSE_BODY_SIZE] = content_length + + # these fields report the total S3 object size, not the HTTP body size. + object_size_field = _RESPONSE_OBJECT_SIZE_FIELDS.get(operation_name) + if object_size_field is not None: + object_size = _as_integer(response.get(object_size_field)) + if object_size is not None and object_size >= 0: + attributes[SPANDATA.AWS_S3_OBJECT_SIZE] = object_size + + if ( + operation_name == "HeadObject" + and "Range" not in ctx.params + and "PartNumber" not in ctx.params + ): + # an un-ranged `HEAD` has no body, so `ContentLength` is the object size. + object_size = _as_integer(response.get("ContentLength")) + if object_size is not None and object_size >= 0: + attributes[SPANDATA.AWS_S3_OBJECT_SIZE] = object_size + + return attributes diff --git a/sentry_sdk/integrations/boto3/consts.py b/sentry_sdk/integrations/boto3/consts.py new file mode 100644 index 0000000000..7d0d441257 --- /dev/null +++ b/sentry_sdk/integrations/boto3/consts.py @@ -0,0 +1,12 @@ +IDENTIFIER = "boto3" +ORIGIN = f"auto.http.{IDENTIFIER}" + +# value is used by `rpc.system` (deprecated in OTel, but we still support it for now) and `rpc.system.name` +# https://opentelemetry.io/docs/specs/semconv/cloud-providers/aws-sdk/#aws-sdk-spans +AWS_RPC_SYSTEM_NAME = "aws-api" + +# default ports for HTTP and HTTPS +DEFAULT_PORTS = { + "http": 80, + "https": 443, +} diff --git a/sentry_sdk/integrations/stdlib.py b/sentry_sdk/integrations/stdlib.py index 02f8b245f7..35157fdd97 100644 --- a/sentry_sdk/integrations/stdlib.py +++ b/sentry_sdk/integrations/stdlib.py @@ -288,7 +288,15 @@ def putrequest( breadcrumb[SPANDATA.HTTP_REQUEST_METHOD] = method breadcrumb.update(url_attributes) - if sentry_sdk.traces.get_current_span() is not None: + parent_span = sentry_sdk.traces.get_current_span() + if parent_span is not None: + is_inactive_boto3_span = ( + client.get_integration("boto3") is not None + and parent_span.get_attributes().get(SPANDATA.SENTRY_ORIGIN) + == getattr(client.get_integration("boto3"), "origin", None) + and not getattr(parent_span, "active", True) + ) + # fmt: off span = sentry_sdk.traces.start_span( name="%s %s" % ( @@ -300,7 +308,11 @@ def putrequest( "sentry.op": OP.HTTP_CLIENT, SPANDATA.HTTP_REQUEST_METHOD: method, }, + # boto3 integration owns span's lifecycle; keep child inactive so it + # can't restore boto3 span later on. + active = not is_inactive_boto3_span, ) + # fmt: on for key, value in url_attributes.items(): span.set_attribute(key, value) diff --git a/tests/integrations/boto3/test_client.py b/tests/integrations/boto3/test_client.py new file mode 100644 index 0000000000..c309549316 --- /dev/null +++ b/tests/integrations/boto3/test_client.py @@ -0,0 +1,897 @@ +from http.server import BaseHTTPRequestHandler, HTTPServer +from threading import Thread + +import boto3 +import pytest +from botocore.awsrequest import AWSResponse +from botocore.config import Config +from botocore.exceptions import ClientError, EndpointConnectionError +from botocore.response import StreamingBody +from botocore.stub import Stubber + +import sentry_sdk +from sentry_sdk.consts import OP, SPANDATA +from sentry_sdk.integrations.boto3 import Boto3Integration +from sentry_sdk.integrations.boto3._services.base import _ServiceExtension +from sentry_sdk.integrations.boto3._services.registry import _SERVICE_EXTENSIONS +from sentry_sdk.integrations.boto3.consts import AWS_RPC_SYSTEM_NAME, ORIGIN +from sentry_sdk.integrations.stdlib import StdlibIntegration +from sentry_sdk.traces import StreamedSpan +from sentry_sdk.tracing import Span +from tests.integrations.boto3.aws_mock import Body, MockResponse + +session = boto3.Session( # type: ignore[attr-defined] + aws_access_key_id="-", + aws_secret_access_key="-", + region_name="eu-north-1", +) + + +@pytest.fixture +def streaming_s3_server(): + class StreamingS3Handler(BaseHTTPRequestHandler): + def do_GET(self): + self.send_response(200) + self.send_header("Content-Length", "1") + self.send_header("Content-Type", "application/octet-stream") + self.end_headers() + self.wfile.write(b"x") + self.wfile.flush() + + def log_message(self, *args): + pass + + server = HTTPServer(("127.0.0.1", 0), StreamingS3Handler) + thread = Thread(target=server.serve_forever, daemon=True) + thread.start() + + try: + yield server + finally: + server.shutdown() + server.server_close() + thread.join() + + +@pytest.mark.parametrize( + "consume", + ["read", "read_exact", "context", "close"], +) +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_streaming_span_order_and_scope( + sentry_init, + capture_items, + streaming_s3_server, + consume, + span_streaming, +): + sentry_init( + traces_sample_rate=1.0, + trace_lifecycle="stream" if span_streaming else "static", + default_integrations=False, + integrations=[Boto3Integration(), StdlibIntegration()], + server_name="", + ) + server = streaming_s3_server + client = session.client( + "s3", + endpoint_url="http://127.0.0.1:%s" % server.server_port, + config=Config( + retries={"total_max_attempts": 1, "mode": "standard"}, + s3={"addressing_style": "path"}, + ), + ) + request_client_spans = [] + + def record_client_span(request, **kwargs): + request_client_spans.append(request.context["_sentrysdk_span"]) + + client.meta.events.register("request-created", record_client_span) + items = capture_items() + + parent_context = ( + sentry_sdk.traces.start_span(name="parent") # type: ignore[attr-defined] + if span_streaming + else sentry_sdk.start_transaction(name="parent") + ) + with parent_context as parent: + body = client.get_object(Bucket="bucket", Key="key")["Body"] + assert len(request_client_spans) == 1 + request_client_span = request_client_spans[0] + if span_streaming: + assert isinstance(request_client_span, StreamedSpan) + assert request_client_span.end_timestamp is None + assert sentry_sdk.traces.get_current_span() is parent # type: ignore[attr-defined] + else: + assert isinstance(request_client_span, Span) + assert not isinstance(request_client_span, StreamedSpan) + assert request_client_span.timestamp is None + assert sentry_sdk.get_current_span() is parent + + if consume == "read": + assert body.read() == b"x" + elif consume == "read_exact": + assert body.read(1) == b"x" + elif consume == "context": + if not hasattr(body, "__enter__"): + body.close() + pytest.skip("`StreamingBody` context manager is unavailable.") + with body as raw_stream: + assert raw_stream.read() == b"x" + else: + body.close() + + if span_streaming: + assert request_client_span.end_timestamp is not None + assert sentry_sdk.traces.get_current_span() is parent # type: ignore[attr-defined] + + probe = sentry_sdk.traces.start_span(name="probe") # type: ignore[attr-defined] + assert probe._parent_span_id == parent.span_id + probe.end() + + body.close() + assert sentry_sdk.traces.get_current_span() is parent # type: ignore[attr-defined] + else: + assert request_client_span.timestamp is not None + assert sentry_sdk.get_current_span() is parent + + sentry_sdk.flush() + if span_streaming: + spans = [item.payload for item in items] + else: + transaction = next(item.payload for item in items if item.type == "transaction") + spans = transaction["spans"] + client_spans = [ + span + for span in spans + if span.get("name", span.get("description")) == "aws.s3.GetObject" + and ( + span["attributes"].get(SPANDATA.SENTRY_ORIGIN) == ORIGIN + and span["attributes"].get(SPANDATA.SENTRY_OP) == OP.HTTP_CLIENT + if span_streaming + else span["origin"] == ORIGIN and span["op"] == OP.HTTP_CLIENT + ) + ] + http_spans = [ + span + for span in spans + if ( + span["attributes"].get(SPANDATA.SENTRY_ORIGIN) == "auto.http.stdlib.httplib" + if span_streaming + else span["origin"] == "auto.http.stdlib.httplib" + ) + ] + stream_spans = [ + span + for span in spans + if span.get("name", span.get("description")) == "aws.s3.GetObject" + and ( + span["attributes"].get(SPANDATA.SENTRY_OP) == OP.HTTP_CLIENT_STREAM + if span_streaming + else span["op"] == OP.HTTP_CLIENT_STREAM + ) + ] + assert len(client_spans) == 1 + assert len(http_spans) == 1 + assert len(stream_spans) == 1 + client_span = client_spans[0] + http_span = http_spans[0] + stream_span = stream_spans[0] + + assert http_span["parent_span_id"] == client_span["span_id"] + assert stream_span["parent_span_id"] == client_span["span_id"] + assert client_span["span_id"] == request_client_span.span_id + end_timestamp = "end_timestamp" if span_streaming else "timestamp" + for span in (client_span, http_span, stream_span): + assert span[end_timestamp] is not None + + +def test_non_body_stream_does_not_delay_client_span(sentry_init, capture_items): + sentry_init( + traces_sample_rate=1.0, + trace_lifecycle="stream", + integrations=[Boto3Integration()], + server_name="", + ) + client = session.client("lambda") + + def respond(request, **kwargs): + return AWSResponse( + request.url, + 200, + {"content-length": "1"}, + Body(b"x"), + ) + + client.meta.events.register("before-send", respond) + items = capture_items("span") + + with sentry_sdk.traces.start_span(name="parent") as parent: # type: ignore[attr-defined] + response = client.invoke(FunctionName="function") + assert isinstance(response["Payload"], StreamingBody) + assert sentry_sdk.traces.get_current_span() is parent # type: ignore[attr-defined] + + sentry_sdk.flush() + spans = [item.payload for item in items] + boto_spans = [ + span + for span in spans + if span["attributes"].get(SPANDATA.SENTRY_ORIGIN) == ORIGIN + ] + assert len(boto_spans) == 1 + assert boto_spans[0]["attributes"].get(SPANDATA.SENTRY_OP) == OP.HTTP_CLIENT + response["Payload"].close() + + +@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 _mock_responses(client, status_codes): + request_span_ids = [] + + def record_request(request, **kwargs): + span = request.context.get("_sentrysdk_span") + assert span is not None + request_span_ids.append(span.span_id) + + def respond(request, **kwargs): + # `request_created` runs before `before_send`, so use zero-based index for current + # attempt; `min(..., len(status_codes) - 1)` clamps to last status to avoid `IndexError`. + response_index = min(len(request_span_ids) - 1, len(status_codes) - 1) + return AWSResponse(request.url, status_codes[response_index], {}, Body(b"")) + + client.meta.events.register("request-created", record_request) + client.meta.events.register("before-send", respond) + return request_span_ids + + +def _capture_boto3_spans_by_op( + invoke_client_method, + capture_items, + span_streaming, + expected_origin=ORIGIN, +): + 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(SPANDATA.SENTRY_ORIGIN) + == expected_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"] == expected_origin + ] + + spans_by_op = {} + for span in spans: + op = ( + span["attributes"].get(SPANDATA.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_attributes(spans[0], span_streaming)[SPANDATA.ERROR_TYPE] + _assert_span_finished(spans[0], span_streaming) + + +def _capture_stubbed_client_span( + client, + method_name, + api_params, + capture_items, + span_streaming, + response=None, +): + with Stubber(client) as stubber: + stubber.add_response( + method_name, response if response is not None else {}, api_params + ) + spans_by_op = _capture_boto3_spans_by_op( + lambda: getattr(client, method_name)(**api_params), + capture_items, + span_streaming, + ) + + client_spans = spans_by_op.get(OP.HTTP_CLIENT, []) + assert len(client_spans) == 1 + return client_spans[0] + + +def _span_attributes(span, span_streaming): + return span["attributes"] if span_streaming else span["data"] + + +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_service_extension_customizes_client_span( + capture_items, + client_factory, + monkeypatch, + span_streaming, +): + class TestServiceExtension(_ServiceExtension): + def get_span_op(self, ctx): + return "aws.test" + + def get_span_origin(self, ctx): + return "auto.aws.test" + + def get_request_attributes(self, ctx): + return { + "aws.test.request": ctx.params["Key"], + SPANDATA.SENTRY_KIND: "producer", + } + + def get_response_attributes(self, ctx, response): + return { + "aws.test.response": response["ResponseMetadata"]["RequestId"], + SPANDATA.HTTP_STATUS_CODE: 418, + } + + monkeypatch.setitem(_SERVICE_EXTENSIONS, "s3", TestServiceExtension()) + client = client_factory() + api_params = {"Bucket": "bucket", "Key": "foo"} + + with Stubber(client) as stubber: + stubber.add_response( + "head_object", + { + "ResponseMetadata": { + "HTTPStatusCode": 200, + "RequestId": "request-id", + "HostId": "extended-request-id", + } + }, + api_params, + ) + spans_by_op = _capture_boto3_spans_by_op( + lambda: client.head_object(**api_params), + capture_items, + span_streaming, + expected_origin="auto.aws.test", + ) + + spans = spans_by_op.get("aws.test", []) + assert len(spans) == 1 + attributes = _span_attributes(spans[0], span_streaming) + assert attributes["aws.test.request"] == "foo" + assert attributes["aws.test.response"] == "request-id" + assert attributes[SPANDATA.SENTRY_KIND] == "producer" + assert attributes[SPANDATA.RPC_METHOD] == "HeadObject" + assert attributes[SPANDATA.HTTP_STATUS_CODE] == 200 + assert attributes[SPANDATA.AWS_EXTENDED_REQUEST_ID] == "extended-request-id" + _assert_span_finished(spans[0], span_streaming) + if span_streaming: + assert attributes[SPANDATA.SENTRY_ORIGIN] == "auto.aws.test" + else: + assert spans[0]["origin"] == "auto.aws.test" + + +@pytest.mark.parametrize( + ( + "service_name", + "method_name", + "api_params", + "span_name", + "rpc_service", + "rpc_method", + "endpoint_url", + "server_address", + "server_port", + ), + [ + ( + "s3", + "head_object", + {"Bucket": "bucket", "Key": "foo"}, + "aws.s3.HeadObject", + "S3", + "HeadObject", + "http://localhost:4566", + "localhost", + 4566, + ), + ( + "events", + "list_event_buses", + {}, + "aws.eventbridge.ListEventBuses", + "EventBridge", + "ListEventBuses", + None, + "events.eu-north-1.amazonaws.com", + 443, + ), + ], +) +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_client_call_has_common_attributes( + capture_items, + client_factory, + span_streaming, + service_name, + method_name, + api_params, + span_name, + rpc_service, + rpc_method, + endpoint_url, + server_address, + server_port, +): + client = client_factory(service_name=service_name, endpoint_url=endpoint_url) + span = _capture_stubbed_client_span( + client, + method_name, + api_params, + capture_items, + span_streaming, + response={ + "ResponseMetadata": { + "HTTPStatusCode": 200, + "RequestId": "request-id", + "RetryAttempts": 0, + } + }, + ) + attributes = _span_attributes(span, span_streaming) + + assert span["name" if span_streaming else "description"] == span_name + assert attributes[SPANDATA.RPC_SERVICE] == rpc_service + assert attributes[SPANDATA.RPC_METHOD] == rpc_method + assert attributes[SPANDATA.RPC_SYSTEM_NAME] == AWS_RPC_SYSTEM_NAME + assert attributes[SPANDATA.SENTRY_KIND] == "client" + assert attributes[SPANDATA.CLOUD_REGION] == "eu-north-1" + assert attributes[SPANDATA.SERVER_ADDRESS] == server_address + assert attributes[SPANDATA.SERVER_PORT] == server_port + assert attributes[SPANDATA.HTTP_STATUS_CODE] == 200 + assert attributes[SPANDATA.AWS_REQUEST_ID] == "request-id" + assert SPANDATA.HTTP_REQUEST_RESEND_COUNT not in attributes + assert SPANDATA.ERROR_TYPE not in attributes + _assert_span_finished(span, span_streaming) + + +def test_client_call_attributes_are_available_at_span_creation( + sentry_init, capture_items +): + # attribute-based filtering happens during span creation, at the same boundary + # where creation attributes are made available for sampling decisions. + sentry_init( + traces_sample_rate=1.0, + integrations=[Boto3Integration()], + trace_lifecycle="stream", + ignore_spans=[ + { + "attributes": { + SPANDATA.RPC_METHOD: "HeadObject", + SPANDATA.RPC_SERVICE: "S3", + SPANDATA.RPC_SYSTEM_NAME: AWS_RPC_SYSTEM_NAME, + SPANDATA.SERVER_ADDRESS: "s3.eu-north-1.amazonaws.com", + SPANDATA.SERVER_PORT: 443, + } + } + ], + ) + client = session.client("s3") + items = capture_items("span") + + with MockResponse(client, 200, {}, b""): + with sentry_sdk.traces.start_span(name="parent") as parent: + response = client.head_object(Bucket="bucket", Key="foo") + assert response["ResponseMetadata"]["HTTPStatusCode"] == 200 + assert sentry_sdk.traces.get_current_span() is parent + assert SPANDATA.RPC_METHOD not in parent.get_attributes() + assert SPANDATA.HTTP_REQUEST_METHOD not in parent.get_attributes() + + sentry_sdk.flush() + client_spans = [ + item.payload + for item in items + if item.payload["attributes"].get(SPANDATA.SENTRY_ORIGIN) == ORIGIN + ] + assert client_spans == [] + + +@pytest.mark.parametrize("span_streaming", [True, False]) +@pytest.mark.parametrize( + "request_id_header", ["x-amzn-requestid", "x-amzn-request-id", "x-amz-request-id"] +) +def test_client_call_has_response_header_attributes( + capture_items, client_factory, span_streaming, request_id_header +): + client = client_factory() + headers = {request_id_header: "request-id", "x-amz-id-2": "extended-request-id"} + with MockResponse(client, 200, headers, b""): + spans_by_op = _capture_boto3_spans_by_op( + lambda: client.head_object(Bucket="bucket", Key="foo"), + capture_items, + span_streaming, + ) + + spans = spans_by_op[OP.HTTP_CLIENT] + assert len(spans) == 1 + attributes = _span_attributes(spans[0], span_streaming) + assert attributes[SPANDATA.HTTP_STATUS_CODE] == 200 + assert attributes[SPANDATA.AWS_REQUEST_ID] == "request-id" + assert attributes[SPANDATA.AWS_EXTENDED_REQUEST_ID] == "extended-request-id" + assert SPANDATA.HTTP_REQUEST_RESEND_COUNT not in attributes + + +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_retry_attempts_share_one_client_span( + capture_items, + client_factory, + span_streaming, +): + attempt_count = 3 + client = client_factory(attempt_count=attempt_count) + request_span_ids = _mock_responses(client, [500] * (attempt_count - 1) + [200]) + + spans_by_op = _capture_boto3_spans_by_op( + lambda: client.head_object(Bucket="bucket", Key="foo"), + capture_items, + span_streaming, + ) + client_spans = spans_by_op.get(OP.HTTP_CLIENT, []) + + assert len(request_span_ids) == attempt_count + # all `AWSRequest` instances created during retries reference the same client span. + assert len(set(request_span_ids)) == 1 + assert len(client_spans) == 1 + attributes = _span_attributes(client_spans[0], span_streaming) + assert attributes[SPANDATA.HTTP_REQUEST_RESEND_COUNT] == attempt_count - 1 + + +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_retries_exhausted_has_one_failed_client_span( + capture_items, + client_factory, + span_streaming, +): + client = client_factory(attempt_count=2) + request_span_ids = _mock_responses(client, [500]) + + def attempt_failed_head_object_call(): + with pytest.raises(ClientError): + client.head_object(Bucket="bucket", Key="foo.pdf") + + spans_by_op = _capture_boto3_spans_by_op( + attempt_failed_head_object_call, capture_items, span_streaming + ) + client_spans = spans_by_op.get(OP.HTTP_CLIENT, []) + + assert len(request_span_ids) == 2 + assert len(set(request_span_ids)) == 1 + _assert_one_failed_span(client_spans, span_streaming) + attributes = _span_attributes(client_spans[0], span_streaming) + assert attributes[SPANDATA.HTTP_STATUS_CODE] == 500 + assert attributes[SPANDATA.HTTP_REQUEST_RESEND_COUNT] == 1 + + +@pytest.mark.parametrize("span_streaming", [True, False]) +@pytest.mark.parametrize("with_service_extension", [False, True]) +def test_client_error_has_response_attributes_and_is_unchanged( + capture_items, + client_factory, + monkeypatch, + span_streaming, + with_service_extension, +): + class TestServiceExtension(_ServiceExtension): + def get_response_attributes(self, ctx, response): + return { + "aws.test.error": response["Error"]["Code"], + SPANDATA.ERROR_TYPE: "must-not-override", + SPANDATA.HTTP_STATUS_CODE: 418, + } + + if with_service_extension: + monkeypatch.setitem(_SERVICE_EXTENSIONS, "s3", TestServiceExtension()) + client = client_factory() + original_exception = ClientError( + { + "Error": { + "Code": "AccessDeniedException", + "Message": "must not become a span attribute", + }, + "ResponseMetadata": { + "RequestId": "request-id", + "HTTPStatusCode": 403, + "RetryAttempts": 1, + }, + }, + "HeadObject", + ) + + def raise_client_error(**kwargs): + raise original_exception + + client.meta.events.register("before-parameter-build", raise_client_error) + + def invoke_failing_client_method(): + with pytest.raises(ClientError) as exc_info: + client.head_object(Bucket="bucket", Key="foo") + assert exc_info.value is original_exception + + spans_by_op = _capture_boto3_spans_by_op( + invoke_failing_client_method, capture_items, span_streaming + ) + client_spans = spans_by_op.get(OP.HTTP_CLIENT, []) + _assert_one_failed_span(client_spans, span_streaming) + attributes = _span_attributes(client_spans[0], span_streaming) + + assert attributes[SPANDATA.AWS_REQUEST_ID] == "request-id" + assert attributes[SPANDATA.HTTP_STATUS_CODE] == 403 + assert attributes[SPANDATA.HTTP_REQUEST_RESEND_COUNT] == 1 + assert attributes[SPANDATA.ERROR_TYPE] == "AccessDeniedException" + if with_service_extension: + assert attributes["aws.test.error"] == "AccessDeniedException" + assert "Error.Message" not in attributes + assert "exception.message" not in attributes + assert "error.message" not in attributes + + +@pytest.mark.parametrize( + "event_name", + [ + pytest.param("before-parameter-build"), + pytest.param("before-send"), + ], +) +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_client_call_exception_is_unchanged_and_finishes_span( + capture_items, + client_factory, + span_streaming, + event_name, +): + client = client_factory() + if event_name == "before-send": + original_exception = EndpointConnectionError( + endpoint_url="https://s3.eu-north-1.amazonaws.com" + ) + else: + original_exception = ValueError("parameter processing failed") + + def raise_original_exception(**kwargs): + raise original_exception + + client.meta.events.register(event_name, raise_original_exception) + + def invoke_failing_client_method(): + with pytest.raises(type(original_exception)) as exc_info: + client.head_object(Bucket="bucket", Key="foo") + assert exc_info.value is original_exception + + spans_by_op = _capture_boto3_spans_by_op( + invoke_failing_client_method, capture_items, span_streaming + ) + client_spans = spans_by_op.get(OP.HTTP_CLIENT, []) + _assert_one_failed_span(client_spans, span_streaming) + + attributes = _span_attributes(client_spans[0], span_streaming) + expected_error_type = ( + "botocore.exceptions.EndpointConnectionError" + if event_name == "before-send" + else "ValueError" + ) + assert attributes[SPANDATA.ERROR_TYPE] == expected_error_type + + +@pytest.mark.tests_internal_exceptions +@pytest.mark.parametrize("span_streaming", [True, False]) +@pytest.mark.parametrize( + "failing_instrumentation", + [ + "_start_client_span", + "_get_response_attributes", + ], +) +def test_instrumentation_failure_does_not_change_response( + capture_items, + client_factory, + monkeypatch, + span_streaming, + failing_instrumentation, +): + client = client_factory() + api_params = {"Bucket": "bucket", "Key": "foo"} + original_response = {"ResponseMetadata": {"HTTPStatusCode": 200}} + returned_responses = [] + + def fail_instrumentation(*args, **kwargs): + raise RuntimeError("instrumentation failed") + + monkeypatch.setattr( + f"sentry_sdk.integrations.boto3._client.{failing_instrumentation}", + fail_instrumentation, + ) + + def invoke_client_method(): + returned_responses.append(client.head_object(**api_params)) + + with Stubber(client) as stubber: + stubber.add_response("head_object", original_response, api_params) + spans_by_op = _capture_boto3_spans_by_op( + invoke_client_method, capture_items, span_streaming + ) + + client_spans = spans_by_op.get(OP.HTTP_CLIENT, []) + assert returned_responses == [original_response] + assert returned_responses[0] is original_response + if failing_instrumentation == "_get_response_attributes": + assert len(client_spans) == 1 + _assert_span_finished(client_spans[0], span_streaming) + else: + assert client_spans == [] + + +@pytest.mark.tests_internal_exceptions +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_error_attribute_extraction_failure_does_not_replace_original_exception( + capture_items, + client_factory, + monkeypatch, + span_streaming, +): + client = client_factory() + original_exception = ValueError("parameter processing failed") + + def raise_original_exception(**kwargs): + raise original_exception + + def fail_attribute_extraction(exception): + raise RuntimeError("attribute extraction failed") + + client.meta.events.register("before-parameter-build", raise_original_exception) + monkeypatch.setattr( + "sentry_sdk.integrations.boto3._client._get_error_attributes", + fail_attribute_extraction, + ) + + def invoke_failing_client_method(): + with pytest.raises(ValueError) as exc_info: + client.head_object(Bucket="bucket", Key="foo") + assert exc_info.value is original_exception + + spans_by_op = _capture_boto3_spans_by_op( + invoke_failing_client_method, capture_items, span_streaming + ) + client_spans = spans_by_op.get(OP.HTTP_CLIENT, []) + + assert len(client_spans) == 1 + assert client_spans[0]["status"] in ("error", "internal_error") + _assert_span_finished(client_spans[0], span_streaming) + + +@pytest.mark.parametrize("span_streaming", [True, False]) +def test_streaming_response_attributes_belong_to_client_span( + capture_items, + client_factory, + span_streaming, +): + client = client_factory() + + def respond(request, **kwargs): + return AWSResponse( + request.url, + 200, + { + "content-length": "5", + "x-amz-request-id": "request-id", + }, + Body(b"hello"), + ) + + client.meta.events.register("before-send", respond) + + def invoke_client_method_and_read_body(): + body = client.get_object(Bucket="bucket", Key="foo")["Body"] + assert body.read() == b"hello" + assert body.read() == b"" + + 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 len(stream_spans) == 1 + client_attributes = _span_attributes(client_spans[0], span_streaming) + stream_attributes = _span_attributes(stream_spans[0], span_streaming) + assert client_attributes[SPANDATA.AWS_REQUEST_ID] == "request-id" + assert client_attributes[SPANDATA.HTTP_STATUS_CODE] == 200 + assert SPANDATA.HTTP_REQUEST_RESEND_COUNT not in client_attributes + assert SPANDATA.AWS_REQUEST_ID not in stream_attributes + assert SPANDATA.HTTP_STATUS_CODE not in stream_attributes + + +@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): + # urllib3 closes the response before propagating some read failures. + self.close() + 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(client_spans, span_streaming) + _assert_one_failed_span(stream_spans, span_streaming) + attributes = _span_attributes(stream_spans[0], span_streaming) + assert attributes[SPANDATA.ERROR_TYPE] == "OSError" diff --git a/tests/integrations/boto3/test_s3.py b/tests/integrations/boto3/test_s3.py index 888e44ee4c..4af2e55ee2 100644 --- a/tests/integrations/boto3/test_s3.py +++ b/tests/integrations/boto3/test_s3.py @@ -7,6 +7,7 @@ from sentry_sdk import capture_message from sentry_sdk.consts import SPANDATA from sentry_sdk.integrations.boto3 import Boto3Integration +from sentry_sdk.integrations.boto3.consts import ORIGIN from tests.conftest import ApproxDict from tests.integrations.boto3 import read_fixture from tests.integrations.boto3.aws_mock import MockResponse @@ -110,16 +111,27 @@ def test_streaming( spans = [item.payload for item in items] assert len(spans) == 3 - span1 = spans[0] - assert span1["attributes"]["sentry.op"] == "http.client" - assert span1["name"] == "aws.s3.GetObject" + stream_span, client_span, parent_span = spans + assert stream_span["attributes"]["sentry.op"] == "http.client.stream" + assert stream_span["name"] == "aws.s3.GetObject" + assert stream_span["parent_span_id"] == client_span["span_id"] + + assert client_span["attributes"]["sentry.op"] == "http.client" + assert client_span["name"] == "aws.s3.GetObject" + assert client_span["parent_span_id"] == parent_span["span_id"] + + assert parent_span["name"] == "custom parent" + assert parent_span["start_timestamp"] <= client_span["start_timestamp"] + assert client_span["start_timestamp"] <= stream_span["start_timestamp"] + assert stream_span["end_timestamp"] <= client_span["end_timestamp"] expected_attrs = { "http.request.method": "GET", - "rpc.method": "S3/GetObject", + "rpc.method": "GetObject", + "rpc.service": "S3", "sentry.environment": "production", "sentry.op": "http.client", - "sentry.origin": "auto.http.boto3", + "sentry.origin": ORIGIN, "sentry.release": mock.ANY, "sentry.sdk.name": "sentry.python", "sentry.sdk.version": mock.ANY, @@ -131,17 +143,12 @@ def test_streaming( } if send_default_pii: expected_attrs["url.full"] = "https://bucket.s3.amazonaws.com/foo.pdf" - assert span1["attributes"] == ApproxDict(expected_attrs) + assert client_span["attributes"] == ApproxDict(expected_attrs) - assert "url.fragment" not in span1["attributes"] - assert "url.query" not in span1["attributes"] + assert "url.fragment" not in client_span["attributes"] + assert "url.query" not in client_span["attributes"] if not send_default_pii: - assert "url.full" not in span1["attributes"] - - span2 = spans[1] - assert span2["attributes"]["sentry.op"] == "http.client.stream" - assert span2["name"] == "aws.s3.GetObject" - assert span2["parent_span_id"] == span1["span_id"] + assert "url.full" not in client_span["attributes"] else: events = capture_events() @@ -207,10 +214,20 @@ def test_streaming_close( sentry_sdk.flush() spans = [item.payload for item in items] assert len(spans) == 3 - span1 = spans[0] - assert span1["attributes"]["sentry.op"] == "http.client" - span2 = spans[1] - assert span2["attributes"]["sentry.op"] == "http.client.stream" + + stream_span, client_span, parent_span = spans + assert stream_span["attributes"]["sentry.op"] == "http.client.stream" + assert stream_span["name"] == "aws.s3.GetObject" + assert stream_span["parent_span_id"] == client_span["span_id"] + + assert client_span["attributes"]["sentry.op"] == "http.client" + assert client_span["name"] == "aws.s3.GetObject" + assert client_span["parent_span_id"] == parent_span["span_id"] + + assert parent_span["name"] == "custom parent" + assert parent_span["start_timestamp"] <= client_span["start_timestamp"] + assert client_span["start_timestamp"] <= stream_span["start_timestamp"] + assert stream_span["end_timestamp"] <= client_span["end_timestamp"] else: events = capture_events() @@ -253,7 +270,7 @@ def test_omit_url_data_if_parsing_fails( items = capture_items("span") with mock.patch( - "sentry_sdk.integrations.boto3.parse_url", + "sentry_sdk.integrations.boto3._instrumentation.parse_url", side_effect=ValueError, ): with sentry_sdk.traces.start_span( @@ -272,10 +289,11 @@ def test_omit_url_data_if_parsing_fails( assert spans[0]["attributes"] == ApproxDict( { "http.request.method": "GET", - "rpc.method": "S3/ListObjects", + "rpc.method": "ListObjects", + "rpc.service": "S3", "sentry.environment": "production", "sentry.op": "http.client", - "sentry.origin": "auto.http.boto3", + "sentry.origin": ORIGIN, "sentry.release": mock.ANY, "sentry.sdk.name": "sentry.python", "sentry.sdk.version": mock.ANY, @@ -294,7 +312,7 @@ def test_omit_url_data_if_parsing_fails( events = capture_events() with mock.patch( - "sentry_sdk.integrations.boto3.parse_url", + "sentry_sdk.integrations.boto3._instrumentation.parse_url", side_effect=ValueError, ): with sentry_sdk.start_transaction() as transaction, MockResponse( @@ -347,7 +365,7 @@ def test_span_origin( spans = [item.payload for item in items] assert spans[1]["attributes"]["sentry.origin"] == "manual" - assert spans[0]["attributes"]["sentry.origin"] == "auto.http.boto3" + assert spans[0]["attributes"]["sentry.origin"] == ORIGIN else: events = capture_events() @@ -359,7 +377,7 @@ def test_span_origin( (event,) = events assert event["contexts"]["trace"]["origin"] == "manual" - assert event["spans"][0]["origin"] == "auto.http.boto3" + assert event["spans"][0]["origin"] == ORIGIN def test_breadcrumb(sentry_init, capture_events):