From 778f72b085e5a7800656b8f14c141d72746a1ce2 Mon Sep 17 00:00:00 2001 From: Charles Cheng Date: Mon, 3 Aug 2026 13:29:04 +0800 Subject: [PATCH 1/3] opentelemetry-instrumentation-boto3sqs: migrate to current messaging semantic conventions --- .../instrumentation/boto3sqs/__init__.py | 68 +++++++++--------- .../tests/test_boto3sqs_instrumentation.py | 69 ++++++++++--------- 2 files changed, 70 insertions(+), 67 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py b/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py index 443d8e8d4f..b4f606ddb0 100644 --- a/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py @@ -32,11 +32,16 @@ unwrap, ) from opentelemetry.propagators.textmap import CarrierT, Getter, Setter -from opentelemetry.semconv.trace import ( - MessagingDestinationKindValues, - MessagingOperationValues, - SpanAttributes, +from opentelemetry.semconv._incubating.attributes.messaging_attributes import ( + MESSAGING_DESTINATION_NAME, + MESSAGING_MESSAGE_ID, + MESSAGING_OPERATION_NAME, + MESSAGING_OPERATION_TYPE, + MESSAGING_SYSTEM, + MessagingOperationTypeValues, + MessagingSystemValues, ) +from opentelemetry.semconv.schemas import Schemas from opentelemetry.trace import Link, Span, SpanKind, Tracer, TracerProvider from .package import _instruments @@ -132,31 +137,21 @@ def instrumentation_dependencies(self) -> Collection[str]: def _enrich_span( span: Span, queue_name: str, - queue_url: str, - conversation_id: Optional[str] = None, - operation: Optional[MessagingOperationValues] = None, + operation_name: str, + operation_type: MessagingOperationTypeValues, message_id: Optional[str] = None, ) -> None: if not span.is_recording(): return - span.set_attribute(SpanAttributes.MESSAGING_SYSTEM, "aws.sqs") - span.set_attribute(SpanAttributes.MESSAGING_DESTINATION, queue_name) span.set_attribute( - SpanAttributes.MESSAGING_DESTINATION_KIND, - MessagingDestinationKindValues.QUEUE.value, + MESSAGING_SYSTEM, MessagingSystemValues.AWS_SQS.value ) - span.set_attribute(SpanAttributes.MESSAGING_URL, queue_url) + span.set_attribute(MESSAGING_DESTINATION_NAME, queue_name) + span.set_attribute(MESSAGING_OPERATION_NAME, operation_name) + span.set_attribute(MESSAGING_OPERATION_TYPE, operation_type.value) - if operation: - span.set_attribute( - SpanAttributes.MESSAGING_OPERATION, operation.value - ) - if conversation_id: - span.set_attribute( - SpanAttributes.MESSAGING_CONVERSATION_ID, conversation_id - ) if message_id: - span.set_attribute(SpanAttributes.MESSAGING_MESSAGE_ID, message_id) + span.set_attribute(MESSAGING_MESSAGE_ID, message_id) @staticmethod def _safe_end_processing_span(receipt_handle: str) -> None: @@ -181,7 +176,6 @@ def _extract_queue_name_from_url(queue_url: str) -> str: def _create_processing_span( self, queue_name: str, - queue_url: str, receipt_handle: str, message: Dict[str, Any], ) -> None: @@ -201,9 +195,9 @@ def _create_processing_span( Boto3SQSInstrumentor._enrich_span( span, queue_name, - queue_url, + "process", + MessagingOperationTypeValues.PROCESS, message_id=message_id, - operation=MessagingOperationValues.PROCESS, ) def _wrap_send_message(self, sqs_class: type) -> None: @@ -221,16 +215,19 @@ def send_wrapper(wrapped, instance, args, kwargs): kind=SpanKind.PRODUCER, end_on_exit=True, ) as span: - Boto3SQSInstrumentor._enrich_span(span, queue_name, queue_url) + Boto3SQSInstrumentor._enrich_span( + span, + queue_name, + "send", + MessagingOperationTypeValues.PUBLISH, + ) attributes = kwargs.pop("MessageAttributes", {}) propagate.inject(attributes, setter=boto3sqs_setter) retval = wrapped(*args, MessageAttributes=attributes, **kwargs) message_id = retval.get("MessageId") if message_id: if span.is_recording(): - span.set_attribute( - SpanAttributes.MESSAGING_MESSAGE_ID, message_id - ) + span.set_attribute(MESSAGING_MESSAGE_ID, message_id) return retval wrap_function_wrapper(sqs_class, "send_message", send_wrapper) @@ -258,7 +255,10 @@ def send_batch_wrapper(wrapped, instance, args, kwargs): ) ids_to_spans[entry_id] = span Boto3SQSInstrumentor._enrich_span( - span, queue_name, queue_url, conversation_id=entry_id + span, + queue_name, + "send", + MessagingOperationTypeValues.PUBLISH, ) with trace.use_span(span): if "MessageAttributes" not in entry: @@ -273,7 +273,7 @@ def send_batch_wrapper(wrapped, instance, args, kwargs): if message_span: if message_span.is_recording(): message_span.set_attribute( - SpanAttributes.MESSAGING_MESSAGE_ID, + MESSAGING_MESSAGE_ID, successful_messages.get("MessageId"), ) for span in ids_to_spans.values(): @@ -302,8 +302,8 @@ def receive_message_wrapper(wrapped, instance, args, kwargs): Boto3SQSInstrumentor._enrich_span( span, queue_name, - queue_url, - operation=MessagingOperationValues.RECEIVE, + "receive", + MessagingOperationTypeValues.RECEIVE, ) retval = wrapped( *args, @@ -321,7 +321,7 @@ def receive_message_wrapper(wrapped, instance, args, kwargs): receipt_handle ) self._create_processing_span( - queue_name, queue_url, receipt_handle, message + queue_name, receipt_handle, message ) retval["Messages"] = Boto3SQSInstrumentor.ContextableList( messages @@ -415,7 +415,7 @@ def _instrument(self, **kwargs: Dict[str, Any]) -> None: __name__, __version__, self._tracer_provider, - schema_url="https://opentelemetry.io/schemas/1.11.0", + schema_url=Schemas.V1_27_0.value, ) self._wrap_client_creation() diff --git a/instrumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py b/instrumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py index 58d968b094..5da54034c7 100644 --- a/instrumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py @@ -16,10 +16,14 @@ Boto3SQSInstrumentor, Boto3SQSSetter, ) -from opentelemetry.semconv.trace import ( - MessagingDestinationKindValues, - MessagingOperationValues, - SpanAttributes, +from opentelemetry.semconv._incubating.attributes.messaging_attributes import ( + MESSAGING_DESTINATION_NAME, + MESSAGING_MESSAGE_ID, + MESSAGING_OPERATION_NAME, + MESSAGING_OPERATION_TYPE, + MESSAGING_SYSTEM, + MessagingOperationTypeValues, + MessagingSystemValues, ) from opentelemetry.test.test_base import TestBase from opentelemetry.trace import SpanKind, TraceFlags @@ -210,12 +214,12 @@ def _assert_injected_span(self, msg_attrs: Dict[str, Any], span: Span): trace_parent.lower(), ) - def _default_span_attrs(self): + def _default_span_attrs(self, operation_name, operation_type): return { - SpanAttributes.MESSAGING_SYSTEM: "aws.sqs", - SpanAttributes.MESSAGING_DESTINATION: self._queue_name, - SpanAttributes.MESSAGING_DESTINATION_KIND: MessagingDestinationKindValues.QUEUE.value, - SpanAttributes.MESSAGING_URL: self._queue_url, + MESSAGING_SYSTEM: MessagingSystemValues.AWS_SQS.value, + MESSAGING_DESTINATION_NAME: self._queue_name, + MESSAGING_OPERATION_NAME: operation_name, + MESSAGING_OPERATION_TYPE: operation_type.value, } @staticmethod @@ -275,8 +279,10 @@ def test_send_message(self): self.assertEqual(SpanKind.PRODUCER, span.kind) self.assertEqual( { - SpanAttributes.MESSAGING_MESSAGE_ID: message_id, - **self._default_span_attrs(), + MESSAGING_MESSAGE_ID: message_id, + **self._default_span_attrs( + "send", MessagingOperationTypeValues.PUBLISH + ), }, span.attributes, ) @@ -303,22 +309,20 @@ def test_send_message_batch(self): spans = self.get_finished_spans() self.assertEqual(2, len(spans)) - spans_by_entry_id = { - span.attributes[SpanAttributes.MESSAGING_CONVERSATION_ID]: span - for span in spans + spans_by_message_id = { + span.attributes[MESSAGING_MESSAGE_ID]: span for span in spans } for entry in entries: - entry_id = entry["Id"] - span = spans_by_entry_id[entry_id] + message_id = expected_message_ids[entry["Id"]] + span = spans_by_message_id[message_id] self.assertEqual(f"{self._queue_name} send", span.name) self.assertEqual(SpanKind.PRODUCER, span.kind) self.assertEqual( { - SpanAttributes.MESSAGING_CONVERSATION_ID: entry_id, - SpanAttributes.MESSAGING_MESSAGE_ID: expected_message_ids[ - entry_id - ], - **self._default_span_attrs(), + MESSAGING_MESSAGE_ID: message_id, + **self._default_span_attrs( + "send", MessagingOperationTypeValues.PUBLISH + ), }, span.attributes, ) @@ -346,13 +350,12 @@ def test_send_message_batch_all_failed(self): self.assertEqual(f"{self._queue_name} send", span.name) self.assertEqual(SpanKind.PRODUCER, span.kind) self.assertEqual( - { - SpanAttributes.MESSAGING_CONVERSATION_ID: "1", - **self._default_span_attrs(), - }, + self._default_span_attrs( + "send", MessagingOperationTypeValues.PUBLISH + ), span.attributes, ) - self.assertNotIn(SpanAttributes.MESSAGING_MESSAGE_ID, span.attributes) + self.assertNotIn(MESSAGING_MESSAGE_ID, span.attributes) self._assert_injected_span(entries[0]["MessageAttributes"], span) def test_receive_message(self): @@ -386,10 +389,9 @@ def test_receive_message(self): self.assertEqual(f"{self._queue_name} receive", span.name) self.assertEqual(SpanKind.CONSUMER, span.kind) self.assertEqual( - { - SpanAttributes.MESSAGING_OPERATION: MessagingOperationValues.RECEIVE.value, - **self._default_span_attrs(), - }, + self._default_span_attrs( + "receive", MessagingOperationTypeValues.RECEIVE + ), span.attributes, ) @@ -411,9 +413,10 @@ def test_receive_message(self): # processing span attributes self.assertEqual( { - SpanAttributes.MESSAGING_MESSAGE_ID: msg_id, - SpanAttributes.MESSAGING_OPERATION: MessagingOperationValues.PROCESS.value, - **self._default_span_attrs(), + MESSAGING_MESSAGE_ID: msg_id, + **self._default_span_attrs( + "process", MessagingOperationTypeValues.PROCESS + ), }, span.attributes, ) From 6f859dc9c8e831e70235cc13be01ed16e92b2a1f Mon Sep 17 00:00:00 2001 From: Charles Cheng Date: Mon, 3 Aug 2026 13:29:48 +0800 Subject: [PATCH 2/3] Add changelog fragment --- .changelog/4920.changed | 1 + 1 file changed, 1 insertion(+) create mode 100644 .changelog/4920.changed diff --git a/.changelog/4920.changed b/.changelog/4920.changed new file mode 100644 index 0000000000..cb242e0c28 --- /dev/null +++ b/.changelog/4920.changed @@ -0,0 +1 @@ +`opentelemetry-instrumentation-boto3sqs`: migrate to current messaging semantic conventions From 5804e8c649347ccdda7437223e30810b657e54e7 Mon Sep 17 00:00:00 2001 From: Charles Cheng Date: Mon, 3 Aug 2026 20:34:11 +0800 Subject: [PATCH 3/3] Set server.address and server.port from the queue URL --- .../instrumentation/boto3sqs/__init__.py | 19 +++++++++++++++- .../tests/test_boto3sqs_instrumentation.py | 22 +++++++++++++++++++ 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py b/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py index b4f606ddb0..60a05321f5 100644 --- a/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py @@ -20,6 +20,7 @@ import logging from typing import Any, Collection, Dict, Generator, List, Mapping, Optional +from urllib.parse import urlparse import boto3.session import botocore.client @@ -41,6 +42,10 @@ MessagingOperationTypeValues, MessagingSystemValues, ) +from opentelemetry.semconv.attributes.server_attributes import ( + SERVER_ADDRESS, + SERVER_PORT, +) from opentelemetry.semconv.schemas import Schemas from opentelemetry.trace import Link, Span, SpanKind, Tracer, TracerProvider @@ -137,6 +142,7 @@ def instrumentation_dependencies(self) -> Collection[str]: def _enrich_span( span: Span, queue_name: str, + queue_url: str, operation_name: str, operation_type: MessagingOperationTypeValues, message_id: Optional[str] = None, @@ -150,6 +156,12 @@ def _enrich_span( span.set_attribute(MESSAGING_OPERATION_NAME, operation_name) span.set_attribute(MESSAGING_OPERATION_TYPE, operation_type.value) + parsed_url = urlparse(queue_url) + if parsed_url.hostname: + span.set_attribute(SERVER_ADDRESS, parsed_url.hostname) + if parsed_url.port: + span.set_attribute(SERVER_PORT, parsed_url.port) + if message_id: span.set_attribute(MESSAGING_MESSAGE_ID, message_id) @@ -176,6 +188,7 @@ def _extract_queue_name_from_url(queue_url: str) -> str: def _create_processing_span( self, queue_name: str, + queue_url: str, receipt_handle: str, message: Dict[str, Any], ) -> None: @@ -195,6 +208,7 @@ def _create_processing_span( Boto3SQSInstrumentor._enrich_span( span, queue_name, + queue_url, "process", MessagingOperationTypeValues.PROCESS, message_id=message_id, @@ -218,6 +232,7 @@ def send_wrapper(wrapped, instance, args, kwargs): Boto3SQSInstrumentor._enrich_span( span, queue_name, + queue_url, "send", MessagingOperationTypeValues.PUBLISH, ) @@ -257,6 +272,7 @@ def send_batch_wrapper(wrapped, instance, args, kwargs): Boto3SQSInstrumentor._enrich_span( span, queue_name, + queue_url, "send", MessagingOperationTypeValues.PUBLISH, ) @@ -302,6 +318,7 @@ def receive_message_wrapper(wrapped, instance, args, kwargs): Boto3SQSInstrumentor._enrich_span( span, queue_name, + queue_url, "receive", MessagingOperationTypeValues.RECEIVE, ) @@ -321,7 +338,7 @@ def receive_message_wrapper(wrapped, instance, args, kwargs): receipt_handle ) self._create_processing_span( - queue_name, receipt_handle, message + queue_name, queue_url, receipt_handle, message ) retval["Messages"] = Boto3SQSInstrumentor.ContextableList( messages diff --git a/instrumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py b/instrumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py index 5da54034c7..ee11661611 100644 --- a/instrumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py @@ -25,6 +25,10 @@ MessagingOperationTypeValues, MessagingSystemValues, ) +from opentelemetry.semconv.attributes.server_attributes import ( + SERVER_ADDRESS, + SERVER_PORT, +) from opentelemetry.test.test_base import TestBase from opentelemetry.trace import SpanKind, TraceFlags from opentelemetry.trace.span import Span, format_span_id, format_trace_id @@ -220,6 +224,7 @@ def _default_span_attrs(self, operation_name, operation_type): MESSAGING_DESTINATION_NAME: self._queue_name, MESSAGING_OPERATION_NAME: operation_name, MESSAGING_OPERATION_TYPE: operation_type.value, + SERVER_ADDRESS: "sqs.us-east-1.amazonaws.com", } @staticmethod @@ -288,6 +293,23 @@ def test_send_message(self): ) self._assert_injected_span(message_attrs, span) + def test_send_message_custom_endpoint_with_port(self): + message_id = "123456789" + mock_response = { + "MD5OfMessageBody": "1234", + "MessageId": message_id, + } + + with self._mocked_endpoint(mock_response): + self._client.send_message( + QueueUrl=f"http://localhost:4566/123456789012/{self._queue_name}", + MessageBody="hello msg", + ) + + span = self._get_only_span() + self.assertEqual("localhost", span.attributes[SERVER_ADDRESS]) + self.assertEqual(4566, span.attributes[SERVER_PORT]) + def test_send_message_batch(self): expected_message_ids = {"1": "msg-1", "2": "msg-2"} mock_response = {