From c3003ce7ff72204ced3da3fa3303248eaa96d081 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 07:14:02 +0000 Subject: [PATCH 1/3] feat(aws-lambda): Add queue.process spans for messaging triggers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit For an SQS, SNS, Kafka, ActiveMQ or RabbitMQ delivery, the handler runs inside a queue.process child span (kind consumer) with messaging.system, messaging.operation.type and messaging.operation.name set to process, the batch message count, and the destination name when all records share it. The span links to each distinct producer context found in the records' Sentry message attributes: SQS messageAttributes, SNS MessageAttributes, and SQS bodies that wrap an SNS notification. Co-Authored-By: Johannes Daxböck <114897+dingsdax@users.noreply.github.com> --- sentry_sdk/consts.py | 12 ++ .../integrations/aws_lambda/_handler.py | 10 +- .../integrations/aws_lambda/_trigger.py | 108 ++++++++++++- .../aws_lambda/test_aws_lambda.py | 146 +++++++++++++++++- 4 files changed, 268 insertions(+), 8 deletions(-) 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 f22c174cf3..bae67cfcbb 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 @@ -15,6 +16,7 @@ from sentry_sdk.integrations.aws_lambda._trigger import ( _get_messaging_batch_size, _get_trigger, + _start_process_span, ) from sentry_sdk.integrations.aws_lambda.consts import ( CLOUD_PLATFORM, @@ -33,7 +35,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 @@ -216,7 +218,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 a43dbe5ccf..bf7a9eabf5 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 @@ -28,6 +39,13 @@ "aws:rmq": "rmqMessagesByQueue", } +# `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( ( @@ -121,3 +139,91 @@ def _get_messaging_batch_size(aws_event: "Any") -> "Optional[int]": return None return sum(len(messages) for messages in aws_event[grouped_messages_key].values()) + + +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), + SPANDATA.FAAS_TRIGGER: "pubsub", + }, + 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 0487c7a304..6404e41df9 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(): @@ -978,6 +999,121 @@ def test_trigger(lambda_client, test_environment, aws_event, trigger, batch_coun ) +@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_trace_continuation(lambda_client, test_environment): trace_id = "471a43a4192642f0b136d5159a501701" parent_span_id = "6e8f22c393e68f19" From 5a479b9624163f12c7438393be0e830a936065a6 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 07:59:42 +0000 Subject: [PATCH 2/3] test(aws-lambda): Cover queue.process when the handler raises The error is captured, both the process span and the invocation span are marked as errors, and the exception still reaches the Lambda runtime. Co-Authored-By: johdax <114897+dingsdax@users.noreply.github.com> --- .../aws_lambda/test_aws_lambda.py | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/tests/integrations/aws_lambda/test_aws_lambda.py b/tests/integrations/aws_lambda/test_aws_lambda.py index 49df8435cd..97ec409a09 100644 --- a/tests/integrations/aws_lambda/test_aws_lambda.py +++ b/tests/integrations/aws_lambda/test_aws_lambda.py @@ -1089,6 +1089,32 @@ def test_no_process_span_without_messages(lambda_client, test_environment): 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" From feaef1c6429c83309228b22a618c02eb5d3cff65 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Mon, 9 Nov 2026 12:19:13 +0000 Subject: [PATCH 3/3] ref(aws-lambda): Keep faas.trigger on the invocation span only MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OTel sets faas.trigger on the FaaS span, which already has it. Also make the producer link cap private, like the other module constants. Co-Authored-By: Johannes Daxböck <114897+dingsdax@users.noreply.github.com> --- sentry_sdk/integrations/aws_lambda/_trigger.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/sentry_sdk/integrations/aws_lambda/_trigger.py b/sentry_sdk/integrations/aws_lambda/_trigger.py index eea0f525fd..bbb81585f4 100644 --- a/sentry_sdk/integrations/aws_lambda/_trigger.py +++ b/sentry_sdk/integrations/aws_lambda/_trigger.py @@ -35,7 +35,7 @@ # 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 +_MAX_PRODUCER_LINKS = 128 # https://docs.aws.amazon.com/eventbridge/latest/ref/overiew-event-structure.html _EVENTBRIDGE_KEYS = frozenset( @@ -154,7 +154,7 @@ def _get_producer_links(records: "list[Any]") -> "list[SpanLink]": if producer["parent_sampled"] is not None: link["sampled"] = bool(producer["parent_sampled"]) links.append(link) - if len(links) == MAX_PRODUCER_LINKS: + if len(links) == _MAX_PRODUCER_LINKS: break return links @@ -189,7 +189,6 @@ def _start_process_span(aws_event: "Any") -> "ContextManager[Any]": SPANDATA.MESSAGING_OPERATION_NAME: "process", SPANDATA.MESSAGING_DESTINATION_NAME: destination, SPANDATA.MESSAGING_BATCH_MESSAGE_COUNT: len(records), - SPANDATA.FAAS_TRIGGER: "pubsub", }, links=_get_producer_links(records), )