pipeline

Documentation

Overview

Package pipeline combines all publisher functionality (processors, broker, 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 common.BeatInfo, monitoring *monitoring.Registry, config Config) (*Pipeline, error)
  • func New(metrics *monitoring.Registry, brokerFactory brokerFactory, out outputs.Group, ...) (*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 {
	WaitShutdown time.Duration          `config:"wait_shutdown"`
	Broker       common.ConfigNamespace `config:"broker"`
	Output       common.ConfigNamespace `config:"output"`
}

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 broker, an output controller and the actual outputs. The broker implementing the broker.Broker 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 broker. 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 broker 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 broker. No filtering/processing will occur on the output side.

func Load

func Load(
	beatInfo common.BeatInfo,
	monitoring *monitoring.Registry,
	config Config,
) (*Pipeline, error)

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

func New

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

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

func (*Pipeline) Close

func (p *Pipeline) Close() error

Close stops the pipeline, outputs and broker. 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
)