diff --git a/sentry_sdk/integrations/aws_lambda/_handler.py b/sentry_sdk/integrations/aws_lambda/_handler.py index 2178eff6c3..0a8607a6eb 100644 --- a/sentry_sdk/integrations/aws_lambda/_handler.py +++ b/sentry_sdk/integrations/aws_lambda/_handler.py @@ -1,6 +1,7 @@ import functools import json import sys +from contextlib import nullcontext from os import environ from typing import TYPE_CHECKING from urllib.parse import urlencode @@ -32,7 +33,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, SpanLink @@ -43,6 +44,8 @@ # OTel's default link 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 +# `messaging.system` of the messaging event sources that get a `queue.process` span. +MESSAGING_SYSTEMS = {"aws:sqs": "aws_sqs", "aws:sns": "aws.sns"} def _get_producer_links(aws_event: "Any", trace_id: str) -> "list[SpanLink]": @@ -267,7 +270,8 @@ def sentry_handler( else: sentry_sdk.continue_trace(headers) - # SQS and SNS messages link to their producers; they never parent the invocation. + # SQS and SNS messages link to their producers from the `queue.process` span; + # they never parent the invocation. producer_links: "list[SpanLink]" = [] with capture_internal_exceptions(): producer_links = _get_producer_links( @@ -282,10 +286,52 @@ def sentry_handler( name=function_name, parent_span=None, attributes=attributes, - links=producer_links, ): try: - return handler(aws_event, aws_context, *args, **kwargs) + # a `queue.process` span around the handler for an SQS or SNS delivery. + process_span: "ContextManager[Any]" = nullcontext() + with capture_internal_exceptions(): + records: "list[Any]" = [] + if isinstance(aws_event, dict): + records = aws_event.get("Records") or [] + + event_source = None + if records: + # SNS spells it `EventSource`. + event_source = records[0].get("eventSource") or records[ + 0 + ].get("EventSource") + + if event_source in MESSAGING_SYSTEMS: + # 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] + process_span = 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 + ), + }, + # OTel links a consumer's process span to each message's creation context. + # https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/ + links=producer_links, + ) + 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/tests/integrations/aws_lambda/test_aws_lambda.py b/tests/integrations/aws_lambda/test_aws_lambda.py index 5b63696bc4..102e58fb9e 100644 --- a/tests/integrations/aws_lambda/test_aws_lambda.py +++ b/tests/integrations/aws_lambda/test_aws_lambda.py @@ -916,6 +916,81 @@ 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", + [ + pytest.param( + {"Records": [SQS_RECORD, SQS_RECORD]}, "aws_sqs", "orders", id="sqs" + ), + pytest.param({"Records": [SNS_RECORD]}, "aws.sns", "alerts", id="sns"), + ], +) +def test_messaging_process_span( + lambda_client, test_environment, aws_event, system, destination +): + 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 + ] + 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}" + + 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" @@ -1085,12 +1160,17 @@ def test_producer_links(lambda_client, test_environment, aws_event, links): Payload=json.dumps(aws_event), ) - (segment_span,) = [ - s for s in test_environment["server"].span_items if s["is_segment"] + 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. + # messages are linked from the process span, never the parent of the invocation. assert "parent_span_id" not in segment_span - assert segment_span["links"] == links + assert "links" not in segment_span + assert process_span["links"] == links @pytest.mark.parametrize( @@ -1124,12 +1204,16 @@ def test_trace_continuation_from_client_context( lambda_client.invoke(FunctionName="ClientContextTrace", Payload=json.dumps(payload)) - (segment_span,) = [ - s for s in test_environment["server"].span_items if s["is_segment"] + 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 segment_span["trace_id"] == "471a43a4192642f0b136d5159a501701" assert segment_span["parent_span_id"] == "6e8f22c393e68f19" - assert segment_span["links"] == [ + assert process_span["links"] == [ { "trace_id": "5c6d7e8f9a0b4c1d8e2f3a4b5c6d7e8f", "span_id": "2b3c4d5e6f7a8b9c",