Beats event processing and default fields

filebeat/channel/factory.go

@@ -125,13 +125,15 @@ func (f *OutletFactory) Create(p beat.Pipeline, cfg *common.Config, dynFields *c
 	}
 
 	client, err := p.ConnectWith(beat.ClientConfig{
-		PublishMode:   beat.GuaranteedSend,
-		EventMetadata: config.EventMetadata,
-		DynamicFields: dynFields,
-		Meta:          meta,
-		Fields:        fields,
-		Processor:     processors,
-		Events:        f.eventer,
+		PublishMode: beat.GuaranteedSend,
+		Processing: beat.ProcessingConfig{
+			EventMetadata: config.EventMetadata,
+			DynamicFields: dynFields,
+			Meta:          meta,
+			Fields:        fields,
+			Processor:     processors,
+		},
+		Events: f.eventer,
 	})
 	if err != nil {
 		return nil, err

heartbeat/monitors/task.go

@@ -139,9 +139,11 @@ func (t *configuredJob) Start() {
 	}
 
 	t.client, err = t.monitor.pipelineConnector.ConnectWith(beat.ClientConfig{
-		EventMetadata: t.config.EventMetadata,
-		Processor:     t.processors,
-		Fields:        fields,
+		Processing: beat.ProcessingConfig{
+			EventMetadata: t.config.EventMetadata,
+			Processor:     t.processors,
+			Fields:        fields,
+		},
 	})
 	if err != nil {
 		logp.Err("could not start monitor: %v", err)

journalbeat/input/input.go

@@ -125,10 +125,12 @@ func New(
 func (i *Input) Run() {
 	var err error
 	i.client, err = i.pipeline.ConnectWith(beat.ClientConfig{
-		PublishMode:   beat.GuaranteedSend,
-		EventMetadata: i.eventMeta,
-		Meta:          nil,
-		Processor:     i.processors,
+		PublishMode: beat.GuaranteedSend,
+		Processing: beat.ProcessingConfig{
+			EventMetadata: i.eventMeta,
+			Meta:          nil,
+			Processor:     i.processors,
+		},
 		ACKCount: func(n int) {
 			i.logger.Infof("journalbeat successfully published %d events", n)
 		},

libbeat/beat/pipeline.go

@@ -46,22 +46,7 @@ type Client interface {
 type ClientConfig struct {
 	PublishMode PublishMode
 
-	// EventMetadata configures additional fields/tags to be added to published events.
-	EventMetadata common.EventMetadata
-
-	// Meta provides additional meta data to be added to the Meta field in the beat.Event
-	// structure.
-	Meta common.MapStr
-
-	// Fields provides additional 'global' fields to be added to every event
-	Fields common.MapStr
-
-	// DynamicFields provides additional fields to be added to every event, supporting live updates
-	DynamicFields *common.MapStrPointer
-
-	// Processors passes additional processor to the client, to be executed before
-	// the pipeline processors.
-	Processor ProcessorList
+	Processing ProcessingConfig
 
 	// WaitClose sets the maximum duration to wait on ACK, if client still has events
 	// active non-acknowledged events in the publisher pipeline.
@@ -72,14 +57,6 @@ type ClientConfig struct {
 	// Events configures callbacks for common client callbacks
 	Events ClientEventer
 
-	// By default events are normalized within processor pipeline,
-	// if the normalization step should be skipped set this to true.
-	SkipNormalization bool
-
-	// By default events are decorated with agent metadata.
-	// To skip adding that metadata set this to true.
-	SkipAgentMetadata bool
-
 	// ACK handler strategies.
 	// Note: ack handlers are run in another go-routine owned by the publisher pipeline.
 	//       They should not block for to long, to not block the internal buffers for
@@ -101,6 +78,31 @@ type ClientConfig struct {
 	ACKLastEvent func(interface{})
 }
 
+// ProcessingConfig provides additional event processing settings a client can
+// pass to the publisher pipeline on Connect.
+type ProcessingConfig struct {
+	// EventMetadata configures additional fields/tags to be added to published events.
+	EventMetadata common.EventMetadata
+
+	// Meta provides additional meta data to be added to the Meta field in the beat.Event
+	// structure.
+	Meta common.MapStr
+
+	// Fields provides additional 'global' fields to be added to every event
+	Fields common.MapStr
+
+	// DynamicFields provides additional fields to be added to every event, supporting live updates
+	DynamicFields *common.MapStrPointer
+
+	// Processors passes additional processor to the client, to be executed before
+	// the pipeline processors.
+	Processor ProcessorList
+
+	// Private contains additional information to be passed to the processing
+	// pipeline builder.
+	Private interface{}
+}
+
 // ClientEventer provides access to internal client events.
 type ClientEventer interface {
 	Closing() // Closing indicates the client is being shutdown next

libbeat/cmd/instance/beat.go

@@ -62,6 +62,7 @@ import (
 	"github.com/elastic/beats/libbeat/paths"
 	"github.com/elastic/beats/libbeat/plugin"
 	"github.com/elastic/beats/libbeat/publisher/pipeline"
+	"github.com/elastic/beats/libbeat/publisher/processing"
 	svc "github.com/elastic/beats/libbeat/service"
 	"github.com/elastic/beats/libbeat/version"
 	sysinfo "github.com/elastic/go-sysinfo"
@@ -78,6 +79,8 @@ type Beat struct {
 
 	keystore keystore.Keystore
 	index    idxmgmt.Supporter
+
+	processing processing.Supporter
 }
 
 type beatConfig struct {
@@ -310,6 +313,7 @@ func (b *Beat) createBeater(bt beat.Creator) (beat.Beater, error) {
 			Logger:    logp.L().Named("publisher"),
 		},
 		b.Config.Pipeline,
+		b.processing,
 		b.makeOutputFactory(b.Config.Output),
 	)
 
@@ -593,6 +597,16 @@ func (b *Beat) configure(settings Settings) error {
 		imFactory = idxmgmt.MakeDefaultSupport(settings.ILM)
 	}
 	b.index, err = imFactory(nil, b.Beat.Info, b.RawConfig)
+	if err != nil {
+		return err
+	}
+
+	processingFactory := settings.Processing
+	if processingFactory == nil {
+		processingFactory = processing.MakeDefaultBeatSupport(true)
+	}
+	b.processing, err = processingFactory(b.Info, logp.L().Named("processors"), b.RawConfig)
+
 	return err
 }
 

libbeat/cmd/instance/settings.go

@@ -24,6 +24,7 @@ import (
 	"github.com/elastic/beats/libbeat/idxmgmt"
 	"github.com/elastic/beats/libbeat/idxmgmt/ilm"
 	"github.com/elastic/beats/libbeat/monitoring/report"
+	"github.com/elastic/beats/libbeat/publisher/processing"
 )
 
 // Settings contains basic settings for any beat to pass into GenRootCmd
@@ -40,4 +41,6 @@ type Settings struct {
 	// load custom index manager. The config object will be the Beats root configuration.
 	IndexManagement idxmgmt.SupportFactory
 	ILM             ilm.SupportFactory
+
+	Processing processing.SupportFactory
 }

libbeat/monitoring/report/elasticsearch/elasticsearch.go

@@ -37,6 +37,7 @@ import (
 	"github.com/elastic/beats/libbeat/outputs/outil"
 	"github.com/elastic/beats/libbeat/outputs/transport"
 	"github.com/elastic/beats/libbeat/publisher/pipeline"
+	"github.com/elastic/beats/libbeat/publisher/processing"
 	"github.com/elastic/beats/libbeat/publisher/queue"
 	"github.com/elastic/beats/libbeat/publisher/queue/memqueue"
 )
@@ -169,11 +170,16 @@ func makeReporter(beat beat.Info, settings report.Settings, cfg *common.Config)
 	outClient := outputs.NewFailoverClient(clients)
 	outClient = outputs.WithBackoff(outClient, config.Backoff.Init, config.Backoff.Max)
 
+	processing, err := processing.MakeDefaultSupport(true)(beat, log, common.NewConfig())
+	if err != nil {
+		return nil, err
+	}
+
 	pipeline, err := pipeline.New(
 		beat,
 		pipeline.Monitors{
 			Metrics: monitoring,
-			Logger:  logp.NewLogger(selector),
+			Logger:  log,
 		},
 		queueFactory,
 		outputs.Group{
@@ -184,6 +190,7 @@ func makeReporter(beat beat.Info, settings report.Settings, cfg *common.Config)
 		pipeline.Settings{
 			WaitClose:     0,
 			WaitCloseMode: pipeline.NoWaitOnClose,
+			Processors:    processing,
 		})
 	if err != nil {
 		return nil, err

libbeat/publisher/pipeline/module.go

@@ -26,7 +26,7 @@ import (
 	"github.com/elastic/beats/libbeat/logp"
 	"github.com/elastic/beats/libbeat/monitoring"
 	"github.com/elastic/beats/libbeat/outputs"
-	"github.com/elastic/beats/libbeat/processors"
+	"github.com/elastic/beats/libbeat/publisher/processing"
 	"github.com/elastic/beats/libbeat/publisher/queue"
 )
 
@@ -60,6 +60,7 @@ func Load(
 	beatInfo beat.Info,
 	monitors Monitors,
 	config Config,
+	processors processing.Supporter,
 	makeOutput func(outputs.Observer) (string, outputs.Group, error),
 ) (*Pipeline, error) {
 	log := monitors.Logger
@@ -71,28 +72,11 @@ func Load(
 		log.Info("Dry run mode. All output types except the file based one are disabled.")
 	}
 
-	processors, err := processors.New(config.Processors)
-	if err != nil {
-		return nil, fmt.Errorf("error initializing processors: %v", err)
-	}
-
 	name := beatInfo.Name
 	settings := Settings{
 		WaitClose:     0,
 		WaitCloseMode: NoWaitOnClose,
-		Disabled:      publishDisabled,
 		Processors:    processors,
-		Annotations: Annotations{
-			Event: config.EventMetadata,
-			Builtin: common.MapStr{
-				"host": common.MapStr{
-					"name": name,
-				},
-				"ecs": common.MapStr{
-					"version": "1.0.0-beta2",
-				},
-			},
-		},
 	}
 
 	queueBuilder, err := createQueueBuilder(config.Queue, monitors)