diff --git a/sentry_sdk/consts.py b/sentry_sdk/consts.py index a87a4e4243..bc2e786907 100644 --- a/sentry_sdk/consts.py +++ b/sentry_sdk/consts.py @@ -1126,6 +1126,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` diff --git a/sentry_sdk/integrations/aws_lambda/_handler.py b/sentry_sdk/integrations/aws_lambda/_handler.py index 9485f10169..9c55842300 100644 --- a/sentry_sdk/integrations/aws_lambda/_handler.py +++ b/sentry_sdk/integrations/aws_lambda/_handler.py @@ -1,5 +1,6 @@ import functools import sys +from contextlib import nullcontext from os import environ from typing import TYPE_CHECKING from urllib.parse import urlencode @@ -12,7 +13,10 @@ _get_user_from_event, _make_request_event_processor, ) -from sentry_sdk.integrations.aws_lambda._trigger import _get_trigger +from sentry_sdk.integrations.aws_lambda._trigger import ( + _get_trigger, + _start_process_span, +) from sentry_sdk.integrations.aws_lambda.consts import ( CLOUD_PLATFORM, CLOUD_PROVIDER, @@ -30,7 +34,7 @@ ) if TYPE_CHECKING: - from typing import Any, Callable, Mapping, TypeVar + from typing import Any, Callable, ContextManager, Mapping, TypeVar from sentry_sdk._types import Attributes @@ -204,7 +208,11 @@ def sentry_handler( attributes=attributes, ): try: - return handler(aws_event, aws_context, *args, **kwargs) + process_span: "ContextManager[Any]" = nullcontext() + with capture_internal_exceptions(): + process_span = _start_process_span(aws_event) + with process_span: + return handler(aws_event, aws_context, *args, **kwargs) except Exception: exc_info = sys.exc_info() sentry_event, hint = event_from_exception( diff --git a/sentry_sdk/integrations/aws_lambda/_trigger.py b/sentry_sdk/integrations/aws_lambda/_trigger.py index 1d3d527a87..bbb81585f4 100644 --- a/sentry_sdk/integrations/aws_lambda/_trigger.py +++ b/sentry_sdk/integrations/aws_lambda/_trigger.py @@ -1,7 +1,18 @@ +import json +from contextlib import nullcontext from typing import TYPE_CHECKING +import sentry_sdk +from sentry_sdk.consts import OP, SPANDATA +from sentry_sdk.integrations.aws_lambda.consts import ORIGIN +from sentry_sdk.traces import SENTRY_TRACE_HEADER_NAME +from sentry_sdk.tracing_utils import extract_sentrytrace_data +from sentry_sdk.utils import capture_internal_exceptions + if TYPE_CHECKING: - from typing import Any, Optional + from typing import Any, ContextManager, Optional + + from sentry_sdk._types import SpanLink # `faas.trigger` of each `eventSource` of `Records` or of a top-level `eventSource`. # https://docs.aws.amazon.com/lambda/latest/dg/lambda-services.html @@ -19,6 +30,13 @@ "aws:docdb": "datasource", } +# `messaging.system` of the event sources whose producers can propagate a trace. +_MESSAGING_SYSTEMS = {"aws:sqs": "aws_sqs", "aws:sns": "aws.sns"} + +# OTel's default limit; Lambda delivers up to 10,000 SQS messages at once. +# https://opentelemetry.io/docs/specs/otel/trace/sdk/#span-limits +_MAX_PRODUCER_LINKS = 128 + # https://docs.aws.amazon.com/eventbridge/latest/ref/overiew-event-structure.html _EVENTBRIDGE_KEYS = frozenset( ( @@ -87,3 +105,90 @@ def _get_trigger(aws_event: "Any") -> str: return "pubsub" return _TRIGGERS.get(aws_event.get("eventSource"), "other") + + +def _get_producer_trace(record: "dict[str, Any]") -> "Optional[str]": + """The `sentry-trace` the producer put in a record's message attributes.""" + # https://docs.aws.amazon.com/lambda/latest/dg/with-sns.html + if record.get("EventSource") == "aws:sns": + attribute = ( + record["Sns"].get("MessageAttributes", {}).get(SENTRY_TRACE_HEADER_NAME, {}) + ) + return attribute.get("Value") + + # https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html + attribute = record.get("messageAttributes", {}).get(SENTRY_TRACE_HEADER_NAME) + if attribute is not None: + return attribute.get("stringValue") + + # without raw message delivery, an SNS subscription delivers the notification, + # with its message attributes, as the body. + # https://docs.aws.amazon.com/sns/latest/dg/sns-sqs-as-subscriber.html + if '"TopicArn"' in record["body"]: + notification = json.loads(record["body"]) + attribute = notification.get("MessageAttributes", {}).get( + SENTRY_TRACE_HEADER_NAME, {} + ) + return attribute.get("Value") + + return None + + +def _get_producer_links(records: "list[Any]") -> "list[SpanLink]": + """One link per distinct producer context; messages never parent the invocation.""" + links: "list[SpanLink]" = [] + linked = set() + for record in records: + with capture_internal_exceptions(): + producer = extract_sentrytrace_data(_get_producer_trace(record)) + if producer is None: + continue + + trace_id = str(producer["trace_id"]) + span_id = str(producer["parent_span_id"]) + if (trace_id, span_id) in linked: + continue + linked.add((trace_id, span_id)) + + link: "SpanLink" = {"trace_id": trace_id, "span_id": span_id} + if producer["parent_sampled"] is not None: + link["sampled"] = bool(producer["parent_sampled"]) + links.append(link) + if len(links) == _MAX_PRODUCER_LINKS: + break + + return links + + +def _start_process_span(aws_event: "Any") -> "ContextManager[Any]": + """A `queue.process` span around the handler for an SQS or SNS delivery.""" + records = aws_event.get("Records") if isinstance(aws_event, dict) else None + if not records: + return nullcontext() + + event_source = _get_records_source(records) + if event_source not in _MESSAGING_SYSTEMS: + return nullcontext() + + # one queue or topic per event: `arn:aws:sqs:::`. + arn = ( + records[0]["eventSourceARN"] + if event_source == "aws:sqs" + else records[0]["Sns"]["TopicArn"] + ) + destination = arn.split(":")[5] + + return sentry_sdk.start_span( + name=f"process {destination}", + attributes={ + SPANDATA.SENTRY_OP: OP.QUEUE_PROCESS, + SPANDATA.SENTRY_ORIGIN: ORIGIN, + SPANDATA.SENTRY_KIND: "consumer", + SPANDATA.MESSAGING_SYSTEM: _MESSAGING_SYSTEMS[event_source], + SPANDATA.MESSAGING_OPERATION_TYPE: "process", + SPANDATA.MESSAGING_OPERATION_NAME: "process", + SPANDATA.MESSAGING_DESTINATION_NAME: destination, + SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT: len(records), + }, + links=_get_producer_links(records), + ) diff --git a/tests/integrations/aws_lambda/test_aws_lambda.py b/tests/integrations/aws_lambda/test_aws_lambda.py index ccee9d0cb4..2f202bb208 100644 --- a/tests/integrations/aws_lambda/test_aws_lambda.py +++ b/tests/integrations/aws_lambda/test_aws_lambda.py @@ -815,20 +815,41 @@ def test_error(lambda_client, test_environment): assert SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT not in attrs -def _sqs_record(): - return { +def _sqs_record(sentry_trace=None): + record = { "messageId": "059f36b4-87a3-44ab-83d2-661975830a7d", "body": "{}", + "messageAttributes": {}, "eventSource": "aws:sqs", "eventSourceARN": "arn:aws:sqs:us-east-1:012345678912:orders", } + if sentry_trace is not None: + record["messageAttributes"]["sentry-trace"] = { + "stringValue": sentry_trace, + "stringListValues": [], + "binaryListValues": [], + "dataType": "String", + } + return record -def _sns_record(): - return { +def _sns_record(sentry_trace=None): + record = { "EventSource": "aws:sns", - "Sns": {"TopicArn": "arn:aws:sns:us-east-1:012345678912:alerts"}, + "Sns": { + "Type": "Notification", + "MessageId": "95df01b4-ee98-5cb9-9903-4c221d41eb5e", + "TopicArn": "arn:aws:sns:us-east-1:012345678912:alerts", + "Message": "{}", + "MessageAttributes": {}, + }, } + if sentry_trace is not None: + record["Sns"]["MessageAttributes"]["sentry-trace"] = { + "Type": "String", + "Value": sentry_trace, + } + return record def _appsync_event(): @@ -914,6 +935,147 @@ def test_trigger(lambda_client, test_environment, aws_event, trigger): assert _get_span_attr(attrs, SPANDATA.FAAS_TRIGGER) == trigger +@pytest.mark.parametrize( + "aws_event,system,destination,links", + [ + pytest.param( + { + "Records": [ + _sqs_record("471a43a4192642f0b136d5159a501701-6e8f22c393e68f19-1"), + # the same send context: linked once. + _sqs_record("471a43a4192642f0b136d5159a501701-6e8f22c393e68f19-1"), + _sqs_record("0f4c2a5b8d9e4f1a2b3c4d5e6f7a8b9c-1a2b3c4d5e6f7a8b-0"), + _sqs_record(), + ] + }, + "aws_sqs", + "orders", + [ + { + "trace_id": "471a43a4192642f0b136d5159a501701", + "span_id": "6e8f22c393e68f19", + "sampled": True, + }, + { + "trace_id": "0f4c2a5b8d9e4f1a2b3c4d5e6f7a8b9c", + "span_id": "1a2b3c4d5e6f7a8b", + "sampled": False, + }, + ], + id="sqs", + ), + pytest.param( + { + "Records": [ + _sns_record("471a43a4192642f0b136d5159a501701-6e8f22c393e68f19-1") + ] + }, + "aws.sns", + "alerts", + [ + { + "trace_id": "471a43a4192642f0b136d5159a501701", + "span_id": "6e8f22c393e68f19", + "sampled": True, + } + ], + id="sns", + ), + pytest.param( + { + "Records": [ + { + **_sqs_record(), + "body": json.dumps( + _sns_record( + "471a43a4192642f0b136d5159a501701-6e8f22c393e68f19-1" + )["Sns"] + ), + } + ] + }, + "aws_sqs", + "orders", + [ + { + "trace_id": "471a43a4192642f0b136d5159a501701", + "span_id": "6e8f22c393e68f19", + "sampled": True, + } + ], + id="sns-to-sqs", + ), + ], +) +def test_messaging_process_span( + lambda_client, test_environment, aws_event, system, destination, links +): + lambda_client.invoke( + FunctionName="BasicOkSpanStreaming", + Payload=json.dumps(aws_event), + ) + + span_items = test_environment["server"].span_items + (segment_span,) = [s for s in span_items if s["is_segment"]] + (process_span,) = [ + s + for s in span_items + if _get_span_attr(s["attributes"], SPANDATA.SENTRY_OP) == OP.QUEUE_PROCESS + ] + # messages are linked, never the parent of the invocation. + assert "parent_span_id" not in segment_span + assert process_span["parent_span_id"] == segment_span["span_id"] + assert process_span["trace_id"] == segment_span["trace_id"] + assert process_span["name"] == f"process {destination}" + assert process_span["links"] == links + + attrs = process_span["attributes"] + assert _get_span_attr(attrs, SPANDATA.SENTRY_KIND) == "consumer" + assert _get_span_attr(attrs, SPANDATA.MESSAGING_SYSTEM) == system + assert _get_span_attr(attrs, SPANDATA.MESSAGING_OPERATION_TYPE) == "process" + assert _get_span_attr(attrs, SPANDATA.MESSAGING_OPERATION_NAME) == "process" + assert _get_span_attr(attrs, SPANDATA.MESSAGING_DESTINATION_NAME) == destination + assert _get_span_attr(attrs, SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT) == len( + aws_event["Records"] + ) + + +def test_no_process_span_without_messages(lambda_client, test_environment): + lambda_client.invoke( + FunctionName="BasicOkSpanStreaming", + Payload=json.dumps({"Records": [{"eventSource": "aws:kinesis"}]}), + ) + + (span,) = test_environment["server"].span_items + assert span["is_segment"] + + +def test_process_span_error(lambda_client, test_environment): + response = lambda_client.invoke( + FunctionName="RaiseErrorSpanStreaming", + Payload=json.dumps({"Records": [_sqs_record()]}), + ) + + # the handler's exception still reaches the Lambda runtime. + assert json.loads(response["Payload"].read())["errorMessage"] == "Oh!" + + (error_event,) = test_environment["server"].envelopes + (exception,) = error_event["exception"]["values"] + assert exception["value"] == "Oh!" + assert not exception["mechanism"]["handled"] + + span_items = test_environment["server"].span_items + (segment_span,) = [s for s in span_items if s["is_segment"]] + (process_span,) = [ + s + for s in span_items + if _get_span_attr(s["attributes"], SPANDATA.SENTRY_OP) == OP.QUEUE_PROCESS + ] + assert process_span["parent_span_id"] == segment_span["span_id"] + assert process_span["status"] == "error" + assert segment_span["status"] == "error" + + def test_trace_continuation(lambda_client, test_environment): trace_id = "471a43a4192642f0b136d5159a501701" parent_span_id = "6e8f22c393e68f19"