tracing

Documentation

Index

• type EventType
• type NodeEvent
  • func EventConsumeOf(part *message.Part) NodeEvent
  • func EventDeleteOf() NodeEvent
  • func EventErrorOf(err error) NodeEvent
  • func EventProduceOf(part *message.Part) NodeEvent
• type Summary
  • func NewSummary() *Summary
  • func TracedBundle(b *bundle.Environment) (*bundle.Environment, *Summary)
  • func (s *Summary) InputEvents(flush bool) map[string][]NodeEvent
  • func (s *Summary) OutputEvents(flush bool) map[string][]NodeEvent
  • func (s *Summary) ProcessorEvents(flush bool) map[string][]NodeEvent
  • func (s *Summary) SetEnabled(e bool)
  • func (s *Summary) SetEventLimit(n int64)

Types

type EventType

type EventType string

EventType describes the type of event a component might experience during a config run.

var (
	EventProduce EventType = "PRODUCE"
	EventConsume EventType = "CONSUME"
	EventDelete  EventType = "DELETE"
	EventError   EventType = "ERROR"
)

Various event types.

type NodeEvent

type NodeEvent struct {
	Type    EventType
	Content string
	Meta    map[string]any
}

NodeEvent represents a single event that occurred within the stream.

func EventConsumeOf

func EventConsumeOf(part *message.Part) NodeEvent

EventConsumeOf creates a consumed event from a message part.

func EventDeleteOf

func EventDeleteOf() NodeEvent

EventDeleteOf creates a deleted event from a message part.

func EventErrorOf

func EventErrorOf(err error) NodeEvent

EventErrorOf creates an error event from a message part.

func EventProduceOf

func EventProduceOf(part *message.Part) NodeEvent

EventProduceOf creates a produce event from a message part.

type Summary

type Summary struct {
	Input           uint64
	Output          uint64
	ProcessorErrors uint64
	// contains filtered or unexported fields
}

Summary is a high level description of all traced events.

func NewSummary

func NewSummary() *Summary

NewSummary creates a new tracing summary that can be passed to component constructors for adding traces.

func TracedBundle

func TracedBundle(b *bundle.Environment) (*bundle.Environment, *Summary)

TracedBundle modifies a provided bundle environment so that traceable components are wrapped by components that add trace events to the returned summary.

func (*Summary) InputEvents

func (s *Summary) InputEvents(flush bool) map[string][]NodeEvent

InputEvents returns a map of input labels to events traced during the execution of a stream pipeline. Set flush to true in order to clear the events after obtaining them.

func (*Summary) OutputEvents

func (s *Summary) OutputEvents(flush bool) map[string][]NodeEvent

OutputEvents returns a map of output labels to events traced during the execution of a stream pipeline.

func (*Summary) ProcessorEvents

func (s *Summary) ProcessorEvents(flush bool) map[string][]NodeEvent

ProcessorEvents returns a map of processor labels to events traced during the execution of a stream pipeline.

func (*Summary) SetEnabled

func (s *Summary) SetEnabled(e bool)

SetEnabled sets whether tracing events are enabled across the stream.

func (*Summary) SetEventLimit

func (s *Summary) SetEventLimit(n int64)

SetEventLimit sets a limit as to how many event traces are stored, this limit is per component that's traced.