otelkafka

Open Telemetry instrumentation for confluent-kafka-go.

Installation

go get -u github.com/jurabek/otelkafka

Usage

package main

import (
 "context"
 "fmt"
 "log"
 "os"
 "time"

 "github.com/confluentinc/confluent-kafka-go/kafka"
 "github.com/otelkafka/otelkafka"
 "go.opentelemetry.io/otel"
 "go.opentelemetry.io/otel/trace"
)

func main() {
 // Create a new Kafka producer
 producer, err := otelkafka.NewProducer(&kafka.ConfigMap{
 "bootstrap.servers": "localhost:9092",
 })
 if err != nil {
 log.Fatalf("Failed to create producer: %v", err)
 }
 defer producer.Close()

 // Produce a message
 topic := "my-topic"
 message := &kafka.Message{
 TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
 Value: []byte("Hello, World!"),
 }

 // make sure the context is propagated before produce the message
	otel.GetTextMapPropagator().Inject(ctx, otelkafka.NewMessageCarrier(message))
 if err := producer.Produce(message, nil); err != nil {
 log.Fatalf("Failed to produce message: %v", err)
 }

 // Wait for the message to be delivered
 producer.Flush(15 * 1000)

 // Create a new Kafka consumer
 consumer, err := otelkafka.NewConsumer(&kafka.ConfigMap{
 "bootstrap.servers": "localhost:9092",
 "group.id": "my-group",
 "auto.offset.reset": "earliest",
 })
 if err != nil {
 log.Fatalf("Failed to create consumer: %v", err)
 }
 defer consumer.Close()

 // Subscribe to the topic
 if err := consumer.Subscribe(topic, nil); err != nil {
 log.Fatalf("Failed to subscribe to topic: %v", err)
 }

 // Consume messages
 for {
 message, err := consumer.ReadMessage(-1)
 if err != nil {
 log.Fatalf("Failed to read message: %v", err)
 }
 fmt.Printf("Message: %s\n", message.Value)
 }
}

Metrics list

Table below lists the metrics that are collected by the instrumentation, and can exported using Otel Metric Exporter

Constants ¶

Variables ¶

Functions ¶

func Version() string

Types ¶

type Consumer struct {
	*kafka.Consumer
	// contains filtered or unexported fields
}

func NewConsumer(conf *kafka.ConfigMap, opts ...Option) (*Consumer, error)

func WrapConsumer(c *kafka.Consumer, opts ...Option) (*Consumer, error)

func (c *Consumer) Close() error

func (c *Consumer) Poll(timeoutMs int) (event kafka.Event)

func (c *Consumer) ReadMessage(timeout time.Duration) (*kafka.Message, error)

type MessageCarrier struct {
	// contains filtered or unexported fields
}

func NewMessageCarrier(msg *kafka.Message) *MessageCarrier

func (c MessageCarrier) Get(key string) string

func (c MessageCarrier) Keys() []string

func (c MessageCarrier) Set(key string, value string)

type Option interface {
	// contains filtered or unexported methods
}

func WithCustomAttributeInjector(fn func(msg *kafka.Message) []attribute.KeyValue) Option

func WithMessageCarrier(carrier func(*kafka.Message) propagation.TextMapCarrier) Option

func WithMeterProvider(meterProvider metric.MeterProvider) Option

func WithPropagators(propagators propagation.TextMapPropagator) Option

func WithTracerProvider(provider trace.TracerProvider) Option

type Producer struct {
	*kafka.Producer
	// contains filtered or unexported fields
}

func NewProducer(conf *kafka.ConfigMap, opts ...Option) (*Producer, error)

func (p *Producer) Close()

func (p *Producer) Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error