From 808bce47870eef6b940763a2f116e109fc5b5202 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Carlos=20P=C3=A9rez-Aradros=20Herce?= <carlos@elastic.co>
Date: Fri, 18 May 2018 10:47:24 +0200
Subject: [PATCH] Keep event when `add_docker_metadata` fails to extract
 container id

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.
---
 CHANGELOG.asciidoc                            |  1 +
 libbeat/processors/actions/extract_field.go   |  6 +--
 .../processors/actions/extract_field_test.go  | 37 +++++++++++++------
 3 files changed, 29 insertions(+), 15 deletions(-)

diff --git a/CHANGELOG.asciidoc b/CHANGELOG.asciidoc
index a20f5c778048..ea10da196677 100644
--- a/CHANGELOG.asciidoc
+++ b/CHANGELOG.asciidoc
@@ -61,6 +61,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Ensure that the dashboard zip files can't contain files outside of the kibana directory. {pull}6921[6921]
 - Fix map overwrite panics by cloning shared structs before doing the update. {pull}6947[6947]
 - Fix error if lz4 compression is used with the kafka output. {pull}7025[7025]
+- Preserve the event when source matching fails in `add_docker_metadata`. {pull}7133[7133]
 
 *Auditbeat*
 
diff --git a/libbeat/processors/actions/extract_field.go b/libbeat/processors/actions/extract_field.go
index 95af9ae49aed..68ee10f3b1aa 100644
--- a/libbeat/processors/actions/extract_field.go
+++ b/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])
diff --git a/libbeat/processors/actions/extract_field_test.go b/libbeat/processors/actions/extract_field_test.go
index f20e393a218e..e7fcf6795ecf 100644
--- a/libbeat/processors/actions/extract_field_test.go
+++ b/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,17 +73,25 @@ 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)
@@ -81,10 +99,5 @@ func runExtractField(t *testing.T, config *common.Config, input common.MapStr) c
 		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})
 }