Center more functionality around RunnerFactory

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)
 		}()

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
 		}

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.

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}
 }
 

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)

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)

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

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
+}

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.

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"

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
 }