Small pipeline enhancements (#12996)

Note: these changes are required to prepare for some refactoring in
filebeat.

Introduce OutputChooses publisher mode. Normally this is equivalent to
the default mode for most Beats. But we introduce a new mode here, as filebeat sets the default mode
to GuaranteedSend. This allows custom inputs to overwrite the default
mode in filebeat in the future.

Using beats.ProcessorList and the processors list to combine processors
from different configurations is a little tricky. To simplify the task,
we make sure that beat.ProcessorList implements processors.Processor.
We also export processors.NewList, that can be used to compose a set of
custom processors.Processor and beats.ProcessorList. The list returned
by processors.NewList also implements beats.ProcessorList.
It's not nice, but at least makes processors composable.

Author: Steffen Siering

CHANGELOG-developer.next.asciidoc

@@ -44,3 +44,5 @@ The list below covers the major changes between 7.0.0-rc2 and master only.
 - New ReporterV2 interfaces that can receive a context on `Fetch(ctx, reporter)`, or `Run(ctx, reporter)`. {pull}11981[11981]
 - Generate configuration from `mage` for all Beats. {pull}12618[12618]
 - Add ClientFactory to TCP input source to add SplitFunc/NetworkFuncs per client. {pull}8543[8543]
+- Introduce beat.OutputChooses publisher mode. {pull}12996[12996]
+- Ensure that beat.Processor, beat.ProcessorList, and processors.ProcessorList are compatible and can be composed more easily. {pull}12996[12996]

libbeat/beat/pipeline.go

@@ -129,6 +129,7 @@ type PipelineACKHandler struct {
 }
 
 type ProcessorList interface {
+	Processor
 	All() []Processor
 }
 
@@ -144,10 +145,16 @@ type Processor interface {
 type PublishMode uint8
 
 const (
-	// DefaultGuarantees are up to the pipeline configuration, as configured by the
-	// operator.
+	// DefaultGuarantees are up to the pipeline configuration itself.
 	DefaultGuarantees PublishMode = iota
 
+	// OutputChooses mode fully depends on the output and its configuration.
+	// Events might be dropped based on the users output configuration.
+	// In this mode no events are dropped within the pipeline. Events are only removed
+	// after the output has ACKed the events to the pipeline, even if the output
+	// did drop the events.
+	OutputChooses
+
 	// GuaranteedSend ensures events are retried until acknowledged by the output.
 	// Normally guaranteed sending should be used with some client ACK-handling
 	// to update state keeping track of the sending status.

libbeat/processors/processor.go

@@ -40,10 +40,18 @@ type Processor interface {
 	String() string
 }
 
-func New(config PluginConfig) (*Processors, error) {
-	procs := &Processors{
-		log: logp.NewLogger(logName),
+// NewList creates a new empty processor list.
+// Additional processors can be added to the List field.
+func NewList(log *logp.Logger) *Processors {
+	if log == nil {
+		log = logp.NewLogger(logName)
 	}
+	return &Processors{log: log}
+}
+
+// New creates a list of processors from a list of free user configurations.
+func New(config PluginConfig) (*Processors, error) {
+	procs := NewList(nil)
 
 	for _, procConfig := range config {
 		// Handle if/then/else processor which has multiple top-level keys.

libbeat/publisher/pipeline/config.go

@@ -41,7 +41,7 @@ func validateClientConfig(c *beat.ClientConfig) error {
 	withDrop := false
 
 	switch m := c.PublishMode; m {
-	case beat.DefaultGuarantees, beat.GuaranteedSend:
+	case beat.DefaultGuarantees, beat.GuaranteedSend, beat.OutputChooses:
 	case beat.DropIfFull:
 		withDrop = true
 	default:

libbeat/publisher/pipeline/pipeline.go

@@ -49,6 +49,9 @@ import (
 // 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.
+//
+// For client connecting to this pipeline, the default PublishMode is
+// OutputChooses.
 type Pipeline struct {
 	beatInfo beat.Info
 
@@ -273,14 +276,15 @@ func (p *Pipeline) Close() error {
 	return nil
 }
 
-// Connect creates a new client with default settings
+// Connect creates a new client with default settings.
 func (p *Pipeline) Connect() (beat.Client, error) {
 	return p.ConnectWith(beat.ClientConfig{})
 }
 
 // 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.
+// If not set otherwise the defaut publish mode is OutputChooses.
 func (p *Pipeline) ConnectWith(cfg beat.ClientConfig) (beat.Client, error) {
 	var (
 		canDrop      bool