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/__init__.py b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py index f694fb51be..8558a6a35d 100644 --- a/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-aiokafka/src/opentelemetry/instrumentation/aiokafka/__init__.py @@ -126,6 +126,7 @@ class InstrumentKwargs(TypedDict, total=False): tracer_provider: trace.TracerProvider async_produce_hook: ProduceHookT async_consume_hook: ConsumeHookT + capture_experimental_span_attributes: bool class UninstrumentKwargs(TypedDict, total=False): pass @@ -147,6 +148,8 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): ``tracer_provider``: a TracerProvider, defaults to global. ``async_produce_hook``: a callable to be executed just before producing a message ``async_consume_hook``: a callable to be executed just after consuming a message + ``capture_experimental_span_attributes``: if True, emit experimental + ``messaging.cluster.id`` on producer and consumer spans. Defaults to False. """ tracer_provider = kwargs.get("tracer_provider") @@ -158,6 +161,10 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): if not iscoroutinefunction(async_consume_hook): async_consume_hook = None + capture_experimental_span_attributes = kwargs.get( + "capture_experimental_span_attributes", False + ) + tracer = trace.get_tracer( __name__, __version__, @@ -168,17 +175,29 @@ def _instrument(self, **kwargs: Unpack[InstrumentKwargs]): wrap_function_wrapper( aiokafka.AIOKafkaProducer, "send", - _wrap_send(tracer, async_produce_hook), + _wrap_send( + tracer, + async_produce_hook, + capture_experimental_span_attributes, + ), ) wrap_function_wrapper( aiokafka.AIOKafkaConsumer, "getone", - _wrap_getone(tracer, async_consume_hook), + _wrap_getone( + tracer, + async_consume_hook, + capture_experimental_span_attributes, + ), ) wrap_function_wrapper( aiokafka.AIOKafkaConsumer, "getmany", - _wrap_getmany(tracer, async_consume_hook), + _wrap_getmany( + tracer, + async_consume_hook, + capture_experimental_span_attributes, + ), ) def _uninstrument(self, **kwargs: Unpack[UninstrumentKwargs]): 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..09a0e4b2fc 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.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,9 @@ 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, + capture_experimental_span_attributes: bool = False, ) -> Callable[..., Awaitable[asyncio.Future[RecordMetadata]]]: async def _traced_send( func: AIOKafkaSendProto, @@ -439,6 +475,11 @@ 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) + if capture_experimental_span_attributes + else None + ) span_name = _get_span_name("send", topic) with tracer.start_as_current_span( span_name, kind=trace.SpanKind.PRODUCER @@ -450,6 +491,7 @@ async def _traced_send( topic=topic, partition=partition, key=key, + cluster_id=cluster_id, ) propagate.inject( headers, @@ -461,8 +503,14 @@ 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 and metadata is populated. + # Set cluster_id now if it wasn't available at span start. + if capture_experimental_span_attributes: + 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 +523,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 +544,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 +558,9 @@ 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, + capture_experimental_span_attributes: bool = False, ) -> Callable[..., Awaitable[aiokafka.ConsumerRecord[object, object]]]: async def _traced_getone( func: AIOKafkaGetOneProto, @@ -522,6 +574,11 @@ 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) + if capture_experimental_span_attributes + else None + ) extracted_context = propagate.extract( record.headers, getter=_aiokafka_getter @@ -534,6 +591,7 @@ async def _traced_getone( bootstrap_servers, client_id, consumer_group, + cluster_id, args, kwargs, ) @@ -543,7 +601,9 @@ async def _traced_getone( def _wrap_getmany( # type: ignore[reportUnusedFunction] - tracer: Tracer, async_consume_hook: ConsumeHookT | None + tracer: Tracer, + async_consume_hook: ConsumeHookT | None, + capture_experimental_span_attributes: bool = False, ) -> Callable[ ..., Awaitable[ @@ -567,6 +627,11 @@ 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) + if capture_experimental_span_attributes + else None + ) span_name = _get_span_name( "receive", @@ -581,6 +646,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 +662,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 +677,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..4806b778a2 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, @@ -139,6 +141,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) @@ -214,6 +217,7 @@ async def test_wrap_getone( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -295,6 +299,7 @@ async def test_wrap_getmany( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -328,6 +333,7 @@ async def test_create_consumer_span( bootstrap_servers, client_id, consumer_group, + None, self.args, self.kwargs, ) @@ -352,12 +358,141 @@ 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, capture_experimental_span_attributes=True + ) + 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, capture_experimental_span_attributes=True + ) + 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) + + async def test_cluster_id_attribute_absent_by_default(self) -> None: + """No cluster ID attribute is set when capture_experimental_span_attributes is False (default).""" + 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" + producer.client._client_id = "test-client" + producer.client._wait_on_metadata = mock.AsyncMock() + producer.client.cluster.cluster_id = "some-cluster-uuid" + producer._key_serializer = None + producer._value_serializer = None + producer._partition.return_value = 0 + + wrapped_send = _wrap_send(tracer, None) # default: flag=False + 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..a67268bc4b 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,114 @@ _LOG = getLogger(__name__) +_MESSAGING_CLUSTER_ID = "messaging.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: + # Store auth config from the first call (constructor) so TTL re-fetches without + # credentials (hot path) can still connect to SSL/SASL-protected clusters. + 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 — fall through to re-fetch + elif existing is not None: + return # "" sentinel — fetch already in progress + _kafka_cluster_id_cache[cache_key] = "" # sentinel + + 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 +271,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 +289,15 @@ 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) + # Trigger a background re-fetch if the cached value has expired (TTL enforcement). + _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 16113338fe..96dcd756c9 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 @@ -95,7 +95,12 @@ def process_msg(message): _instruments_kafka_python, _instruments_kafka_python_ng, ) -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 @@ -145,6 +150,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) ) @@ -155,5 +186,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..08b90b52dc 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,104 @@ _LOG = getLogger(__name__) +_MESSAGING_CLUSTER_ID = "messaging.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 — fall through to re-fetch + elif existing is not None: + return # "" sentinel — fetch already in progress + _kafka_cluster_id_cache[cache_key] = "" # sentinel + + 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 +235,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 +259,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 +274,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 +317,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 )