processor

Documentation

Index

• type KafkaSpanProcessor
  • func NewSpanProcessor(params SpanProcessorParams) *KafkaSpanProcessor
  • func (s KafkaSpanProcessor) Process(message Message) error
• type Message
• type ParallelProcessor
  • func NewParallelProcessor(processor SpanProcessor, parallelism int, logger *zap.Logger) *ParallelProcessor
  • func (k *ParallelProcessor) Close() error
  • func (k *ParallelProcessor) Process(message Message) error
  • func (k *ParallelProcessor) Start()
• type SpanProcessor
  • func NewDecoratedProcessor(f metrics.Factory, processor SpanProcessor) SpanProcessor
• type SpanProcessorParams

Types

type KafkaSpanProcessor

type KafkaSpanProcessor struct {
	io.Closer
	// contains filtered or unexported fields
}

KafkaSpanProcessor implements SpanProcessor for Kafka messages

func NewSpanProcessor

func NewSpanProcessor(params SpanProcessorParams) *KafkaSpanProcessor

NewSpanProcessor creates a new KafkaSpanProcessor

func (KafkaSpanProcessor) Process

func (s KafkaSpanProcessor) Process(message Message) error

Process unmarshals and writes a single kafka message

type Message

type Message interface {
	Value() []byte
}

Message contains the fields of the kafka message that the span processor uses

type ParallelProcessor

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

ParallelProcessor is a processor that processes in parallel using a pool of goroutines

func NewParallelProcessor

func NewParallelProcessor(
	processor SpanProcessor,
	parallelism int,
	logger *zap.Logger) *ParallelProcessor

NewParallelProcessor creates a new parallel processor

func (*ParallelProcessor) Close

func (k *ParallelProcessor) Close() error

Close terminates all running goroutines

func (*ParallelProcessor) Process

func (k *ParallelProcessor) Process(message Message) error

Process queues a message for processing

func (*ParallelProcessor) Start

func (k *ParallelProcessor) Start()

Start begins processing queued messages

type SpanProcessor

type SpanProcessor interface {
	Process(input Message) error
	io.Closer
}

SpanProcessor processes kafka spans

func NewDecoratedProcessor

func NewDecoratedProcessor(f metrics.Factory, processor SpanProcessor) SpanProcessor

NewDecoratedProcessor returns a processor with metrics

type SpanProcessorParams

type SpanProcessorParams struct {
	Writer       spanstore.Writer
	Unmarshaller kafka.Unmarshaller
}

SpanProcessorParams stores the necessary parameters for a SpanProcessor