aggregator

Documentation

Index

• func AddFlags(flags *pflag.FlagSet)
• type BufferedAggregator
  • func NewBuffered(kevents chan *kevent.Kevent, errs chan error, config Config, ...) (*BufferedAggregator, error)
  • func (agg *BufferedAggregator) Stop() error
• type Config
  • func (c *Config) InitFromViper(v *viper.Viper)

Functions

func AddFlags

func AddFlags(flags *pflag.FlagSet)

AddFlags registers persistent aggregator flags.

Types

type BufferedAggregator

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

BufferedAggregator collects events from the inbound channel and produces batches on regular intervals. The batches are pushed to the work queue from which load-balanced configured workers consume the batches and publish to the outputs.

func NewBuffered

func NewBuffered(
	kevents chan *kevent.Kevent,
	errs chan error,
	config Config,
	outputConfig outputs.Config,
	transformerConfigs []transformers.Config,
	alertsenderConfigs []alertsender.Config,
) (*BufferedAggregator, error)

NewBuffered creates a new instance of the event aggregator.

func (*BufferedAggregator) Stop

func (agg *BufferedAggregator) Stop() error

Stop flushes pending event batches and instructs the aggregator to stop processing events.

type Config

type Config struct {
	// FlushPeriod determines the period for flushing batches to outputs.
	FlushPeriod time.Duration `json:"aggregator.flush-period" yaml:"aggregator.flush-period"`
	// FlushTimeout represents the max time to wait before announcing failed flushing of enqueued events
	FlushTimeout time.Duration `json:"aggregator.flush-timeout" yaml:"aggregator.flush-timeout"`
}

Config contains aggregator-specific configuration tweaks.

func (*Config) InitFromViper

func (c *Config) InitFromViper(v *viper.Viper)

InitFromViper initializes aggregator flags from viper.