tracing

Documentation

Index

• func ComponentName(v CKGoVersion) string
• func IntegrationName(v CKGoVersion) string
• func WrapConsumeEventsChannel[E any, TE Event](tr *KafkaTracer, in chan E, consumer Consumer, translateFn func(E) TE) chan E
• func WrapDeliveryChannel[E any, TE Event](tr *KafkaTracer, deliveryChan chan E, span ddtrace.Span, ...) (chan E, chan error)
• func WrapProduceChannel[M any, TM Message](tr *KafkaTracer, out chan M, translateFn func(M) TM) chan M
• func WrapProduceEventsChannel[E any, TE Event](tr *KafkaTracer, in chan E, translateFn func(E) TE) chan E
• type CKGoVersion
• type ConfigMap
• type Consumer
• type Event
• type Header
• type KafkaHeader
  • func (h KafkaHeader) GetKey() string
  • func (h KafkaHeader) GetValue() []byte
• type KafkaTracer
  • func NewKafkaTracer(ckgoVersion CKGoVersion, librdKafkaVersion int, opts ...Option) *KafkaTracer
  • func (tr *KafkaTracer) DSMEnabled() bool
  • func (tr *KafkaTracer) SetConsumeCheckpoint(msg Message)
  • func (tr *KafkaTracer) SetProduceCheckpoint(msg Message)
  • func (tr *KafkaTracer) StartConsumeSpan(msg Message) ddtrace.Span
  • func (tr *KafkaTracer) StartProduceSpan(msg Message) ddtrace.Span
  • func (tr *KafkaTracer) TrackCommitOffsets(offsets []TopicPartition, err error)
  • func (tr *KafkaTracer) TrackHighWatermarkOffset(offsets []TopicPartition, consumer Consumer)
  • func (tr *KafkaTracer) TrackProduceOffsets(msg Message)
• 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 OffsetsCommitted
• type Option
  • func WithAnalytics(on bool) Option
  • func WithAnalyticsRate(rate float64) Option
  • func WithConfig(cg ConfigMap) Option
  • func WithContext(ctx context.Context) Option
  • func WithCustomTag(tag string, tagFn func(msg Message) interface{}) Option
  • func WithDataStreams() Option
  • func WithServiceName(serviceName string) Option
• type SpanStore
• type TopicPartition

Functions

func ComponentName

func ComponentName(v CKGoVersion) string

func IntegrationName

func IntegrationName(v CKGoVersion) string

func WrapConsumeEventsChannel

func WrapConsumeEventsChannel[E any, TE Event](tr *KafkaTracer, in chan E, consumer Consumer, translateFn func(E) TE) chan E

func WrapDeliveryChannel

func WrapDeliveryChannel[E any, TE Event](tr *KafkaTracer, deliveryChan chan E, span ddtrace.Span, translateFn func(E) TE) (chan E, chan error)

func WrapProduceChannel

func WrapProduceChannel[M any, TM Message](tr *KafkaTracer, out chan M, translateFn func(M) TM) chan M

func WrapProduceEventsChannel

func WrapProduceEventsChannel[E any, TE Event](tr *KafkaTracer, in chan E, translateFn func(E) TE) chan E

Types

type CKGoVersion

type CKGoVersion int32

const (
	CKGoVersion1 CKGoVersion = 1
	CKGoVersion2 CKGoVersion = 2
)

type ConfigMap

type ConfigMap interface {
	Get(key string, defval any) (any, error)
}

type Consumer

type Consumer interface {
	GetWatermarkOffsets(topic string, partition int32) (low int64, high int64, err error)
}

type Event

type Event interface {
	KafkaMessage() (Message, bool)
	KafkaOffsetsCommitted() (OffsetsCommitted, bool)
}

type Header

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

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 KafkaTracer

type KafkaTracer struct {
	PrevSpan ddtrace.Span
	// contains filtered or unexported fields
}

func NewKafkaTracer

func NewKafkaTracer(ckgoVersion CKGoVersion, librdKafkaVersion int, opts ...Option) *KafkaTracer

func (*KafkaTracer) DSMEnabled

func (tr *KafkaTracer) DSMEnabled() bool

func (*KafkaTracer) SetConsumeCheckpoint

func (tr *KafkaTracer) SetConsumeCheckpoint(msg Message)

func (*KafkaTracer) SetProduceCheckpoint

func (tr *KafkaTracer) SetProduceCheckpoint(msg Message)

func (*KafkaTracer) StartConsumeSpan

func (tr *KafkaTracer) StartConsumeSpan(msg Message) ddtrace.Span

func (*KafkaTracer) StartProduceSpan

func (tr *KafkaTracer) StartProduceSpan(msg Message) ddtrace.Span

func (*KafkaTracer) TrackCommitOffsets

func (tr *KafkaTracer) TrackCommitOffsets(offsets []TopicPartition, err error)

func (*KafkaTracer) TrackHighWatermarkOffset

func (tr *KafkaTracer) TrackHighWatermarkOffset(offsets []TopicPartition, consumer Consumer)

func (*KafkaTracer) TrackProduceOffsets

func (tr *KafkaTracer) TrackProduceOffsets(msg Message)

type Message

type Message interface {
	GetValue() []byte
	GetKey() []byte
	GetHeaders() []Header
	SetHeaders([]Header)
	GetTopicPartition() TopicPartition
	Unwrap() any
}

type MessageCarrier

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

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

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 OffsetsCommitted

type OffsetsCommitted interface {
	GetError() error
	GetOffsets() []TopicPartition
}

type Option

type Option func(tr *KafkaTracer)

An Option customizes the KafkaTracer.

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 WithConfig

func WithConfig(cg ConfigMap) Option

WithConfig extracts the config information for the client to be tagged

func WithContext

func WithContext(ctx context.Context) Option

WithContext sets the config context to ctx. Deprecated: This is deprecated in favor of passing the context via the message headers

func WithCustomTag

func WithCustomTag(tag string, tagFn func(msg Message) interface{}) Option

WithCustomTag will cause the given tagFn to be evaluated after executing a query and attach the result to the span tagged by the key.

func WithDataStreams

func WithDataStreams() Option

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

func WithServiceName

func WithServiceName(serviceName string) Option

WithServiceName sets the config service name to serviceName.

type SpanStore

type SpanStore struct {
	Prev ddtrace.Span
}

type TopicPartition

type TopicPartition interface {
	GetTopic() string
	GetPartition() int32
	GetOffset() int64
	GetError() error
}