Add json.ignore_decoding_error to not log decoding errors

Closes #6045

CHANGELOG.asciidoc

@@ -209,6 +209,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Add Logstash module support for main log and the slow log, support the plain text or structured JSON format {pull}5481[5481]
 - Add stream filtering when using `docker` prospector. {pull}6057[6057]
 - Add support for CRI logs format. {issue}5630[5630]
+- Add json.ignore_decoding_error config to not log json decoding erors. {issue}6547[6547]
 
 *Heartbeat*
 

filebeat/docs/filebeat-options.asciidoc

@@ -422,6 +422,9 @@ key must be at the top level in the JSON object and the value associated with
 the key must be a string, otherwise no filtering or multiline aggregation will
 occur.
 
+*`ignore_decoding_error`*:: An optional configuration setting that specifies if JSON decoding errors should be
+logged or not. If set to true, errors will not be logged. The default is false.
+
 [float]
 ==== `multiline`
 

filebeat/harvester/reader/json.go

@@ -29,7 +29,9 @@ func (r *JSON) decodeJSON(text []byte) ([]byte, common.MapStr) {
 
 	err := unmarshal(text, &jsonFields)
 	if err != nil || jsonFields == nil {
-		logp.Err("Error decoding JSON: %v", err)
+		if !r.cfg.IgnoreDecodingError {
+			logp.Err("Error decoding JSON: %v", err)
+		}
 		if r.cfg.AddErrorKey {
 			jsonFields = common.MapStr{"error": createJSONError(fmt.Sprintf("Error decoding JSON: %v", err))}
 		}

filebeat/harvester/reader/json_config.go

@@ -1,10 +1,11 @@
 package reader
 
 type JSONConfig struct {
-	MessageKey    string `config:"message_key"`
-	KeysUnderRoot bool   `config:"keys_under_root"`
-	OverwriteKeys bool   `config:"overwrite_keys"`
-	AddErrorKey   bool   `config:"add_error_key"`
+	MessageKey          string `config:"message_key"`
+	KeysUnderRoot       bool   `config:"keys_under_root"`
+	OverwriteKeys       bool   `config:"overwrite_keys"`
+	AddErrorKey         bool   `config:"add_error_key"`
+	IgnoreDecodingError bool   `config:"ignore_decoding_error"`
 }
 
 func (c *JSONConfig) Validate() error {

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

@@ -74,6 +74,7 @@ filebeat.{{input_config | default("inputs")}}:
     {% if json.keys_under_root %}keys_under_root: true{% endif %}
     {% if json.overwrite_keys %}overwrite_keys: true{% endif %}
     {% if json.add_error_key %}add_error_key: true{% endif %}
+    {% if json.ignore_decoding_error %}ignore_decoding_error: true{% endif %}
   {% endif %}
 
   {% if multiline %}

filebeat/tests/system/test_json.py

@@ -316,6 +316,72 @@ def test_with_generic_filtering(self):
         # We drop null values during the generic event conversion.
         assert "res" not in o
 
+    def test_json_decoding_error_true(self):
+        """
+        Test if json_decoding_error is set to true, that no errors are logged.
+        """
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/log/*",
+            json=dict(
+                message_key="message",
+                ignore_decoding_error=True
+            ),
+        )
+
+        os.mkdir(self.working_dir + "/log/")
+
+        testfile1 = self.working_dir + "/log/test.log"
+
+        message = "invalidjson"
+        with open(testfile1, 'a') as f:
+            f.write(message + "\n")
+
+        proc = self.start_beat()
+        self.wait_until(
+            lambda: self.output_has(lines=1),
+            max_timeout=10)
+        proc.check_kill_and_wait()
+
+        output = self.read_output(
+            required_fields=["@timestamp"],
+        )
+        assert len(output) == 1
+        assert output[0]["message"] == message
+        assert False == self.log_contains_count("Error decoding JSON")
+
+    def test_json_decoding_error_false(self):
+        """
+        Test if json_decoding_error is set to false, that an errors is logged.
+        """
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/log/*",
+            json=dict(
+                message_key="message",
+                ignore_decoding_error=False
+            ),
+        )
+
+        os.mkdir(self.working_dir + "/log/")
+
+        testfile1 = self.working_dir + "/log/test.log"
+
+        message = "invalidjson"
+        with open(testfile1, 'a') as f:
+            f.write(message + "\n")
+
+        proc = self.start_beat()
+        self.wait_until(
+            lambda: self.output_has(lines=1),
+            max_timeout=10)
+        proc.check_kill_and_wait()
+
+        output = self.read_output(
+            required_fields=["@timestamp"],
+        )
+        assert len(output) == 1
+        assert output[0]["message"] == message
+        assert True == self.log_contains_count("Error decoding JSON")
+
     def test_with_generic_filtering_remove_headers(self):
         """
         It should work fine to combine JSON decoding with

libbeat/tests/system/beat/beat.py

@@ -210,7 +210,7 @@ def render_config_template(self, template_name=None,
 
         output_path = os.path.join(self.working_dir, output)
         with open(output_path, "wb") as f:
-            os.chmod(output_path, 0600)
+            os.chmod(output_path, 0o600)
             f.write(output_str.encode('utf8'))
 
     # Returns output as JSON object with flattened fields (. notation)