Cherry-pick #19135 to 7.x: Apply Autodiscovery dynamic fields in auto…

…reloading (#19708)
elastic · Jul 7, 2020 · 8f5355d · 8f5355d
1 parent d2c02aa
commit 8f5355d
Show file tree

Hide file tree

Showing 42 changed files with 232 additions and 170 deletions.
diff --git a/CHANGELOG-developer.next.asciidoc b/CHANGELOG-developer.next.asciidoc
@@ -45,6 +45,7 @@ The list below covers the major changes between 7.0.0-rc2 and master only.
 - The Elasticsearch client settings expect the API key to be raw (not base64-encoded). {issue}18939[18939] {pull}18945[18945]
 - `management.ConfigManager` has been renamed to `management.Manager`. {pull}19114[19114]
 - `UpdateStatus` has been added to the `management.Manager` interface. {pull}19114[19114]
+- Remove `common.MapStrPointer` parameter from `cfgfile.Runnerfactory` interface. {pull}19135[19135]
 
 ==== Bugfixes
 

diff --git a/filebeat/beater/crawler.go b/filebeat/beater/crawler.go
@@ -128,7 +128,7 @@ func (c *crawler) startInput(
 		return fmt.Errorf("input with same ID already exists: %v", id)
 	}
 
-	runner, err := c.inputsFactory.Create(pipeline, config, nil)
+	runner, err := c.inputsFactory.Create(pipeline, config)
 	if err != nil {
 		return fmt.Errorf("Error while initializing input: %+v", err)
 	}

diff --git a/filebeat/channel/runner.go b/filebeat/channel/runner.go
@@ -32,7 +32,7 @@ type onCreateFactory struct {
 	create  onCreateWrapper
 }
 
-type onCreateWrapper func(cfgfile.RunnerFactory, beat.PipelineConnector, *common.Config, *common.MapStrPointer) (cfgfile.Runner, error)
+type onCreateWrapper func(cfgfile.RunnerFactory, beat.PipelineConnector, *common.Config) (cfgfile.Runner, error)
 
 // commonInputConfig defines common input settings
 // for the publisher pipeline.
@@ -63,12 +63,8 @@ func (f *onCreateFactory) CheckConfig(cfg *common.Config) error {
 	return f.factory.CheckConfig(cfg)
 }
 
-func (f *onCreateFactory) Create(
-	pipeline beat.PipelineConnector,
-	cfg *common.Config,
-	meta *common.MapStrPointer,
-) (cfgfile.Runner, error) {
-	return f.create(f.factory, pipeline, cfg, meta)
+func (f *onCreateFactory) Create(pipeline beat.PipelineConnector, cfg *common.Config) (cfgfile.Runner, error) {
+	return f.create(f.factory, pipeline, cfg)
 }
 
 // RunnerFactoryWithCommonInputSettings wraps a runner factory, such that all runners
@@ -93,14 +89,13 @@ func RunnerFactoryWithCommonInputSettings(info beat.Info, f cfgfile.RunnerFactor
 			f cfgfile.RunnerFactory,
 			pipeline beat.PipelineConnector,
 			cfg *common.Config,
-			meta *common.MapStrPointer,
 		) (runner cfgfile.Runner, err error) {
 			pipeline, err = withClientConfig(info, pipeline, cfg)
 			if err != nil {
 				return nil, err
 			}
 
-			return f.Create(pipeline, cfg, meta)
+			return f.Create(pipeline, cfg)
 		})
 }
 

diff --git a/filebeat/fileset/factory.go b/filebeat/fileset/factory.go
@@ -76,7 +76,7 @@ func NewFactory(
 }
 
 // Create creates a module based on a config
-func (f *Factory) Create(p beat.PipelineConnector, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) {
+func (f *Factory) Create(p beat.PipelineConnector, c *common.Config) (cfgfile.Runner, error) {
 	// Start a registry of one module:
 	m, err := NewModuleRegistry([]*common.Config{c}, f.beatInfo, false)
 	if err != nil {
@@ -98,7 +98,7 @@ func (f *Factory) Create(p beat.PipelineConnector, c *common.Config, meta *commo
 
 	inputs := make([]cfgfile.Runner, len(pConfigs))
 	for i, pConfig := range pConfigs {
-		inputs[i], err = f.inputFactory.Create(p, pConfig, meta)
+		inputs[i], err = f.inputFactory.Create(p, pConfig)
 		if err != nil {
 			logp.Err("Error creating input: %s", err)
 			return nil, err
@@ -116,7 +116,7 @@ func (f *Factory) Create(p beat.PipelineConnector, c *common.Config, meta *commo
 }
 
 func (f *Factory) CheckConfig(c *common.Config) error {
-	_, err := f.Create(pubpipeline.NewNilPipeline(), c, nil)
+	_, err := f.Create(pubpipeline.NewNilPipeline(), c)
 	return err
 }
 

diff --git a/filebeat/fileset/setup.go b/filebeat/fileset/setup.go
@@ -42,7 +42,7 @@ func NewSetupFactory(beatInfo beat.Info, pipelineLoaderFactory PipelineLoaderFac
 }
 
 // Create creates a new SetupCfgRunner to setup module configuration.
-func (sf *SetupFactory) Create(_ beat.PipelineConnector, c *common.Config, _ *common.MapStrPointer) (cfgfile.Runner, error) {
+func (sf *SetupFactory) Create(_ beat.PipelineConnector, c *common.Config) (cfgfile.Runner, error) {
 	m, err := NewModuleRegistry([]*common.Config{c}, sf.beatInfo, false)
 	if err != nil {
 		return nil, err
@@ -56,7 +56,7 @@ func (sf *SetupFactory) Create(_ beat.PipelineConnector, c *common.Config, _ *co
 }
 
 func (sf *SetupFactory) CheckConfig(c *common.Config) error {
-	_, err := sf.Create(pubpipeline.NewNilPipeline(), c, nil)
+	_, err := sf.Create(pubpipeline.NewNilPipeline(), c)
 	return err
 }
 

diff --git a/filebeat/input/input.go b/filebeat/input/input.go
@@ -60,7 +60,6 @@ func New(
 	connector channel.Connector,
 	beatDone chan struct{},
 	states []file.State,
-	dynFields *common.MapStrPointer,
 ) (*Runner, error) {
 	input := &Runner{
 		config:   defaultConfig,
@@ -82,11 +81,10 @@ func New(
 	}
 
 	context := Context{
-		States:        states,
-		Done:          input.done,
-		BeatDone:      input.beatDone,
-		DynamicFields: dynFields,
-		Meta:          nil,
+		States:   states,
+		Done:     input.done,
+		BeatDone: input.beatDone,
+		Meta:     nil,
 	}
 	var ipt Input
 	ipt, err = f(conf, connector, context)

diff --git a/filebeat/input/kafka/input.go b/filebeat/input/kafka/input.go
@@ -69,9 +69,6 @@ func NewInput(
 	}
 
 	out, err := connector.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: inputContext.DynamicFields,
-		},
 		ACKEvents: func(events []interface{}) {
 			for _, event := range events {
 				if meta, ok := event.(eventMeta); ok {

diff --git a/filebeat/input/log/input.go b/filebeat/input/log/input.go
@@ -107,11 +107,7 @@ func NewInput(
 	//  The outlet generated here is the underlying outlet, only closed
 	//  once all workers have been shut down.
 	//  For state updates and events, separate sub-outlets will be used.
-	out, err := outlet.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: context.DynamicFields,
-		},
-	})
+	out, err := outlet.Connect(cfg)
 	if err != nil {
 		return nil, err
 	}

diff --git a/filebeat/input/mqtt/client_mocked.go b/filebeat/input/mqtt/client_mocked.go
@@ -190,8 +190,8 @@ type mockedConnector struct {
 
 var _ channel.Connector = new(mockedConnector)
 
-func (m *mockedConnector) Connect(*common.Config) (channel.Outleter, error) {
-	panic("implement me")
+func (m *mockedConnector) Connect(c *common.Config) (channel.Outleter, error) {
+	return m.ConnectWith(c, beat.ClientConfig{})
 }
 
 func (m *mockedConnector) ConnectWith(*common.Config, beat.ClientConfig) (channel.Outleter, error) {

diff --git a/filebeat/input/mqtt/input.go b/filebeat/input/mqtt/input.go
@@ -79,11 +79,7 @@ func newInput(
 		return nil, errors.Wrap(err, "reading mqtt input config")
 	}
 
-	out, err := connector.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: inputContext.DynamicFields,
-		},
-	})
+	out, err := connector.Connect(cfg)
 	if err != nil {
 		return nil, err
 	}

diff --git a/filebeat/input/redis/input.go b/filebeat/input/redis/input.go
@@ -26,7 +26,6 @@ import (
 	"github.com/elastic/beats/v7/filebeat/harvester"
 	"github.com/elastic/beats/v7/filebeat/input"
 	"github.com/elastic/beats/v7/filebeat/input/file"
-	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/cfgwarn"
 	"github.com/elastic/beats/v7/libbeat/logp"
@@ -59,11 +58,7 @@ func NewInput(cfg *common.Config, connector channel.Connector, context input.Con
 		return nil, err
 	}
 
-	out, err := connector.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: context.DynamicFields,
-		},
-	})
+	out, err := connector.Connect(cfg)
 	if err != nil {
 		return nil, err
 	}

diff --git a/filebeat/input/registry.go b/filebeat/input/registry.go
@@ -27,11 +27,10 @@ import (
 )
 
 type Context struct {
-	States        []file.State
-	Done          chan struct{}
-	BeatDone      chan struct{}
-	DynamicFields *common.MapStrPointer
-	Meta          map[string]string
+	States   []file.State
+	Done     chan struct{}
+	BeatDone chan struct{}
+	Meta     map[string]string
 }
 
 // Factory is used to register functions creating new Input instances.

diff --git a/filebeat/input/runnerfactory.go b/filebeat/input/runnerfactory.go
@@ -46,10 +46,9 @@ func NewRunnerFactory(outlet channel.Factory, registrar *registrar.Registrar, be
 func (r *RunnerFactory) Create(
 	pipeline beat.PipelineConnector,
 	c *common.Config,
-	meta *common.MapStrPointer,
 ) (cfgfile.Runner, error) {
 	connector := r.outlet(pipeline)
-	p, err := New(c, connector, r.beatDone, r.registrar.GetStates(), meta)
+	p, err := New(c, connector, r.beatDone, r.registrar.GetStates())
 	if err != nil {
 		// In case of error with loading state, input is still returned
 		return p, err
@@ -59,6 +58,6 @@ func (r *RunnerFactory) Create(
 }
 
 func (r *RunnerFactory) CheckConfig(cfg *common.Config) error {
-	_, err := r.Create(pipeline.NewNilPipeline(), cfg, nil)
+	_, err := r.Create(pipeline.NewNilPipeline(), cfg)
 	return err
 }
diff --git a/filebeat/input/stdin/input.go b/filebeat/input/stdin/input.go
@@ -25,7 +25,6 @@ import (
 	"github.com/elastic/beats/v7/filebeat/input"
 	"github.com/elastic/beats/v7/filebeat/input/file"
 	"github.com/elastic/beats/v7/filebeat/input/log"
-	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/logp"
 )
@@ -49,11 +48,7 @@ type Input struct {
 // NewInput creates a new stdin input
 // This input contains one harvester which is reading from stdin
 func NewInput(cfg *common.Config, outlet channel.Connector, context input.Context) (input.Input, error) {
-	out, err := outlet.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: context.DynamicFields,
-		},
-	})
+	out, err := outlet.Connect(cfg)
 	if err != nil {
 		return nil, err
 	}

diff --git a/filebeat/input/syslog/input.go b/filebeat/input/syslog/input.go
@@ -112,11 +112,7 @@ func NewInput(
 
 	log := logp.NewLogger("syslog")
 
-	out, err := outlet.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: context.DynamicFields,
-		},
-	})
+	out, err := outlet.Connect(cfg)
 	if err != nil {
 		return nil, err
 	}

diff --git a/filebeat/input/tcp/input.go b/filebeat/input/tcp/input.go
@@ -57,11 +57,7 @@ func NewInput(
 	context input.Context,
 ) (input.Input, error) {
 
-	out, err := connector.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: context.DynamicFields,
-		},
-	})
+	out, err := connector.Connect(cfg)
 	if err != nil {
 		return nil, err
 	}

diff --git a/filebeat/input/udp/input.go b/filebeat/input/udp/input.go
@@ -53,11 +53,7 @@ func NewInput(
 	context input.Context,
 ) (input.Input, error) {
 
-	out, err := outlet.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: context.DynamicFields,
-		},
-	})
+	out, err := outlet.Connect(cfg)
 	if err != nil {
 		return nil, err
 	}

diff --git a/filebeat/input/unix/input.go b/filebeat/input/unix/input.go
@@ -59,11 +59,7 @@ func NewInput(
 ) (input.Input, error) {
 	cfgwarn.Beta("Unix socket support is beta.")
 
-	out, err := connector.ConnectWith(cfg, beat.ClientConfig{
-		Processing: beat.ProcessingConfig{
-			DynamicFields: context.DynamicFields,
-		},
-	})
+	out, err := connector.Connect(cfg)
 	if err != nil {
 		return nil, err
 	}

diff --git a/heartbeat/beater/heartbeat.go b/heartbeat/beater/heartbeat.go
@@ -126,7 +126,7 @@ func (bt *Heartbeat) RunStaticMonitors(b *beat.Beat) error {
 	factory := monitors.NewFactory(bt.scheduler, true)
 
 	for _, cfg := range bt.config.Monitors {
-		created, err := factory.Create(b.Publisher, cfg, nil)
+		created, err := factory.Create(b.Publisher, cfg)
 		if err != nil {
 			return errors.Wrap(err, "could not create monitor")
 		}

diff --git a/heartbeat/monitors/factory.go b/heartbeat/monitors/factory.go
@@ -37,8 +37,8 @@ func NewFactory(sched *scheduler.Scheduler, allowWatches bool) *RunnerFactory {
 }
 
 // Create makes a new Runner for a new monitor with the given Config.
-func (f *RunnerFactory) Create(p beat.PipelineConnector, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) {
-	monitor, err := newMonitor(c, globalPluginsReg, p, f.sched, f.allowWatches, meta)
+func (f *RunnerFactory) Create(p beat.PipelineConnector, c *common.Config) (cfgfile.Runner, error) {
+	monitor, err := newMonitor(c, globalPluginsReg, p, f.sched, f.allowWatches)
 	return monitor, err
 }
 

diff --git a/heartbeat/monitors/monitor.go b/heartbeat/monitors/monitor.go
@@ -62,8 +62,7 @@ type Monitor struct {
 
 	// stats is the countersRecorder used to record lifecycle events
 	// for global metrics + telemetry
-	stats           registryRecorder
-	factoryMetadata *common.MapStrPointer
+	stats registryRecorder
 }
 
 // String prints a description of the monitor in a threadsafe way. It is important that this use threadsafe
@@ -73,7 +72,7 @@ func (m *Monitor) String() string {
 }
 
 func checkMonitorConfig(config *common.Config, registrar *pluginsReg, allowWatches bool) error {
-	m, err := newMonitor(config, registrar, nil, nil, allowWatches, nil)
+	m, err := newMonitor(config, registrar, nil, nil, allowWatches)
 	if m != nil {
 		m.Stop() // Stop the monitor to free up the ID from uniqueness checks
 	}
@@ -101,9 +100,8 @@ func newMonitor(
 	pipelineConnector beat.PipelineConnector,
 	scheduler *scheduler.Scheduler,
 	allowWatches bool,
-	factoryMetadata *common.MapStrPointer,
 ) (*Monitor, error) {
-	m, err := newMonitorUnsafe(config, registrar, pipelineConnector, scheduler, allowWatches, factoryMetadata)
+	m, err := newMonitorUnsafe(config, registrar, pipelineConnector, scheduler, allowWatches)
 	if m != nil && err != nil {
 		m.Stop()
 	}
@@ -118,7 +116,6 @@ func newMonitorUnsafe(
 	pipelineConnector beat.PipelineConnector,
 	scheduler *scheduler.Scheduler,
 	allowWatches bool,
-	factoryMetadata *common.MapStrPointer,
 ) (*Monitor, error) {
 	// Extract just the Id, Type, and Enabled fields from the config
 	// We'll parse things more precisely later once we know what exact type of
@@ -145,7 +142,6 @@ func newMonitorUnsafe(
 		internalsMtx:      sync.Mutex{},
 		config:            config,
 		stats:             monitorPlugin.stats,
-		factoryMetadata:   factoryMetadata,
 	}
 
 	if m.id != "" {