New processor: truncate_fields

CHANGELOG.next.asciidoc

@@ -269,6 +269,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Gracefully shut down on SIGHUP {pull}10704[10704]
 - New processor: `copy_fields`. {pull}11303[11303]
 - Add `error.message` to events when `fail_on_error` is set in `rename` and `copy_fields` processors. {pull}11303[11303]
+- New processor: `truncate_fields`. {pull}11297[11297]
 
 *Auditbeat*
 

auditbeat/auditbeat.reference.yml

@@ -339,6 +339,32 @@ auditbeat.modules:
 #          to: message_copied
 #    fail_on_error: true
 #    ignore_missing: false
+#
+# The following example truncates the value of message to 1024 bytes
+#
+#processors:
+#- truncate_fields:
+#    fields:
+#      - message
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
+#
+# The following example preserves the raw message under event.original
+#
+#processors:
+#- copy_fields:
+#    fields:
+#        - from: message
+#          to: event.original
+#    fail_on_error: false
+#    ignore_missing: true
+#- truncate_fields:
+#    fields:
+#      - event.original
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
 
 #============================= Elastic Cloud ==================================
 

filebeat/filebeat.reference.yml

@@ -1058,6 +1058,32 @@ filebeat.inputs:
 #          to: message_copied
 #    fail_on_error: true
 #    ignore_missing: false
+#
+# The following example truncates the value of message to 1024 bytes
+#
+#processors:
+#- truncate_fields:
+#    fields:
+#      - message
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
+#
+# The following example preserves the raw message under event.original
+#
+#processors:
+#- copy_fields:
+#    fields:
+#        - from: message
+#          to: event.original
+#    fail_on_error: false
+#    ignore_missing: true
+#- truncate_fields:
+#    fields:
+#      - event.original
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
 
 #============================= Elastic Cloud ==================================
 

filebeat/tests/system/test_processors.py

@@ -1,4 +1,6 @@
+# -*- coding: utf-8 -*-
 from filebeat import BaseTest
+import io
 import os
 
 """
@@ -192,3 +194,74 @@ def test_dissect_bad_tokenizer(self):
         )[0]
         assert "extracted.key" not in output
         assert output["message"] == "Hello world"
+
+    def test_truncate_bytes(self):
+        """
+        Check if truncate_fields with max_bytes can truncate long lines and leave short lines as is
+        """
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/test.log",
+            processors=[{
+                "truncate_fields": {
+                    "max_bytes": 10,
+                    "fields": ["message"],
+                },
+            }]
+        )
+
+        self._init_and_read_test_input([
+            u"This is my super long line\n",
+            u"This is an even longer long line\n",
+            u"A végrehajtás során hiba történt\n",  # Error occured during execution (Hungarian)
+            u"This is OK\n",
+        ])
+
+        self._assert_expected_lines([
+            u"This is my",
+            u"This is an",
+            u"A végreha",
+            u"This is OK",
+        ])
+
+    def test_truncate_characters(self):
+        """
+        Check if truncate_fields with max_charaters can truncate long lines and leave short lines as is
+        """
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/test.log",
+            processors=[{
+                "truncate_fields": {
+                    "max_characters": 10,
+                    "fields": ["message"],
+                },
+            }]
+        )
+
+        self._init_and_read_test_input([
+            u"This is my super long line\n",
+            u"A végrehajtás során hiba történt\n",  # Error occured during execution (Hungarian)
+            u"This is OK\n",
+        ])
+
+        self._assert_expected_lines([
+            u"This is my",
+            u"A végrehaj",
+            u"This is OK",
+        ])
+
+    def _init_and_read_test_input(self, input_lines):
+        with io.open(self.working_dir + "/test.log", "w", encoding="utf-8") as f:
+            for line in input_lines:
+                f.write((line))
+
+        filebeat = self.start_beat()
+        self.wait_until(lambda: self.output_has(lines=len(input_lines)))
+        filebeat.check_kill_and_wait()
+
+    def _assert_expected_lines(self, expected_lines):
+        output = self.read_output()
+
+        assert len(output) == len(expected_lines)
+
+        for i in range(len(expected_lines)):
+            assert output[i]["message"] == expected_lines[i]

heartbeat/heartbeat.reference.yml

@@ -483,6 +483,32 @@ heartbeat.scheduler:
 #          to: message_copied
 #    fail_on_error: true
 #    ignore_missing: false
+#
+# The following example truncates the value of message to 1024 bytes
+#
+#processors:
+#- truncate_fields:
+#    fields:
+#      - message
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
+#
+# The following example preserves the raw message under event.original
+#
+#processors:
+#- copy_fields:
+#    fields:
+#        - from: message
+#          to: event.original
+#    fail_on_error: false
+#    ignore_missing: true
+#- truncate_fields:
+#    fields:
+#      - event.original
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
 
 #============================= Elastic Cloud ==================================
 

journalbeat/journalbeat.reference.yml

@@ -279,6 +279,32 @@ setup.template.settings:
 #          to: message_copied
 #    fail_on_error: true
 #    ignore_missing: false
+#
+# The following example truncates the value of message to 1024 bytes
+#
+#processors:
+#- truncate_fields:
+#    fields:
+#      - message
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
+#
+# The following example preserves the raw message under event.original
+#
+#processors:
+#- copy_fields:
+#    fields:
+#        - from: message
+#          to: event.original
+#    fail_on_error: false
+#    ignore_missing: true
+#- truncate_fields:
+#    fields:
+#      - event.original
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
 
 #============================= Elastic Cloud ==================================
 

libbeat/_meta/config.reference.yml

@@ -227,6 +227,32 @@
 #          to: message_copied
 #    fail_on_error: true
 #    ignore_missing: false
+#
+# The following example truncates the value of message to 1024 bytes
+#
+#processors:
+#- truncate_fields:
+#    fields:
+#      - message
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
+#
+# The following example preserves the raw message under event.original
+#
+#processors:
+#- copy_fields:
+#    fields:
+#        - from: message
+#          to: event.original
+#    fail_on_error: false
+#    ignore_missing: true
+#- truncate_fields:
+#    fields:
+#      - event.original
+#    max_bytes: 1024
+#    fail_on_error: false
+#    ignore_missing: true
 
 #============================= Elastic Cloud ==================================
 

libbeat/processors/actions/checks.go

@@ -79,3 +79,25 @@ func allowedFields(fields ...string) func(*common.Config) error {
 		return nil
 	}
 }
+
+func mutuallyExclusiveRequiredFields(fields ...string) func(*common.Config) error {
+	return func(cfg *common.Config) error {
+		var foundField string
+		for _, field := range cfg.GetFields() {
+			for _, f := range fields {
+				if field == f {
+					if len(foundField) == 0 {
+						foundField = field
+					} else {
+						return fmt.Errorf("field %s and %s are mutually exclusive", foundField, field)
+					}
+				}
+			}
+		}
+
+		if len(foundField) == 0 {
+			return fmt.Errorf("missing option, select one from %v", fields)
+		}
+		return nil
+	}
+}