pipeline

Documentation

Overview

Package pipeline combines all publisher functionality (processors, queue, outputs) to create instances of complete publisher pipelines, beats can connect to publish events to.

Index

• type Annotations
• type Batch
  • 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 Pipeline
  • func Load(beatInfo beat.Info, config Config, outcfg common.ConfigNamespace) (*Pipeline, error)
  • func New(beat beat.Info, metrics *monitoring.Registry, queueFactory queueFactory, ...) (*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 WaitCloseMode

Types

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 (*Batch) ACK

func (b *Batch) ACK()

func (*Batch) Cancelled

func (b *Batch) Cancelled()

func (*Batch) CancelledEvents

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

func (*Batch) Drop

func (b *Batch) Drop()

func (*Batch) Events

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

func (*Batch) Retry

func (b *Batch) Retry()

func (*Batch) RetryEvents

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

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

Load uses a Config object to create a new complete Pipeline instance with configured queue and outputs.

func New

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

New create a new Pipeline instance from a queue instance and a set of outputs. The new pipeline will take ownership of queue and outputs. On Close, the queue and outputs will be closed.

func (*Pipeline) Close

func (p *Pipeline) Close() error

Close stops the pipeline, outputs and queue. If WaitClose with WaitOnPipelineClose mode is configured, Close will block for a duration of WaitClose, if there are still active events in the pipeline. Note: clients must be closed before calling Close.

func (*Pipeline) Connect

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

Connect creates a new client with default settings

func (*Pipeline) ConnectWith

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

ConnectWith create a new Client for publishing events to the pipeline. The client behavior on close and ACK handling can be configured by setting the appropriate fields in the passed ClientConfig.

func (*Pipeline) SetACKHandler

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

SetACKHandler sets a global ACK handler on all events published to the pipeline. SetACKHandler must be called before any connection is made.

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
)