diff --git a/.changelog/4648.fixed b/.changelog/4648.fixed new file mode 100644 index 0000000000..a744d9de2a --- /dev/null +++ b/.changelog/4648.fixed @@ -0,0 +1 @@ +`opentelemetry-instrumentation-confluent-kafka`: forward unproxied methods to the wrapped producer/consumer 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 3637f9b638..b1436605a5 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 @@ -168,76 +168,78 @@ def close(self): # pylint: disable=useless-super-delegation return super().close() -class ProxiedProducer(Producer): +class ProxiedProducer(wrapt.ObjectProxy): + # ObjectProxy transparently forwards every attribute and method not + # overridden below to the wrapped producer, so calls such as list_topics() + # or set_sasl_credentials() reach the underlying object instead of raising + # AttributeError / segfaulting on the C-extension type (see #4278). def __init__(self, producer: Producer, tracer: Tracer): - self._producer = producer - self._tracer = tracer - # Surface the wrapped producer's config (if any) so that - # KafkaPropertiesExtractor.extract_bootstrap_servers can read it - # through this proxy. - self.config = getattr(producer, "config", None) - - def flush(self, timeout=-1): - return self._producer.flush(timeout) - - def poll(self, timeout=-1): - return self._producer.poll(timeout) - - def purge(self, in_queue=True, in_flight=True, blocking=True): - self._producer.purge(in_queue, in_flight, blocking) + super().__init__(producer) + self._self_tracer = tracer def produce(self, topic, value=None, *args, **kwargs): # pylint: disable=keyword-arg-before-vararg new_kwargs = kwargs.copy() new_kwargs["topic"] = topic new_kwargs["value"] = value - return ConfluentKafkaInstrumentor.wrap_produce(self._producer.produce, self, self._tracer, args, new_kwargs) + return ConfluentKafkaInstrumentor.wrap_produce( + self.__wrapped__.produce, self, self._self_tracer, args, new_kwargs + ) - def original_producer(self): - return self._producer + def original_producer(self) -> Producer: + return self.__wrapped__ -class ProxiedConsumer(Consumer): +class ProxiedConsumer(wrapt.ObjectProxy): + # See ProxiedProducer for the transparent-forwarding rationale. def __init__(self, consumer: Consumer, tracer: Tracer): - self._consumer = consumer - self._tracer = tracer - self._current_consume_span = None - self._current_context_token = None - # See ProxiedProducer.__init__ for rationale. - self.config = getattr(consumer, "config", None) + super().__init__(consumer) + self._self_tracer = tracer + self._self_current_consume_span = None + self._self_current_context_token = None + + # The consume-tracking helpers in utils.py read and write these two + # attributes on the instance. Back them with ObjectProxy's `_self_` + # storage via properties so the state lives on the proxy rather than being + # forwarded onto the wrapped consumer, whose C-extension type does not + # accept arbitrary attributes. + @property + def _current_consume_span(self): + return self._self_current_consume_span + + @_current_consume_span.setter + def _current_consume_span(self, value) -> None: + self._self_current_consume_span = value + + @property + def _current_context_token(self): + return self._self_current_context_token + + @_current_context_token.setter + def _current_context_token(self, value) -> None: + self._self_current_context_token = value def close(self, *args, **kwargs): - return ConfluentKafkaInstrumentor.wrap_close(self._consumer.close, self, args, kwargs) - - def committed(self, partitions, timeout=-1): - return self._consumer.committed(partitions, timeout) - - def commit(self, *args, **kwargs): - return self._consumer.commit(*args, **kwargs) + return ConfluentKafkaInstrumentor.wrap_close( + self.__wrapped__.close, self, args, kwargs + ) def consume(self, *args, **kwargs): return ConfluentKafkaInstrumentor.wrap_consume( - self._consumer.consume, + self.__wrapped__.consume, self, - self._tracer, + self._self_tracer, args, kwargs, ) - def get_watermark_offsets(self, partition, timeout=-1, *args, **kwargs): # pylint: disable=keyword-arg-before-vararg - return self._consumer.get_watermark_offsets(partition, timeout, *args, **kwargs) - - def offsets_for_times(self, partitions, timeout=-1): - return self._consumer.offsets_for_times(partitions, timeout) - def poll(self, timeout=-1): - return ConfluentKafkaInstrumentor.wrap_poll(self._consumer.poll, self, self._tracer, [timeout], {}) - - def subscribe(self, topics, on_assign=lambda *args: None, *args, **kwargs): # pylint: disable=keyword-arg-before-vararg - self._consumer.subscribe(topics, on_assign, *args, **kwargs) + return ConfluentKafkaInstrumentor.wrap_poll( + self.__wrapped__.poll, self, self._self_tracer, [timeout], {} + ) - def original_consumer(self): - return self._consumer + def original_consumer(self) -> Consumer: + return self.__wrapped__ class ConfluentKafkaInstrumentor(BaseInstrumentor): diff --git a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py index 78fbead5e7..2d8abf8b44 100644 --- a/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py +++ b/instrumentation/opentelemetry-instrumentation-confluent-kafka/tests/test_instrumentation.py @@ -41,7 +41,11 @@ def test_instrument_api(self) -> None: producer = Producer({"bootstrap.servers": "localhost:29092"}) producer = instrumentation.instrument_producer(producer) - self.assertEqual(producer.__class__, ProxiedProducer) + # The proxies are wrapt.ObjectProxy subclasses, which deliberately + # report the *wrapped* object's __class__, so assert on the proxy type + # rather than comparing __class__ directly. Checks against the + # unwrapped class below are unaffected. + self.assertIsInstance(producer, ProxiedProducer) producer = instrumentation.uninstrument_producer(producer) self.assertEqual(producer.__class__, Producer) @@ -55,7 +59,7 @@ def test_instrument_api(self) -> None: ) consumer = instrumentation.instrument_consumer(consumer) - self.assertEqual(consumer.__class__, ProxiedConsumer) + self.assertIsInstance(consumer, ProxiedConsumer) consumer = instrumentation.uninstrument_consumer(consumer) self.assertEqual(consumer.__class__, Consumer) @@ -69,7 +73,7 @@ def test_instrument_api(self) -> None: ) consumer = instrumentation.instrument_consumer(consumer) - self.assertEqual(consumer.__class__, ProxiedConsumer) + self.assertIsInstance(consumer, ProxiedConsumer) consumer = instrumentation.uninstrument_consumer(consumer) self.assertEqual(consumer.__class__, Consumer) @@ -116,7 +120,9 @@ def test_consumer_commit_method_exists(self) -> None: ) consumer = instrumentation.instrument_consumer(consumer) - self.assertEqual(consumer.__class__, ProxiedConsumer) + # See test_instrument_api: ObjectProxy reports the wrapped __class__. + self.assertIsInstance(consumer, ProxiedConsumer) + # commit() is no longer declared on the proxy; ObjectProxy forwards it. self.assertTrue(hasattr(consumer, "commit")) def test_context_setter(self) -> None: @@ -470,3 +476,46 @@ def test_consumer_sets_bootstrap_servers_attributes(self) -> None: process_span = next(s for s in self.memory_exporter.get_finished_spans() if s.name == "topic-1 process") self.assertEqual(process_span.attributes[SERVER_ADDRESS], "broker-1") self.assertEqual(process_span.attributes[SERVER_PORT], 9092) + + def test_proxied_producer_delegates_unproxied_methods(self) -> None: + """ProxiedProducer (a wrapt.ObjectProxy) must forward methods not + overridden on the proxy to the underlying producer, preventing + AttributeError or segfaults (see #4278).""" + from confluent_kafka import Producer # noqa: PLC0415 + + instrumentation = ConfluentKafkaInstrumentor() + producer = MockedProducer([], {"bootstrap.servers": "localhost:9092"}) + # A method that is *not* overridden on ProxiedProducer. + producer.custom_metadata = lambda: {"clusters": 1} + proxied = instrumentation.instrument_producer(producer) + self.assertIsInstance(proxied, ProxiedProducer) + # ObjectProxy exposes the wrapped object and keeps its identity. + self.assertIs(proxied.__wrapped__, producer) + self.assertIsInstance(proxied, Producer) + # Transparently forwarded, not raising AttributeError. + self.assertEqual(proxied.custom_metadata(), {"clusters": 1}) + + def test_proxied_consumer_delegates_unproxied_methods(self) -> None: + """ProxiedConsumer (a wrapt.ObjectProxy) must forward methods not + overridden on the proxy to the underlying consumer, preventing + AttributeError or segfaults (see #4278).""" + from confluent_kafka import Consumer # noqa: PLC0415 + + instrumentation = ConfluentKafkaInstrumentor() + consumer = MockConsumer( + [], + { + "bootstrap.servers": "localhost:9092", + "group.id": "test-group", + "auto.offset.reset": "earliest", + }, + ) + # A method that is *not* overridden on ProxiedConsumer. + consumer.custom_assignment = lambda: [] + proxied = instrumentation.instrument_consumer(consumer) + self.assertIsInstance(proxied, ProxiedConsumer) + # ObjectProxy exposes the wrapped object and keeps its identity. + self.assertIs(proxied.__wrapped__, consumer) + self.assertIsInstance(proxied, Consumer) + # Transparently forwarded, not raising AttributeError. + self.assertEqual(proxied.custom_assignment(), [])