sarama

README

support async producer and group consumer

Documentation

Index

• func InjectCtx(ctx context.Context, msg *sarama.ProducerMessage) *sarama.ProducerMessage
• func WrapHandler(handler func(ctx context.Context, data []byte), tracer aitracer.Tracer, ...) func(msg *sarama.ConsumerMessage)
• func WrapProducer(cfg *sarama.Config, p sarama.AsyncProducer, tracer aitracer.Tracer, ...) sarama.AsyncProducer
• type Config
• type Option
  • func WithAdditionalTags(tags map[string]string) Option

Functions

func InjectCtx

func InjectCtx(ctx context.Context, msg *sarama.ProducerMessage) *sarama.ProducerMessage

InjectCtx inject ctx into msg to generate clientSpan

func WrapHandler

func WrapHandler(handler func(ctx context.Context, data []byte), tracer aitracer.Tracer, opts ...Option) func(msg *sarama.ConsumerMessage)

WrapHandler wrap func(ctx context.Context, data []byte) for sarama.ConsumerGroupHandler in order to extract tracing info from msg and generate serverSpan handler take sarama.ConsumerMessage.Value as param, and ctx contains serverSpan

func WrapProducer

func WrapProducer(cfg *sarama.Config, p sarama.AsyncProducer, tracer aitracer.Tracer, opts ...Option) sarama.AsyncProducer

WrapProducer wrap inner sarama.AsyncProducer to generate client span. due to sarama.AsyncProducer.Input() return a chan, we can not instrument Input method so a outerInput chan is provided, when put msg into outerInput chan, we can intercept and trace it

Types

type Config

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

type Option

type Option func(*Config)

func WithAdditionalTags

func WithAdditionalTags(tags map[string]string) Option