elasticapm

Documentation

Index

• type AcceptedDetails
• type Config
• type InvalidInputError
  • func (e *InvalidInputError) Error() string
• type Processor
  • func NewProcessor(cfg Config) *Processor
  • func (p *Processor) HandleStream(ctx context.Context, baseEvent *modelpb.APMEvent, reader io.Reader, ...) error
• type Result
• type StreamHandler

Types

type AcceptedDetails

type AcceptedDetails struct {
	Transaction int
	Span        int
	Metric      int
	Log         int
	Error       int
}

ProcessedDetail holds the number of events processed for each type.

type Config

type Config struct {
	// Semaphore holds a semaphore on which Processor.HandleStream will acquire a
	// token before proceeding, to limit concurrency.
	Semaphore input.Semaphore
	// Logger holds a logger for the processor. If Logger is nil,
	// then no logging will be performed.
	Logger *zap.Logger
	// MaxEventSize holds the maximum event size, in bytes.
	MaxEventSize int
}

Config holds configuration for Processor constructors.

type InvalidInputError

type InvalidInputError struct {
	Message  string
	Document string
	TooLarge bool
}

func (*InvalidInputError) Error

func (e *InvalidInputError) Error() string

type Processor

type Processor struct {
	MaxEventSize int
	// contains filtered or unexported fields
}

Processor decodes a streams and is safe for concurrent use. The processor accepts a channel that is used as a semaphore to control the maximum concurrent number of stream decode operations that can happen at any time. The buffered channel is meant to be shared between all the processors so the concurrency limit is shared between all the intake endpoints.

func NewProcessor

func NewProcessor(cfg Config) *Processor

NewProcessor returns a new Processor for processing an event stream from Elastic APM agents.

func (*Processor) HandleStream

func (p *Processor) HandleStream(
	ctx context.Context,
	baseEvent *modelpb.APMEvent,
	reader io.Reader,
	batchSize int,
	processor modelpb.BatchProcessor,
	result *Result,
) error

HandleStream processes a stream of events in batches of batchSize at a time, updating result as events are accepted, or per-event errors occur.

HandleStream will return an error when a terminal stream-level error occurs, such as the rate limit being exceeded, or due to authorization errors. In this case the result will only cover the subset of events accepted.

Callers must not access result concurrently with HandleStream.

type Result

type Result struct {

	// Errors holds a limited number of errors that occurred while
	// processing the event stream. If the limit is reached, the
	// counters above are still incremented.
	Errors []error

	// Accepted holds the total number of valid events accepted.
	Accepted int

	// AcceptedDetails provides a detailed breakdown of the count
	// of valid events categorized by signal type.
	AcceptedDetails AcceptedDetails

	// TooLarge holds the number of events that were rejected due
	// to exceeding the event size limit.
	TooLarge int

	// Invalid holds the number of events that were rejected due
	// to being invalid, excluding those that are counted by TooLarge.
	Invalid int
	// contains filtered or unexported fields
}

type StreamHandler

type StreamHandler interface {
	HandleStream(
		ctx context.Context,
		baseEvent *modelpb.APMEvent,
		stream io.Reader,
		batchSize int,
		processor modelpb.BatchProcessor,
		out *Result,
	) error
}

StreamHandler is an interface for handling an Elastic APM agent ND-JSON event stream, implemented by processor/stream.