Introducing preprocessors on the module input override

filebeat/docs/filebeat-modules-options.asciidoc

@@ -121,4 +121,39 @@ You can also enable `close_eof` for all inputs created by any of the modules:
 -M "*.*.input.close_eof=true"
 ----------------------------------------------------------------------
 
+==== Processors
+
+As described in <<where-valid,Where are processors valid?>> you can define `processors`
+under the input section of the module definition. These processors are run _after_ any processors defined in the
+module.
+
+[source,yaml]
+----------------------------------------------------------------------
+- module: nginx
+  access:
+    input:
+      processors:
+        - drop_event:
+            when:
+              contains:
+                source: "test"
+----------------------------------------------------------------------
+
+Alternatively you can define `preprocessors` which are run after the input but _before_ the processors defined
+in the module. Preprocessors can be useful if we don't retrieve the data directly from the log source and the
+intermediate channel introduces additional 'wrapping'.
+
+[source,yaml]
+----------------------------------------------------------------------
+- module: nginx
+  access:
+    input:
+      preprocessors:
+        - drop_event:
+            when:
+               contains:
+                 source: "test"
+----------------------------------------------------------------------
+
+
 :modulename!:

filebeat/fileset/fileset.go

@@ -373,7 +373,21 @@ func (fs *Fileset) getInputConfig() (*common.Config, error) {
 		if err != nil {
 			return nil, fmt.Errorf("Error creating config from input overrides: %v", err)
 		}
+
 		cfg, err = common.MergeConfigsWithOptions([]*common.Config{cfg, overrides}, ucfg.FieldReplaceValues("**.paths"), ucfg.FieldAppendValues("**.processors"))
+		if overrides.HasField("preprocessors") {
+			preprocessors, err := overrides.Child("preprocessors", -1)
+			if err != nil {
+				return nil, fmt.Errorf("Error creating config from input overrides: %v", err)
+			}
+			overrides, err = common.NewConfigFrom(map[string]interface{}{
+				"processors": preprocessors,
+			})
+			if err != nil {
+				return nil, fmt.Errorf("Error creating config from input overrides: %v", err)
+			}
+			cfg, err = common.MergeConfigsWithOptions([]*common.Config{cfg, overrides}, ucfg.FieldReplaceValues("**.paths"), ucfg.FieldPrependValues("**.processors"))
+		}
 		if err != nil {
 			return nil, fmt.Errorf("Error applying config overrides: %v", err)
 		}

filebeat/fileset/fileset_test.go

@@ -234,6 +234,100 @@ func TestGetInputConfigNginxOverrides(t *testing.T) {
 				require.Equal(t, "foobar", v)
 			},
 		},
+		"processors": {
+			map[string]interface{}{
+				"processors": []map[string]interface{}{
+					map[string]interface{}{
+						"drop_event": map[string]interface{}{
+							"when": map[string]interface{}{
+								"contains": map[string]interface{}{
+									"source": "test",
+								},
+							},
+						},
+					},
+				},
+			},
+			func(t require.TestingT, cfg interface{}, rest ...interface{}) {
+				c, ok := cfg.(*common.Config)
+				if !ok {
+					t.FailNow()
+				}
+
+				require.True(t, c.HasField("processors"))
+				count, _ := c.CountField("processors")
+				require.Equal(t, 3, count)
+				processors, err := c.Child("processors", -1)
+				require.NoError(t, err)
+				dropProcessor, err := processors.Child("", 2)
+				require.NoError(t, err)
+				require.True(t, dropProcessor.HasField("drop_event"), "should contain 'drop_event' last")
+			},
+		},
+		"preprocessors": {
+			map[string]interface{}{
+				"close_eof": true,
+				"preprocessors": []map[string]interface{}{
+					map[string]interface{}{
+						"drop_event": map[string]interface{}{
+							"when": map[string]interface{}{
+								"contains": map[string]interface{}{
+									"source": "test",
+								},
+							},
+						},
+					},
+				},
+			},
+			func(t require.TestingT, cfg interface{}, rest ...interface{}) {
+				c, ok := cfg.(*common.Config)
+				if !ok {
+					t.FailNow()
+				}
+
+				require.True(t, c.HasField("processors"))
+				count, _ := c.CountField("processors")
+				require.Equal(t, 3, count, "Expecting processors to be merged 2 from module 1 from preprocessor")
+				processors, err := c.Child("processors", -1)
+				require.NoError(t, err)
+				addFieldsProcessor, err := processors.Child("", 2)
+				require.NoError(t, err)
+				require.True(t, addFieldsProcessor.HasField("add_fields"), "should contain 'add_fields' last")
+			},
+		},
+		"combinedProcessors": {
+			map[string]interface{}{
+				"close_eof": true,
+				"processors": []map[string]interface{}{
+					map[string]interface{}{
+						"drop_event": map[string]interface{}{},
+					},
+				},
+				"preprocessors": []map[string]interface{}{
+					map[string]interface{}{
+						"drop_event": map[string]interface{}{},
+					},
+				},
+			},
+			func(t require.TestingT, cfg interface{}, rest ...interface{}) {
+				c, ok := cfg.(*common.Config)
+				if !ok {
+					t.FailNow()
+				}
+
+				require.True(t, c.HasField("processors"))
+				count, _ := c.CountField("processors")
+				require.Equal(t, 4, count, "Expecting processors to be merged: 2 from module, 1 from preprocessor, 1 from processor")
+				processors, err := c.Child("processors", -1)
+				require.NoError(t, err)
+				dropProcessor, err := processors.Child("", 0)
+				require.NoError(t, err)
+				require.True(t, dropProcessor.HasField("drop_event"), "should contain 'drop_event' first")
+				dropProcessor, err = processors.Child("", 3)
+				require.NoError(t, err)
+				require.True(t, dropProcessor.HasField("drop_event"), "should contain 'drop_event' last")
+			},
+		},
 	}
 
 	for name, test := range tests {