include_fields processor does not work with dotted keys

libbeat/common/mapstr.go

@@ -318,37 +318,48 @@ func tryToMapStr(v interface{}) (MapStr, bool) {
 // walkMap walks the data MapStr to arrive at the value specified by the key.
 // The key is expressed in dot-notation (eg. x.y.z). When the key is found then
 // the given mapStrOperation is invoked.
-func walkMap(key string, data MapStr, op mapStrOperation) (interface{}, error) {
-	var err error
-	keyParts := strings.Split(key, ".")
-
-	// Walk maps until reaching a leaf object.
-	m := data
-	for i, k := range keyParts[0 : len(keyParts)-1] {
-		v, exists := m[k]
-		if !exists {
-			if op.CreateMissingKeys {
-				newMap := MapStr{}
-				m[k] = newMap
-				m = newMap
-				continue
-			}
-			return nil, errors.Wrapf(ErrKeyNotFound, "key=%v", strings.Join(keyParts[0:i+1], "."))
-		}
+func walkMapRecursive(key string, data MapStr, op mapStrOperation) (interface{}, error) {
+
+	// Splits up the key in two parts: full key and first part before the dot
+	keyParts := strings.SplitN(key, ".", 2)
+	_, exists := data[key]
 
-		m, err = toMapStr(v)
+	// If leave node or key exists directly
+	if len(keyParts) == 1 || exists {
+		// Execute the mapStrOperator on the leaf object.
+		v, err := op.Do(key, data)
 		if err != nil {
-			return nil, errors.Wrapf(err, "key=%v", strings.Join(keyParts[0:i+1], "."))
+			return nil, err
 		}
+		return v, nil
 	}
 
-	// Execute the mapStrOperator on the leaf object.
-	v, err := op.Do(keyParts[len(keyParts)-1], m)
+	// Checks if first part of the key exists
+	k := keyParts[0]
+	_, keyExist := data[k]
+	if !keyExist {
+		if op.CreateMissingKeys {
+			data[k] = MapStr{}
+		} else {
+			return nil, ErrKeyNotFound
+		}
+	}
+
+	data, err := toMapStr(data[k])
 	if err != nil {
-		return nil, errors.Wrapf(err, "key=%v", key)
+		return nil, err
 	}
 
-	return v, nil
+	return walkMapRecursive(keyParts[1], data, op)
+}
+
+func walkMap(key string, data MapStr, op mapStrOperation) (interface{}, error) {
+	v, err := walkMapRecursive(key, data, op)
+	if err != nil {
+		// Add key to error
+		err = errors.Wrapf(err, "key=%v", key)
+	}
+	return v, err
 }
 
 // mapStrOperation types

libbeat/common/mapstr_test.go

@@ -60,6 +60,16 @@ func TestMapStrDeepUpdate(t *testing.T) {
 			MapStr{"a": 1},
 			MapStr{"a": 1},
 		},
+		{
+			MapStr{"a.b": 1},
+			MapStr{"a": 1},
+			MapStr{"a": 1, "a.b": 1},
+		},
+		{
+			MapStr{"a": 1},
+			MapStr{"a.b": 1},
+			MapStr{"a": 1, "a.b": 1},
+		},
 	}
 
 	for i, test := range tests {
@@ -173,7 +183,9 @@ func TestHasKey(t *testing.T) {
 				"c31": 1,
 				"c32": 2,
 			},
+			"c4.f": 19,
 		},
+		"d.f": 1,
 	}
 
 	hasKey, err := m.HasKey("c.c2")
@@ -191,6 +203,14 @@ func TestHasKey(t *testing.T) {
 	hasKey, err = m.HasKey("dd")
 	assert.Equal(nil, err)
 	assert.Equal(false, hasKey)
+
+	//hasKey, err = m.HasKey("d.f")
+	//assert.Equal(nil, err)
+	//assert.Equal(true, hasKey)
+
+	/*hasKey, err = m.HasKey("c.c4.f")
+	assert.Equal(nil, err)
+	assert.Equal(true, hasKey)*/
 }
 
 func TestMapStrPut(t *testing.T) {
@@ -569,3 +589,83 @@ func BenchmarkMapStrLogging(b *testing.B) {
 		logger.Infow("test", "mapstr", m)
 	}
 }
+
+func BenchmarkWalkMap(b *testing.B) {
+
+	b.Run("Get", func(b *testing.B) {
+		m := MapStr{
+			"test": 15,
+			"hello": MapStr{
+				"world": MapStr{
+					"ok": "test",
+				},
+			},
+			"elastic": MapStr{
+				"for": "search",
+			},
+		}
+		b.ResetTimer()
+
+		for i := 0; i < b.N; i++ {
+			m.GetValue("test.world.ok")
+		}
+	})
+
+	b.Run("Put", func(b *testing.B) {
+
+		for i := 0; i < b.N; i++ {
+			m := MapStr{
+				"test": 15,
+				"hello": MapStr{
+					"world": MapStr{
+						"ok": "test",
+					},
+				},
+				"elastic": MapStr{
+					"for": "search",
+				},
+			}
+
+			m.Put("test.world.tt", 17)
+		}
+	})
+
+	b.Run("HasKey", func(b *testing.B) {
+		b.ResetTimer()
+
+		for i := 0; i < b.N; i++ {
+			m := MapStr{
+				"test": 15,
+				"hello": MapStr{
+					"world": MapStr{
+						"ok": "test",
+					},
+				},
+				"elastic": MapStr{
+					"for": "search",
+				},
+			}
+			m.HasKey("test.world.tt")
+		}
+	})
+
+	b.Run("Delete", func(b *testing.B) {
+		b.ResetTimer()
+
+		for i := 0; i < b.N; i++ {
+			m := MapStr{
+				"test": 15,
+				"hello": MapStr{
+					"world": MapStr{
+						"ok": "test",
+					},
+				},
+				"elastic": MapStr{
+					"for": "search",
+				},
+			}
+			m.Put("test.world.tt", 17)
+		}
+	})
+
+}

libbeat/processors/actions/include_fields_test.go

@@ -0,0 +1,59 @@
+package actions
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+)
+
+func TestIncludeFields(t *testing.T) {
+
+	var tests = []struct {
+		Fields []string
+		Input  common.MapStr
+		Output common.MapStr
+	}{
+		{
+			Fields: []string{"test"},
+			Input: common.MapStr{
+				"hello": "world",
+				"test":  17,
+			},
+			Output: common.MapStr{
+				"test": 17,
+			},
+		},
+		{
+			Fields: []string{"test", "a.b"},
+			Input: common.MapStr{
+				"a.b":  "b",
+				"a.c":  "c",
+				"test": 17,
+			},
+			Output: common.MapStr{
+				"test": 17,
+				"a": common.MapStr{
+					"b": "b",
+				},
+			},
+		},
+	}
+
+	for _, test := range tests {
+		p := includeFields{
+			Fields: test.Fields,
+		}
+
+		event := &beat.Event{
+			Fields: test.Input,
+		}
+
+		newEvent, err := p.Run(event)
+		assert.NoError(t, err)
+
+		assert.Equal(t, test.Output, newEvent.Fields)
+	}
+}