From 5e12d136f4957b256b605e27babf354fee81f857 Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Tue, 10 Nov 2026 13:16:21 -0600 Subject: [PATCH 1/4] feat(kafka-python): honour the messaging semconv stability opt-in The instrumentation emitted the pre-1.17 messaging attributes unconditionally and passed a hardcoded 1.11.0 schema URL, so OTEL_SEMCONV_STABILITY_OPT_IN had no effect on it. Read the MESSAGING opt-in mode and pass it to the producer and consumer wrappers. The default mode keeps emitting messaging.destination, messaging.kafka.partition and messaging.url unchanged. The messaging mode emits messaging.destination.name, messaging.destination.partition.id, server.address, server.port, messaging.operation.name and messaging.operation.type instead, and messaging/dup emits both generations. messaging.system is emitted in every mode. messaging.url carries the whole bootstrap_servers value JSON-encoded. The stable conventions define server.address as one host and server.port as one port, so the host and the port of the first entry are parsed out of it rather than the JSON being copied across, the way opentelemetry-instrumentation-confluent-kafka already does it. An entry naming no port, or a port that is not a number, leaves server.port unset instead of recording a wrong one. The schema URL now follows the mode in force rather than being a literal, so a consumer is told which generation of attributes it received. Refs #3495 --- .../instrumentation/kafka/__init__.py | 19 ++- .../instrumentation/kafka/utils.py | 122 ++++++++++++-- .../tests/test_utils.py | 154 ++++++++++++++++-- 3 files changed, 269 insertions(+), 26 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py index dd2d159c78f..b3f3713df6b 100644 --- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py @@ -88,6 +88,11 @@ def process_msg(message): from wrapt import wrap_function_wrapper from opentelemetry import trace +from opentelemetry.instrumentation._semconv import ( + _get_schema_url_for_signal_types, + _get_semconv_opt_in_modes, + _OpenTelemetryStabilitySignalType, +) from opentelemetry.instrumentation.instrumentor import BaseInstrumentor from opentelemetry.instrumentation.kafka.package import _instruments from opentelemetry.instrumentation.kafka.utils import _wrap_next, _wrap_send @@ -116,18 +121,26 @@ def _instrument(self, **kwargs): produce_hook = kwargs.get("produce_hook") consume_hook = kwargs.get("consume_hook") + sem_conv_opt_in_mode = _get_semconv_opt_in_modes((_OpenTelemetryStabilitySignalType.MESSAGING,))[ + _OpenTelemetryStabilitySignalType.MESSAGING + ] + tracer = trace.get_tracer( __name__, __version__, tracer_provider=tracer_provider, - schema_url="https://opentelemetry.io/schemas/1.11.0", + schema_url=_get_schema_url_for_signal_types([_OpenTelemetryStabilitySignalType.MESSAGING]), ) - wrap_function_wrapper(kafka.KafkaProducer, "send", _wrap_send(tracer, produce_hook)) + wrap_function_wrapper( + kafka.KafkaProducer, + "send", + _wrap_send(tracer, produce_hook, sem_conv_opt_in_mode), + ) wrap_function_wrapper( kafka.KafkaConsumer, "__next__", - _wrap_next(tracer, consume_hook), + _wrap_next(tracer, consume_hook, sem_conv_opt_in_mode), ) def _uninstrument(self, **kwargs): diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py index 479186e9b7a..47b7bd919f6 100644 --- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/utils.py @@ -4,12 +4,20 @@ import json from collections.abc import Callable from logging import getLogger +from re import compile as re_compile +from socket import AF_INET6, inet_pton from kafka.record.abc import ABCRecord from opentelemetry import context, propagate, trace +from opentelemetry.instrumentation._semconv import ( + _report_new, + _report_old, + _StabilityMode, +) from opentelemetry.propagators import textmap from opentelemetry.semconv._incubating.attributes import messaging_attributes +from opentelemetry.semconv.attributes import server_attributes from opentelemetry.semconv.trace import SpanAttributes from opentelemetry.trace import Tracer from opentelemetry.trace.span import Span @@ -17,6 +25,8 @@ _LOG = getLogger(__name__) +_SECURITY_PROTOCOL_PREFIX = re_compile("^.*://") + class KafkaPropertiesExtractor: @staticmethod @@ -89,18 +99,93 @@ def set(self, carrier: textmap.CarrierT, key: str, value: str) -> None: _kafka_setter = KafkaContextSetter() +def _bootstrap_servers_attributes(bootstrap_servers: str | list[str] | None) -> dict[str, AttributeValue]: + """Derive ``server.address`` and ``server.port`` from kafka-python's + ``bootstrap_servers``. + + The value is one ``host:port`` entry, a comma-separated string of them, or + a list of them, and each entry may carry a leading security protocol as in + ``SASL_SSL://broker:9093``. Only the first entry is reported, the way + ``opentelemetry-instrumentation-confluent-kafka`` reports it. An entry + carrying no port leaves ``server.port`` out rather than assuming 9092, so + the span reports what was configured. + """ + if not bootstrap_servers: + return {} + + entries = bootstrap_servers.split(",") if isinstance(bootstrap_servers, str) else bootstrap_servers + + entry = "" + for candidate in entries: + entry = _SECURITY_PROTOCOL_PREFIX.sub("", str(candidate).strip()) + if entry: + break + + if not entry: + return {} + + if entry.startswith("["): + host, _, rest = entry[1:].partition("]") + port = rest[1:] if rest.startswith(":") else "" + elif ":" not in entry: + host, port = entry, "" + else: + try: + inet_pton(AF_INET6, entry) + except (OSError, ValueError): + host, _, port = entry.rpartition(":") + else: + host, port = entry, "" + + if not host: + return {} + + attributes: dict[str, AttributeValue] = {server_attributes.SERVER_ADDRESS: host} + if port.isdecimal(): + attributes[server_attributes.SERVER_PORT] = int(port) + elif port: + _LOG.debug("non-numeric port in bootstrap_servers: %r", port) + return attributes + + +def _partition_attributes(partition: int, sem_conv_opt_in_mode: _StabilityMode) -> dict[str, AttributeValue]: + """Return the partition attributes of each generation the mode asks for. + + The producer only learns the partition after ``send`` returns, so this is + applied both when the span is created and afterwards. + """ + attributes: dict[str, AttributeValue] = {} + if _report_old(sem_conv_opt_in_mode): + attributes[SpanAttributes.MESSAGING_KAFKA_PARTITION] = partition + if _report_new(sem_conv_opt_in_mode): + attributes[messaging_attributes.MESSAGING_DESTINATION_PARTITION_ID] = str(partition) + return attributes + + def _get_attributes( - bootstrap_servers: list[str], + bootstrap_servers: str | list[str] | None, topic: str, + operation: messaging_attributes.MessagingOperationTypeValues, + sem_conv_opt_in_mode: _StabilityMode, partition: int | None = None, ) -> dict[str, AttributeValue]: attributes: dict[str, AttributeValue] = { messaging_attributes.MESSAGING_SYSTEM: messaging_attributes.MessagingSystemValues.KAFKA.value, - SpanAttributes.MESSAGING_DESTINATION: topic, - SpanAttributes.MESSAGING_URL: json.dumps(bootstrap_servers), } + + if _report_old(sem_conv_opt_in_mode): + attributes[SpanAttributes.MESSAGING_DESTINATION] = topic + attributes[SpanAttributes.MESSAGING_URL] = json.dumps(bootstrap_servers) + + if _report_new(sem_conv_opt_in_mode): + attributes[messaging_attributes.MESSAGING_DESTINATION_NAME] = topic + attributes[messaging_attributes.MESSAGING_OPERATION_NAME] = operation.value + attributes[messaging_attributes.MESSAGING_OPERATION_TYPE] = operation.value + attributes.update(_bootstrap_servers_attributes(bootstrap_servers)) + if partition is not None: - attributes[SpanAttributes.MESSAGING_KAFKA_PARTITION] = partition + attributes.update(_partition_attributes(partition, sem_conv_opt_in_mode)) + return attributes @@ -108,7 +193,13 @@ def _get_span_name(operation: str, topic: str): return f"{topic} {operation}" -def _wrap_send(tracer: Tracer, produce_hook: ProduceHookT) -> Callable: +def _wrap_send( + tracer: Tracer, + produce_hook: ProduceHookT, + sem_conv_opt_in_mode: _StabilityMode, +) -> Callable: + operation = messaging_attributes.MessagingOperationTypeValues.SEND + def _traced_send(func, instance, args, kwargs): headers = KafkaPropertiesExtractor.extract_send_headers(args, kwargs) if headers is None: @@ -117,8 +208,8 @@ def _traced_send(func, instance, args, kwargs): topic = KafkaPropertiesExtractor.extract_send_topic(args, kwargs) bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers(instance) - span_name = _get_span_name("send", topic) - attributes = _get_attributes(bootstrap_servers, topic) + span_name = _get_span_name(operation.value, topic) + attributes = _get_attributes(bootstrap_servers, topic, operation, sem_conv_opt_in_mode) with tracer.start_as_current_span(span_name, kind=trace.SpanKind.PRODUCER, attributes=attributes) as span: propagate.inject( headers, @@ -134,7 +225,8 @@ def _traced_send(func, instance, args, kwargs): future = func(*args, **kwargs) partition = KafkaPropertiesExtractor.extract_send_partition(future) if partition is not None: - span.set_attribute(SpanAttributes.MESSAGING_KAFKA_PARTITION, partition) + for key, value in _partition_attributes(partition, sem_conv_opt_in_mode).items(): + span.set_attribute(key, value) return future return _traced_send @@ -148,9 +240,17 @@ def _create_consumer_span( bootstrap_servers, args, kwargs, + sem_conv_opt_in_mode: _StabilityMode, ): - span_name = _get_span_name("receive", record.topic) - attributes = _get_attributes(bootstrap_servers, record.topic, record.partition) + operation = messaging_attributes.MessagingOperationTypeValues.RECEIVE + span_name = _get_span_name(operation.value, record.topic) + attributes = _get_attributes( + bootstrap_servers, + record.topic, + operation, + sem_conv_opt_in_mode, + record.partition, + ) with tracer.start_as_current_span( span_name, context=extracted_context, @@ -171,6 +271,7 @@ def _create_consumer_span( def _wrap_next( tracer: Tracer, consume_hook: ConsumeHookT, + sem_conv_opt_in_mode: _StabilityMode, ) -> Callable: def _traced_next(func, instance, args, kwargs): record = func(*args, **kwargs) @@ -187,6 +288,7 @@ def _traced_next(func, instance, args, kwargs): bootstrap_servers, args, kwargs, + sem_conv_opt_in_mode, ) return record diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py index 4c6e6a97dc7..14cdcf89bd5 100644 --- a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py @@ -7,8 +7,10 @@ from kafka.producer.future import FutureProduceResult, FutureRecordMetadata +from opentelemetry.instrumentation._semconv import _StabilityMode from opentelemetry.instrumentation.kafka.utils import ( KafkaPropertiesExtractor, + _bootstrap_servers_attributes, _create_consumer_span, _get_attributes, _get_span_name, @@ -18,6 +20,7 @@ _wrap_send, ) from opentelemetry.semconv._incubating.attributes import messaging_attributes +from opentelemetry.semconv.attributes import server_attributes from opentelemetry.semconv.trace import SpanAttributes from opentelemetry.test.samplertestutil import CapturingSampler from opentelemetry.test.test_base import TestBase @@ -90,7 +93,7 @@ def wrap_send_helper( kafka_producer = mock.MagicMock() expected_span_name = _get_span_name("send", self.topic_name) - wrapped_send = _wrap_send(tracer, produce_hook) + wrapped_send = _wrap_send(tracer, produce_hook, _StabilityMode.DEFAULT) retval = wrapped_send(original_send_callback, kafka_producer, self.args, self.kwargs) extract_bootstrap_servers.assert_called_once_with(kafka_producer) @@ -102,7 +105,12 @@ def wrap_send_helper( ) span = tracer.start_as_current_span().__enter__.return_value - get_attributes.assert_called_once_with(extract_bootstrap_servers.return_value, self.topic_name) + get_attributes.assert_called_once_with( + extract_bootstrap_servers.return_value, + self.topic_name, + messaging_attributes.MessagingOperationTypeValues.SEND, + _StabilityMode.DEFAULT, + ) span.set_attribute.assert_called_once_with( SpanAttributes.MESSAGING_KAFKA_PARTITION, extract_send_partition.return_value ) @@ -130,7 +138,7 @@ def test_wrap_next( original_next_callback = mock.MagicMock() kafka_consumer = mock.MagicMock() - wrapped_next = _wrap_next(tracer, consume_hook) + wrapped_next = _wrap_next(tracer, consume_hook, _StabilityMode.DEFAULT) record = wrapped_next(original_next_callback, kafka_consumer, self.args, self.kwargs) extract_bootstrap_servers.assert_called_once_with(kafka_consumer) @@ -150,6 +158,7 @@ def test_wrap_next( bootstrap_servers, self.args, self.kwargs, + _StabilityMode.DEFAULT, ) @mock.patch("opentelemetry.trace.set_span_in_context") @@ -177,6 +186,7 @@ def test_create_consumer_span( bootstrap_servers, self.args, self.kwargs, + _StabilityMode.DEFAULT, ) expected_span_name = _get_span_name("receive", record.topic) @@ -191,7 +201,13 @@ def test_create_consumer_span( set_span_in_context.assert_called_once_with(span, extracted_context) attach.assert_called_once_with(set_span_in_context.return_value) - get_attributes.assert_called_once_with(bootstrap_servers, record.topic, record.partition) + get_attributes.assert_called_once_with( + bootstrap_servers, + record.topic, + messaging_attributes.MessagingOperationTypeValues.RECEIVE, + _StabilityMode.DEFAULT, + record.partition, + ) consume_hook.assert_called_once_with(span, record, self.args, self.kwargs) detach.assert_called_once_with(attach.return_value) @@ -240,14 +256,50 @@ class _NoInternals: self.assertIsNone(partition) def test_get_attributes_records_partition_when_present(self) -> None: - attributes = _get_attributes(["localhost:9092"], self.topic_name, 2) + attributes = _get_attributes( + ["localhost:9092"], + self.topic_name, + messaging_attributes.MessagingOperationTypeValues.SEND, + _StabilityMode.DEFAULT, + 2, + ) self.assertEqual(attributes[SpanAttributes.MESSAGING_KAFKA_PARTITION], 2) self.assertIs(type(attributes[SpanAttributes.MESSAGING_KAFKA_PARTITION]), int) def test_get_attributes_omits_partition_when_none(self) -> None: - attributes = _get_attributes(["localhost:9092"], self.topic_name, None) + attributes = _get_attributes( + ["localhost:9092"], + self.topic_name, + messaging_attributes.MessagingOperationTypeValues.SEND, + _StabilityMode.DEFAULT, + None, + ) + + self.assertNotIn(SpanAttributes.MESSAGING_KAFKA_PARTITION, attributes) + + def test_get_attributes_omits_partition_when_none_in_messaging_mode(self) -> None: + attributes = _get_attributes( + ["localhost:9092"], + self.topic_name, + messaging_attributes.MessagingOperationTypeValues.SEND, + _StabilityMode.MESSAGING, + None, + ) + + self.assertNotIn(messaging_attributes.MESSAGING_DESTINATION_PARTITION_ID, attributes) + + def test_get_attributes_records_partition_as_a_string_in_messaging_mode(self) -> None: + attributes = _get_attributes( + ["localhost:9092"], + self.topic_name, + messaging_attributes.MessagingOperationTypeValues.SEND, + _StabilityMode.MESSAGING, + 2, + ) + self.assertEqual(attributes[messaging_attributes.MESSAGING_DESTINATION_PARTITION_ID], "2") + self.assertIs(type(attributes[messaging_attributes.MESSAGING_DESTINATION_PARTITION_ID]), str) self.assertNotIn(SpanAttributes.MESSAGING_KAFKA_PARTITION, attributes) @@ -274,7 +326,7 @@ def test_wrap_send_records_exception_raised_by_send(self, _extract_send_partitio def failing_send(*args, **kwargs): raise error - wrapped_send = _wrap_send(self.tracer, None) + wrapped_send = _wrap_send(self.tracer, None, _StabilityMode.DEFAULT) with self.assertRaises(ConnectionError) as raised: wrapped_send(failing_send, self.producer, self.args, self.kwargs) @@ -294,7 +346,7 @@ def failing_send(*args, **kwargs): def test_wrap_send_leaves_successful_send_unchanged(self, _extract_send_partition: mock.MagicMock) -> None: original_send = mock.MagicMock() - wrapped_send = _wrap_send(self.tracer, None) + wrapped_send = _wrap_send(self.tracer, None, _StabilityMode.DEFAULT) retval = wrapped_send(original_send, self.producer, self.args, self.kwargs) original_send.assert_called_once_with(*self.args, **self.kwargs) @@ -330,7 +382,9 @@ def test_send_attributes_passed_to_sampler(self) -> None: produce_hook = mock.Mock() kwargs = {"partition": 0, "headers": []} - result = _wrap_send(self.tracer, produce_hook)(send, self.client, ("topic-1",), kwargs) + result = _wrap_send(self.tracer, produce_hook, _StabilityMode.DEFAULT)( + send, self.client, ("topic-1",), kwargs + ) expected = { messaging_attributes.MESSAGING_SYSTEM: messaging_attributes.MessagingSystemValues.KAFKA.value, @@ -358,7 +412,7 @@ def test_consumer_attributes_passed_to_sampler(self) -> None: receive = mock.Mock(return_value=record) consume_hook = mock.Mock() - result = _wrap_next(self.tracer, consume_hook)(receive, self.client, (), {}) + result = _wrap_next(self.tracer, consume_hook, _StabilityMode.DEFAULT)(receive, self.client, (), {}) expected = { messaging_attributes.MESSAGING_SYSTEM: messaging_attributes.MessagingSystemValues.KAFKA.value, @@ -385,12 +439,14 @@ def test_attributes_control_sampling(self) -> None: if method == "send": future = FutureRecordMetadata(FutureProduceResult((topic, 0)), 0, None, None, -1, -1, -1) callback = mock.Mock(return_value=future) - result = _wrap_send(self.tracer, None)(callback, self.client, (topic,), {}) + result = _wrap_send(self.tracer, None, _StabilityMode.DEFAULT)( + callback, self.client, (topic,), {} + ) self.assertIs(result, future) else: record = SimpleNamespace(topic=topic, partition=0, headers=[]) callback = mock.Mock(return_value=record) - result = _wrap_next(self.tracer, None)(callback, self.client, (), {}) + result = _wrap_next(self.tracer, None, _StabilityMode.DEFAULT)(callback, self.client, (), {}) self.assertIs(result, record) callback.assert_called_once() self.assertEqual(len(self.memory_exporter.get_finished_spans()), expected_count) @@ -401,7 +457,79 @@ def test_consumer_error_propagates(self) -> None: receive = mock.Mock(side_effect=error) with self.assertRaises(ConnectionError) as raised: - _wrap_next(self.tracer, None)(receive, self.client, (), {}) + _wrap_next(self.tracer, None, _StabilityMode.DEFAULT)(receive, self.client, (), {}) self.assertIs(raised.exception, error) self.assertEqual(len(self.memory_exporter.get_finished_spans()), 0) + + +class TestBootstrapServersAttributes(TestCase): + """``bootstrap_servers`` is a ``host:port`` entry, a comma-separated string + of them, or a list of them, so ``server.address`` and ``server.port`` have + to be parsed out of it rather than copied from it.""" + + def test_list_entry_splits_into_host_and_port(self) -> None: + self.assertEqual( + _bootstrap_servers_attributes(["kafka-1:9092", "kafka-2:9092"]), + { + server_attributes.SERVER_ADDRESS: "kafka-1", + server_attributes.SERVER_PORT: 9092, + }, + ) + + def test_comma_separated_string_uses_the_first_entry(self) -> None: + self.assertEqual( + _bootstrap_servers_attributes("kafka-1:9092,kafka-2:9092"), + { + server_attributes.SERVER_ADDRESS: "kafka-1", + server_attributes.SERVER_PORT: 9092, + }, + ) + + def test_bare_host_omits_the_port(self) -> None: + """``localhost`` is kafka-python's own default for ``bootstrap_servers``.""" + self.assertEqual( + _bootstrap_servers_attributes("localhost"), + {server_attributes.SERVER_ADDRESS: "localhost"}, + ) + + def test_security_protocol_prefix_is_stripped(self) -> None: + self.assertEqual( + _bootstrap_servers_attributes("SASL_SSL://broker:9093"), + { + server_attributes.SERVER_ADDRESS: "broker", + server_attributes.SERVER_PORT: 9093, + }, + ) + + def test_bracketed_ipv6_literal_with_a_port(self) -> None: + self.assertEqual( + _bootstrap_servers_attributes("[::1]:9092"), + { + server_attributes.SERVER_ADDRESS: "::1", + server_attributes.SERVER_PORT: 9092, + }, + ) + + def test_bracketed_ipv6_literal_without_a_port(self) -> None: + self.assertEqual( + _bootstrap_servers_attributes("[::1]"), + {server_attributes.SERVER_ADDRESS: "::1"}, + ) + + def test_bare_ipv6_literal_is_not_split_on_its_last_colon(self) -> None: + self.assertEqual( + _bootstrap_servers_attributes("::1"), + {server_attributes.SERVER_ADDRESS: "::1"}, + ) + + def test_non_numeric_port_is_omitted_rather_than_emitted(self) -> None: + self.assertEqual( + _bootstrap_servers_attributes("broker:not-a-port"), + {server_attributes.SERVER_ADDRESS: "broker"}, + ) + + def test_empty_values_record_nothing(self) -> None: + for bootstrap_servers in (None, "", [], [""], ","): + with self.subTest(bootstrap_servers=bootstrap_servers): + self.assertEqual(_bootstrap_servers_attributes(bootstrap_servers), {}) From 96b92d5a06bb3ae2a2f2d5a3ec28f284b4675996 Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Tue, 10 Nov 2026 13:16:21 -0600 Subject: [PATCH 2/4] test(kafka-python): cover each messaging semconv stability mode Instrument a producer and a consumer under each of the three modes and assert the exact attribute names, the value types and the schema URL the span carries. messaging.destination.partition.id is a string while messaging.kafka.partition was an integer, so both are checked. --- .../tests/test_semconv_opt_in.py | 194 ++++++++++++++++++ 1 file changed, 194 insertions(+) create mode 100644 instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_semconv_opt_in.py diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_semconv_opt_in.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_semconv_opt_in.py new file mode 100644 index 00000000000..00619fc0a91 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_semconv_opt_in.py @@ -0,0 +1,194 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from os import environ +from unittest import mock + +from kafka import KafkaConsumer, KafkaProducer + +from opentelemetry.instrumentation._semconv import ( + OTEL_SEMCONV_STABILITY_OPT_IN, + _OpenTelemetrySemanticConventionStability, +) +from opentelemetry.instrumentation.kafka import KafkaInstrumentor +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.semconv._incubating.attributes import messaging_attributes +from opentelemetry.semconv.attributes import server_attributes +from opentelemetry.semconv.trace import SpanAttributes +from opentelemetry.test.test_base import TestBase +from opentelemetry.trace import SpanKind + +_TOPIC: str = "test_topic" +_PARTITION: int = 3 +_BOOTSTRAP_SERVERS: list[str] = ["localhost:9092"] +_BOOTSTRAP_SERVERS_JSON: str = '["localhost:9092"]' +_HOST: str = "localhost" +_PORT: int = 9092 +_LEGACY_SCHEMA_URL: str = "https://opentelemetry.io/schemas/1.11.0" +_MESSAGING_SCHEMA_URL: str = "https://opentelemetry.io/schemas/1.44.0" + +_OLD_ATTRIBUTES: set[str] = { + SpanAttributes.MESSAGING_DESTINATION, + SpanAttributes.MESSAGING_KAFKA_PARTITION, + SpanAttributes.MESSAGING_URL, +} +_NEW_ATTRIBUTES: set[str] = { + messaging_attributes.MESSAGING_DESTINATION_NAME, + messaging_attributes.MESSAGING_DESTINATION_PARTITION_ID, + messaging_attributes.MESSAGING_OPERATION_NAME, + messaging_attributes.MESSAGING_OPERATION_TYPE, + server_attributes.SERVER_ADDRESS, + server_attributes.SERVER_PORT, +} + + +def _do_nothing(*args: object, **kwargs: object) -> None: + """Stand in for ``KafkaProducer.close``. + + ``KafkaProducer.__del__`` calls it, and the producer below is built with + ``__new__`` and a config holding only ``bootstrap_servers``, so the real + one raises out of the garbage collector. + """ + + +class TestMessagingSemanticConventionOptIn(TestBase): + def setUp(self) -> None: + super().setUp() + _OpenTelemetrySemanticConventionStability._initialized = False + + def tearDown(self) -> None: + super().tearDown() + _OpenTelemetrySemanticConventionStability._initialized = False + + def produce_and_consume(self, opt_in: str | None) -> tuple[ReadableSpan, ReadableSpan]: + """Instrument kafka-python with ``OTEL_SEMCONV_STABILITY_OPT_IN`` set to + ``opt_in``, send and receive one message, and return the producer span + and the consumer span that it emitted. + + ``opt_in`` of ``None`` leaves the environment variable unset, which is + the default mode. + """ + record = mock.MagicMock() + record.topic = _TOPIC + record.partition = _PARTITION + record.headers = [] + + overrides = {} if opt_in is None else {OTEL_SEMCONV_STABILITY_OPT_IN: opt_in} + with mock.patch.dict("os.environ", overrides): + if opt_in is None: + environ.pop(OTEL_SEMCONV_STABILITY_OPT_IN, None) + + with ( + mock.patch.object(KafkaProducer, "send", autospec=True) as send, + mock.patch.object(KafkaConsumer, "__next__", autospec=True) as next_record, + ): + send.return_value._produce_future.topic_partition = ( + _TOPIC, + _PARTITION, + ) + next_record.return_value = record + + instrumentor = KafkaInstrumentor() + instrumentor.instrument(tracer_provider=self.tracer_provider) + try: + producer = KafkaProducer.__new__(KafkaProducer) + producer.config = {"bootstrap_servers": _BOOTSTRAP_SERVERS} + producer.close = _do_nothing + producer.send(_TOPIC, b"value") + + consumer = KafkaConsumer.__new__(KafkaConsumer) + consumer.config = {"bootstrap_servers": _BOOTSTRAP_SERVERS} + next(consumer) + finally: + instrumentor.uninstrument() + + producer_span, consumer_span = self.memory_exporter.get_finished_spans() + self.assertEqual(producer_span.kind, SpanKind.PRODUCER) + self.assertEqual(consumer_span.kind, SpanKind.CONSUMER) + return producer_span, consumer_span + + def test_default_mode_emits_only_the_old_attributes(self) -> None: + spans = self.produce_and_consume(None) + + for span in spans: + self.assertEqual(span.instrumentation_scope.schema_url, _LEGACY_SCHEMA_URL) + self.assertEqual( + span.attributes[messaging_attributes.MESSAGING_SYSTEM], + messaging_attributes.MessagingSystemValues.KAFKA.value, + ) + self.assertEqual(span.attributes[SpanAttributes.MESSAGING_DESTINATION], _TOPIC) + self.assertEqual( + span.attributes[SpanAttributes.MESSAGING_KAFKA_PARTITION], + _PARTITION, + ) + self.assertIsInstance( + span.attributes[SpanAttributes.MESSAGING_KAFKA_PARTITION], + int, + ) + self.assertEqual( + span.attributes[SpanAttributes.MESSAGING_URL], + _BOOTSTRAP_SERVERS_JSON, + ) + self.assertEqual(_NEW_ATTRIBUTES & set(span.attributes), set()) + + def test_messaging_mode_emits_only_the_stable_attributes(self) -> None: + producer_span, consumer_span = self.produce_and_consume("messaging") + operations = ( + ( + producer_span, + messaging_attributes.MessagingOperationTypeValues.SEND.value, + ), + ( + consumer_span, + messaging_attributes.MessagingOperationTypeValues.RECEIVE.value, + ), + ) + + for span, operation in operations: + self.assertEqual(span.instrumentation_scope.schema_url, _MESSAGING_SCHEMA_URL) + self.assertEqual( + span.attributes[messaging_attributes.MESSAGING_SYSTEM], + messaging_attributes.MessagingSystemValues.KAFKA.value, + ) + self.assertEqual( + span.attributes[messaging_attributes.MESSAGING_DESTINATION_NAME], + _TOPIC, + ) + self.assertEqual( + span.attributes[messaging_attributes.MESSAGING_DESTINATION_PARTITION_ID], + str(_PARTITION), + ) + self.assertIsInstance( + span.attributes[messaging_attributes.MESSAGING_DESTINATION_PARTITION_ID], + str, + ) + self.assertEqual(span.attributes[server_attributes.SERVER_ADDRESS], _HOST) + self.assertEqual(span.attributes[server_attributes.SERVER_PORT], _PORT) + self.assertIsInstance(span.attributes[server_attributes.SERVER_PORT], int) + self.assertEqual( + span.attributes[messaging_attributes.MESSAGING_OPERATION_NAME], + operation, + ) + self.assertEqual( + span.attributes[messaging_attributes.MESSAGING_OPERATION_TYPE], + operation, + ) + self.assertEqual(_OLD_ATTRIBUTES & set(span.attributes), set()) + + def test_dup_mode_emits_both_generations_of_attributes(self) -> None: + spans = self.produce_and_consume("messaging/dup") + + for span in spans: + self.assertEqual(span.instrumentation_scope.schema_url, _MESSAGING_SCHEMA_URL) + self.assertTrue(_OLD_ATTRIBUTES.issubset(span.attributes)) + self.assertTrue(_NEW_ATTRIBUTES.issubset(span.attributes)) + self.assertEqual( + span.attributes[SpanAttributes.MESSAGING_KAFKA_PARTITION], + _PARTITION, + ) + self.assertEqual( + span.attributes[messaging_attributes.MESSAGING_DESTINATION_PARTITION_ID], + str(_PARTITION), + ) From 48c705aab3d57558f15b62f46d183c849e06b54b Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Tue, 10 Nov 2026 13:16:21 -0600 Subject: [PATCH 3/4] docs(kafka-python): document OTEL_SEMCONV_STABILITY_OPT_IN Describe the messaging opt-in values the instrumentation now reads, and give the mapping from each old attribute to the stable one that replaces it. --- .../instrumentation/kafka/__init__.py | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py index b3f3713df6b..29d9c03bc8c 100644 --- a/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/src/opentelemetry/instrumentation/kafka/__init__.py @@ -78,6 +78,48 @@ def process_msg(message): # process message process_msg(message) +Stable Semantic Conventions +*************************** + +This instrumentation supports the messaging semantic convention migration plan. +You can control which conventions are emitted by setting the +``OTEL_SEMCONV_STABILITY_OPT_IN`` environment variable to one of these values: + +- ``messaging`` - emit the stable messaging conventions and stop emitting the + old experimental conventions. +- ``messaging/dup`` - emit both the old experimental and stable messaging + conventions during a transition period. + +The environment variable accepts a comma-separated list of opt-in values. For +example, ``messaging,http/dup`` enables stable messaging conventions and emits +both old and stable HTTP conventions. + +By default, when the environment variable is not set, the old experimental +messaging conventions are emitted. + +The old and stable attributes correspond as follows: + +============================= ====================================== +old stable +============================= ====================================== +``messaging.destination`` ``messaging.destination.name`` +``messaging.kafka.partition`` ``messaging.destination.partition.id`` +``messaging.url`` ``server.address`` and ``server.port`` +============================= ====================================== + +``messaging.destination.partition.id`` is a string, while +``messaging.kafka.partition`` was an integer. + +``messaging.url`` carries the whole ``bootstrap_servers`` value JSON-encoded. +``server.address`` and ``server.port`` carry the host and the port of its +first entry instead, parsed out of it, because the stable conventions define +them as one host and one port rather than as a connection string. +``server.port`` is omitted when that entry names no port. + +The stable conventions also add ``messaging.operation.name`` and +``messaging.operation.type``, which the old conventions did not have. +``messaging.system`` is emitted in every mode. + API --- """ From 71e88b43862b6293809a19cbdefe695fea3b3c16 Mon Sep 17 00:00:00 2001 From: Diego Hurtado Date: Tue, 10 Nov 2026 13:20:52 -0600 Subject: [PATCH 4/4] chore: add changelog fragment for the kafka-python messaging opt-in --- .changelog/5318.added | 1 + 1 file changed, 1 insertion(+) create mode 100644 .changelog/5318.added diff --git a/.changelog/5318.added b/.changelog/5318.added new file mode 100644 index 00000000000..38f0cc6b1f8 --- /dev/null +++ b/.changelog/5318.added @@ -0,0 +1 @@ +`opentelemetry-instrumentation-kafka-python`: support the messaging semantic convention stability opt-in through `OTEL_SEMCONV_STABILITY_OPT_IN`