Center more functionality around RunnerFactory (elastic#16715)

* Center more functionality around RunnerFactory Remove CheckableRunnerFactory and require RunnerFactory to implement CheckConfig. CheckableRunnerFactory more and more superseded RunnerFactory. As we want more config validation support in the future as well I combined the two into RunnerFactory Remove autodiscover.Adapter. The adapter did inherit from CheckableRunnerFactory, giving us some inheritance chain between RunnerFactory, CheckableRunnerFactory, and Adapter for autodiscovery. By removig the Adapter and CheckableRunnerFactory we have one common type (RunnerFactory) to integrate with config file reloading, static input/module setup, and autodiscovery. Add selectors for autodiscovery event selection that are used as additional parameters when creating a new Autodiscover instance. This gives us some more composability, yet I wonder if we can even remove those, as every instance of NewAutodiscover did look for events with a 'config' field of type []*common.Config.
leweafan · Mar 26, 2020 · 257fdfc · 257fdfc
1 parent 778156c
commit 257fdfc
Show file tree

Hide file tree

Showing 26 changed files with 336 additions and 290 deletions.
diff --git a/filebeat/autodiscover/autodiscover.go b/filebeat/autodiscover/autodiscover.go
diff --git a/filebeat/autodiscover/include.go b/filebeat/autodiscover/include.go
diff --git a/filebeat/beater/crawler.go b/filebeat/beater/crawler.go
@@ -76,27 +76,32 @@ func (c *crawler) Start(
 	for _, inputConfig := range c.inputConfigs {
 		err := c.startInput(pipeline, inputConfig, r.GetStates())
 		if err != nil {
-			return err
+			return fmt.Errorf("starting input failed: %+v", err)
 		}
 	}
 
 	if configInputs.Enabled() {
 		c.inputReloader = cfgfile.NewReloader(pipeline, configInputs)
 		if err := c.inputReloader.Check(c.inputsFactory); err != nil {
-			return err
+			return fmt.Errorf("creating input reloader failed: %+v", err)
 		}
 
-		go func() {
-			c.inputReloader.Run(c.inputsFactory)
-		}()
 	}
 
 	if configModules.Enabled() {
 		c.modulesReloader = cfgfile.NewReloader(pipeline, configModules)
 		if err := c.modulesReloader.Check(c.modulesFactory); err != nil {
-			return err
+			return fmt.Errorf("creating module reloader failed: %+v", err)
 		}
 
+	}
+
+	if c.inputReloader != nil {
+		go func() {
+			c.inputReloader.Run(c.inputsFactory)
+		}()
+	}
+	if c.modulesReloader != nil {
 		go func() {
 			c.modulesReloader.Run(c.modulesFactory)
 		}()

diff --git a/filebeat/beater/filebeat.go b/filebeat/beater/filebeat.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/pkg/errors"
 
-	fbautodiscover "github.com/elastic/beats/v7/filebeat/autodiscover"
 	"github.com/elastic/beats/v7/filebeat/channel"
 	cfg "github.com/elastic/beats/v7/filebeat/config"
 	"github.com/elastic/beats/v7/filebeat/fileset"
@@ -43,9 +42,14 @@ import (
 	"github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
 
+	_ "github.com/elastic/beats/v7/filebeat/include"
+
 	// Add filebeat level processors
 	_ "github.com/elastic/beats/v7/filebeat/processor/add_kubernetes_metadata"
 	_ "github.com/elastic/beats/v7/libbeat/processors/decode_csv_fields"
+
+	// include all filebeat specific builders
+	_ "github.com/elastic/beats/v7/filebeat/autodiscover/builder/hints"
 )
 
 const pipelinesWarning = "Filebeat is unable to load the Ingest Node pipelines for the configured" +
@@ -282,7 +286,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 	err = crawler.Start(b.Publisher, registrar, config.ConfigInput, config.ConfigModules)
 	if err != nil {
 		crawler.Stop()
-		return err
+		return fmt.Errorf("Failed to start crawler: %+v", err)
 	}
 
 	// If run once, add crawler completion check as alternative to done signal
@@ -304,8 +308,16 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 
 	var adiscover *autodiscover.Autodiscover
 	if fb.config.Autodiscover != nil {
-		adapter := fbautodiscover.NewAutodiscoverAdapter(inputLoader, moduleLoader)
-		adiscover, err = autodiscover.NewAutodiscover("filebeat", b.Publisher, adapter, config.Autodiscover)
+		adiscover, err = autodiscover.NewAutodiscover(
+			"filebeat",
+			b.Publisher,
+			cfgfile.MultiplexedRunnerFactory(
+				cfgfile.MatchHasField("module", moduleLoader),
+				cfgfile.MatchDefault(inputLoader),
+			),
+			autodiscover.QueryConfig(),
+			config.Autodiscover,
+		)
 		if err != nil {
 			return err
 		}

diff --git a/filebeat/channel/connector.go b/filebeat/channel/connector.go
@@ -30,7 +30,7 @@ type ConnectorFunc func(*common.Config, beat.ClientConfig) (Outleter, error)
 
 type pipelineConnector struct {
 	parent   *OutletFactory
-	pipeline beat.Pipeline
+	pipeline beat.PipelineConnector
 }
 
 // Connect passes the cfg and the zero value of beat.ClientConfig to the underlying function.

diff --git a/filebeat/channel/factory.go b/filebeat/channel/factory.go
@@ -87,7 +87,7 @@ func NewOutletFactory(
 // Inputs and all harvesters use the same pipeline client instance.
 // This guarantees ordering between events as required by the registrar for
 // file.State updates
-func (f *OutletFactory) Create(p beat.Pipeline) Connector {
+func (f *OutletFactory) Create(p beat.PipelineConnector) Connector {
 	return &pipelineConnector{parent: f, pipeline: p}
 }
 

diff --git a/filebeat/channel/interface.go b/filebeat/channel/interface.go
@@ -23,7 +23,7 @@ import (
 )
 
 // Factory is used to create a new Outlet instance
-type Factory func(beat.Pipeline) Connector
+type Factory func(beat.PipelineConnector) Connector
 
 // Connector creates an Outlet connecting the event publishing with some internal pipeline.
 // type Connector func(*common.Config, *common.MapStrPointer) (Outleter, error)

diff --git a/filebeat/fileset/factory.go b/filebeat/fileset/factory.go
@@ -27,6 +27,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/logp"
 	"github.com/elastic/beats/v7/libbeat/monitoring"
 	"github.com/elastic/beats/v7/libbeat/outputs/elasticsearch"
+	pubpipeline "github.com/elastic/beats/v7/libbeat/publisher/pipeline"
 
 	"github.com/mitchellh/hashstructure"
 )
@@ -75,7 +76,7 @@ func NewFactory(
 }
 
 // Create creates a module based on a config
-func (f *Factory) Create(p beat.Pipeline, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) {
+func (f *Factory) Create(p beat.PipelineConnector, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) {
 	// Start a registry of one module:
 	m, err := NewModuleRegistry([]*common.Config{c}, f.beatInfo, false)
 	if err != nil {
@@ -114,6 +115,11 @@ func (f *Factory) Create(p beat.Pipeline, c *common.Config, meta *common.MapStrP
 	}, nil
 }
 
+func (f *Factory) CheckConfig(c *common.Config) error {
+	_, err := f.Create(pubpipeline.NewNilPipeline(), c, nil)
+	return err
+}
+
 func (p *inputsRunner) Start() {
 	// Load pipelines
 	if p.pipelineLoaderFactory != nil {
@@ -153,6 +159,7 @@ func (p *inputsRunner) Start() {
 		moduleList.Add(m)
 	}
 }
+
 func (p *inputsRunner) Stop() {
 	if p.pipelineCallbackID != uuid.Nil {
 		elasticsearch.DeregisterConnectCallback(p.pipelineCallbackID)

diff --git a/filebeat/fileset/setup.go b/filebeat/fileset/setup.go
@@ -22,6 +22,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/logp"
+	pubpipeline "github.com/elastic/beats/v7/libbeat/publisher/pipeline"
 )
 
 // SetupFactory is for loading module assets when running setup subcommand.
@@ -41,7 +42,7 @@ func NewSetupFactory(beatInfo beat.Info, pipelineLoaderFactory PipelineLoaderFac
 }
 
 // Create creates a new SetupCfgRunner to setup module configuration.
-func (sf *SetupFactory) Create(_ beat.Pipeline, c *common.Config, _ *common.MapStrPointer) (cfgfile.Runner, error) {
+func (sf *SetupFactory) Create(_ beat.PipelineConnector, c *common.Config, _ *common.MapStrPointer) (cfgfile.Runner, error) {
 	m, err := NewModuleRegistry([]*common.Config{c}, sf.beatInfo, false)
 	if err != nil {
 		return nil, err
@@ -54,6 +55,11 @@ func (sf *SetupFactory) Create(_ beat.Pipeline, c *common.Config, _ *common.MapS
 	}, nil
 }
 
+func (sf *SetupFactory) CheckConfig(c *common.Config) error {
+	_, err := sf.Create(pubpipeline.NewNilPipeline(), c, nil)
+	return err
+}
+
 // SetupCfgRunner is for loading assets of modules.
 type SetupCfgRunner struct {
 	moduleRegistry        *ModuleRegistry

diff --git a/filebeat/input/runnerfactory.go b/filebeat/input/runnerfactory.go
@@ -23,6 +23,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/beat"
 	"github.com/elastic/beats/v7/libbeat/cfgfile"
 	"github.com/elastic/beats/v7/libbeat/common"
+	"github.com/elastic/beats/v7/libbeat/publisher/pipeline"
 )
 
 // RunnerFactory is a factory for registrars
@@ -43,7 +44,7 @@ func NewRunnerFactory(outlet channel.Factory, registrar *registrar.Registrar, be
 
 // Create creates a input based on a config
 func (r *RunnerFactory) Create(
-	pipeline beat.Pipeline,
+	pipeline beat.PipelineConnector,
 	c *common.Config,
 	meta *common.MapStrPointer,
 ) (cfgfile.Runner, error) {
@@ -56,3 +57,8 @@ func (r *RunnerFactory) Create(
 
 	return p, nil
 }
+
+func (r *RunnerFactory) CheckConfig(cfg *common.Config) error {
+	_, err := r.Create(pipeline.NewNilPipeline(), cfg, nil)
+	return err
+}
diff --git a/heartbeat/autodiscover/include.go b/heartbeat/autodiscover/include.go
diff --git a/heartbeat/beater/heartbeat.go b/heartbeat/beater/heartbeat.go
@@ -156,14 +156,7 @@ func (bt *Heartbeat) RunReloadableMonitors(b *beat.Beat) (err error) {
 
 // makeAutodiscover creates an autodiscover object ready to be started.
 func (bt *Heartbeat) makeAutodiscover(b *beat.Beat) (*autodiscover.Autodiscover, error) {
-	adapter := autodiscover.NewFactoryAdapter(bt.dynamicFactory)
-
-	ad, err := autodiscover.NewAutodiscover("heartbeat", b.Publisher, adapter, bt.config.Autodiscover)
-	if err != nil {
-		return nil, err
-	}
-
-	return ad, nil
+	return autodiscover.NewAutodiscover("heartbeat", b.Publisher, bt.dynamicFactory, autodiscover.QueryConfig(), bt.config.Autodiscover)
 }
 
 // Stop stops the beat.

diff --git a/heartbeat/cmd/root.go b/heartbeat/cmd/root.go
@@ -20,9 +20,11 @@ package cmd
 import (
 	"fmt"
 
-	_ "github.com/elastic/beats/v7/heartbeat/autodiscover"
 	"github.com/elastic/beats/v7/heartbeat/beater"
 
+	// include all heartbeat specific autodiscovery builders
+	_ "github.com/elastic/beats/v7/heartbeat/autodiscover/builder/hints"
+
 	// register default heartbeat monitors
 	_ "github.com/elastic/beats/v7/heartbeat/monitors/defaults"
 	cmd "github.com/elastic/beats/v7/libbeat/cmd"

diff --git a/heartbeat/monitors/factory.go b/heartbeat/monitors/factory.go
@@ -37,7 +37,7 @@ 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.Pipeline, c *common.Config, meta *common.MapStrPointer) (cfgfile.Runner, error) {
+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)
 	return monitor, err
 }