From 15e62a31f4536b6b9668e0daccd6436f22553feb Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Mon, 13 Jul 2026 23:56:53 +0530 Subject: [PATCH 1/9] feat(kafka): emit messaging.kafka.cluster.id unconditionally across kafka-python, confluent-kafka, and aiokafka Add always-on messaging.kafka.cluster.id span attribute to all three Python Kafka instrumentation libraries. Cluster ID is read lazily from the client instance (list_topics() for kafka-python/confluent-kafka, metadata() for aiokafka) with a 1-hour TTL cache per client. Removes the capture_experimental_span_attributes gate and promotes the attribute to default-on behavior matching the semconv stable promotion. Assisted-by: Claude Sonnet 4.6 --- .../.changelog/.gitignore | 1 + .../.changelog/4727.added | 1 + .../instrumentation/aiokafka/utils.py | 62 ++++++++- .../tests/test_utils.py | 105 +++++++++++++++ .../confluent_kafka/__init__.py | 33 +++++ .../instrumentation/confluent_kafka/utils.py | 122 +++++++++++++++++- .../instrumentation/kafka/__init__.py | 35 ++++- .../instrumentation/kafka/utils.py | 115 ++++++++++++++++- 8 files changed, 462 insertions(+), 12 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore create mode 100644 instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore new file mode 100644 index 0000000000..f935021a8f --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore @@ -0,0 +1 @@ +!.gitignore diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added new file mode 100644 index 0000000000..11db77dcc7 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added @@ -0,0 +1 @@ +`opentelemetry-instrumentation-aiokafka`: add `capture_experimental_span_attributes` option to gate `messaging.cluster.id` on producer and consumer spans diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py index ccfe157129..8b27ac78b3 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -79,6 +79,8 @@ async def __call__( _LOG = getLogger(__name__) +_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" + def _extract_bootstrap_servers( client: aiokafka.AIOKafkaClient, @@ -90,6 +92,24 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str: return client._client_id +def _extract_cluster_id_from_client( + client: aiokafka.AIOKafkaClient, +) -> str | None: + """Read cluster ID from the aiokafka client's cached cluster metadata. + + aiokafka sets AIOKafkaClient.cluster.cluster_id after the first successful + broker metadata response — no extra connection or background thread needed. + Returns None if metadata has not been received yet. + """ + try: + cluster_id = getattr( + getattr(client, "cluster", None), "cluster_id", None + ) + return cluster_id if cluster_id else None + except Exception: # pylint: disable=broad-except + return None + + def _extract_consumer_group( consumer: aiokafka.AIOKafkaConsumer, ) -> str | None: @@ -237,6 +257,7 @@ def _enrich_base_span( topic: str, partition: int | None, key: str | None, + cluster_id: str | None = None, ) -> None: span.set_attribute( messaging_attributes.MESSAGING_SYSTEM, @@ -259,6 +280,9 @@ def _enrich_base_span( messaging_attributes.MESSAGING_KAFKA_MESSAGE_KEY, key ) + if cluster_id is not None: + span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + def _enrich_send_span( span: Span, @@ -268,6 +292,7 @@ def _enrich_send_span( topic: str, partition: int | None, key: str | None, + cluster_id: str | None = None, ) -> None: if not span.is_recording(): return @@ -279,6 +304,7 @@ def _enrich_send_span( topic=topic, partition=partition, key=key, + cluster_id=cluster_id, ) span.set_attribute(messaging_attributes.MESSAGING_OPERATION_NAME, "send") @@ -298,6 +324,7 @@ def _enrich_getone_span( partition: int | None, key: str | None, offset: int, + cluster_id: str | None = None, ) -> None: if not span.is_recording(): return @@ -309,6 +336,7 @@ def _enrich_getone_span( topic=topic, partition=partition, key=key, + cluster_id=cluster_id, ) if consumer_group is not None: @@ -344,6 +372,7 @@ def _enrich_getmany_poll_span( client_id: str, consumer_group: str | None, message_count: int, + cluster_id: str | None = None, ) -> None: if not span.is_recording(): return @@ -357,6 +386,9 @@ def _enrich_getmany_poll_span( ) span.set_attribute(messaging_attributes.MESSAGING_CLIENT_ID, client_id) + if cluster_id is not None: + span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + if consumer_group is not None: span.set_attribute( messaging_attributes.MESSAGING_CONSUMER_GROUP_NAME, consumer_group @@ -384,6 +416,7 @@ def _enrich_getmany_topic_span( topic: str, partition: int, message_count: int, + cluster_id: str | None = None, ) -> None: if not span.is_recording(): return @@ -395,6 +428,7 @@ def _enrich_getmany_topic_span( topic=topic, partition=partition, key=None, + cluster_id=cluster_id, ) if consumer_group is not None: @@ -420,7 +454,8 @@ def _get_span_name(operation: str, topic: str): def _wrap_send( # type: ignore[reportUnusedFunction] - tracer: Tracer, async_produce_hook: ProduceHookT | None + tracer: Tracer, + async_produce_hook: ProduceHookT | None, ) -> Callable[..., Awaitable[asyncio.Future[RecordMetadata]]]: async def _traced_send( func: AIOKafkaSendProto, @@ -439,6 +474,7 @@ async def _traced_send( client_id = _extract_client_id(instance.client) key = _deserialize_key(_extract_send_key(args, kwargs)) partition = await _extract_send_partition(instance, args, kwargs) + cluster_id = _extract_cluster_id_from_client(instance.client) span_name = _get_span_name("send", topic) with tracer.start_as_current_span( span_name, kind=trace.SpanKind.PRODUCER @@ -450,6 +486,7 @@ async def _traced_send( topic=topic, partition=partition, key=key, + cluster_id=cluster_id, ) propagate.inject( headers, @@ -461,8 +498,13 @@ async def _traced_send( await async_produce_hook(span, args, kwargs) except Exception as hook_exception: # pylint: disable=W0703 _LOG.exception(hook_exception) - - return await func(*args, **kwargs) + result = await func(*args, **kwargs) + # After send(), broker has responded — refresh cluster ID in case + # metadata was not yet populated before the send started. + cluster_id = _extract_cluster_id_from_client(instance.client) + if cluster_id is not None and span.is_recording(): + span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + return result return _traced_send @@ -475,6 +517,7 @@ async def _create_consumer_span( bootstrap_servers: str | list[str], client_id: str, consumer_group: str | None, + cluster_id: str | None, args: tuple[aiokafka.TopicPartition, ...], kwargs: dict[str, Any], ) -> trace.Span: @@ -495,6 +538,7 @@ async def _create_consumer_span( partition=record.partition, key=_deserialize_key(record.key), offset=record.offset, + cluster_id=cluster_id, ) try: if async_consume_hook is not None: @@ -508,7 +552,8 @@ async def _create_consumer_span( def _wrap_getone( # type: ignore[reportUnusedFunction] - tracer: Tracer, async_consume_hook: ConsumeHookT | None + tracer: Tracer, + async_consume_hook: ConsumeHookT | None, ) -> Callable[..., Awaitable[aiokafka.ConsumerRecord[object, object]]]: async def _traced_getone( func: AIOKafkaGetOneProto, @@ -522,6 +567,7 @@ async def _traced_getone( bootstrap_servers = _extract_bootstrap_servers(instance._client) client_id = _extract_client_id(instance._client) consumer_group = _extract_consumer_group(instance) + cluster_id = _extract_cluster_id_from_client(instance._client) extracted_context = propagate.extract( record.headers, getter=_aiokafka_getter @@ -534,6 +580,7 @@ async def _traced_getone( bootstrap_servers, client_id, consumer_group, + cluster_id, args, kwargs, ) @@ -543,7 +590,8 @@ async def _traced_getone( def _wrap_getmany( # type: ignore[reportUnusedFunction] - tracer: Tracer, async_consume_hook: ConsumeHookT | None + tracer: Tracer, + async_consume_hook: ConsumeHookT | None, ) -> Callable[ ..., Awaitable[ @@ -567,6 +615,7 @@ async def _traced_getmany( bootstrap_servers = _extract_bootstrap_servers(instance._client) client_id = _extract_client_id(instance._client) consumer_group = _extract_consumer_group(instance) + cluster_id = _extract_cluster_id_from_client(instance._client) span_name = _get_span_name( "receive", @@ -581,6 +630,7 @@ async def _traced_getmany( client_id=client_id, consumer_group=consumer_group, message_count=sum(len(r) for r in records.values()), + cluster_id=cluster_id, ) for topic, topic_records in records.items(): @@ -596,6 +646,7 @@ async def _traced_getmany( topic=topic.topic, partition=topic.partition, message_count=len(topic_records), + cluster_id=cluster_id, ) for record in topic_records: @@ -610,6 +661,7 @@ async def _traced_getmany( bootstrap_servers, client_id, consumer_group, + cluster_id, args, kwargs, ) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index 0856d1389d..e363cd239a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -8,11 +8,13 @@ import aiokafka from opentelemetry.instrumentation.aiokafka.utils import ( + _MESSAGING_CLUSTER_ID, AIOKafkaContextGetter, AIOKafkaContextSetter, _aiokafka_getter, _aiokafka_setter, _create_consumer_span, + _extract_cluster_id_from_client, _extract_send_partition, _get_span_name, _wrap_getmany, @@ -114,6 +116,7 @@ async def wrap_send_helper( produce_hook = mock.AsyncMock() original_send_callback = mock.AsyncMock() kafka_producer = mock.MagicMock() + kafka_producer.client.cluster.cluster_id = None expected_span_name = _get_span_name("send", self.topic_name) wrapped_send = _wrap_send(tracer, produce_hook) @@ -139,6 +142,7 @@ async def wrap_send_helper( topic=self.topic_name, partition=extract_send_partition.return_value, key=None, + cluster_id=None, ) set_span_in_context.assert_called_once_with(span) @@ -179,6 +183,7 @@ async def test_wrap_getone( consume_hook = mock.AsyncMock() original_getone_callback = mock.AsyncMock() kafka_consumer = mock.MagicMock() + kafka_consumer._client.cluster.cluster_id = None wrapped_getone = _wrap_getone(tracer, consume_hook) record = await wrapped_getone( @@ -214,6 +219,7 @@ async def test_wrap_getone( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -259,6 +265,7 @@ async def test_wrap_getmany( } ) kafka_consumer = mock.MagicMock() + kafka_consumer._client.cluster.cluster_id = None _create_consumer_span.return_value = mock.MagicMock() wrapped_getmany = _wrap_getmany(tracer, consume_hook) @@ -295,6 +302,7 @@ async def test_wrap_getmany( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -328,6 +336,7 @@ async def test_create_consumer_span( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -352,12 +361,108 @@ async def test_create_consumer_span( partition=record.partition, key=str(record.key), offset=record.offset, + cluster_id=None, ) consume_hook.assert_awaited_once_with( span, record, self.args, self.kwargs ) detach.assert_called_once_with(attach.return_value) + async def test_cluster_id_attribute_set_on_send_span(self) -> None: + """Cluster ID is added to producer span when client metadata is available.""" + tracer = mock.MagicMock() + span = mock.MagicMock() + span.is_recording.return_value = True + tracer.start_as_current_span.return_value.__enter__ = mock.Mock( + return_value=span + ) + tracer.start_as_current_span.return_value.__exit__ = mock.Mock( + return_value=False + ) + + producer = mock.MagicMock() + producer.client._bootstrap_servers = "broker1:9092,broker2:9092" + producer.client._client_id = "test-client" + producer.client._wait_on_metadata = mock.AsyncMock() + producer.client.cluster.cluster_id = "test-cluster-uuid" + producer._key_serializer = None + producer._value_serializer = None + producer._partition.return_value = 0 + + wrapped_send = _wrap_send(tracer, None) + await wrapped_send(mock.AsyncMock(), producer, [self.topic_name], {}) + + set_attribute_calls = { + call.args[0]: call.args[1] + for call in span.set_attribute.call_args_list + } + self.assertEqual( + set_attribute_calls.get(_MESSAGING_CLUSTER_ID), + "test-cluster-uuid", + ) + + async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None: + """No cluster ID attribute is set when client metadata is not yet available.""" + tracer = mock.MagicMock() + span = mock.MagicMock() + span.is_recording.return_value = True + tracer.start_as_current_span.return_value.__enter__ = mock.Mock( + return_value=span + ) + tracer.start_as_current_span.return_value.__exit__ = mock.Mock( + return_value=False + ) + + producer = mock.MagicMock() + producer.client._bootstrap_servers = "unknown-broker:9092" + producer.client._client_id = "test-client" + producer.client._wait_on_metadata = mock.AsyncMock() + producer.client.cluster.cluster_id = None + producer._key_serializer = None + producer._value_serializer = None + producer._partition.return_value = 0 + + wrapped_send = _wrap_send(tracer, None) + await wrapped_send(mock.AsyncMock(), producer, [self.topic_name], {}) + + attribute_keys = [ + call.args[0] for call in span.set_attribute.call_args_list + ] + self.assertNotIn(_MESSAGING_CLUSTER_ID, attribute_keys) + + def test_extract_cluster_id_from_client_returns_cluster_id(self) -> None: + """Returns cluster ID from client.cluster.cluster_id when available.""" + client = mock.MagicMock() + client.cluster.cluster_id = "abc-uuid-1234" + self.assertEqual( + _extract_cluster_id_from_client(client), "abc-uuid-1234" + ) + + def test_extract_cluster_id_from_client_returns_none_when_cluster_id_none( + self, + ) -> None: + """Returns None when cluster_id is None (metadata not yet received).""" + client = mock.MagicMock() + client.cluster.cluster_id = None + self.assertIsNone(_extract_cluster_id_from_client(client)) + + def test_extract_cluster_id_from_client_returns_none_when_no_cluster_attr( + self, + ) -> None: + """Returns None when client has no cluster attribute.""" + client = mock.MagicMock(spec=[]) # no attributes + self.assertIsNone(_extract_cluster_id_from_client(client)) + + def test_extract_cluster_id_from_client_returns_none_on_exception( + self, + ) -> None: + """Returns None if attribute access raises unexpectedly.""" + client = mock.MagicMock() + type(client).cluster = mock.PropertyMock( + side_effect=RuntimeError("boom") + ) + self.assertIsNone(_extract_cluster_id_from_client(client)) + async def test_kafka_properties_extractor(self): aiokafka_instance_mock = mock.Mock() aiokafka_instance_mock._key_serializer = None diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py index 08aa4c3c7d..d75792553b 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/__init__.py @@ -117,6 +117,8 @@ def instrument_consumer(consumer: Consumer, tracer_provider=None): ... _create_new_consume_span, _end_current_consume_span, _enrich_span, + _fetch_cluster_id_background, + _get_real_instance, _get_span_name, _kafka_setter, ) @@ -143,6 +145,13 @@ class AutoInstrumentedProducer(Producer): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.config = _capture_config(args, kwargs) + bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers( + self + ) + if bootstrap_servers: + _fetch_cluster_id_background( + bootstrap_servers, self.config, instance=self + ) # This method is deliberately implemented in order to allow wrapt to wrap this function def produce(self, topic, value=None, *args, **kwargs): # pylint: disable=keyword-arg-before-vararg,useless-super-delegation @@ -154,6 +163,13 @@ def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.config = _capture_config(args, kwargs) self._current_consume_span = None + bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers( + self + ) + if bootstrap_servers: + _fetch_cluster_id_background( + bootstrap_servers, self.config, instance=self + ) # This method is deliberately implemented in order to allow wrapt to wrap this function def poll(self, timeout=-1): # pylint: disable=useless-super-delegation @@ -176,6 +192,13 @@ def __init__(self, producer: Producer, tracer: Tracer): # KafkaPropertiesExtractor.extract_bootstrap_servers can read it # through this proxy. self.config = getattr(producer, "config", None) + bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers( + self + ) + if bootstrap_servers: + _fetch_cluster_id_background( + bootstrap_servers, self.config, instance=producer + ) def flush(self, timeout=-1): return self._producer.flush(timeout) @@ -207,6 +230,13 @@ def __init__(self, consumer: Consumer, tracer: Tracer): self._current_context_token = None # See ProxiedProducer.__init__ for rationale. self.config = getattr(consumer, "config", None) + bootstrap_servers = KafkaPropertiesExtractor.extract_bootstrap_servers( + self + ) + if bootstrap_servers: + _fetch_cluster_id_background( + bootstrap_servers, self.config, instance=consumer + ) def close(self, *args, **kwargs): return ConfluentKafkaInstrumentor.wrap_close( @@ -397,6 +427,7 @@ def wrap_produce(func, instance, tracer, args, kwargs): topic, operation=MessagingOperationTypeValues.PUBLISH, bootstrap_servers=bootstrap_servers, + instance=_get_real_instance(instance), ) # Publish propagate.inject( headers, @@ -425,6 +456,7 @@ def wrap_poll(func, instance, tracer, args, kwargs): record.offset(), operation=MessagingOperationTypeValues.PROCESS, bootstrap_servers=bootstrap_servers, + instance=_get_real_instance(instance), ) instance._current_context_token = context.attach( trace.set_span_in_context(instance._current_consume_span) @@ -451,6 +483,7 @@ def wrap_consume(func, instance, tracer, args, kwargs): records[0].topic(), operation=MessagingOperationTypeValues.PROCESS, bootstrap_servers=bootstrap_servers, + instance=_get_real_instance(instance), ) instance._current_context_token = context.attach( trace.set_span_in_context(instance._current_consume_span) diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py index 27a188e7c7..55eeae6e4a 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py @@ -1,8 +1,10 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +import threading +import time from logging import getLogger -from typing import List, Optional +from typing import Any, Dict, List, Optional from opentelemetry import context, propagate from opentelemetry.propagators import textmap @@ -24,6 +26,116 @@ _LOG = getLogger(__name__) +_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" + +_CLUSTER_ID_TTL_SECONDS = 60 * 60 + +_kafka_cluster_id_cache: Dict[str, object] = {} +_kafka_cluster_id_lock = threading.Lock() +# Auth config stored from the first fetch per broker key; used for TTL re-fetches. +_kafka_cluster_id_config_cache: Dict[str, Optional[Dict[str, str]]] = {} + + +def _get_real_instance(instance: Any) -> Any: + """Unwrap Proxied* wrappers to get the underlying confluent-kafka Producer/Consumer.""" + return ( + getattr(instance, "_producer", None) + or getattr(instance, "_consumer", None) + or instance + ) + + +def _bootstrap_cache_key(bootstrap_servers: Optional[str]) -> str: + if not bootstrap_servers: + return "" + parts = [s.strip() for s in bootstrap_servers.split(",") if s.strip()] + return ",".join(sorted(parts)) + + +def _fetch_cluster_id_background( + bootstrap_servers: Optional[str], + base_config: Optional[Dict[str, str]] = None, + instance: Optional[Any] = None, +) -> None: + """Fetch cluster UUID in a daemon thread. Uses instance.list_topics() when available; falls back to AdminClient.""" + if not bootstrap_servers: + return + cache_key = _bootstrap_cache_key(bootstrap_servers) + if not cache_key: + return + + with _kafka_cluster_id_lock: + if base_config is not None: + _kafka_cluster_id_config_cache.setdefault(cache_key, base_config) + resolved_config = ( + _kafka_cluster_id_config_cache.get(cache_key) or base_config + ) + + existing = _kafka_cluster_id_cache.get(cache_key) + if isinstance(existing, tuple): + if time.monotonic() - existing[1] <= _CLUSTER_ID_TTL_SECONDS: + return # still fresh; stale value stays until re-fetch succeeds + # TTL expired — leave stale tuple in cache so callers get the old value + # while the background refresh runs, then the refresh will overwrite it. + elif existing is not None: + return # "" sentinel — first fetch already in progress + else: + _kafka_cluster_id_cache[cache_key] = ( + "" # mark first fetch in-flight + ) + + def _run() -> None: + try: + if instance is not None: + cluster_metadata = instance.list_topics(timeout=10) + else: + from confluent_kafka.admin import ( # noqa: PLC0415 # pylint: disable=import-outside-toplevel + AdminClient, + ) + + admin = AdminClient( + { + **(resolved_config or {}), + "bootstrap.servers": bootstrap_servers, + } + ) + try: + cluster_metadata = admin.list_topics(timeout=10) + finally: + # confluent_kafka.AdminClient has no explicit close(); deleting the reference + # allows librdkafka to release native resources via __del__ rather than waiting for GC. + del admin + cluster_id = getattr(cluster_metadata, "cluster_id", None) + if cluster_id: + with _kafka_cluster_id_lock: + _kafka_cluster_id_cache[cache_key] = ( + cluster_id, + time.monotonic(), + ) + else: + with _kafka_cluster_id_lock: + _kafka_cluster_id_cache.pop(cache_key, None) + except Exception: # pylint: disable=broad-except + with _kafka_cluster_id_lock: + _kafka_cluster_id_cache.pop(cache_key, None) + + thread = threading.Thread( + target=_run, daemon=True, name="otel-confluent-kafka-cluster-id" + ) + try: + thread.start() + except Exception: # pylint: disable=broad-except + with _kafka_cluster_id_lock: + _kafka_cluster_id_cache.pop(cache_key, None) + + +def _get_cluster_id(bootstrap_servers: Optional[str]) -> Optional[str]: + if not bootstrap_servers: + return None + cache_key = _bootstrap_cache_key(bootstrap_servers) + val = _kafka_cluster_id_cache.get(cache_key) + return val[0] if isinstance(val, tuple) else None + class KafkaPropertiesExtractor: @staticmethod @@ -161,6 +273,7 @@ def _enrich_span( offset: Optional[int] = None, operation: Optional[MessagingOperationTypeValues] = None, bootstrap_servers: Optional[str] = None, + instance: Optional[Any] = None, ): if not span.is_recording(): return @@ -178,11 +291,14 @@ def _enrich_span( if operation: span.set_attribute(MESSAGING_OPERATION, operation.value) - else: - span.set_attribute(SpanAttributes.MESSAGING_TEMP_DESTINATION, True) _set_bootstrap_servers_attributes(span, bootstrap_servers) + _fetch_cluster_id_background(bootstrap_servers, instance=instance) + cluster_id = _get_cluster_id(bootstrap_servers) + if cluster_id: + span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + # https://stackoverflow.com/questions/65935155/identify-and-find-specific-message-in-kafka-topic # A message within Kafka is uniquely defined by its topic name, topic partition and offset. if partition is not None and offset is not None and topic: 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 5dd0df77e6..38ad1e2e15 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 @@ -90,7 +90,12 @@ def process_msg(message): from opentelemetry import trace from opentelemetry.instrumentation.instrumentor import BaseInstrumentor from opentelemetry.instrumentation.kafka.package import _instruments -from opentelemetry.instrumentation.kafka.utils import _wrap_next, _wrap_send +from opentelemetry.instrumentation.kafka.utils import ( + KafkaPropertiesExtractor, + _fetch_cluster_id_background, + _wrap_next, + _wrap_send, +) from opentelemetry.instrumentation.kafka.version import __version__ from opentelemetry.instrumentation.utils import unwrap @@ -123,6 +128,32 @@ def _instrument(self, **kwargs): schema_url="https://opentelemetry.io/schemas/1.11.0", ) + def _wrap_producer_init(func, instance, args, kwargs): + func(*args, **kwargs) + bootstrap_servers = ( + KafkaPropertiesExtractor.extract_bootstrap_servers(instance) + ) + if bootstrap_servers: + _fetch_cluster_id_background( + bootstrap_servers, getattr(instance, "config", None) + ) + + def _wrap_consumer_init(func, instance, args, kwargs): + func(*args, **kwargs) + bootstrap_servers = ( + KafkaPropertiesExtractor.extract_bootstrap_servers(instance) + ) + if bootstrap_servers: + _fetch_cluster_id_background( + bootstrap_servers, getattr(instance, "config", None) + ) + + wrap_function_wrapper( + kafka.KafkaProducer, "__init__", _wrap_producer_init + ) + wrap_function_wrapper( + kafka.KafkaConsumer, "__init__", _wrap_consumer_init + ) wrap_function_wrapper( kafka.KafkaProducer, "send", _wrap_send(tracer, produce_hook) ) @@ -133,5 +164,7 @@ def _instrument(self, **kwargs): ) def _uninstrument(self, **kwargs): + unwrap(kafka.KafkaProducer, "__init__") + unwrap(kafka.KafkaConsumer, "__init__") unwrap(kafka.KafkaProducer, "send") unwrap(kafka.KafkaConsumer, "__next__") 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 00da544325..2cd972b082 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 @@ -2,6 +2,8 @@ # SPDX-License-Identifier: Apache-2.0 import json +import threading +import time from logging import getLogger from typing import Callable, Dict, List, Optional @@ -15,6 +17,108 @@ _LOG = getLogger(__name__) +_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" + +_SECURITY_CONFIG_KEYS = frozenset( + { + "ssl_cafile", + "ssl_certfile", + "ssl_keyfile", + "ssl_password", + "ssl_crlfile", + "ssl_check_hostname", + "ssl_context", + "security_protocol", + "sasl_mechanism", + "sasl_plain_username", + "sasl_plain_password", + "sasl_kerberos_service_name", + "sasl_kerberos_domain_name", + "sasl_oauth_token_provider", + } +) + +_CLUSTER_ID_TTL_SECONDS = 60 * 60 + +_kafka_cluster_id_cache: Dict[str, object] = {} +_kafka_cluster_id_lock = threading.Lock() + + +def _bootstrap_cache_key(servers) -> str: + if isinstance(servers, (list, tuple)): + return ",".join(sorted(str(s) for s in servers)) + parts = [s.strip() for s in str(servers).split(",") if s.strip()] + return ",".join(sorted(parts)) + + +def _fetch_cluster_id_background(bootstrap_servers, extra_config=None) -> None: + """Fetch cluster UUID via KafkaAdminClient in a daemon thread. Caches result by broker key.""" + cache_key = _bootstrap_cache_key(bootstrap_servers) + with _kafka_cluster_id_lock: + existing = _kafka_cluster_id_cache.get(cache_key) + if isinstance(existing, tuple): + if time.monotonic() - existing[1] <= _CLUSTER_ID_TTL_SECONDS: + return # still fresh; stale value stays until re-fetch succeeds + # TTL expired — leave stale tuple in cache so callers get the old value + # while the background refresh runs, then the refresh will overwrite it. + elif existing is not None: + return # "" sentinel — first fetch already in progress + else: + _kafka_cluster_id_cache[cache_key] = ( + "" # mark first fetch in-flight + ) + + def _run() -> None: + admin = None + try: + from kafka.admin import ( # noqa: PLC0415 # pylint: disable=import-outside-toplevel + KafkaAdminClient, + ) + + security_kwargs = { + k: v + for k, v in (extra_config or {}).items() + if k in _SECURITY_CONFIG_KEYS and v is not None + } + admin = KafkaAdminClient( + bootstrap_servers=bootstrap_servers, **security_kwargs + ) + info = admin.describe_cluster() + cluster_id = info.get("cluster_id") + if cluster_id: + with _kafka_cluster_id_lock: + _kafka_cluster_id_cache[cache_key] = ( + cluster_id, + time.monotonic(), + ) + else: + with _kafka_cluster_id_lock: + _kafka_cluster_id_cache.pop(cache_key, None) + except Exception: # pylint: disable=broad-except + with _kafka_cluster_id_lock: + _kafka_cluster_id_cache.pop(cache_key, None) + finally: + if admin is not None: + try: + admin.close() + except Exception: # pylint: disable=broad-except + pass + + thread = threading.Thread( + target=_run, daemon=True, name="otel-kafka-cluster-id" + ) + try: + thread.start() + except Exception: # pylint: disable=broad-except + with _kafka_cluster_id_lock: + _kafka_cluster_id_cache.pop(cache_key, None) + + +def _get_cluster_id(bootstrap_servers) -> Optional[str]: + cache_key = _bootstrap_cache_key(bootstrap_servers) + val = _kafka_cluster_id_cache.get(cache_key) + return val[0] if isinstance(val, tuple) else None + class KafkaPropertiesExtractor: @staticmethod @@ -135,6 +239,9 @@ def _enrich_span( span.set_attribute( SpanAttributes.MESSAGING_URL, json.dumps(bootstrap_servers) ) + cluster_id = _get_cluster_id(bootstrap_servers) + if cluster_id: + span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) def _get_span_name(operation: str, topic: str): @@ -156,6 +263,7 @@ def _traced_send(func, instance, args, kwargs): instance, args, kwargs ) span_name = _get_span_name("send", topic) + _fetch_cluster_id_background(bootstrap_servers, dict(instance.config)) with tracer.start_as_current_span( span_name, kind=trace.SpanKind.PRODUCER ) as span: @@ -170,8 +278,7 @@ def _traced_send(func, instance, args, kwargs): produce_hook(span, args, kwargs) except Exception as hook_exception: # pylint: disable=W0703 _LOG.exception(hook_exception) - - return func(*args, **kwargs) + return func(*args, **kwargs) return _traced_send @@ -214,7 +321,9 @@ def _traced_next(func, instance, args, kwargs): bootstrap_servers = ( KafkaPropertiesExtractor.extract_bootstrap_servers(instance) ) - + _fetch_cluster_id_background( + bootstrap_servers, dict(instance.config) + ) extracted_context = propagate.extract( record.headers, getter=_kafka_getter ) From 193777e4c69d3708274441af88f1d3d48d11ac7d Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Tue, 14 Jul 2026 12:35:29 +0530 Subject: [PATCH 2/9] feat(instrumentation/aiokafka): add messaging.kafka.cluster.id to producer/consumer spans aiokafka's ClusterMetadata receives cluster_id in every MetadataResponse but does not persist it as an attribute. Wrap cluster.update_metadata before start() to cache the cluster_id; read it from _extract_cluster_id_from_client at span creation time. Also refresh the attribute after send() in case the first metadata response arrived mid-send. Add _wrap_start_producer / _wrap_start_consumer wrappers and register them on AIOKafkaProducer.start / AIOKafkaConsumer.start in _instrument / _uninstrument. Tested E2E against PLAINTEXT, SASL/PLAIN, and SASL/SCRAM-SHA-256 listeners; messaging.kafka.cluster.id appears in all producer and consumer spans. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/aiokafka/__init__.py | 14 +++++ .../instrumentation/aiokafka/utils.py | 54 ++++++++++++++++++- .../tests/test_instrumentation.py | 12 +++++ .../tests/test_utils.py | 48 +++++++++++++++++ 4 files changed, 126 insertions(+), 2 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py index f694fb51be..cac438be0f 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -109,6 +109,8 @@ async def produce(): _wrap_getmany, _wrap_getone, _wrap_send, + _wrap_start_consumer, + _wrap_start_producer, ) from opentelemetry.instrumentation.aiokafka.version import __version__ from opentelemetry.instrumentation.instrumentor import BaseInstrumentor @@ -165,6 +167,16 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): schema_url=Schemas.V1_27_0.value, ) + wrap_function_wrapper( + aiokafka.AIOKafkaProducer, + "start", + _wrap_start_producer(), + ) + wrap_function_wrapper( + aiokafka.AIOKafkaConsumer, + "start", + _wrap_start_consumer(), + ) wrap_function_wrapper( aiokafka.AIOKafkaProducer, "send", @@ -182,6 +194,8 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): ) def _uninstrument(self, **kwargs: Unpack[UninstrumentKwargs]): + unwrap(aiokafka.AIOKafkaProducer, "start") + unwrap(aiokafka.AIOKafkaConsumer, "start") unwrap(aiokafka.AIOKafkaProducer, "send") unwrap(aiokafka.AIOKafkaConsumer, "getone") unwrap(aiokafka.AIOKafkaConsumer, "getmany") diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py index 8b27ac78b3..5edd61b538 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -92,13 +92,33 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str: return client._client_id +def _patch_cluster_id_capture(client: aiokafka.AIOKafkaClient) -> None: + """Wrap cluster.update_metadata once to populate cluster.cluster_id. + + aiokafka's ClusterMetadata receives cluster_id in every MetadataResponse + but does not store it as an attribute. This one-time patch intercepts + update_metadata so that _extract_cluster_id_from_client can read it. + """ + cluster = getattr(client, "cluster", None) + if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): + return + original_update = cluster.update_metadata + + def _patched_update(metadata: Any) -> None: + cluster_id = getattr(metadata, "cluster_id", None) + if cluster_id: + cluster.cluster_id = cluster_id + original_update(metadata) + + cluster.update_metadata = _patched_update + cluster._otel_cluster_id_patched = True + + def _extract_cluster_id_from_client( client: aiokafka.AIOKafkaClient, ) -> str | None: """Read cluster ID from the aiokafka client's cached cluster metadata. - aiokafka sets AIOKafkaClient.cluster.cluster_id after the first successful - broker metadata response — no extra connection or background thread needed. Returns None if metadata has not been received yet. """ try: @@ -669,3 +689,33 @@ async def _traced_getmany( return records return _traced_getmany + + +def _wrap_start_producer() -> Callable[..., Awaitable[None]]: + """Wrap AIOKafkaProducer.start to install the cluster_id capture patch.""" + + async def _traced_start( + func: Callable[..., Awaitable[None]], + instance: aiokafka.AIOKafkaProducer, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: + _patch_cluster_id_capture(instance.client) + return await func(*args, **kwargs) + + return _traced_start + + +def _wrap_start_consumer() -> Callable[..., Awaitable[None]]: + """Wrap AIOKafkaConsumer.start to install the cluster_id capture patch.""" + + async def _traced_start( + func: Callable[..., Awaitable[None]], + instance: aiokafka.AIOKafkaConsumer, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> None: + _patch_cluster_id_capture(instance._client) + return await func(*args, **kwargs) + + return _traced_start diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py index 35ec4bcb09..108bfdd266 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_instrumentation.py @@ -30,6 +30,12 @@ def test_instrument_api(self) -> None: instrumentation = AIOKafkaInstrumentor() instrumentation.instrument() + self.assertTrue( + isinstance(AIOKafkaProducer.start, BoundFunctionWrapper) + ) + self.assertTrue( + isinstance(AIOKafkaConsumer.start, BoundFunctionWrapper) + ) self.assertTrue( isinstance(AIOKafkaProducer.send, BoundFunctionWrapper) ) @@ -41,6 +47,12 @@ def test_instrument_api(self) -> None: ) instrumentation.uninstrument() + self.assertFalse( + isinstance(AIOKafkaProducer.start, BoundFunctionWrapper) + ) + self.assertFalse( + isinstance(AIOKafkaConsumer.start, BoundFunctionWrapper) + ) self.assertFalse( isinstance(AIOKafkaProducer.send, BoundFunctionWrapper) ) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index e363cd239a..4bc39a049e 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -17,6 +17,7 @@ _extract_cluster_id_from_client, _extract_send_partition, _get_span_name, + _patch_cluster_id_capture, _wrap_getmany, _wrap_getone, _wrap_send, @@ -430,6 +431,53 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None: ] self.assertNotIn(_MESSAGING_CLUSTER_ID, attribute_keys) + def test_patch_cluster_id_capture_sets_cluster_id_from_metadata( + self, + ) -> None: + """_patch_cluster_id_capture intercepts update_metadata and sets cluster_id.""" + cluster = mock.MagicMock(spec=[]) + cluster.cluster_id = None + update_calls: list[object] = [] + + def original_update(metadata: object) -> None: + update_calls.append(metadata) + + cluster.update_metadata = original_update + client = mock.MagicMock() + client.cluster = cluster + + _patch_cluster_id_capture(client) + + metadata = mock.MagicMock() + metadata.cluster_id = "test-cluster-uuid" + cluster.update_metadata(metadata) + + self.assertEqual(cluster.cluster_id, "test-cluster-uuid") + self.assertEqual(update_calls, [metadata]) + + def test_patch_cluster_id_capture_is_idempotent(self) -> None: + """Calling _patch_cluster_id_capture twice does not double-wrap.""" + cluster = mock.MagicMock(spec=[]) + update_calls: list[object] = [] + cluster.update_metadata = lambda m: update_calls.append(m) + client = mock.MagicMock() + client.cluster = cluster + + _patch_cluster_id_capture(client) + _patch_cluster_id_capture(client) + + metadata = mock.MagicMock() + metadata.cluster_id = "id-1" + cluster.update_metadata(metadata) + + # original called exactly once despite two patch calls + self.assertEqual(len(update_calls), 1) + + def test_patch_cluster_id_capture_ignores_none_cluster(self) -> None: + """_patch_cluster_id_capture is a no-op when client has no cluster.""" + client = mock.MagicMock(spec=[]) # no attributes + _patch_cluster_id_capture(client) # must not raise + def test_extract_cluster_id_from_client_returns_cluster_id(self) -> None: """Returns cluster ID from client.cluster.cluster_id when available.""" client = mock.MagicMock() From 1fe9ae1a3d2eec761e4c7341ffa8960c0672b92c Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Tue, 14 Jul 2026 12:58:14 +0530 Subject: [PATCH 3/9] fix(instrumentation/aiokafka): move start wrappers inline to fix pyright/pylint Move _start_producer_wrapper and _start_consumer_wrapper from utils.py into __init__.py where they are used, eliminating the unnecessary factory pattern (no captured variables) and resolving: - pyright reportUnusedFunction: functions were flagged as unused because pyright checks within-file usage for module-level private functions - pylint W0108/R6301: remove unnecessary lambda, add @staticmethod to test_patch_cluster_id_capture_ignores_none_cluster Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/aiokafka/__init__.py | 29 ++++++++++++++---- .../instrumentation/aiokafka/utils.py | 30 ------------------- .../tests/test_utils.py | 5 ++-- 3 files changed, 27 insertions(+), 37 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py index cac438be0f..daa30f9e3f 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -96,7 +96,7 @@ async def produce(): from __future__ import annotations from inspect import iscoroutinefunction -from typing import TYPE_CHECKING, Collection +from typing import TYPE_CHECKING, Any, Awaitable, Callable, Collection import aiokafka from wrapt import ( @@ -106,11 +106,10 @@ async def produce(): from opentelemetry import trace from opentelemetry.instrumentation.aiokafka.package import _instruments from opentelemetry.instrumentation.aiokafka.utils import ( + _patch_cluster_id_capture, _wrap_getmany, _wrap_getone, _wrap_send, - _wrap_start_consumer, - _wrap_start_producer, ) from opentelemetry.instrumentation.aiokafka.version import __version__ from opentelemetry.instrumentation.instrumentor import BaseInstrumentor @@ -133,6 +132,26 @@ class UninstrumentKwargs(TypedDict, total=False): pass +async def _start_producer_wrapper( + func: Callable[..., Awaitable[None]], + instance: aiokafka.AIOKafkaProducer, + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> None: + _patch_cluster_id_capture(instance.client) + await func(*args, **kwargs) + + +async def _start_consumer_wrapper( + func: Callable[..., Awaitable[None]], + instance: aiokafka.AIOKafkaConsumer, + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> None: + _patch_cluster_id_capture(instance._client) + await func(*args, **kwargs) + + class AIOKafkaInstrumentor(BaseInstrumentor): """An instrumentor for kafka module See `BaseInstrumentor` @@ -170,12 +189,12 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): wrap_function_wrapper( aiokafka.AIOKafkaProducer, "start", - _wrap_start_producer(), + _start_producer_wrapper, ) wrap_function_wrapper( aiokafka.AIOKafkaConsumer, "start", - _wrap_start_consumer(), + _start_consumer_wrapper, ) wrap_function_wrapper( aiokafka.AIOKafkaProducer, diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py index 5edd61b538..d279b7575a 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -689,33 +689,3 @@ async def _traced_getmany( return records return _traced_getmany - - -def _wrap_start_producer() -> Callable[..., Awaitable[None]]: - """Wrap AIOKafkaProducer.start to install the cluster_id capture patch.""" - - async def _traced_start( - func: Callable[..., Awaitable[None]], - instance: aiokafka.AIOKafkaProducer, - args: tuple[Any, ...], - kwargs: dict[str, Any], - ) -> None: - _patch_cluster_id_capture(instance.client) - return await func(*args, **kwargs) - - return _traced_start - - -def _wrap_start_consumer() -> Callable[..., Awaitable[None]]: - """Wrap AIOKafkaConsumer.start to install the cluster_id capture patch.""" - - async def _traced_start( - func: Callable[..., Awaitable[None]], - instance: aiokafka.AIOKafkaConsumer, - args: tuple[Any, ...], - kwargs: dict[str, Any], - ) -> None: - _patch_cluster_id_capture(instance._client) - return await func(*args, **kwargs) - - return _traced_start diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index 4bc39a049e..d95d1e7569 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -459,7 +459,7 @@ def test_patch_cluster_id_capture_is_idempotent(self) -> None: """Calling _patch_cluster_id_capture twice does not double-wrap.""" cluster = mock.MagicMock(spec=[]) update_calls: list[object] = [] - cluster.update_metadata = lambda m: update_calls.append(m) + cluster.update_metadata = update_calls.append client = mock.MagicMock() client.cluster = cluster @@ -473,7 +473,8 @@ def test_patch_cluster_id_capture_is_idempotent(self) -> None: # original called exactly once despite two patch calls self.assertEqual(len(update_calls), 1) - def test_patch_cluster_id_capture_ignores_none_cluster(self) -> None: + @staticmethod + def test_patch_cluster_id_capture_ignores_none_cluster() -> None: """_patch_cluster_id_capture is a no-op when client has no cluster.""" client = mock.MagicMock(spec=[]) # no attributes _patch_cluster_id_capture(client) # must not raise From e4c312b84190e38aebb26660d6c6db75e106f587 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Tue, 14 Jul 2026 13:04:52 +0530 Subject: [PATCH 4/9] fix(instrumentation/aiokafka): move _patch_cluster_id_capture to __init__ to satisfy pyright pyright reportUnusedFunction flags any module-level private function that is not accessed within the same file. _patch_cluster_id_capture was in utils.py but only called from __init__.py, triggering the error. Moving it to __init__.py where it is defined and called keeps all cross-file import graphs clean without requiring type: ignore annotations (prohibited by AGENTS.md). Update test_utils.py to import _patch_cluster_id_capture from __init__ instead. Assisted-by: Claude Sonnet 4.6 --- .../instrumentation/aiokafka/__init__.py | 17 +++++++++++++- .../instrumentation/aiokafka/utils.py | 22 ------------------- .../tests/test_utils.py | 2 +- 3 files changed, 17 insertions(+), 24 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py index daa30f9e3f..c55a439ecb 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -106,7 +106,6 @@ async def produce(): from opentelemetry import trace from opentelemetry.instrumentation.aiokafka.package import _instruments from opentelemetry.instrumentation.aiokafka.utils import ( - _patch_cluster_id_capture, _wrap_getmany, _wrap_getone, _wrap_send, @@ -132,6 +131,22 @@ class UninstrumentKwargs(TypedDict, total=False): pass +def _patch_cluster_id_capture(client: aiokafka.AIOKafkaClient) -> None: + cluster = getattr(client, "cluster", None) + if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): + return + original_update = cluster.update_metadata + + def _patched_update(metadata: Any) -> None: + cluster_id = getattr(metadata, "cluster_id", None) + if cluster_id: + cluster.cluster_id = cluster_id + original_update(metadata) + + cluster.update_metadata = _patched_update + cluster._otel_cluster_id_patched = True + + async def _start_producer_wrapper( func: Callable[..., Awaitable[None]], instance: aiokafka.AIOKafkaProducer, diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py index d279b7575a..30042f30f5 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -92,28 +92,6 @@ def _extract_client_id(client: aiokafka.AIOKafkaClient) -> str: return client._client_id -def _patch_cluster_id_capture(client: aiokafka.AIOKafkaClient) -> None: - """Wrap cluster.update_metadata once to populate cluster.cluster_id. - - aiokafka's ClusterMetadata receives cluster_id in every MetadataResponse - but does not store it as an attribute. This one-time patch intercepts - update_metadata so that _extract_cluster_id_from_client can read it. - """ - cluster = getattr(client, "cluster", None) - if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): - return - original_update = cluster.update_metadata - - def _patched_update(metadata: Any) -> None: - cluster_id = getattr(metadata, "cluster_id", None) - if cluster_id: - cluster.cluster_id = cluster_id - original_update(metadata) - - cluster.update_metadata = _patched_update - cluster._otel_cluster_id_patched = True - - def _extract_cluster_id_from_client( client: aiokafka.AIOKafkaClient, ) -> str | None: diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index d95d1e7569..95a2d27ae9 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -7,6 +7,7 @@ import aiokafka +from opentelemetry.instrumentation.aiokafka import _patch_cluster_id_capture from opentelemetry.instrumentation.aiokafka.utils import ( _MESSAGING_CLUSTER_ID, AIOKafkaContextGetter, @@ -17,7 +18,6 @@ _extract_cluster_id_from_client, _extract_send_partition, _get_span_name, - _patch_cluster_id_capture, _wrap_getmany, _wrap_getone, _wrap_send, From e1ceb1b969c8070c7191ab50a13b78af33736900 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Tue, 14 Jul 2026 18:29:36 +0530 Subject: [PATCH 5/9] fix(instrumentation/aiokafka): correct changelog entry for PR 4727 Assisted-by: Claude Sonnet 4.6 --- .../.changelog/4727.added | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added index 11db77dcc7..f63468f4f4 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added @@ -1 +1 @@ -`opentelemetry-instrumentation-aiokafka`: add `capture_experimental_span_attributes` option to gate `messaging.cluster.id` on producer and consumer spans +`opentelemetry-instrumentation-aiokafka`: emit `messaging.kafka.cluster.id` on producer and consumer spans From f26d6101351c6d459f0477cf9b7de1ca222eb6c1 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Wed, 15 Jul 2026 23:44:39 +0530 Subject: [PATCH 6/9] fix(instrumentation/kafka-python): source cluster id from client metadata, not a separate admin client Mirror the aiokafka instrumentation: read messaging.kafka.cluster.id from the client's own already-resolved metadata instead of opening a separate KafkaAdminClient in a background thread. Removes the hand-maintained security-config allowlist (which also omitted ssl_ciphers) and opens no extra broker connection. The id is captured from the MetadataResponse via update_metadata, so it works on kafka-python 2.0.x (which does not persist cluster_id on ClusterMetadata) as well as 2.1+. Assisted-by: Claude Opus 4.8 --- .../instrumentation/kafka/__init__.py | 19 +- .../instrumentation/kafka/utils.py | 163 ++++++------------ .../tests/test_utils.py | 67 ++++++- 3 files changed, 125 insertions(+), 124 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 38ad1e2e15..26d57f5afd 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 @@ -91,8 +91,7 @@ def process_msg(message): from opentelemetry.instrumentation.instrumentor import BaseInstrumentor from opentelemetry.instrumentation.kafka.package import _instruments from opentelemetry.instrumentation.kafka.utils import ( - KafkaPropertiesExtractor, - _fetch_cluster_id_background, + _patch_cluster_id_capture, _wrap_next, _wrap_send, ) @@ -130,23 +129,11 @@ def _instrument(self, **kwargs): def _wrap_producer_init(func, instance, args, kwargs): func(*args, **kwargs) - bootstrap_servers = ( - KafkaPropertiesExtractor.extract_bootstrap_servers(instance) - ) - if bootstrap_servers: - _fetch_cluster_id_background( - bootstrap_servers, getattr(instance, "config", None) - ) + _patch_cluster_id_capture(instance) def _wrap_consumer_init(func, instance, args, kwargs): func(*args, **kwargs) - bootstrap_servers = ( - KafkaPropertiesExtractor.extract_bootstrap_servers(instance) - ) - if bootstrap_servers: - _fetch_cluster_id_background( - bootstrap_servers, getattr(instance, "config", None) - ) + _patch_cluster_id_capture(instance) wrap_function_wrapper( kafka.KafkaProducer, "__init__", _wrap_producer_init 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 2cd972b082..59e0563dfc 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 @@ -2,8 +2,6 @@ # SPDX-License-Identifier: Apache-2.0 import json -import threading -import time from logging import getLogger from typing import Callable, Dict, List, Optional @@ -19,105 +17,47 @@ _MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" -_SECURITY_CONFIG_KEYS = frozenset( - { - "ssl_cafile", - "ssl_certfile", - "ssl_keyfile", - "ssl_password", - "ssl_crlfile", - "ssl_check_hostname", - "ssl_context", - "security_protocol", - "sasl_mechanism", - "sasl_plain_username", - "sasl_plain_password", - "sasl_kerberos_service_name", - "sasl_kerberos_domain_name", - "sasl_oauth_token_provider", - } -) - -_CLUSTER_ID_TTL_SECONDS = 60 * 60 - -_kafka_cluster_id_cache: Dict[str, object] = {} -_kafka_cluster_id_lock = threading.Lock() - - -def _bootstrap_cache_key(servers) -> str: - if isinstance(servers, (list, tuple)): - return ",".join(sorted(str(s) for s in servers)) - parts = [s.strip() for s in str(servers).split(",") if s.strip()] - return ",".join(sorted(parts)) - - -def _fetch_cluster_id_background(bootstrap_servers, extra_config=None) -> None: - """Fetch cluster UUID via KafkaAdminClient in a daemon thread. Caches result by broker key.""" - cache_key = _bootstrap_cache_key(bootstrap_servers) - with _kafka_cluster_id_lock: - existing = _kafka_cluster_id_cache.get(cache_key) - if isinstance(existing, tuple): - if time.monotonic() - existing[1] <= _CLUSTER_ID_TTL_SECONDS: - return # still fresh; stale value stays until re-fetch succeeds - # TTL expired — leave stale tuple in cache so callers get the old value - # while the background refresh runs, then the refresh will overwrite it. - elif existing is not None: - return # "" sentinel — first fetch already in progress - else: - _kafka_cluster_id_cache[cache_key] = ( - "" # mark first fetch in-flight - ) - def _run() -> None: - admin = None - try: - from kafka.admin import ( # noqa: PLC0415 # pylint: disable=import-outside-toplevel - KafkaAdminClient, - ) +def _get_cluster_metadata(instance): + """Return the kafka-python ``ClusterMetadata`` for a producer or consumer. - security_kwargs = { - k: v - for k, v in (extra_config or {}).items() - if k in _SECURITY_CONFIG_KEYS and v is not None - } - admin = KafkaAdminClient( - bootstrap_servers=bootstrap_servers, **security_kwargs - ) - info = admin.describe_cluster() - cluster_id = info.get("cluster_id") - if cluster_id: - with _kafka_cluster_id_lock: - _kafka_cluster_id_cache[cache_key] = ( - cluster_id, - time.monotonic(), - ) - else: - with _kafka_cluster_id_lock: - _kafka_cluster_id_cache.pop(cache_key, None) - except Exception: # pylint: disable=broad-except - with _kafka_cluster_id_lock: - _kafka_cluster_id_cache.pop(cache_key, None) - finally: - if admin is not None: - try: - admin.close() - except Exception: # pylint: disable=broad-except - pass - - thread = threading.Thread( - target=_run, daemon=True, name="otel-kafka-cluster-id" - ) - try: - thread.start() - except Exception: # pylint: disable=broad-except - with _kafka_cluster_id_lock: - _kafka_cluster_id_cache.pop(cache_key, None) - - -def _get_cluster_id(bootstrap_servers) -> Optional[str]: - cache_key = _bootstrap_cache_key(bootstrap_servers) - val = _kafka_cluster_id_cache.get(cache_key) - return val[0] if isinstance(val, tuple) else None + ``KafkaProducer`` exposes it as ``_metadata``; ``KafkaConsumer`` as + ``_client.cluster``. + """ + cluster = getattr(instance, "_metadata", None) + if cluster is not None: + return cluster + return getattr(getattr(instance, "_client", None), "cluster", None) + + +def _patch_cluster_id_capture(instance) -> None: + """Capture the cluster id from the client's own metadata responses. + + Reads from the client's already-resolved metadata; opens no extra broker + connection. kafka-python < 2.1 does not persist ``cluster_id`` on + ``ClusterMetadata``, but the ``MetadataResponse`` (v2+) passed to + ``update_metadata`` carries it, so wrap ``update_metadata`` to store it. + Guarded so each client's metadata object is patched at most once. + """ + cluster = _get_cluster_metadata(instance) + if cluster is None or getattr(cluster, "_otel_cluster_id_patched", False): + return + original_update = cluster.update_metadata + + def _patched_update(metadata): + result = original_update(metadata) + cluster_id = getattr(metadata, "cluster_id", None) + if cluster_id: + cluster.cluster_id = cluster_id + return result + + cluster.update_metadata = _patched_update + cluster._otel_cluster_id_patched = True + + +def _extract_cluster_id(instance) -> Optional[str]: + cluster_id = getattr(_get_cluster_metadata(instance), "cluster_id", None) + return cluster_id if cluster_id else None class KafkaPropertiesExtractor: @@ -230,7 +170,11 @@ def set(self, carrier: textmap.CarrierT, key: str, value: str) -> None: def _enrich_span( - span, bootstrap_servers: List[str], topic: str, partition: int + span, + bootstrap_servers: List[str], + topic: str, + partition: int, + cluster_id: Optional[str] = None, ): if span.is_recording(): span.set_attribute(SpanAttributes.MESSAGING_SYSTEM, "kafka") @@ -239,7 +183,6 @@ def _enrich_span( span.set_attribute( SpanAttributes.MESSAGING_URL, json.dumps(bootstrap_servers) ) - cluster_id = _get_cluster_id(bootstrap_servers) if cluster_id: span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) @@ -262,12 +205,12 @@ def _traced_send(func, instance, args, kwargs): partition = KafkaPropertiesExtractor.extract_send_partition( instance, args, kwargs ) + cluster_id = _extract_cluster_id(instance) span_name = _get_span_name("send", topic) - _fetch_cluster_id_background(bootstrap_servers, dict(instance.config)) with tracer.start_as_current_span( span_name, kind=trace.SpanKind.PRODUCER ) as span: - _enrich_span(span, bootstrap_servers, topic, partition) + _enrich_span(span, bootstrap_servers, topic, partition, cluster_id) propagate.inject( headers, context=trace.set_span_in_context(span), @@ -289,6 +232,7 @@ def _create_consumer_span( record, extracted_context, bootstrap_servers, + cluster_id, args, kwargs, ): @@ -300,7 +244,13 @@ def _create_consumer_span( ) as span: new_context = trace.set_span_in_context(span, extracted_context) token = context.attach(new_context) - _enrich_span(span, bootstrap_servers, record.topic, record.partition) + _enrich_span( + span, + bootstrap_servers, + record.topic, + record.partition, + cluster_id, + ) try: if callable(consume_hook): consume_hook(span, record, args, kwargs) @@ -321,9 +271,7 @@ def _traced_next(func, instance, args, kwargs): bootstrap_servers = ( KafkaPropertiesExtractor.extract_bootstrap_servers(instance) ) - _fetch_cluster_id_background( - bootstrap_servers, dict(instance.config) - ) + cluster_id = _extract_cluster_id(instance) extracted_context = propagate.extract( record.headers, getter=_kafka_getter ) @@ -333,6 +281,7 @@ def _traced_next(func, instance, args, kwargs): record, extracted_context, bootstrap_servers, + cluster_id, args, kwargs, ) diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py index 3d5935a25f..f02625f1be 100644 --- a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py @@ -7,9 +7,11 @@ from opentelemetry.instrumentation.kafka.utils import ( KafkaPropertiesExtractor, _create_consumer_span, + _extract_cluster_id, _get_span_name, _kafka_getter, _kafka_setter, + _patch_cluster_id_capture, _wrap_next, _wrap_send, ) @@ -30,6 +32,9 @@ def setUp(self) -> None: @mock.patch( "opentelemetry.instrumentation.kafka.utils.KafkaPropertiesExtractor.extract_send_partition" ) + @mock.patch( + "opentelemetry.instrumentation.kafka.utils._extract_cluster_id" + ) @mock.patch("opentelemetry.instrumentation.kafka.utils._enrich_span") @mock.patch("opentelemetry.trace.set_span_in_context") @mock.patch("opentelemetry.propagate.inject") @@ -38,6 +43,7 @@ def test_wrap_send_with_topic_as_arg( inject: mock.MagicMock, set_span_in_context: mock.MagicMock, enrich_span: mock.MagicMock, + extract_cluster_id: mock.MagicMock, extract_send_partition: mock.MagicMock, extract_bootstrap_servers: mock.MagicMock, ) -> None: @@ -45,6 +51,7 @@ def test_wrap_send_with_topic_as_arg( inject, set_span_in_context, enrich_span, + extract_cluster_id, extract_send_partition, extract_bootstrap_servers, ) @@ -55,6 +62,9 @@ def test_wrap_send_with_topic_as_arg( @mock.patch( "opentelemetry.instrumentation.kafka.utils.KafkaPropertiesExtractor.extract_send_partition" ) + @mock.patch( + "opentelemetry.instrumentation.kafka.utils._extract_cluster_id" + ) @mock.patch("opentelemetry.instrumentation.kafka.utils._enrich_span") @mock.patch("opentelemetry.trace.set_span_in_context") @mock.patch("opentelemetry.propagate.inject") @@ -63,6 +73,7 @@ def test_wrap_send_with_topic_as_kwarg( inject: mock.MagicMock, set_span_in_context: mock.MagicMock, enrich_span: mock.MagicMock, + extract_cluster_id: mock.MagicMock, extract_send_partition: mock.MagicMock, extract_bootstrap_servers: mock.MagicMock, ) -> None: @@ -72,6 +83,7 @@ def test_wrap_send_with_topic_as_kwarg( inject, set_span_in_context, enrich_span, + extract_cluster_id, extract_send_partition, extract_bootstrap_servers, ) @@ -81,6 +93,7 @@ def wrap_send_helper( inject: mock.MagicMock, set_span_in_context: mock.MagicMock, enrich_span: mock.MagicMock, + extract_cluster_id: mock.MagicMock, extract_send_partition: mock.MagicMock, extract_bootstrap_servers: mock.MagicMock, ) -> None: @@ -109,6 +122,7 @@ def wrap_send_helper( extract_bootstrap_servers.return_value, self.topic_name, extract_send_partition.return_value, + extract_cluster_id.return_value, ) set_span_in_context.assert_called_once_with(span) @@ -124,6 +138,9 @@ def wrap_send_helper( ) self.assertEqual(retval, original_send_callback.return_value) + @mock.patch( + "opentelemetry.instrumentation.kafka.utils._extract_cluster_id" + ) @mock.patch("opentelemetry.propagate.extract") @mock.patch( "opentelemetry.instrumentation.kafka.utils._create_consumer_span" @@ -136,6 +153,7 @@ def test_wrap_next( extract_bootstrap_servers: mock.MagicMock, _create_consumer_span: mock.MagicMock, extract: mock.MagicMock, + extract_cluster_id: mock.MagicMock, ) -> None: tracer = mock.MagicMock() consume_hook = mock.MagicMock() @@ -164,6 +182,7 @@ def test_wrap_next( record, context, bootstrap_servers, + extract_cluster_id.return_value, self.args, self.kwargs, ) @@ -184,6 +203,7 @@ def test_create_consumer_span( bootstrap_servers = mock.MagicMock() extracted_context = mock.MagicMock() record = mock.MagicMock() + cluster_id = mock.MagicMock() _create_consumer_span( tracer, @@ -191,6 +211,7 @@ def test_create_consumer_span( record, extracted_context, bootstrap_servers, + cluster_id, self.args, self.kwargs, ) @@ -207,7 +228,11 @@ def test_create_consumer_span( attach.assert_called_once_with(set_span_in_context.return_value) enrich_span.assert_called_once_with( - span, bootstrap_servers, record.topic, record.partition + span, + bootstrap_servers, + record.topic, + record.partition, + cluster_id, ) consume_hook.assert_called_once_with( span, record, self.args, self.kwargs @@ -238,3 +263,43 @@ def test_kafka_properties_extractor( ) is None ) + + def test_extract_cluster_id_from_producer_metadata(self) -> None: + producer = mock.MagicMock() + producer._metadata.cluster_id = "test-cluster-id" + self.assertEqual(_extract_cluster_id(producer), "test-cluster-id") + + def test_extract_cluster_id_from_consumer_client(self) -> None: + consumer = mock.MagicMock(spec=["_client"]) + consumer._client.cluster.cluster_id = "test-cluster-id" + self.assertEqual(_extract_cluster_id(consumer), "test-cluster-id") + + def test_extract_cluster_id_absent_returns_none(self) -> None: + instance = mock.MagicMock(spec=[]) + self.assertIsNone(_extract_cluster_id(instance)) + + def test_patch_cluster_id_capture_captures_and_is_idempotent( + self, + ) -> None: + class FakeCluster: + def __init__(self) -> None: + self.update_calls = 0 + + def update_metadata(self, metadata) -> None: + self.update_calls += 1 + + class FakeMetadataResponse: + cluster_id = "test-cluster-id" + + cluster = FakeCluster() + instance = mock.MagicMock() + instance._metadata = cluster + + _patch_cluster_id_capture(instance) + # Second call must not double-wrap update_metadata. + _patch_cluster_id_capture(instance) + + cluster.update_metadata(FakeMetadataResponse()) + + self.assertEqual(cluster.update_calls, 1) + self.assertEqual(_extract_cluster_id(instance), "test-cluster-id") From d61b42821885dbd2a953e32bef113dca08f80f7e Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Thu, 16 Jul 2026 00:00:01 +0530 Subject: [PATCH 7/9] fix: place PR 4727 changelog fragment in root .changelog/ for all three kafka packages aiokafka/confluent-kafka/kafka-python are coordinated packages, so their towncrier fragment belongs in the root .changelog/ directory (per CONTRIBUTING.md), not a package-level one. Move the fragment to .changelog/4727.added as a single entry with comma-separated package prefixes (matching the existing 4613.fixed fragment for the same packages), covering all three packages the PR touches. Remove the misplaced package-level .changelog/ directory and its self-referential .gitignore. Assisted-by: Claude Opus 4.8 --- .changelog/4727.added | 1 + .../opentelemetry-instrumentation-aiokafka/.changelog/.gitignore | 1 - .../opentelemetry-instrumentation-aiokafka/.changelog/4727.added | 1 - 3 files changed, 1 insertion(+), 2 deletions(-) create mode 100644 .changelog/4727.added delete mode 100644 instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore delete mode 100644 instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added diff --git a/.changelog/4727.added b/.changelog/4727.added new file mode 100644 index 0000000000..d6c17de1ed --- /dev/null +++ b/.changelog/4727.added @@ -0,0 +1 @@ +`opentelemetry-instrumentation-aiokafka`, `opentelemetry-instrumentation-confluent-kafka`, `opentelemetry-instrumentation-kafka-python`: emit `messaging.kafka.cluster.id` on producer and consumer spans diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore deleted file mode 100644 index f935021a8f..0000000000 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/.gitignore +++ /dev/null @@ -1 +0,0 @@ -!.gitignore diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added b/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added deleted file mode 100644 index f63468f4f4..0000000000 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/.changelog/4727.added +++ /dev/null @@ -1 +0,0 @@ -`opentelemetry-instrumentation-aiokafka`: emit `messaging.kafka.cluster.id` on producer and consumer spans From a419ba548788456381999a6a3a1e9b064d790982 Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Thu, 16 Jul 2026 00:33:50 +0530 Subject: [PATCH 8/9] fix(instrumentation/kafka-python): keep test helper under pylint too-many-locals Adding the extract_cluster_id mock parameter pushed wrap_send_helper to 16 locals (pylint limit 15). Inline the single-use expected_span_name local to stay within the limit; no behavioral change. Assisted-by: Claude Opus 4.8 --- .../tests/test_utils.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py index f02625f1be..9b4841c846 100644 --- a/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-kafka-python/tests/test_utils.py @@ -101,7 +101,6 @@ def wrap_send_helper( produce_hook = mock.MagicMock() original_send_callback = mock.MagicMock() kafka_producer = mock.MagicMock() - expected_span_name = _get_span_name("send", self.topic_name) wrapped_send = _wrap_send(tracer, produce_hook) retval = wrapped_send( @@ -113,7 +112,7 @@ def wrap_send_helper( kafka_producer, self.args, self.kwargs ) tracer.start_as_current_span.assert_called_once_with( - expected_span_name, kind=SpanKind.PRODUCER + _get_span_name("send", self.topic_name), kind=SpanKind.PRODUCER ) span = tracer.start_as_current_span().__enter__.return_value From eefa7a5ee58a295297e9463391b83de2bbb20c7a Mon Sep 17 00:00:00 2001 From: Pulipelly Shashank Reddy Date: Thu, 16 Jul 2026 01:14:16 +0530 Subject: [PATCH 9/9] refactor(kafka): rename cluster-id constant to _MESSAGING_KAFKA_CLUSTER_ID + add semconv TODO The private constant was _MESSAGING_CLUSTER_ID, which dropped the 'kafka' segment. Rename it to _MESSAGING_KAFKA_CLUSTER_ID across the aiokafka, confluent-kafka and kafka-python instrumentations so it matches the attribute key (messaging.kafka.cluster.id) and the eventual generated semconv constant (messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID). Add a TODO to switch to that constant once it is generated in opentelemetry-semantic-conventions (semconv spec PR #3819). Assisted-by: Claude Opus 4.8 --- .../opentelemetry/instrumentation/aiokafka/utils.py | 10 ++++++---- .../tests/test_utils.py | 6 +++--- .../instrumentation/confluent_kafka/utils.py | 6 ++++-- .../src/opentelemetry/instrumentation/kafka/utils.py | 6 ++++-- 4 files changed, 17 insertions(+), 11 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py index 30042f30f5..78ca1691c2 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/utils.py @@ -79,7 +79,9 @@ async def __call__( _LOG = getLogger(__name__) -_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" +# TODO(semconv #3819): once generated in opentelemetry-semantic-conventions, +# use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. +_MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" def _extract_bootstrap_servers( @@ -279,7 +281,7 @@ def _enrich_base_span( ) if cluster_id is not None: - span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) def _enrich_send_span( @@ -385,7 +387,7 @@ def _enrich_getmany_poll_span( span.set_attribute(messaging_attributes.MESSAGING_CLIENT_ID, client_id) if cluster_id is not None: - span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) if consumer_group is not None: span.set_attribute( @@ -501,7 +503,7 @@ async def _traced_send( # metadata was not yet populated before the send started. cluster_id = _extract_cluster_id_from_client(instance.client) if cluster_id is not None and span.is_recording(): - span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) return result return _traced_send diff --git a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py index 95a2d27ae9..d3a637a309 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -9,7 +9,7 @@ from opentelemetry.instrumentation.aiokafka import _patch_cluster_id_capture from opentelemetry.instrumentation.aiokafka.utils import ( - _MESSAGING_CLUSTER_ID, + _MESSAGING_KAFKA_CLUSTER_ID, AIOKafkaContextGetter, AIOKafkaContextSetter, _aiokafka_getter, @@ -398,7 +398,7 @@ async def test_cluster_id_attribute_set_on_send_span(self) -> None: for call in span.set_attribute.call_args_list } self.assertEqual( - set_attribute_calls.get(_MESSAGING_CLUSTER_ID), + set_attribute_calls.get(_MESSAGING_KAFKA_CLUSTER_ID), "test-cluster-uuid", ) @@ -429,7 +429,7 @@ async def test_cluster_id_attribute_absent_when_not_resolved(self) -> None: attribute_keys = [ call.args[0] for call in span.set_attribute.call_args_list ] - self.assertNotIn(_MESSAGING_CLUSTER_ID, attribute_keys) + self.assertNotIn(_MESSAGING_KAFKA_CLUSTER_ID, attribute_keys) def test_patch_cluster_id_capture_sets_cluster_id_from_metadata( self, diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py index 55eeae6e4a..1cb6e59e14 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/src/opentelemetry/instrumentation/confluent_kafka/utils.py @@ -26,7 +26,9 @@ _LOG = getLogger(__name__) -_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" +# TODO(semconv #3819): once generated in opentelemetry-semantic-conventions, +# use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. +_MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" _CLUSTER_ID_TTL_SECONDS = 60 * 60 @@ -297,7 +299,7 @@ def _enrich_span( _fetch_cluster_id_background(bootstrap_servers, instance=instance) cluster_id = _get_cluster_id(bootstrap_servers) if cluster_id: - span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) # https://stackoverflow.com/questions/65935155/identify-and-find-specific-message-in-kafka-topic # A message within Kafka is uniquely defined by its topic name, topic partition and offset. 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 59e0563dfc..3ccc3a9979 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 @@ -15,7 +15,9 @@ _LOG = getLogger(__name__) -_MESSAGING_CLUSTER_ID = "messaging.kafka.cluster.id" +# TODO(semconv #3819): once generated in opentelemetry-semantic-conventions, +# use messaging_attributes.MESSAGING_KAFKA_CLUSTER_ID instead of this literal. +_MESSAGING_KAFKA_CLUSTER_ID = "messaging.kafka.cluster.id" def _get_cluster_metadata(instance): @@ -184,7 +186,7 @@ def _enrich_span( SpanAttributes.MESSAGING_URL, json.dumps(bootstrap_servers) ) if cluster_id: - span.set_attribute(_MESSAGING_CLUSTER_ID, cluster_id) + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) def _get_span_name(operation: str, topic: str):