tracing

Documentation

Overview

Package tracing contains tracing logic for the segmentio/kafka-go.v0 instrumentation.

WARNING: this package SHOULD NOT import segmentio/kafka-go.

The motivation of this package is to support orchestrion, which cannot use the main package because it imports the segmentio/kafka-go package, and since orchestrion modifies the library code itself, this would cause an import cycle.

Index

• func Logger() instrumentation.Logger
• type Header
• type KafkaConfig
• type KafkaHeader
  • func (h KafkaHeader) GetKey() string
  • func (h KafkaHeader) GetValue() []byte
• type Message
• type MessageCarrier
  • func NewMessageCarrier(msg Message) MessageCarrier
  • func (c MessageCarrier) ForeachKey(handler func(key, val string) error) error
  • func (c MessageCarrier) Set(key, val string)
• type Option
  • func WithAnalytics(on bool) Option
  • func WithAnalyticsRate(rate float64) Option
  • func WithDataStreams() Option
  • func WithService(serviceName string) Option
• type OptionFn
• type Tracer
  • func NewTracer(kafkaCfg KafkaConfig, opts ...Option) *Tracer
  • func (*Tracer) FinishProduceSpan(span *tracer.Span, partition int, offset int64, err error)
  • func (tr *Tracer) SetConsumeDSMCheckpoint(msg Message)
  • func (tr *Tracer) SetProduceDSMCheckpoint(msg Message, writer Writer)
  • func (tr *Tracer) StartConsumeSpan(ctx context.Context, msg Message) *tracer.Span
  • func (tr *Tracer) StartProduceSpan(ctx context.Context, writer Writer, msg Message, ...) *tracer.Span
• type Writer

Functions

func Logger

func Logger() instrumentation.Logger

Types

type Header

type Header interface {
	GetKey() string
	GetValue() []byte
}

type KafkaConfig

type KafkaConfig struct {
	BootstrapServers string
	ConsumerGroupID  string
}

KafkaConfig holds information from the kafka config for span tags.

type KafkaHeader

type KafkaHeader struct {
	Key   string
	Value []byte
}

func (KafkaHeader) GetKey

func (h KafkaHeader) GetKey() string

func (KafkaHeader) GetValue

func (h KafkaHeader) GetValue() []byte

type Message

type Message interface {
	GetValue() []byte
	GetKey() []byte
	GetHeaders() []Header
	SetHeaders([]Header)
	GetTopic() string
	GetPartition() int
	GetOffset() int64
}

type MessageCarrier

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

A MessageCarrier implements TextMapReader/TextMapWriter for extracting/injecting traces on a kafka.Message

func NewMessageCarrier

func NewMessageCarrier(msg Message) MessageCarrier

func (MessageCarrier) ForeachKey

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

ForeachKey conforms to the TextMapReader interface.

func (MessageCarrier) Set

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

Set implements TextMapWriter

type Option

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

Option describes options for the Kafka integration.

func WithAnalytics

func WithAnalytics(on bool) Option

WithAnalytics enables Trace Analytics for all started spans.

func WithAnalyticsRate

func WithAnalyticsRate(rate float64) Option

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

func WithDataStreams

func WithDataStreams() Option

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

func WithService

func WithService(serviceName string) Option

WithService sets the Tracer service name to serviceName.

type OptionFn

type OptionFn func(*Tracer)

OptionFn represents options applicable to NewReader, NewWriter, WrapReader and WrapWriter.

type Tracer

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

func NewTracer

func NewTracer(kafkaCfg KafkaConfig, opts ...Option) *Tracer

func (*Tracer) FinishProduceSpan

func (*Tracer) FinishProduceSpan(span *tracer.Span, partition int, offset int64, err error)

func (*Tracer) SetConsumeDSMCheckpoint

func (tr *Tracer) SetConsumeDSMCheckpoint(msg Message)

func (*Tracer) SetProduceDSMCheckpoint

func (tr *Tracer) SetProduceDSMCheckpoint(msg Message, writer Writer)

func (*Tracer) StartConsumeSpan

func (tr *Tracer) StartConsumeSpan(ctx context.Context, msg Message) *tracer.Span

func (*Tracer) StartProduceSpan

func (tr *Tracer) StartProduceSpan(ctx context.Context, writer Writer, msg Message, spanOpts ...tracer.StartSpanOption) *tracer.Span

type Writer

type Writer interface {
	GetTopic() string
}