OpenTelemetry kafka-python integration

pypi

Installation

pip install opentelemetry-instrumentation-kafka-python

API

Instrument kafka-python to report instrumentation-kafka produced and consumed messages

Usage

from opentelemetry.instrumentation.kafka import KafkaInstrumentor
from kafka import KafkaProducer, KafkaConsumer

# Instrument kafka
KafkaInstrumentor().instrument()

# report a span of type producer with the default settings
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
producer.send('my-topic', b'raw_bytes')

# report a span of type consumer with the default settings
consumer = KafkaConsumer('my-topic', group_id='my-group', bootstrap_servers=['localhost:9092'])
for message in consumer:
# process message

The _instrument() method accepts the following keyword args: tracer_provider (TracerProvider) - an optional tracer provider produce_hook (Callable) - a function with extra user-defined logic to be performed before sending the message this function signature is: def produce_hook(span: Span, args, kwargs) consume_hook (Callable) - a function with extra user-defined logic to be performed after consuming a message this function signature is: def consume_hook(span: Span, record: kafka.record.ABCRecord, args, kwargs) for example:

from opentelemetry.instrumentation.kafka import KafkaInstrumentor
from kafka import KafkaProducer, KafkaConsumer

def produce_hook(span, args, kwargs):
    if span and span.is_recording():
        span.set_attribute("custom_user_attribute_from_produce_hook", "some-value")
def consume_hook(span, record, args, kwargs):
    if span and span.is_recording():
        span.set_attribute("custom_user_attribute_from_consume_hook", "some-value")

# instrument kafka with produce and consume hooks
KafkaInstrumentor().instrument(produce_hook=produce_hook, consume_hook=consume_hook)

# Using kafka as normal now will automatically generate spans,
# including user custom attributes added from the hooks
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])
producer.send('my-topic', b'raw_bytes')

API

class opentelemetry.instrumentation.kafka.KafkaInstrumentor(*args, **kwargs)[source]

Bases: BaseInstrumentor

An instrumentor for kafka module See BaseInstrumentor

instrumentation_dependencies()[source]

Return a list of python packages with versions that the will be instrumented.

The format should be the same as used in requirements.txt or pyproject.toml.

For example, if an instrumentation instruments requests 1.x, this method should look like: :rtype: Collection[str]

def instrumentation_dependencies(self) -> Collection[str]:

return [‘requests ~= 1.0’]

This will ensure that the instrumentation will only be used when the specified library is present in the environment.