diff --git a/.changelog/5318.added b/.changelog/5318.added new file mode 100644 index 0000000000..38f0cc6b1f --- /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` 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 dd2d159c78..29d9c03bc8 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 --- """ @@ -88,6 +130,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 +163,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 479186e9b7..47b7bd919f 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_semconv_opt_in.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_semconv_opt_in.py new file mode 100644 index 0000000000..00619fc0a9 --- /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), + ) diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py index 4c6e6a97dc..14cdcf89bd 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), {})