pipeline

type Annotations ¶

type Annotations struct {
	Beat  common.MapStr
	Event common.EventMetadata
}

Annotations configures additional metadata to be adde to every single event being published. The meta data will be added before executing the configured processors, so all processors configured with the pipeline or client will see the same/complete event.

type Batch ¶

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

func (b *Batch) ACK()

func (b *Batch) Cancelled()

func (b *Batch) CancelledEvents(events []publisher.Event)

func (b *Batch) Drop()

func (b *Batch) Events() []publisher.Event

func (b *Batch) Retry()

func (b *Batch) RetryEvents(events []publisher.Event)

type Config ¶

type Config struct {
	// Event processing configurations
	common.EventMetadata `config:",inline"`      // Fields and tags to add to each event.
	Processors           processors.PluginConfig `config:"processors"`

	// Event queue
	Queue common.ConfigNamespace `config:"queue"`
}

Config object for loading a pipeline instance via Load.

type Pipeline ¶

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

Pipeline implementation providint all beats publisher functionality. The pipeline consists of clients, processors, a central queue, an output controller and the actual outputs. The queue implementing the queue.Queue interface is the most entral entity to the pipeline, providing support for pushung, batching and pulling events. The pipeline adds different ACKing strategies and wait close support on top of the queue. For handling ACKs, the pipeline keeps track of filtered out events, to be ACKed to the client in correct order. The output controller configures a (potentially reloadable) set of load balanced output clients. Events will be pulled from the queue and pushed to the output clients using a shared work queue for the active outputs.Group. Processors in the pipeline are executed in the clients go-routine, before entering the queue. No filtering/processing will occur on the output side.

func Load(
	beatInfo beat.Info,
	reg *monitoring.Registry,
	config Config,
	outcfg common.ConfigNamespace,
) (*Pipeline, error)

func New(
	beat beat.Info,
	metrics *monitoring.Registry,
	queueFactory queueFactory,
	out outputs.Group,
	settings Settings,
) (*Pipeline, error)

func (p *Pipeline) Close() error

func (p *Pipeline) Connect() (beat.Client, error)

func (p *Pipeline) ConnectWith(cfg beat.ClientConfig) (beat.Client, error)

func (p *Pipeline) SetACKHandler(handler beat.PipelineACKHandler) error

type Settings ¶

type Settings struct {
	// WaitClose sets the maximum duration to block when clients or pipeline itself is closed.
	// When and how WaitClose is applied depends on WaitCloseMode.
	WaitClose time.Duration

	WaitCloseMode WaitCloseMode

	Annotations Annotations
	Processors  *processors.Processors

	Disabled bool
}

Settings is used to pass additional settings to a newly created pipeline instance.

type WaitCloseMode ¶

type WaitCloseMode uint8

WaitCloseMode enumerates the possible behaviors of WaitClose in a pipeline.

const (
	// NoWaitOnClose disable wait close in the pipeline. Clients can still
	// selectively enable WaitClose when connecting to the pipeline.
	NoWaitOnClose WaitCloseMode = iota

	// WaitOnPipelineClose applies WaitClose to the pipeline itself, waiting for outputs
	// to ACK any outstanding events. This is independent of Clients asking for
	// ACK and/or WaitClose. Clients can still optionally configure WaitClose themselves.
	WaitOnPipelineClose

	// WaitOnClientClose applies WaitClose timeout to each client connecting to
	// the pipeline. Clients are still allowed to overwrite WaitClose with a timeout > 0s.
	WaitOnClientClose
)