kafka

Documentation

Index

• func ExtractSpanContext(msg kafka.Message) (pobptrace.SpanContext, error)
• type Option
  • func WithAnalytics(on bool) Option
  • func WithAnalyticsRate(rate float64) Option
  • func WithServiceName(serviceName string) Option
• type Reader
  • func NewReader(conf kafka.ReaderConfig, opts ...Option) *Reader
  • func WrapReader(c *kafka.Reader, opts ...Option) *Reader
  • func (r *Reader) Close() error
  • func (r *Reader) FetchMessage(ctx context.Context) (kafka.Message, error)
  • func (r *Reader) ReadMessage(ctx context.Context) (kafka.Message, error)
• type Writer
  • func NewWriter(conf kafka.WriterConfig, opts ...Option) *Writer
  • func WrapWriter(w *kafka.Writer, opts ...Option) *Writer
  • func (w *Writer) WriteMessages(ctx context.Context, msgs ...kafka.Message) error

Examples

• Reader
• Writer

Functions

func ExtractSpanContext

func ExtractSpanContext(msg kafka.Message) (pobptrace.SpanContext, error)

ExtractSpanContext retrieves the SpanContext from a kafka.Message

Types

type Option

type Option func(cfg *config)

An Option customizes the config.

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 WithServiceName

func WithServiceName(serviceName string) Option

WithServiceName sets the config service name to serviceName.

type Reader

type Reader struct {
	*kafka.Reader
	// contains filtered or unexported fields
}

A Reader wraps a kafka.Reader.

Example

package main

import (
	"context"
	"log"
	"time"

	kafkatrace "git.proto.group/protoobp/pobp-trace-go/contrib/segmentio/kafka.go.v0"
	"git.proto.group/protoobp/pobp-trace-go/pobptrace/tracer"

	kafka "github.com/segmentio/kafka-go"
)

func main() {
	r := kafkatrace.NewReader(kafka.ReaderConfig{
		Brokers:        []string{"localhost:9092"},
		Topic:          "some-topic",
		GroupID:        "group-id",
		SessionTimeout: 30 * time.Second,
	})
	msg, err := r.ReadMessage(context.Background())
	if err != nil {
		log.Fatal("Failed to read message", err)
	}

	// create a child span using span id and trace id in message header
	spanContext, err := kafkatrace.ExtractSpanContext(msg)
	if err != nil {
		log.Fatal("Failed to extract span context from carrier", err)
	}
	operationName := "child-span"
	s := tracer.StartSpan(operationName, tracer.ChildOf(spanContext))
	defer s.Finish()
}

func NewReader

func NewReader(conf kafka.ReaderConfig, opts ...Option) *Reader

NewReader calls kafka.NewReader and wraps the resulting Consumer.

func WrapReader

func WrapReader(c *kafka.Reader, opts ...Option) *Reader

WrapReader wraps a kafka.Reader so that any consumed events are traced.

func (*Reader) Close

func (r *Reader) Close() error

Close calls the underlying Reader.Close and if polling is enabled, finishes any remaining span.

func (*Reader) FetchMessage

func (r *Reader) FetchMessage(ctx context.Context) (kafka.Message, error)

FetchMessage reads and returns the next message from the reader. Message will be traced.

func (*Reader) ReadMessage

func (r *Reader) ReadMessage(ctx context.Context) (kafka.Message, error)

ReadMessage polls the consumer for a message. Message will be traced.

type Writer

type Writer struct {
	*kafka.Writer
	// contains filtered or unexported fields
}

Writer wraps a kafka.Writer with tracing config data

Example

package main

import (
	"context"
	"log"

	kafkatrace "git.proto.group/protoobp/pobp-trace-go/contrib/segmentio/kafka.go.v0"

	kafka "github.com/segmentio/kafka-go"
)

func main() {
	w := kafkatrace.NewWriter(kafka.WriterConfig{
		Brokers: []string{"localhost:9092"},
		Topic:   "some-topic",
	})

	// use slice as it passes the value by reference if you want message headers updated in kafkatrace
	msgs := []kafka.Message{
		{
			Key:   []byte("key1"),
			Value: []byte("value1"),
		},
	}
	if err := w.WriteMessages(context.Background(), msgs...); err != nil {
		log.Fatal("Failed to write message", err)
	}
}

func NewWriter

func NewWriter(conf kafka.WriterConfig, opts ...Option) *Writer

NewWriter calls kafka.NewWriter and wraps the resulting Producer.

func WrapWriter

func WrapWriter(w *kafka.Writer, opts ...Option) *Writer

WrapWriter wraps a kafka.Writer so requests are traced.

func (*Writer) WriteMessages

func (w *Writer) WriteMessages(ctx context.Context, msgs ...kafka.Message) error

WriteMessages calls kafka.go.v0.Writer.WriteMessages and traces the requests.