Add filebeat.config.path as replacement for filebeat.config_dir (#4051)

The reload feature can also be used to load configuration files once. This enables the feature to load config options once through `filebeat.config.path: ...`. The difference to config_dir is that only the prospectors have to be configured in the file and not the full filebeat config tree.

`filebeat.config_dir` will be removed in a follow up PR.

As the reloading is now also used for just loading, the code could use some refactoring in the future to make this more obvious.

CHANGELOG.asciidoc

@@ -114,6 +114,7 @@ https://github.com/elastic/beats/compare/v5.1.1...master[Check the HEAD diff]
 - Restructure input.Event to be inline with outputs.Data {pull}3823[3823]
 - Add base for supporting prospector level processors {pull}3853[3853]
 - Add auditd module for reading audit logs on Linux. {pull}3750[3750] {pull}3941[3941]
+- Add filebeat.config.path as replacement for config_dir. {pull}4051[4051]
 
 *Heartbeat*
 

filebeat/_meta/common.full.p2.yml

@@ -232,3 +232,10 @@ filebeat.prospectors:
 # How long filebeat waits on shutdown for the publisher to finish.
 # Default is 0, not waiting.
 #filebeat.shutdown_timeout: 0
+
+# Enable filebeat config reloading
+#filebeat.config.prospectors:
+  #enabled: false
+  #path: configs/*.yml
+  #reload.enabled: true
+  #reload.period: 10s

filebeat/beater/filebeat.go

@@ -57,12 +57,12 @@ func New(b *beat.Beat, rawConfig *common.Config) (beat.Beater, error) {
 	// Add prospectors created by the modules
 	config.Prospectors = append(config.Prospectors, moduleProspectors...)
 
-	if !config.ProspectorReload.Enabled() && len(config.Prospectors) == 0 {
+	if !config.ConfigProspector.Enabled() && len(config.Prospectors) == 0 {
 		return nil, errors.New("No prospectors defined. What files do you want me to watch?")
 	}
 
-	if *once && config.ProspectorReload.Enabled() {
-		return nil, errors.New("prospector reloading and -once cannot be used together.")
+	if *once && config.ConfigProspector.Enabled() {
+		return nil, errors.New("prospector configs and -once cannot be used together")
 	}
 
 	fb := &Filebeat{
@@ -182,7 +182,7 @@ func (fb *Filebeat) Run(b *beat.Beat) error {
 		spooler.Stop()
 	}()
 
-	err = crawler.Start(registrar, config.ProspectorReload)
+	err = crawler.Start(registrar, config.ConfigProspector)
 	if err != nil {
 		crawler.Stop()
 		return err

filebeat/config/config.go

@@ -27,7 +27,7 @@ type Config struct {
 	ConfigDir        string           `config:"config_dir"`
 	ShutdownTimeout  time.Duration    `config:"shutdown_timeout"`
 	Modules          []*common.Config `config:"modules"`
-	ProspectorReload *common.Config   `config:"config.prospectors"`
+	ConfigProspector *common.Config   `config:"config.prospectors"`
 }
 
 var (

filebeat/crawler/crawler.go

@@ -33,7 +33,8 @@ func New(out prospector.Outlet, prospectorConfigs []*common.Config, beatDone cha
 	}, nil
 }
 
-func (c *Crawler) Start(r *registrar.Registrar, reloaderConfig *common.Config) error {
+// Start starts the crawler with all prospectors
+func (c *Crawler) Start(r *registrar.Registrar, configProspectors *common.Config) error {
 
 	logp.Info("Loading Prospectors: %v", len(c.prospectorConfigs))
 
@@ -45,10 +46,10 @@ func (c *Crawler) Start(r *registrar.Registrar, reloaderConfig *common.Config) e
 		}
 	}
 
-	if reloaderConfig.Enabled() {
-		logp.Warn("BETA feature dynamic configuration reloading is enabled.")
+	if configProspectors.Enabled() {
+		logp.Beta("Loading separate prospectors is enabled.")
 
-		c.reloader = cfgfile.NewReloader(reloaderConfig)
+		c.reloader = cfgfile.NewReloader(configProspectors)
 		factory := prospector.NewFactory(c.out, r, c.beatDone)
 		go func() {
 			c.reloader.Run(factory)

filebeat/filebeat.full.yml

@@ -362,6 +362,13 @@ filebeat.prospectors:
 # Default is 0, not waiting.
 #filebeat.shutdown_timeout: 0
 
+# Enable filebeat config reloading
+#filebeat.config.prospectors:
+  #enabled: false
+  #path: configs/*.yml
+  #reload.enabled: true
+  #reload.period: 10s
+
 #================================ General ======================================
 
 # The name of the shipper that publishes the network data. It can be used to group

filebeat/tests/system/config/filebeat.yml.j2

@@ -94,11 +94,14 @@ filebeat.registry_file: {{ beat.working_dir + '/' }}{{ registryFile|default("reg
 {%endif%}
 filebeat.publish_async: {{publish_async}}
 
-{% if reload -%}
+{% if reload or reload_path -%}
 filebeat.config.prospectors:
+  enabled: true
   path: {{ reload_path }}
+  {% if reload  -%}
   reload.period: 1s
   reload.enabled: true
+  {% endif -%}
 {% endif -%}
 
 #================================ General =====================================

filebeat/tests/system/test_reload.py

@@ -201,3 +201,53 @@ def test_reload_same_prospector(self):
         assert output[1]["message"] == second_line
         # assert that fields are added
         assert output[1]["fields.hello"] == "world"
+
+    def test_load_configs(self):
+        """
+        Test loading separate prospectors configs
+        """
+        self.render_config_template(
+            reload_path=self.working_dir + "/configs/*.yml",
+            prospectors=False,
+        )
+
+        os.mkdir(self.working_dir + "/logs/")
+        logfile = self.working_dir + "/logs/test.log"
+        os.mkdir(self.working_dir + "/configs/")
+
+        first_line = "First log file"
+        second_line = "Second log file"
+
+        config = prospectorConfigTemplate.format(self.working_dir + "/logs/test.log")
+        config = config + """
+  close_eof: true
+"""
+        with open(self.working_dir + "/configs/prospector.yml", 'w') as f:
+            f.write(config)
+
+        with open(logfile, 'w') as f:
+            f.write(first_line + "\n")
+
+        proc = self.start_beat()
+
+        self.wait_until(lambda: self.output_lines() == 1)
+
+        # Update both log files, only 1 change should be picke dup
+        with open(logfile, 'a') as f:
+            f.write(second_line + "\n")
+
+        self.wait_until(lambda: self.output_lines() == 2)
+
+        proc.check_kill_and_wait()
+
+        output = self.read_output()
+
+        # Reloading stopped.
+        self.wait_until(
+            lambda: self.log_contains("Loading of config files completed."),
+            max_timeout=15)
+
+        # Make sure the correct lines were picked up
+        assert self.output_lines() == 2
+        assert output[0]["message"] == first_line
+        assert output[1]["message"] == second_line

libbeat/cfgfile/reload.go

@@ -87,6 +87,11 @@ func (rl *Reloader) Run(runnerFactory RunnerFactory) {
 
 	gw := NewGlobWatcher(path)
 
+	// If reloading is disable, config files should be loaded immidiately
+	if !rl.config.Reload.Enabled {
+		rl.config.Reload.Period = 0
+	}
+
 	for {
 		select {
 		case <-rl.done:
@@ -153,6 +158,16 @@ func (rl *Reloader) Run(runnerFactory RunnerFactory) {
 			rl.stopRunners(stopList)
 			rl.startRunners(startList)
 		}
+
+		// Path loading is enabled but not reloading. Loads files only once and then stops.
+		if !rl.config.Reload.Enabled {
+			logp.Info("Loading of config files completed.")
+			select {
+			case <-rl.done:
+				logp.Info("Dynamic config reloader stopped")
+				return
+			}
+		}
 	}
 }