logs

Documentation

Overview

nolint:errcheck

nolint:errcheck

Index

• func MetricViews() []*view.View
• type GroupByTraceProcessor
  • func NewGroupByTraceProcessor(logger *zap.Logger, st Storage, nextConsumer consumer.Logs, ...) *GroupByTraceProcessor
  • func (sp *GroupByTraceProcessor) Capabilities() consumer.Capabilities
  • func (sp *GroupByTraceProcessor) ConsumeLogs(_ context.Context, td plog.Logs) error
  • func (sp *GroupByTraceProcessor) Shutdown(_ context.Context) error
  • func (sp *GroupByTraceProcessor) Start(context.Context, component.Host) error
• type MemoryStorage
  • func NewMemoryStorage() *MemoryStorage
• type Storage

Functions

func MetricViews

func MetricViews() []*view.View

MetricViews return the metrics views according to given telemetry level.

Types

type GroupByTraceProcessor

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

GroupByTraceProcessor is a processor that keeps logs in memory for a given duration, with the expectation that the log will be complete once this duration expires. After the duration, the log is sent to the next consumer. This processor uses a buffered event machine, which converts operations into events for non-blocking processing, but keeping all operations serialized per worker scope. This ensures that we don't need locks but that the state is consistent across go routines. Initially, all incoming batches are split into different logs and distributed among workers by a hash of traceID in eventMachine.consume method. Afterwards, the log is registered with a go routine, which will be called after the given duration and dispatched to the event machine for further processing. The typical data flow looks like this: ConsumeLogs -> eventMachine.consume(log) -> event(logReceived) -> onLogReceived -> AfterFunc(duration, event(logExpired)) -> onTraceExpired async markAsReleased -> event(logReleased) -> onLogReleased -> nextConsumer Each worker in the eventMachine also uses a ring buffer to hold the in-flight log IDs, so that we don't hold more than the given maximum number of logs in memory/storage. Items that are evicted from the buffer are discarded without warning.

func NewGroupByTraceProcessor

func NewGroupByTraceProcessor(logger *zap.Logger, st Storage, nextConsumer consumer.Logs, config common.Config) *GroupByTraceProcessor

NewGroupByTraceProcessor returns a new processor.

func (*GroupByTraceProcessor) Capabilities

func (sp *GroupByTraceProcessor) Capabilities() consumer.Capabilities

func (*GroupByTraceProcessor) ConsumeLogs

func (sp *GroupByTraceProcessor) ConsumeLogs(_ context.Context, td plog.Logs) error

func (*GroupByTraceProcessor) Shutdown

func (sp *GroupByTraceProcessor) Shutdown(_ context.Context) error

Shutdown is invoked during service shutdown.

func (*GroupByTraceProcessor) Start

func (sp *GroupByTraceProcessor) Start(context.Context, component.Host) error

Start is invoked during service startup.

type MemoryStorage

type MemoryStorage struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewMemoryStorage

func NewMemoryStorage() *MemoryStorage

type Storage

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

storage is an abstraction for the span storage used by the groupbytrace log processor. Implementations should be safe for concurrent use.