otelsaramax

README

Otelsarama

OpenTelemetry Instrumentation for the Sarama Kafka client.

Attributions

Code taken from opentelemetry-go-contrib v1.17.0

When migrating watermill-kafka to v3.0.0 (Migrate to IBM/Sarama), we could consider using (github.com/dnwe/otelsarama)

Documentation

Index

• func WrapAsyncProducer(saramaConfig *sarama.Config, p sarama.AsyncProducer, opts ...Option) sarama.AsyncProducer
• func WrapConsumer(c sarama.Consumer, consumerInfo ConsumerInfo, opts ...Option) sarama.Consumer
• func WrapConsumerGroupHandler(handler sarama.ConsumerGroupHandler, consumerInfo ConsumerInfo, opts ...Option) sarama.ConsumerGroupHandler
• func WrapPartitionConsumer(pc sarama.PartitionConsumer, consumerInfo ConsumerInfo, opts ...Option) sarama.PartitionConsumer
• func WrapSyncProducer(saramaConfig *sarama.Config, producer sarama.SyncProducer, opts ...Option) sarama.SyncProducer
• type ConsumerInfo
• type ConsumerMessageCarrier
  • func NewConsumerMessageCarrier(msg *sarama.ConsumerMessage) ConsumerMessageCarrier
  • func (c ConsumerMessageCarrier) Get(key string) string
  • func (c ConsumerMessageCarrier) Keys() []string
  • func (c ConsumerMessageCarrier) Set(key, val string)
• type Option
  • func WithPropagators(propagators propagation.TextMapPropagator) Option
  • func WithTracerProvider(provider trace.TracerProvider) Option
• type ProducerMessageCarrier
  • func NewProducerMessageCarrier(msg *sarama.ProducerMessage) ProducerMessageCarrier
  • func (c ProducerMessageCarrier) Get(key string) string
  • func (c ProducerMessageCarrier) Keys() []string
  • func (c ProducerMessageCarrier) Set(key, val string)

Functions

func WrapAsyncProducer

func WrapAsyncProducer(saramaConfig *sarama.Config, p sarama.AsyncProducer, opts ...Option) sarama.AsyncProducer

WrapAsyncProducer wraps a sarama.AsyncProducer so that all produced messages are traced. It requires the underlying sarama Config, so we can know whether or not successes will be returned.

If `Return.Successes` is false, there is no way to know partition and offset of the message.

func WrapConsumer

func WrapConsumer(c sarama.Consumer, consumerInfo ConsumerInfo, opts ...Option) sarama.Consumer

WrapConsumer wraps a sarama.Consumer wrapping any PartitionConsumer created via Consumer.ConsumePartition.

func WrapConsumerGroupHandler

func WrapConsumerGroupHandler(handler sarama.ConsumerGroupHandler, consumerInfo ConsumerInfo, opts ...Option) sarama.ConsumerGroupHandler

WrapConsumerGroupHandler wraps a sarama.ConsumerGroupHandler causing each received message to be traced.

func WrapPartitionConsumer

func WrapPartitionConsumer(pc sarama.PartitionConsumer, consumerInfo ConsumerInfo, opts ...Option) sarama.PartitionConsumer

WrapPartitionConsumer wraps a sarama.PartitionConsumer causing each received message to be traced.

func WrapSyncProducer

func WrapSyncProducer(saramaConfig *sarama.Config, producer sarama.SyncProducer, opts ...Option) sarama.SyncProducer

WrapSyncProducer wraps a sarama.SyncProducer so that all produced messages are traced.

Types

type ConsumerInfo

type ConsumerInfo struct {
	ConsumerGroup string
	Attributes    []attribute.KeyValue
}

ConsumerInfo holds information about the consumer group.

type ConsumerMessageCarrier

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

ConsumerMessageCarrier injects and extracts traces from a sarama.ConsumerMessage.

func NewConsumerMessageCarrier

func NewConsumerMessageCarrier(msg *sarama.ConsumerMessage) ConsumerMessageCarrier

NewConsumerMessageCarrier creates a new ConsumerMessageCarrier.

func (ConsumerMessageCarrier) Get

func (c ConsumerMessageCarrier) Get(key string) string

Get retrieves a single value for a given key.

func (ConsumerMessageCarrier) Keys

func (c ConsumerMessageCarrier) Keys() []string

Keys returns a slice of all key identifiers in the carrier.

func (ConsumerMessageCarrier) Set

func (c ConsumerMessageCarrier) Set(key, val string)

Set sets a header.

type Option

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

Option interface used for setting optional config properties.

func WithPropagators

func WithPropagators(propagators propagation.TextMapPropagator) Option

WithPropagators specifies propagators to use for extracting information from the HTTP requests. If none are specified, global ones will be used.

func WithTracerProvider

func WithTracerProvider(provider trace.TracerProvider) Option

WithTracerProvider specifies a tracer provider to use for creating a tracer. If none is specified, the global provider is used.

type ProducerMessageCarrier

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

ProducerMessageCarrier injects and extracts traces from a sarama.ProducerMessage.

func NewProducerMessageCarrier

func NewProducerMessageCarrier(msg *sarama.ProducerMessage) ProducerMessageCarrier

NewProducerMessageCarrier creates a new ProducerMessageCarrier.

func (ProducerMessageCarrier) Get

func (c ProducerMessageCarrier) Get(key string) string

Get retrieves a single value for a given key.

func (ProducerMessageCarrier) Keys

func (c ProducerMessageCarrier) Keys() []string

Keys returns a slice of all key identifiers in the carrier.

func (ProducerMessageCarrier) Set

func (c ProducerMessageCarrier) Set(key, val string)

Set sets a header.