Keep event when add_docker_metadata fails to extract container id (e…

…lastic#7133) (elastic#7156)

Before this change an error extracting container id from the source path
resulted on lossing the log line.

This change ensures we report the error but still send the log lines
through, without enriching.

(cherry picked from commit b1d8542)

CHANGELOG.asciidoc

@@ -29,6 +29,7 @@ https://github.com/elastic/beats/compare/v6.3.0...6.3[Check the HEAD diff]
 
 *Affecting all Beats*
 
+- Preserve the event when source matching fails in `add_docker_metadata`. {pull}7133[7133]
 - Negotiate Docker API version from our client instead of using a hardcoded one. {pull}7165[7165]
 
 *Auditbeat*

libbeat/processors/actions/extract_field.go

@@ -58,18 +58,18 @@ func NewExtractField(c *common.Config) (processors.Processor, error) {
 func (f *extract_field) Run(event *beat.Event) (*beat.Event, error) {
 	fieldValue, err := event.GetValue(f.Field)
 	if err != nil {
-		return nil, fmt.Errorf("error getting field '%s' from event", f.Field)
+		return event, fmt.Errorf("error getting field '%s' from event", f.Field)
 	}
 
 	value, ok := fieldValue.(string)
 	if !ok {
-		return nil, fmt.Errorf("could not get a string from field '%s'", f.Field)
+		return event, fmt.Errorf("could not get a string from field '%s'", f.Field)
 	}
 
 	parts := strings.Split(value, f.Separator)
 	parts = deleteEmpty(parts)
 	if len(parts) < f.Index+1 {
-		return nil, fmt.Errorf("index is out of range for field '%s'", f.Field)
+		return event, fmt.Errorf("index is out of range for field '%s'", f.Field)
 	}
 
 	event.PutValue(f.Target, parts[f.Index])

libbeat/processors/actions/extract_field_test.go

@@ -14,6 +14,7 @@ func TestCommonPaths(t *testing.T) {
 	var tests = []struct {
 		Value, Field, Separator, Target, Result string
 		Index                                   int
+		Error                                   bool
 	}{
 		// Common docker case
 		{
@@ -48,6 +49,15 @@ func TestCommonPaths(t *testing.T) {
 			Index:     0,
 			Result:    "var",
 		},
+		{
+			Value:     "/var/lib/foo/bar",
+			Field:     "source",
+			Separator: "*",
+			Target:    "destination",
+			Index:     10, // out of range
+			Result:    "var",
+			Error:     true,
+		},
 	}
 
 	for _, test := range tests {
@@ -63,28 +73,31 @@ func TestCommonPaths(t *testing.T) {
 			test.Field: test.Value,
 		}
 
-		actual := runExtractField(t, testConfig, input)
+		event, err := runExtractField(t, testConfig, input)
+		if test.Error {
+			assert.NotNil(t, err)
+		} else {
 
-		result, err := actual.GetValue(test.Target)
-		if err != nil {
-			t.Fatalf("could not get target field: %s", err)
+			assert.Nil(t, err)
+			result, err := event.Fields.GetValue(test.Target)
+			if err != nil {
+				t.Fatalf("could not get target field: %s", err)
+			}
+			assert.Equal(t, result.(string), test.Result)
 		}
-		assert.Equal(t, result.(string), test.Result)
+
+		// Event must be present, even on error
+		assert.NotNil(t, event)
 	}
 }
 
-func runExtractField(t *testing.T, config *common.Config, input common.MapStr) common.MapStr {
+func runExtractField(t *testing.T, config *common.Config, input common.MapStr) (*beat.Event, error) {
 	logp.TestingSetup()
 
 	p, err := NewExtractField(config)
 	if err != nil {
 		t.Fatalf("error initializing extract_field: %s", err)
 	}
 
-	actual, err := p.Run(&beat.Event{Fields: input})
-	if err != nil {
-		t.Fatalf("error running extract_field: %s", err)
-	}
-
-	return actual.Fields
+	return p.Run(&beat.Event{Fields: input})
 }