update filebeat config for the tests (#6284)

Update all python based tests to use `input` instead of prospectors,
All test were changed to use the new options in the YAML configuration.
This commit also fixes an issue when using the
`filebeat.config.inputs` field instead of the
`filebeat.config.prospectors` and add a deprecation test for it.

filebeat/beater/filebeat.go

@@ -62,6 +62,14 @@ func New(b *beat.Beat, rawConfig *common.Config) (beat.Beater, error) {
 		config.Inputs = config.Prospectors
 	}
 
+	if config.ConfigProspector != nil {
+		cfgwarn.Deprecate("7.0.0", "config.prospectors are deprecated, Use `config.inputs` instead.")
+		if config.ConfigInput != nil {
+			return nil, fmt.Errorf("config.prospectors and config.inputs used in the configuration file, define only config.inputs not both")
+		}
+		config.ConfigInput = config.ConfigProspector
+	}
+
 	moduleRegistry, err := fileset.NewModuleRegistry(config.Modules, b.Info.Version, true)
 	if err != nil {
 		return nil, err

filebeat/config/config.go

@@ -21,16 +21,17 @@ const (
 )
 
 type Config struct {
-	Inputs          []*common.Config     `config:"inputs"`
-	Prospectors     []*common.Config     `config:"prospectors"`
-	RegistryFile    string               `config:"registry_file"`
-	RegistryFlush   time.Duration        `config:"registry_flush"`
-	ConfigDir       string               `config:"config_dir"`
-	ShutdownTimeout time.Duration        `config:"shutdown_timeout"`
-	Modules         []*common.Config     `config:"modules"`
-	ConfigInput     *common.Config       `config:"config.prospectors"`
-	ConfigModules   *common.Config       `config:"config.modules"`
-	Autodiscover    *autodiscover.Config `config:"autodiscover"`
+	Inputs           []*common.Config     `config:"inputs"`
+	Prospectors      []*common.Config     `config:"prospectors"`
+	RegistryFile     string               `config:"registry_file"`
+	RegistryFlush    time.Duration        `config:"registry_flush"`
+	ConfigDir        string               `config:"config_dir"`
+	ShutdownTimeout  time.Duration        `config:"shutdown_timeout"`
+	Modules          []*common.Config     `config:"modules"`
+	ConfigInput      *common.Config       `config:"config.inputs"`
+	ConfigProspector *common.Config       `config:"config.prospectors"`
+	ConfigModules    *common.Config       `config:"config.modules"`
+	Autodiscover     *autodiscover.Config `config:"autodiscover"`
 }
 
 var (

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

@@ -1,10 +1,10 @@
 ###################### Filebeat Config Template ###############################
 
-filebeat.{{input_config | default("prospectors")}}:
-{% if prospectors is not defined %}
-{% set prospectors = true %}
+filebeat.{{input_config | default("inputs")}}:
+{% if inputs is not defined %}
+{% set inputs = true %}
 {% endif %}
-{% if prospectors %}
+{% if inputs %}
 - type: {{type | default("log") }}
   input_type: {{input_type_deprecated }}
   # Paths that should be crawled and fetched
@@ -31,9 +31,9 @@ filebeat.{{input_config | default("prospectors")}}:
   harvester_limit: {{harvester_limit | default(0) }}
   symlinks: {{symlinks}}
   pipeline: {{pipeline}}
-  {%- if prospector_processors %}
+  {%- if input_processors %}
   processors:
-    {%- for processor in prospector_processors %}
+    {%- for processor in input_processors %}
     {%- for name, settings in processor.items() %}
     - {{name}}:
       {%- if settings %}
@@ -85,8 +85,8 @@ filebeat.{{input_config | default("prospectors")}}:
     max_lines: {{ max_lines|default(500) }}
   {% endif %}
 {% endif %}
-{% if prospector_raw %}
-{{prospector_raw}}
+{% if input_raw %}
+{{input_raw}}
 {% endif %}
 
 filebeat.shutdown_timeout: {{ shutdown_timeout|default(0) }}
@@ -95,7 +95,7 @@ filebeat.registry_file: {{ beat.working_dir + '/' }}{{ registryFile|default("reg
 {%endif%}
 
 {% if reload or reload_path -%}
-filebeat.config.{{ reload_type|default("prospectors") }}:
+filebeat.config.{{ reload_type|default("inputs") }}:
   path: {{ reload_path }}
   {% if reload  -%}
   reload.period: 1s

filebeat/tests/system/config/filebeat_prospectors.yml.j2 → filebeat/tests/system/config/filebeat_inputs.yml.j2

@@ -1,9 +1,9 @@
-filebeat.prospectors:
-{% for prospector in prospectors %}
+filebeat.inputs:
+{% for input in inputs %}
 - paths:
-    - {{prospector.path}}
+    - {{input.path}}
   scan_frequency: 0.5s
-  encoding: {{prospector.encoding | default("plain") }}
+  encoding: {{input.encoding | default("plain") }}
 {% endfor %}
 filebeat.registry_file: {{ beat.working_dir + '/' }}{{ registryFile|default("registry")}}
 

filebeat/tests/system/module/test/test/manifest.yml

@@ -6,4 +6,4 @@ var:
       - test.log
 
 ingest_pipeline: ingest/default.json
-prospector: config/test.yml
+input: config/test.yml

filebeat/tests/system/test_autodiscover.py

@@ -20,7 +20,7 @@ def test_docker(self):
         docker_client = docker.from_env()
 
         self.render_config_template(
-            prospectors=False,
+            inputs=False,
             autodiscover={
                 'docker': {
                     'templates': '''

filebeat/tests/system/test_base.py

@@ -25,6 +25,7 @@ def test_base(self):
         output = self.read_output()[0]
         assert "@timestamp" in output
         assert "prospector.type" in output
+        assert "input.type" in output
 
     def test_invalid_config_with_removed_settings(self):
         """

filebeat/tests/system/test_crawler.py

@@ -573,15 +573,15 @@ def test_encodings(self):
                 f.write(text + "\n")
 
         # create the config file
-        prospectors = []
+        inputs = []
         for enc_go, enc_py, _ in encodings:
-            prospectors.append({
+            inputs.append({
                 "path": self.working_dir + "/log/test-{}".format(enc_py),
                 "encoding": enc_go
             })
         self.render_config_template(
-            template_name="filebeat_prospectors",
-            prospectors=prospectors
+            template_name="filebeat_inputs",
+            inputs=inputs
         )
 
         # run filebeat

filebeat/tests/system/test_deprecated.py

@@ -63,3 +63,56 @@ def test_prospectors_deprecated(self):
         filebeat.check_kill_and_wait()
 
         assert self.log_contains("DEPRECATED: prospectors are deprecated, Use `inputs` instead.")
+
+    def test_reload_config_prospector_deprecated(self):
+        """
+        Checks that harvesting works with `config.prospectors`
+        """
+
+        inputConfigTemplate = """
+        - type: log
+          paths:
+            - {}
+          scan_frequency: 1s
+        """
+
+        self.render_config_template(
+            reload_type="prospectors",
+            reload=True,
+            reload_path=self.working_dir + "/configs/*.yml",
+            inputs=False,
+        )
+
+        os.mkdir(self.working_dir + "/logs/")
+        logfile1 = self.working_dir + "/logs/test1.log"
+        logfile2 = self.working_dir + "/logs/test2.log"
+        os.mkdir(self.working_dir + "/configs/")
+
+        with open(self.working_dir + "/configs/input.yml", 'w') as f:
+            f.write(inputConfigTemplate.format(self.working_dir + "/logs/test1.log"))
+
+        proc = self.start_beat()
+
+        with open(logfile1, 'w') as f:
+            f.write("Hello world1\n")
+
+        self.wait_until(lambda: self.output_lines() > 0)
+
+        with open(self.working_dir + "/configs/input2.yml", 'w') as f:
+            f.write(inputConfigTemplate.format(self.working_dir + "/logs/test2.log"))
+
+        self.wait_until(
+            lambda: self.log_contains_count("New runner started") == 2,
+            max_timeout=15)
+
+        # Add new log line and see if it is picked up = new input is running
+        with open(logfile1, 'a') as f:
+            f.write("Hello world2\n")
+
+        # Add new log line and see if it is picked up = new input is running
+        with open(logfile2, 'a') as f:
+            f.write("Hello world3\n")
+
+        self.wait_until(lambda: self.output_lines() == 3)
+
+        assert self.log_contains("DEPRECATED: config.prospectors are deprecated, Use `config.inputs` instead.")

filebeat/tests/system/test_prospector.py → filebeat/tests/system/test_input.py

@@ -3,12 +3,11 @@
 from filebeat import BaseTest
 import os
 import time
-import unittest
 
 from beat.beat import Proc
 
 """
-Tests for the prospector functionality.
+Tests for the input functionality.
 """
 
 
@@ -269,12 +268,12 @@ def test_rotating_close_inactive_low_write_rate(self):
 
         filebeat.check_kill_and_wait()
 
-    def test_shutdown_no_prospectors(self):
+    def test_shutdown_no_inputs(self):
         """
-        In case no prospectors are defined, filebeat must shut down and report an error
+        In case no inputs are defined, filebeat must shut down and report an error
         """
         self.render_config_template(
-            prospectors=False,
+            inputs=False,
         )
 
         filebeat = self.start_beat()
@@ -288,7 +287,7 @@ def test_shutdown_no_prospectors(self):
 
     def test_no_paths_defined(self):
         """
-        In case a prospector is defined but doesn't contain any paths, prospector must return error which
+        In case a input is defined but doesn't contain any paths, input must return error which
         leads to shutdown of filebeat because of configuration error
         """
         self.render_config_template(
@@ -299,7 +298,7 @@ def test_no_paths_defined(self):
         # wait for first  "Start next scan" log message
         self.wait_until(
             lambda: self.log_contains(
-                "No paths were defined for input"),
+                "No paths were defined for "),
             max_timeout=10)
 
         self.wait_until(
@@ -311,7 +310,7 @@ def test_no_paths_defined(self):
 
     def test_files_added_late(self):
         """
-        Tests that prospectors stay running even though no harvesters are started yet
+        Tests that inputs stay running even though no harvesters are started yet
         """
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/log/*",
@@ -627,13 +626,13 @@ def test_harvester_limit(self):
 
         filebeat.check_kill_and_wait()
 
-    def test_prospector_filter_dropfields(self):
+    def test_input_filter_dropfields(self):
         """
-        Check drop_fields filtering action at a prospector level
+        Check drop_fields filtering action at a input level
         """
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/test.log",
-            prospector_processors=[{
+            input_processors=[{
                 "drop_fields": {
                     "fields": ["offset"],
                 },
@@ -652,13 +651,13 @@ def test_prospector_filter_dropfields(self):
         assert "offset" not in output
         assert "message" in output
 
-    def test_prospector_filter_includefields(self):
+    def test_input_filter_includefields(self):
         """
-        Check include_fields filtering action at a prospector level
+        Check include_fields filtering action at a input level
         """
         self.render_config_template(
             path=os.path.abspath(self.working_dir) + "/test.log",
-            prospector_processors=[{
+            input_processors=[{
                 "include_fields": {
                     "fields": ["offset"],
                 },

filebeat/tests/system/test_modules.py

@@ -111,7 +111,7 @@ def run_on_file(self, module, fileset, test_file, cfgfile):
             "-M", "{module}.{fileset}.enabled=true".format(module=module, fileset=fileset),
             "-M", "{module}.{fileset}.var.paths=[{test_file}]".format(
                 module=module, fileset=fileset, test_file=test_file),
-            "-M", "*.*.prospector.close_eof=true",
+            "-M", "*.*.input.close_eof=true",
         ]
 
         output_path = os.path.join(self.working_dir, module, fileset, os.path.basename(test_file))
@@ -153,13 +153,13 @@ def run_on_file(self, module, fileset, test_file, cfgfile):
     @unittest.skipIf(not INTEGRATION_TESTS or
                      os.getenv("TESTING_ENVIRONMENT") == "2x",
                      "integration test not available on 2.x")
-    def test_prospector_pipeline_config(self):
+    def test_input_pipeline_config(self):
         """
-        Tests that the pipeline configured in the prospector overwrites
+        Tests that the pipeline configured in the input overwrites
         the one from the output.
         """
         self.init()
-        index_name = "filebeat-test-prospector"
+        index_name = "filebeat-test-input"
         try:
             self.es.indices.delete(index=index_name)
         except:

filebeat/tests/system/test_redis.py

@@ -19,21 +19,21 @@ def init(self):
         return r
 
     @unittest.skipUnless(INTEGRATION_TESTS, "integration test")
-    def test_prospector(self):
+    def test_input(self):
         r = self.init()
         r.set("hello", "world")
 
-        prospector_raw = """
+        input_raw = """
 - type: redis
   hosts: ["{}:{}"]
   enabled: true
   scan_frequency: 1s
 """
-        prospector_raw = prospector_raw.format(self.get_host(), self.get_port())
+        input_raw = input_raw.format(self.get_host(), self.get_port())
 
         self.render_config_template(
-            prospector_raw=prospector_raw,
-            prospectors=False,
+            input_raw=input_raw,
+            inputs=False,
         )
 
         filebeat = self.start_beat()
@@ -45,8 +45,8 @@ def test_prospector(self):
 
         output = self.read_output()[0]
 
-        print output
         assert output["prospector.type"] == "redis"
+        assert output["input.type"] == "redis"
         assert "redis.slowlog.cmd" in output
 
     def get_host(self):