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/src/opentelemetry/instrumentation/aiokafka/__init__.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py index f694fb51be..c55a439ecb 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 ( @@ -131,6 +131,42 @@ 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, + 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` @@ -165,6 +201,16 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): schema_url=Schemas.V1_27_0.value, ) + wrap_function_wrapper( + aiokafka.AIOKafkaProducer, + "start", + _start_producer_wrapper, + ) + wrap_function_wrapper( + aiokafka.AIOKafkaConsumer, + "start", + _start_consumer_wrapper, + ) wrap_function_wrapper( aiokafka.AIOKafkaProducer, "send", @@ -182,6 +228,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 ccfe157129..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,6 +79,10 @@ async def __call__( _LOG = getLogger(__name__) +# 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( client: aiokafka.AIOKafkaClient, @@ -90,6 +94,22 @@ 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. + + 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_KAFKA_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_KAFKA_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_KAFKA_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_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 0856d1389d..d3a637a309 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/tests/test_utils.py @@ -7,12 +7,15 @@ import aiokafka +from opentelemetry.instrumentation.aiokafka import _patch_cluster_id_capture from opentelemetry.instrumentation.aiokafka.utils import ( + _MESSAGING_KAFKA_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 +117,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 +143,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 +184,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 +220,7 @@ async def test_wrap_getone( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -259,6 +266,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 +303,7 @@ async def test_wrap_getmany( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -328,6 +337,7 @@ async def test_create_consumer_span( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -352,12 +362,156 @@ 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_KAFKA_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_KAFKA_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 = update_calls.append + 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) + + @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 + + 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..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 @@ -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,118 @@ _LOG = getLogger(__name__) +# 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 + +_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 +275,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 +293,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_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. 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..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 @@ -90,7 +90,11 @@ 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 ( + _patch_cluster_id_capture, + _wrap_next, + _wrap_send, +) from opentelemetry.instrumentation.kafka.version import __version__ from opentelemetry.instrumentation.utils import unwrap @@ -123,6 +127,20 @@ def _instrument(self, **kwargs): schema_url="https://opentelemetry.io/schemas/1.11.0", ) + def _wrap_producer_init(func, instance, args, kwargs): + func(*args, **kwargs) + _patch_cluster_id_capture(instance) + + def _wrap_consumer_init(func, instance, args, kwargs): + func(*args, **kwargs) + _patch_cluster_id_capture(instance) + + 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 +151,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..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,6 +15,52 @@ _LOG = getLogger(__name__) +# 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): + """Return the kafka-python ``ClusterMetadata`` for a producer or consumer. + + ``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: @staticmethod @@ -126,7 +172,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") @@ -135,6 +185,8 @@ def _enrich_span( span.set_attribute( SpanAttributes.MESSAGING_URL, json.dumps(bootstrap_servers) ) + if cluster_id: + span.set_attribute(_MESSAGING_KAFKA_CLUSTER_ID, cluster_id) def _get_span_name(operation: str, topic: str): @@ -155,11 +207,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) 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), @@ -170,8 +223,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 @@ -182,6 +234,7 @@ def _create_consumer_span( record, extracted_context, bootstrap_servers, + cluster_id, args, kwargs, ): @@ -193,7 +246,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) @@ -214,7 +273,7 @@ def _traced_next(func, instance, args, kwargs): bootstrap_servers = ( KafkaPropertiesExtractor.extract_bootstrap_servers(instance) ) - + cluster_id = _extract_cluster_id(instance) extracted_context = propagate.extract( record.headers, getter=_kafka_getter ) @@ -224,6 +283,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..9b4841c846 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: @@ -88,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( @@ -100,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 @@ -109,6 +121,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 +137,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 +152,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 +181,7 @@ def test_wrap_next( record, context, bootstrap_servers, + extract_cluster_id.return_value, self.args, self.kwargs, ) @@ -184,6 +202,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 +210,7 @@ def test_create_consumer_span( record, extracted_context, bootstrap_servers, + cluster_id, self.args, self.kwargs, ) @@ -207,7 +227,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 +262,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")