diff --git a/sentry_sdk/consts.py b/sentry_sdk/consts.py index f7a673f1bc..8cda281a25 100644 --- a/sentry_sdk/consts.py +++ b/sentry_sdk/consts.py @@ -471,6 +471,12 @@ class SPANDATA: Example: "db-password" """ + AWS_SQS_QUEUE_URL = "aws.sqs.queue.url" + """ + The URL of the AWS SQS queue the request refers to. + Example: "https://sqs.us-east-1.amazonaws.com/123456789012/orders" + """ + CACHE_HIT = "cache.hit" """ A boolean indicating whether the requested data was found in the cache. @@ -1090,6 +1096,18 @@ class SPANDATA: Number of retries/attempts to process a message. """ + MESSAGING_OPERATION_NAME = "messaging.operation.name" + """ + The system-specific name of the messaging operation. + Example: "publish" + """ + + MESSAGING_OPERATION_TYPE = "messaging.operation.type" + """ + The type of the messaging operation: `create`, `send`, `receive`, `process` or `settle`. + Example: "send" + """ + MESSAGING_SYSTEM = "messaging.system" """ The messaging system's name, e.g. `kafka`, `aws_sqs` @@ -1489,6 +1507,8 @@ class OP: OBJECT_UPLOAD_PART = "object.upload_part" QUEUE_PROCESS = "queue.process" QUEUE_PUBLISH = "queue.publish" + QUEUE_RECEIVE = "queue.receive" + QUEUE_SETTLE = "queue.settle" QUEUE_SUBMIT_ARQ = "queue.submit.arq" QUEUE_TASK_ARQ = "queue.task.arq" QUEUE_SUBMIT_CELERY = "queue.submit.celery" diff --git a/sentry_sdk/integrations/boto3/_instrumentation.py b/sentry_sdk/integrations/boto3/_instrumentation.py index 731dac6fa3..8b7e34da87 100644 --- a/sentry_sdk/integrations/boto3/_instrumentation.py +++ b/sentry_sdk/integrations/boto3/_instrumentation.py @@ -180,6 +180,11 @@ def _start_client_span( with capture_internal_exceptions(): attributes.update(service_ext.get_request_attributes(ctx)) + with capture_internal_exceptions(): + service_span_name = service_ext.get_span_name(ctx, attributes) + if service_span_name is not None: + span_name = service_span_name + # `start_span()` evaluates `ignore_spans` against the initial attributes. # https://opentelemetry.io/docs/specs/semconv/rpc/rpc-spans/#rpc-client-span attributes.update( diff --git a/sentry_sdk/integrations/boto3/_services/base.py b/sentry_sdk/integrations/boto3/_services/base.py index ec7ac1df1c..397499bffb 100644 --- a/sentry_sdk/integrations/boto3/_services/base.py +++ b/sentry_sdk/integrations/boto3/_services/base.py @@ -21,6 +21,12 @@ 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_name( + self, ctx: "AwsCallContext", attributes: "Attributes" + ) -> "Optional[str]": + """Return an optional span name 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 {} diff --git a/sentry_sdk/integrations/boto3/_services/sqs.py b/sentry_sdk/integrations/boto3/_services/sqs.py index e9ae02142e..274f6d21dc 100644 --- a/sentry_sdk/integrations/boto3/_services/sqs.py +++ b/sentry_sdk/integrations/boto3/_services/sqs.py @@ -2,36 +2,110 @@ from typing import TYPE_CHECKING from urllib.parse import unquote, urlsplit -from sentry_sdk.consts import SPANDATA +from sentry_sdk.consts import OP, SPANDATA from sentry_sdk.integrations.boto3._services.base import _ServiceExtension from sentry_sdk.integrations.boto3._utils import _get_aws_arn_attributes -from sentry_sdk.utils import capture_internal_exceptions if TYPE_CHECKING: + from typing import Any, Optional + from sentry_sdk._types import Attributes from sentry_sdk.integrations.boto3._context import AwsCallContext +# operation -> (`sentry.op`, `messaging.operation.type`) +_MESSAGING_OPERATIONS = { + "SendMessage": (OP.QUEUE_PUBLISH, "send"), + "SendMessageBatch": (OP.QUEUE_PUBLISH, "send"), + "ReceiveMessage": (OP.QUEUE_RECEIVE, "receive"), + "DeleteMessage": (OP.QUEUE_SETTLE, "settle"), + "DeleteMessageBatch": (OP.QUEUE_SETTLE, "settle"), +} + class _SQSExtension(_ServiceExtension): + def get_span_op(self, ctx: "AwsCallContext") -> "Optional[str]": + if ctx.operation_name in _MESSAGING_OPERATIONS: + return _MESSAGING_OPERATIONS[ctx.operation_name][0] + return None + + def get_span_name( + self, ctx: "AwsCallContext", attributes: "Attributes" + ) -> "Optional[str]": + if ctx.operation_name not in _MESSAGING_OPERATIONS: + return None + # for example: `send ` when `messaging.operation.type=send`. + # https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#span-name + return f"{attributes[SPANDATA.MESSAGING_OPERATION_TYPE]} {attributes[SPANDATA.MESSAGING_DESTINATION_NAME]}" + def get_request_attributes(self, ctx: "AwsCallContext") -> "Attributes": - attributes: "Attributes" = {} - with capture_internal_exceptions(): + attributes = _get_aws_arn_attributes( + (ctx.params.get("DestinationArn"), ctx.params.get("SourceArn")), ("sqs",) + ) + + queue_url = ctx.params.get("QueueUrl") + if queue_url is not None: + attributes[SPANDATA.AWS_SQS_QUEUE_URL] = queue_url + + # `https://sqs.eu-north-1.amazonaws.com/123456789012/orders` => ['123456789012', 'orders'] + path = unquote(urlsplit(queue_url).path).strip("/").split("/") + + attributes[SPANDATA.MESSAGING_DESTINATION_NAME] = path[-1] + if re.fullmatch(r"[0-9]{12}", path[0]): + attributes[SPANDATA.CLOUD_ACCOUNT_ID] = path[0] + elif "QueueName" in ctx.params: + # `CreateQueue` & `GetQueueUrl` operations. + attributes[SPANDATA.MESSAGING_DESTINATION_NAME] = ctx.params["QueueName"] + + if "QueueOwnerAWSAccountId" in ctx.params: + attributes[SPANDATA.CLOUD_ACCOUNT_ID] = ctx.params[ + "QueueOwnerAWSAccountId" + ] + elif "SourceArn" in ctx.params: + # `arn:aws:sqs:eu-north-1:123456789012:orders` => `orders` + attributes[SPANDATA.MESSAGING_DESTINATION_NAME] = ctx.params[ + "SourceArn" + ].split(":")[-1] + + if ctx.operation_name in _MESSAGING_OPERATIONS: + operation_type = _MESSAGING_OPERATIONS[ctx.operation_name][1] attributes.update( - _get_aws_arn_attributes( - ( - ctx.params.get("DestinationArn"), - ctx.params.get("SourceArn"), - ), - ("sqs",), - ) + { + SPANDATA.MESSAGING_SYSTEM: "aws_sqs", + SPANDATA.MESSAGING_OPERATION_TYPE: operation_type, + SPANDATA.MESSAGING_OPERATION_NAME: operation_type, + } ) + if operation_type == "send": + attributes[SPANDATA.SENTRY_KIND] = "producer" + if "Entries" in ctx.params: + attributes[SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT] = len( + ctx.params["Entries"] + ) + return attributes - queue_url = ctx.params.get("QueueUrl") - if queue_url is None: + def get_response_attributes( + self, ctx: "AwsCallContext", response: "Any" + ) -> "Attributes": + attributes: "Attributes" = {} + + # `ClientError` response holds no results. + if "Error" in response: return attributes - with capture_internal_exceptions(): - account_id = unquote(urlsplit(queue_url).path).strip("/").split("/", 1)[0] - if re.fullmatch(r"^[0-9]{12}$", account_id): - attributes[SPANDATA.CLOUD_ACCOUNT_ID] = account_id + if ctx.operation_name == "SendMessage": + attributes[SPANDATA.MESSAGING_MESSAGE_ID] = response["MessageId"] + elif ctx.operation_name == "ReceiveMessage": + # `Messages` is left out when no message was received. + messages = response.get("Messages", ()) + attributes[SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT] = len(messages) + if len(messages) == 1: + attributes[SPANDATA.MESSAGING_MESSAGE_ID] = messages[0]["MessageId"] + + # failed entries have a `200 OK` response; use their code when they share one. + # https://docs.aws.amazon.com/sns/latest/api/API_BatchResultErrorEntry.html + error_codes = {entry["Code"] for entry in response.get("Failed", ())} + if len(error_codes) == 1: + attributes[SPANDATA.ERROR_TYPE] = error_codes.pop() + elif error_codes: + attributes[SPANDATA.ERROR_TYPE] = "aws_batch_partial_failure" return attributes diff --git a/tests/integrations/boto3/test_sqs.py b/tests/integrations/boto3/test_sqs.py index ecc81c69d9..f28f8cb74b 100644 --- a/tests/integrations/boto3/test_sqs.py +++ b/tests/integrations/boto3/test_sqs.py @@ -1,4 +1,5 @@ import pytest +from botocore.exceptions import ClientError from botocore.stub import Stubber from sentry_sdk.consts import OP, SPANDATA @@ -8,24 +9,65 @@ ) from tests.integrations.boto3.helpers import client_factory as client_factory +_BATCH_ENTRIES = [ + {"Id": "1", "MessageBody": "hi"}, + {"Id": "2", "MessageBody": "hi"}, +] -@pytest.mark.parametrize("source", ["arn", "queue-url"]) -def test_request_attributes(client_factory, capture_items, source): - client = client_factory("sqs") - if source == "arn": - require_botocore_model_fields(client, "start_message_move_task") - resource_id = "arn:aws:sqs:eu-north-1:123456789012:orders-dlq" - method = "start_message_move_task" - params = {"SourceArn": resource_id} - response = {"TaskHandle": "task-handle"} - else: - resource_id = None - method = "delete_queue" - params = { - "QueueUrl": "https://sqs.eu-north-1.amazonaws.com/123456789012/orders" - } - response = {} +@pytest.mark.parametrize( + "method,params,response,queue_url,account_id,resource_id", + [ + pytest.param( + "delete_queue", + {"QueueUrl": "https://sqs.eu-north-1.amazonaws.com/123456789012/orders"}, + {}, + "https://sqs.eu-north-1.amazonaws.com/123456789012/orders", + "123456789012", + None, + id="queue-url", + ), + pytest.param( + "create_queue", + {"QueueName": "orders"}, + {"QueueUrl": "https://sqs.eu-north-1.amazonaws.com/123456789012/orders"}, + None, + None, + None, + id="create-queue", + ), + pytest.param( + "get_queue_url", + {"QueueName": "orders", "QueueOwnerAWSAccountId": "123456789012"}, + {"QueueUrl": "https://sqs.eu-north-1.amazonaws.com/123456789012/orders"}, + None, + "123456789012", + None, + id="get-queue-url", + ), + pytest.param( + "start_message_move_task", + {"SourceArn": "arn:aws:sqs:eu-north-1:123456789012:orders"}, + {"TaskHandle": "task-handle"}, + None, + "123456789012", + "arn:aws:sqs:eu-north-1:123456789012:orders", + id="source-arn", + ), + ], +) +def test_request_attributes( + client_factory, + capture_items, + method, + params, + response, + queue_url, + account_id, + resource_id, +): + client = client_factory("sqs") + require_botocore_model_fields(client, method) with Stubber(client) as stubber: stubber.add_response(method, response, params) @@ -35,8 +77,232 @@ def test_request_attributes(client_factory, capture_items, source): (span,) = spans[OP.RPC] attributes = span["attributes"] - assert attributes[SPANDATA.CLOUD_ACCOUNT_ID] == "123456789012" - if resource_id is None: - assert SPANDATA.CLOUD_RESOURCE_ID not in attributes - else: - assert attributes[SPANDATA.CLOUD_RESOURCE_ID] == resource_id + assert attributes[SPANDATA.MESSAGING_DESTINATION_NAME] == "orders" + assert attributes.get(SPANDATA.AWS_SQS_QUEUE_URL) == queue_url + assert attributes.get(SPANDATA.CLOUD_ACCOUNT_ID) == account_id + assert attributes.get(SPANDATA.CLOUD_RESOURCE_ID) == resource_id + assert SPANDATA.MESSAGING_SYSTEM not in attributes + + +@pytest.mark.parametrize( + "method,params,response,message_count,message_id,error_type", + [ + pytest.param( + "send_message", + {"MessageBody": "hi"}, + {"MessageId": "5fea7756-0ea4-451a-a703-a558b933e274"}, + None, + "5fea7756-0ea4-451a-a703-a558b933e274", + None, + id="single", + ), + pytest.param( + "send_message_batch", + {"Entries": _BATCH_ENTRIES}, + { + "Successful": [ + { + "Id": "1", + "MessageId": "message-1", + "MD5OfMessageBody": "49f68a5c8493ec2c0bf489821c21fc3b", + }, + { + "Id": "2", + "MessageId": "message-2", + "MD5OfMessageBody": "49f68a5c8493ec2c0bf489821c21fc3b", + }, + ], + "Failed": [], + }, + 2, + None, + None, + id="batch", + ), + pytest.param( + "send_message_batch", + {"Entries": _BATCH_ENTRIES}, + { + "Successful": [ + { + "Id": "1", + "MessageId": "message-1", + "MD5OfMessageBody": "49f68a5c8493ec2c0bf489821c21fc3b", + } + ], + "Failed": [{"Id": "2", "Code": "InternalError", "SenderFault": False}], + }, + 2, + None, + "InternalError", + id="batch-shared-error-code", + ), + pytest.param( + "send_message_batch", + {"Entries": _BATCH_ENTRIES}, + { + "Successful": [], + "Failed": [ + {"Id": "1", "Code": "InternalError", "SenderFault": False}, + {"Id": "2", "Code": "InvalidParameterValue", "SenderFault": True}, + ], + }, + 2, + None, + "aws_batch_partial_failure", + id="batch-mixed-error-codes", + ), + ], +) +def test_send_message( + client_factory, + capture_items, + method, + params, + response, + message_count, + message_id, + error_type, +): + client = client_factory("sqs") + + with Stubber(client) as stubber: + stubber.add_response(method, response) + spans = capture_spans_by_op( + lambda: getattr(client, method)( + QueueUrl="https://sqs.eu-north-1.amazonaws.com/123456789012/orders", + **params, + ), + capture_items, + ) + + (span,) = spans[OP.QUEUE_PUBLISH] + assert span["name"] == "send orders" + assert span["status"] == ("ok" if error_type is None else "error") + attributes = span["attributes"] + assert attributes[SPANDATA.SENTRY_KIND] == "producer" + assert attributes[SPANDATA.MESSAGING_OPERATION_TYPE] == "send" + assert attributes[SPANDATA.MESSAGING_OPERATION_NAME] == "send" + assert attributes.get(SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT) == message_count + assert attributes.get(SPANDATA.MESSAGING_MESSAGE_ID) == message_id + assert attributes.get(SPANDATA.ERROR_TYPE) == error_type + + +@pytest.mark.parametrize( + "response,message_count,message_id", + [ + pytest.param({}, 0, None, id="empty"), + pytest.param( + {"Messages": [{"MessageId": "1", "ReceiptHandle": "receipt-handle-1"}]}, + 1, + "1", + id="single-message", + ), + pytest.param( + { + "Messages": [ + {"MessageId": "1", "ReceiptHandle": "receipt-handle-1"}, + {"MessageId": "2", "ReceiptHandle": "receipt-handle-2"}, + ] + }, + 2, + None, + id="messages", + ), + ], +) +def test_receive_message( + client_factory, capture_items, response, message_count, message_id +): + client = client_factory("sqs") + + with Stubber(client) as stubber: + stubber.add_response("receive_message", response) + spans = capture_spans_by_op( + lambda: client.receive_message( + QueueUrl="https://sqs.eu-north-1.amazonaws.com/123456789012/orders", + MaxNumberOfMessages=10, + ), + capture_items, + ) + + (span,) = spans[OP.QUEUE_RECEIVE] + assert span["name"] == "receive orders" + attributes = span["attributes"] + assert attributes[SPANDATA.SENTRY_KIND] == "client" + assert attributes[SPANDATA.MESSAGING_OPERATION_TYPE] == "receive" + assert attributes[SPANDATA.MESSAGING_OPERATION_NAME] == "receive" + assert attributes[SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT] == message_count + assert attributes.get(SPANDATA.MESSAGING_MESSAGE_ID) == message_id + assert "receipt-handle" not in str(span) + + +def test_receive_message_error(client_factory, capture_items): + client = client_factory("sqs") + + with Stubber(client) as stubber: + stubber.add_client_error( + "receive_message", + service_error_code="AWS.SimpleQueueService.NonExistentQueue", + http_status_code=400, + ) + + def call(): + with pytest.raises(ClientError): + client.receive_message( + QueueUrl="https://sqs.eu-north-1.amazonaws.com/123456789012/orders" + ) + + spans = capture_spans_by_op(call, capture_items) + + (span,) = spans[OP.QUEUE_RECEIVE] + assert SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT not in span["attributes"] + + +@pytest.mark.parametrize( + "method,params,response,message_count", + [ + pytest.param( + "delete_message", + {"ReceiptHandle": "receipt-handle-1"}, + {}, + None, + id="single", + ), + pytest.param( + "delete_message_batch", + { + "Entries": [ + {"Id": "1", "ReceiptHandle": "receipt-handle-1"}, + {"Id": "2", "ReceiptHandle": "receipt-handle-2"}, + ] + }, + {"Successful": [{"Id": "1"}, {"Id": "2"}], "Failed": []}, + 2, + id="batch", + ), + ], +) +def test_delete_message( + client_factory, capture_items, method, params, response, message_count +): + client = client_factory("sqs") + + with Stubber(client) as stubber: + stubber.add_response(method, response) + spans = capture_spans_by_op( + lambda: getattr(client, method)( + QueueUrl="https://sqs.eu-north-1.amazonaws.com/123456789012/orders", + **params, + ), + capture_items, + ) + + (span,) = spans[OP.QUEUE_SETTLE] + assert span["name"] == "settle orders" + attributes = span["attributes"] + assert attributes[SPANDATA.SENTRY_KIND] == "client" + assert attributes[SPANDATA.MESSAGING_OPERATION_TYPE] == "settle" + assert attributes[SPANDATA.MESSAGING_OPERATION_NAME] == "settle" + assert attributes.get(SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT) == message_count + assert "receipt-handle" not in str(span)