Fix processor failure in Filebeat when using regex, contain, or equals

When using any of those conditions with the `message` field in Filebeat a warning would occur and no processor would be applied. The warning message was:

    WARN unexpected type *string in contains condition as it accepts only strings.

This occurred because Filebeat was passing the message field as a *string (string pointer). The processor code only expected to receive string values.

This PR contains three changes:

- Enhance the processor code to accept *string and string.
- Make filebeat pass the message field as a string rather than *string.
- Modify a test case to work against the message field rather than the source field.

Fixes elastic#2178

CHANGELOG.asciidoc

@@ -39,6 +39,7 @@ https://github.com/elastic/beats/compare/v5.0.0-alpha5...master[Check the HEAD d
 *Topbeat*
 
 *Filebeat*
+- Fix processor failure in Filebeat when using regex, contain, or equals with the message field. {issue}2178[2178]
 
 *Winlogbeat*
 

filebeat/input/event.go

@@ -41,8 +41,8 @@ func (f *Event) ToMapStr() common.MapStr {
 
 	if f.JSONConfig != nil && len(f.JSONFields) > 0 {
 		mergeJSONFields(f, event)
-	} else {
-		event["message"] = f.Text
+	} else if f.Text != nil {
+		event["message"] = *f.Text
 	}
 
 	return event

filebeat/tests/system/test_processors.py

@@ -94,7 +94,7 @@ def test_condition(self):
             path=os.path.abspath(self.working_dir) + "/test*.log",
             processors=[{
                 "drop_event": {
-                    "when": "not.contains.source: test",
+                    "when": "not.contains.message: test",
                 },
             }]
         )

libbeat/processors/config.go

@@ -125,8 +125,10 @@ func extractInt(unk interface{}) (uint64, error) {
 func extractString(unk interface{}) (string, error) {
 	switch s := unk.(type) {
 	case string:
-		return string(s), nil
+		return s, nil
+	case *string:
+		return *s, nil
 	default:
-		return "", fmt.Errorf("unkown type %T passed to extractString", unk)
+		return "", fmt.Errorf("unknown type %T passed to extractString", unk)
 	}
 }

libbeat/processors/config_test.go

@@ -0,0 +1,23 @@
+package processors
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestExtractString(t *testing.T) {
+	input := "test"
+
+	v, err := extractString(input)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, input, v)
+
+	v, err = extractString(&input)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, input, v)
+}