Skip to content

opentelemetry-instrumentation-boto3sqs: migrate to current messaging semantic conventions #4920

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
anneheartrecord wants to merge 3 commits into
base: main
Choose a base branch
from
+105 −62
Open

opentelemetry-instrumentation-boto3sqs: migrate to current messaging semantic conventions #4920

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
778f72b
opentelemetry-instrumentation-boto3sqs: migrate to current messaging …
anneheartrecord Aug 3, 2026
6f859dc
Add changelog fragment
anneheartrecord Aug 3, 2026
5804e8c
Set server.address and server.port from the queue URL
anneheartrecord Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .changelog/4920.changed
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
`opentelemetry-instrumentation-boto3sqs`: migrate to current messaging semantic conventions
75 changes: 46 additions & 29 deletions ...telemetry-instrumentation-boto3sqs/src/opentelemetry/instrumentation/boto3sqs/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -32,11 +33,20 @@
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.attributes.server_attributes import (
SERVER_ADDRESS,
SERVER_PORT,
)
from opentelemetry.semconv.schemas import Schemas
from opentelemetry.trace import Link, Span, SpanKind, Tracer, TracerProvider

from .package import _instruments
Expand Down Expand Up @@ -133,30 +143,27 @@ 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,
Comment on lines +146 to +147

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve the SQS endpoint in the new convention

Removing queue_url from _enrich_span drops the previous endpoint information without replacing it with the current convention's server.address (and server.port for non-default ports), even though every instrumented call already supplies the queue URL. As a result, spans for AWS endpoints and especially LocalStack/custom endpoints can no longer identify the server that handled the operation, and the declared v1.27 messaging telemetry is incomplete; retain and parse the URL using the generated server attribute constants.

AGENTS.md reference: AGENTS.md:L84-L87

Useful? React with 👍 / 👎.

@anneheartrecord anneheartrecord Aug 3, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch — added server.address (and server.port when the URL carries an explicit port, e.g. LocalStack) parsed from the queue URL in _enrich_span, replacing the endpoint info the old messaging.url attribute provided. Covered by the updated default-attrs assertions plus a new custom-endpoint test. 5804e8c

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)

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)
Comment on lines +162 to +163

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Handle malformed queue URLs without changing application errors

When a custom QueueUrl contains a nonnumeric or out-of-range port, accessing parsed_url.port raises ValueError before the wrapped botocore operation runs. Previously this value was forwarded so botocore could produce its own validation or request error; the instrumentation now changes application behavior instead of preserving the underlying exception. Skip the server attributes when URL parsing fails.

AGENTS.md reference: AGENTS.md:L80-L82

Useful? React with 👍 / 👎.

Comment on lines +162 to +163

@chatgpt-codex-connector chatgpt-codex-connector Bot Aug 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Omit default ports from server.port

When a valid queue URL explicitly includes its scheme's default port, such as https://host:443/... or http://host:80/..., this truthiness check emits server.port. The current server semantic convention requires omitting the attribute for the default port, so the newly added endpoint enrichment produces nonconforming telemetry for these URLs; compare the parsed scheme and only set non-default ports.

Useful? React with 👍 / 👎.


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:
Expand Down Expand Up @@ -202,8 +209,9 @@ def _create_processing_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:
Expand All @@ -221,16 +229,20 @@ 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,
queue_url,
"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)
Expand Down Expand Up @@ -258,7 +270,11 @@ 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,
queue_url,
"send",
MessagingOperationTypeValues.PUBLISH,
)
with trace.use_span(span):
if "MessageAttributes" not in entry:
Expand All @@ -273,7 +289,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():
Expand Down Expand Up @@ -303,7 +319,8 @@ def receive_message_wrapper(wrapped, instance, args, kwargs):
span,
queue_name,
queue_url,
operation=MessagingOperationValues.RECEIVE,
"receive",
MessagingOperationTypeValues.RECEIVE,
)
retval = wrapped(
*args,
Expand Down Expand Up @@ -415,7 +432,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()

Expand Down
91 changes: 58 additions & 33 deletions ...rumentation/opentelemetry-instrumentation-boto3sqs/tests/test_boto3sqs_instrumentation.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,10 +16,18 @@
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.semconv.attributes.server_attributes import (
SERVER_ADDRESS,
SERVER_PORT,
)
from opentelemetry.test.test_base import TestBase
from opentelemetry.trace import SpanKind, TraceFlags
Expand Down Expand Up @@ -210,12 +218,13 @@ 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,
SERVER_ADDRESS: "sqs.us-east-1.amazonaws.com",
}

@staticmethod
Expand Down Expand Up @@ -275,13 +284,32 @@ 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,
)
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 = {
Expand All @@ -303,22 +331,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,
)
Expand Down Expand Up @@ -346,13 +372,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):
Expand Down Expand Up @@ -386,10 +411,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,
)

Expand All @@ -411,9 +435,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,
)
Expand Down
Loading