sarama

Documentation

Overview

Package sarama provides functions to trace the IBM/sarama package (https://github.com/IBM/sarama).

Index

• func WrapAsyncProducer(saramaConfig *sarama.Config, p sarama.AsyncProducer, opts ...Option) sarama.AsyncProducer
• func WrapConsumer(c sarama.Consumer, opts ...Option) sarama.Consumer
• func WrapConsumerGroupHandler(handler sarama.ConsumerGroupHandler, opts ...Option) sarama.ConsumerGroupHandler
• func WrapPartitionConsumer(pc sarama.PartitionConsumer, opts ...Option) sarama.PartitionConsumer
• func WrapSyncProducer(saramaConfig *sarama.Config, producer sarama.SyncProducer, opts ...Option) sarama.SyncProducer
• type ConsumerMessageCarrier
  • func NewConsumerMessageCarrier(msg *sarama.ConsumerMessage) ConsumerMessageCarrier
  • func (c ConsumerMessageCarrier) ForeachKey(handler func(key, val string) error) error
  • func (c ConsumerMessageCarrier) Set(key, val string)
• type Option
• type OptionFn
  • func WithAnalytics(on bool) OptionFn
  • func WithAnalyticsRate(rate float64) OptionFn
  • func WithBrokers(addrs []string) OptionFn
  • func WithConsumerCustomTag(tag string, tagFn func(msg *sarama.ConsumerMessage) any) OptionFn
  • func WithDataStreams() OptionFn
  • func WithGroupID(groupID string) OptionFn
  • func WithProducerCustomTag(tag string, tagFn func(msg *sarama.ProducerMessage) any) OptionFn
  • func WithService(name string) OptionFn
• type ProducerMessageCarrier
  • func NewProducerMessageCarrier(msg *sarama.ProducerMessage) ProducerMessageCarrier
  • func (c ProducerMessageCarrier) ForeachKey(handler func(key, val string) error) error
  • func (c ProducerMessageCarrier) Set(key, val string)

Examples

• WrapAsyncProducer
• WrapConsumer
• WrapConsumerGroupHandler
• WrapSyncProducer

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. Tracing requires at least sarama.V0_11_0_0 version which is the first version that supports headers. Only spans of successfully published messages have partition and offset tags set.

Example

package main

import (
	saramatrace "github.com/DataDog/dd-trace-go/contrib/IBM/sarama/v2"

	"github.com/IBM/sarama"
)

func main() {
	cfg := sarama.NewConfig()
	cfg.Version = sarama.V0_11_0_0 // minimum version that supports headers which are required for tracing

	producer, err := sarama.NewAsyncProducer([]string{"localhost:9092"}, cfg)
	if err != nil {
		panic(err)
	}
	defer producer.Close()

	producer = saramatrace.WrapAsyncProducer(cfg, producer)

	msg := &sarama.ProducerMessage{
		Topic: "some-topic",
		Value: sarama.StringEncoder("Hello World"),
	}
	producer.Input() <- msg
}

func WrapConsumer

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

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

Example

package main

import (
	"log"

	saramatrace "github.com/DataDog/dd-trace-go/contrib/IBM/sarama/v2"

	"github.com/DataDog/dd-trace-go/v2/ddtrace/tracer"

	"github.com/IBM/sarama"
)

func main() {
	consumer, err := sarama.NewConsumer([]string{"localhost:9092"}, nil)
	if err != nil {
		panic(err)
	}
	defer consumer.Close()

	consumer = saramatrace.WrapConsumer(consumer)

	partitionConsumer, err := consumer.ConsumePartition("some-topic", 0, sarama.OffsetNewest)
	if err != nil {
		panic(err)
	}
	defer partitionConsumer.Close()

	consumed := 0
	for msg := range partitionConsumer.Messages() {
		// if you want to use the kafka message as a parent span:
		if spanctx, err := tracer.Extract(saramatrace.NewConsumerMessageCarrier(msg)); err == nil {
			// you can create a span using ChildOf(spanctx)
			_ = spanctx
		}

		log.Printf("Consumed message offset %d\n", msg.Offset)
		consumed++
	}
}

func WrapConsumerGroupHandler

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

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

Example

package main

import (
	"context"
	"errors"
	"log"

	saramatrace "github.com/DataDog/dd-trace-go/contrib/IBM/sarama/v2"

	"github.com/IBM/sarama"
)

func main() {
	cfg := sarama.NewConfig()
	cfg.Version = sarama.V0_11_0_0 // first version that supports headers
	cfg.Producer.Return.Successes = true
	cfg.Producer.Flush.Messages = 1

	const groupID = "group-id"

	consumerGroup, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, groupID, cfg)
	if err != nil {
		panic(err)
	}

	// trace your sarama.ConsumerGroupHandler implementation
	var myHandler sarama.ConsumerGroupHandler
	handler := saramatrace.WrapConsumerGroupHandler(myHandler, saramatrace.WithGroupID(groupID))

	ctx := context.Background()
	for {
		// `Consume` should be called inside an infinite loop, when a
		// server-side rebalance happens, the consumer session will need to be
		// recreated to get the new claims
		if err := consumerGroup.Consume(ctx, []string{"my-topic"}, handler); err != nil {
			if errors.Is(err, sarama.ErrClosedConsumerGroup) {
				return
			}
			log.Panicf("Error from consumer: %v", err)
		}
		// check if context was cancelled, signaling that the consumer should stop
		if ctx.Err() != nil {
			return
		}
	}
}

func WrapPartitionConsumer

func WrapPartitionConsumer(pc sarama.PartitionConsumer, 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.

Example

package main

import (
	saramatrace "github.com/DataDog/dd-trace-go/contrib/IBM/sarama/v2"

	"github.com/IBM/sarama"
)

func main() {
	cfg := sarama.NewConfig()
	cfg.Producer.Return.Successes = true

	producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, cfg)
	if err != nil {
		panic(err)
	}
	defer producer.Close()

	producer = saramatrace.WrapSyncProducer(cfg, producer)

	msg := &sarama.ProducerMessage{
		Topic: "some-topic",
		Value: sarama.StringEncoder("Hello World"),
	}
	_, _, err = producer.SendMessage(msg)
	if err != nil {
		panic(err)
	}
}

Types

type ConsumerMessageCarrier

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

A ConsumerMessageCarrier injects and extracts traces from a sarama.ConsumerMessage.

func NewConsumerMessageCarrier

func NewConsumerMessageCarrier(msg *sarama.ConsumerMessage) ConsumerMessageCarrier

NewConsumerMessageCarrier creates a new ConsumerMessageCarrier.

func (ConsumerMessageCarrier) ForeachKey

func (c ConsumerMessageCarrier) ForeachKey(handler func(key, val string) error) error

ForeachKey iterates over every header.

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 describes options for the Sarama integration.

type OptionFn

type OptionFn func(*config)

OptionFn represents options applicable to WrapConsumer, WrapPartitionConsumer, WrapAsyncProducer and WrapSyncProducer.

func WithAnalytics

func WithAnalytics(on bool) OptionFn

WithAnalytics enables Trace Analytics for all started spans.

func WithAnalyticsRate

func WithAnalyticsRate(rate float64) OptionFn

WithAnalyticsRate sets the sampling rate for Trace Analytics events correlated to started spans.

func WithBrokers

func WithBrokers(addrs []string) OptionFn

WithBrokers provides broker addresses for automatic Kafka cluster ID detection for Data Streams Monitoring. The cluster ID is fetched asynchronously in the background when the producer or consumer is wrapped.

func WithConsumerCustomTag

func WithConsumerCustomTag(tag string, tagFn func(msg *sarama.ConsumerMessage) any) OptionFn

WithConsumerCustomTag enables calling a callback func to generate the value for a custom tag on wrapped consumers.

func WithDataStreams

func WithDataStreams() OptionFn

WithDataStreams enables the Data Streams monitoring product features: https://www.datadoghq.com/product/data-streams-monitoring/

func WithGroupID

func WithGroupID(groupID string) OptionFn

WithGroupID tags the produced data streams metrics with the given groupID (aka consumer group)

func WithProducerCustomTag

func WithProducerCustomTag(tag string, tagFn func(msg *sarama.ProducerMessage) any) OptionFn

WithCustomProducerSpanOptions enables calling a callback func to generate the value for a custom tag on wrapped producers.

func WithService

func WithService(name string) OptionFn

WithService sets the given service name for the intercepted client.

type ProducerMessageCarrier

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

A ProducerMessageCarrier injects and extracts traces from a sarama.ProducerMessage.

func NewProducerMessageCarrier

func NewProducerMessageCarrier(msg *sarama.ProducerMessage) ProducerMessageCarrier

NewProducerMessageCarrier creates a new ProducerMessageCarrier.

func (ProducerMessageCarrier) ForeachKey

func (c ProducerMessageCarrier) ForeachKey(handler func(key, val string) error) error

ForeachKey iterates over every header.

func (ProducerMessageCarrier) Set

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

Set sets a header.