Keep original messages in case of Filebeat modules

CHANGELOG.asciidoc

@@ -121,6 +121,7 @@ https://github.com/elastic/beats/compare/v6.4.0...master[Check the HEAD diff]
 - Add tag "multiline" to "log.flags" if event consists of multiple lines. {pull}7997[7997]
 - Add haproxy module. {pull}8014[8014]
 - Release `docker` input as GA. {pull}8328[8328]
+- Keep original messages in case of Filebeat modules. {pull}8448[8448]
 
 *Heartbeat*
 

filebeat/_meta/fields.common.yml

@@ -112,6 +112,12 @@
       description: >
         This field contains the flags of the event.
 
+    - name: log.original
+      type: keyword
+      description: >
+        The unprocessed original log message. This can be used for reprocessing logs.
+      index: false
+
     - name: event.created
       type: date
       description: >

filebeat/channel/factory.go

@@ -43,6 +43,9 @@ type clientEventer struct {
 // inputOutletConfig defines common input settings
 // for the publisher pipeline.
 type inputOutletConfig struct {
+	// KeepOriginalMsg determines if the original message needs to be kept for a module.
+	KeepOriginalMsg bool `config:"keep_original_message"`
+
 	// event processing
 	common.EventMetadata `config:",inline"`      // Fields and tags to add to events.
 	Processors           processors.PluginConfig `config:"processors"`
@@ -59,6 +62,10 @@ type inputOutletConfig struct {
 
 }
 
+var defaultConfig = inputOutletConfig{
+	KeepOriginalMsg: true,
+}
+
 // NewOutletFactory creates a new outlet factory for
 // connecting an input to the publisher pipeline.
 func NewOutletFactory(
@@ -82,7 +89,7 @@ func NewOutletFactory(
 // This guarantees ordering between events as required by the registrar for
 // file.State updates
 func (f *OutletFactory) Create(p beat.Pipeline, cfg *common.Config, dynFields *common.MapStrPointer) (Outleter, error) {
-	config := inputOutletConfig{}
+	config := defaultConfig
 	if err := cfg.Unpack(&config); err != nil {
 		return nil, err
 	}
@@ -101,13 +108,16 @@ func (f *OutletFactory) Create(p beat.Pipeline, cfg *common.Config, dynFields *c
 	meta := common.MapStr{}
 	setMeta(meta, "pipeline", config.Pipeline)
 
+	keepOriginal := false
 	fields := common.MapStr{}
 	setMeta(fields, "module", config.Module)
 	setMeta(fields, "name", config.Fileset)
 	if len(fields) > 0 {
 		fields = common.MapStr{
 			"fileset": fields,
 		}
+		keepOriginal = config.KeepOriginalMsg
+
 	}
 	if config.Type != "" {
 		fields["prospector"] = common.MapStr{
@@ -119,13 +129,14 @@ func (f *OutletFactory) Create(p beat.Pipeline, cfg *common.Config, dynFields *c
 	}
 
 	client, err := p.ConnectWith(beat.ClientConfig{
-		PublishMode:   beat.GuaranteedSend,
-		EventMetadata: config.EventMetadata,
-		DynamicFields: dynFields,
-		Meta:          meta,
-		Fields:        fields,
-		Processor:     processors,
-		Events:        f.eventer,
+		PublishMode:     beat.GuaranteedSend,
+		EventMetadata:   config.EventMetadata,
+		DynamicFields:   dynFields,
+		Meta:            meta,
+		Fields:          fields,
+		KeepOriginalMsg: keepOriginal,
+		Processor:       processors,
+		Events:          f.eventer,
 	})
 	if err != nil {
 		return nil, err

filebeat/docs/fields.asciidoc

@@ -3042,6 +3042,18 @@ Logging level.
 This field contains the flags of the event.
 
 
+--
+
+*`log.original`*::
++
+--
+type: keyword
+
+The unprocessed original log message. This can be used for reprocessing logs.
+
+
+Field is not indexed.
+
 --
 
 *`event.created`*::

filebeat/filebeat.reference.yml

@@ -27,6 +27,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
   # Authorization logs
   #auth:
@@ -42,6 +44,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #------------------------------- Apache2 Module ------------------------------
 #- module: apache2
@@ -56,6 +60,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
   # Error logs
   #error:
@@ -68,6 +74,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #------------------------------- Auditd Module -------------------------------
 #- module: auditd
@@ -81,6 +89,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #---------------------------- elasticsearch Module ---------------------------
 - module: elasticsearch
@@ -142,6 +152,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
   # Debug logs
   #debug:
@@ -154,6 +166,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
   # Startup logs
   #startup:
@@ -166,6 +180,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #--------------------------------- IIS Module --------------------------------
 #- module: iis
@@ -180,6 +196,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
   # Error logs
   #error:
@@ -192,6 +210,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #-------------------------------- Kafka Module -------------------------------
 - module: kafka
@@ -250,6 +270,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #-------------------------------- MySQL Module -------------------------------
 #- module: mysql
@@ -264,6 +286,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
   # Slow logs
   #slowlog:
@@ -276,6 +300,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #-------------------------------- Nginx Module -------------------------------
 #- module: nginx
@@ -290,6 +316,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
   # Error logs
   #error:
@@ -302,6 +330,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #------------------------------- Osquery Module ------------------------------
 - module: osquery
@@ -330,6 +360,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 #-------------------------------- Redis Module -------------------------------
 #- module: redis
@@ -364,6 +396,8 @@ filebeat.modules:
     # Input configuration (advanced). Any input configuration option
     # can be added under this section.
     #input:
+      #Keeps the original message, so the data can be processed again on Ingest Node.
+      #keep_original_message: true
 
 
 #=========================== Filebeat inputs =============================