feat: add otlp kafka exporters

[jackburridge]

Description

Adds an OTLP Kafka exporter

Fixes #4828

Type of change

Please delete options that are not relevant.

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• This change requires a documentation update

How Has This Been Tested?

This has been tested against Kafka, and the OpenTelemetry Collector.

All tests use the same setup. First the OpenTelemetry Collector is configured to debug the traces, logs, and metrics (otel-collector-config.yaml):

receivers:
    kafka:
        brokers: kafka:9093
exporters:
    debug:
        verbosity: detailed
service:
    pipelines:
        traces:
            receivers: [kafka]
            exporters: [debug]
        logs:
            receivers: [kafka]
            exporters: [debug]
        metrics:
            receivers: [kafka]
            exporters: [debug]

With docker compose you bring up a Kafka broker, and the collector (docker-compose.yaml):

services:
    kafka:
        image: apache/kafka:latest
        environment:
            KAFKA_LISTENERS: CONTROLLER://localhost:9091,HOST://0.0.0.0:9092,DOCKER://0.0.0.0:9093
            KAFKA_ADVERTISED_LISTENERS: HOST://localhost:9092,DOCKER://kafka:9093
            KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,DOCKER:PLAINTEXT,HOST:PLAINTEXT
            KAFKA_NODE_ID: 1
            KAFKA_PROCESS_ROLES: broker,controller
            KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
            KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9091
            KAFKA_INTER_BROKER_LISTENER_NAME: DOCKER
            KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
        ports:
            - "9092:9092"
    otel-collector:
        image: otel/opentelemetry-collector-contrib
        volumes:
            - ./otel-collector-config.yaml:/etc/otelcol-contrib/config.yaml

This can then be started with:

docker compose up -d

Logs can be checked with docker logs otel-setup-otel-collector-1 -f

• Test Traces

Send some spans to the Kafka with the following Python:

import time

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.kafka.trace_exporter import (
    OTLPSpanExporter,
)
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import (
    BatchSpanProcessor,
)

provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter())
provider.add_span_processor(processor)

# Sets the global default tracer provider
trace.set_tracer_provider(provider)

# Creates a tracer from the global tracer provider
tracer = trace.get_tracer("my.tracer.name")


while True:
    try:
        time.sleep(1)
        with tracer.start_as_current_span("span-name") as span:
            # do some work that 'span' will track
            print("doing some work...")
            # When the 'with' block goes out of scope, 'span' is closed for you
        provider.force_flush()
    except (KeyboardInterrupt, SystemExit):
        break

Check the logs in the collector looking for span-name

• Test Logs

Send some logs to the Kafka with the following Python:

import logging
import time

from opentelemetry._logs import set_logger_provider
from opentelemetry.exporter.otlp.proto.kafka._log_exporter import (
    OTLPLogExporter,
)
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.sdk._logs.export import BatchLogRecordProcessor

provider = LoggerProvider()
processor = BatchLogRecordProcessor(OTLPLogExporter())
provider.add_log_record_processor(processor)
# Sets the global default logger provider
set_logger_provider(provider)

handler = LoggingHandler(level=logging.INFO, logger_provider=provider)
logging.basicConfig(handlers=[handler], level=logging.INFO)

while True:
    try:
        time.sleep(1)
        logging.getLogger(__name__).info(
            "This is an OpenTelemetry log record!"
        )
        provider.force_flush()
    except (KeyboardInterrupt, SystemExit):
        break

Check the logs in the collector looking for This is an OpenTelemetry log record!

• Test Logs

Send some metrics updates to the Kafka with the following Python:

import time

from opentelemetry import metrics
from opentelemetry.exporter.otlp.proto.kafka.metric_exporter import (
    OTLPMetricExporter,
)
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.metrics.export import (
    PeriodicExportingMetricReader,
)

metric_reader = PeriodicExportingMetricReader(OTLPMetricExporter())
provider = MeterProvider(metric_readers=[metric_reader])

# Sets the global default meter provider
metrics.set_meter_provider(provider)

# Creates a meter from the global meter provider
meter = metrics.get_meter("my.meter.name")

counter = meter.create_counter(
    "my.counter.name",
)


while True:
    try:
        time.sleep(1)
        counter.add(1)
        provider.force_flush()
    except (KeyboardInterrupt, SystemExit):
        break

Check the logs in the collector looking for my.counter.name, and check that it increments.

Does This PR Require a Contrib Repo Change?

• Yes.
• No.

Checklist:

• Followed the style guidelines of this project
• Changelogs have been updated
• Unit tests have been added
• Documentation has been updated

[linux-foundation-easycla]

The committers listed above are authorized under a signed CLA.

• ✅ login: jackburridge / name: Jack Burridge (788a2e7, eb636e6)

[herin049]

This might be more appropriate in https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/exporter

[jackburridge]

This might be more appropriate in https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/exporter

@herin049 the main reason I've put this here is its using the opentelemetry-exporter-otlp-proto-common. This will be released with any release of common, and will receive any updates instantly on release

[herin049]

This might be more appropriate in https://github.com/open-telemetry/opentelemetry-python-contrib/tree/main/exporter

@herin049 the main reason I've put this here is its using the opentelemetry-exporter-otlp-proto-common. This will be released with any release of common, and will receive any updates instantly on release

@jackburridge All libraries in the contrib repo are updated in lockstep with the core repo, so there isn't really a need to keep it here for that reason. Given that I'm new, I'm not sure what the conventions are. Perhaps @xrmx can chime in here