Dedot kubernetes fields (#6203)

* libbeat: dedot kubernetes labels and annotations
* metricbeat: dedot kubernetes event meta labels and annotations
* add and refactor tests for DeDotJSON

CHANGELOG.asciidoc

@@ -21,6 +21,7 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Adding a local keystore to allow user to obfuscate password {pull}5687[5687]
 - Update go-ucfg library to support top level key reference and cyclic key reference for the
   keystore {pull}6098[6098]
+- De dot keys of labels and annotations in kubernetes meta processors to prevent collisions. {pull}6203[6203]
 
 *Auditbeat*
 
@@ -47,7 +48,8 @@ https://github.com/elastic/beats/compare/v6.0.0-beta2...master[Check the HEAD di
 - Rename `golang.heap.system.optained` field to `golang.heap.system.obtained`. {issue}5703[5703]
 - Support haproxy stats gathering using http (additionaly to tcp socket). {pull}5819[5819]
 - De dot keys in jolokia/jmx metricset to prevent collisions. {pull}5957[5957]
-- Support to optionally 'de dot' keys in http/json metricset to prevent collisions. {pull}5957[5957]
+- Support to optionally 'de dot' keys in http/json metricset to prevent collisions. {pull}5970[5970]
+- De dot keys in kubernetes/event metricset to prevent collisions. {pull}6203[6203]
 
 *Packetbeat*
 

libbeat/common/event.go

@@ -262,16 +262,22 @@ func DeDot(s string) string {
 // DeDotJSON replaces in keys all . with _
 // This helps when sending data to Elasticsearch to prevent object and key collisions.
 func DeDotJSON(json interface{}) interface{} {
-	switch json.(type) {
+	switch json := json.(type) {
 	case map[string]interface{}:
 		result := map[string]interface{}{}
-		for key, value := range json.(map[string]interface{}) {
+		for key, value := range json {
+			result[DeDot(key)] = DeDotJSON(value)
+		}
+		return result
+	case MapStr:
+		result := MapStr{}
+		for key, value := range json {
 			result[DeDot(key)] = DeDotJSON(value)
 		}
 		return result
 	case []interface{}:
-		result := make([]interface{}, len(json.([]interface{})))
-		for i, value := range json.([]interface{}) {
+		result := make([]interface{}, len(json))
+		for i, value := range json {
 			result[i] = DeDotJSON(value)
 		}
 		return result

libbeat/common/event_test.go

@@ -7,9 +7,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"io/ioutil"
-	"path/filepath"
-
 	"github.com/elastic/beats/libbeat/logp"
 )
 
@@ -399,49 +396,60 @@ func BenchmarkConvertToGenericEventStringPointer(b *testing.B) {
 		ConvertToGenericEvent(MapStr{"key": &val})
 	}
 }
-
-func TestDeDotJsonMap(t *testing.T) {
-	var actualJSONBody map[string]interface{}
-	var expectedJSONBody map[string]interface{}
-
-	absPath, err := filepath.Abs("./testdata")
-	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
-
-	actualJSONResponse, err := ioutil.ReadFile(absPath + "/json_map_with_dots.json")
-	assert.Nil(t, err)
-	err = json.Unmarshal(actualJSONResponse, &actualJSONBody)
-	assert.Nil(t, err)
-
-	dedottedJSONResponse, err := ioutil.ReadFile(absPath + "/json_map_dedot.json")
-	assert.Nil(t, err)
-	err = json.Unmarshal(dedottedJSONResponse, &expectedJSONBody)
-	assert.Nil(t, err)
-
-	actualJSONBody = DeDotJSON(actualJSONBody).(map[string]interface{})
-
-	assert.Equal(t, expectedJSONBody, actualJSONBody)
-}
-
-func TestDeDotJsonArray(t *testing.T) {
-	var actualJSONBody []interface{}
-	var expectedJSONBody []interface{}
-
-	absPath, err := filepath.Abs("./testdata")
-	assert.NotNil(t, absPath)
-	assert.Nil(t, err)
-
-	actualJSONResponse, err := ioutil.ReadFile(absPath + "/json_array_with_dots.json")
-	assert.Nil(t, err)
-	err = json.Unmarshal(actualJSONResponse, &actualJSONBody)
-	assert.Nil(t, err)
-
-	dedottedJSONResponse, err := ioutil.ReadFile(absPath + "/json_array_dedot.json")
-	assert.Nil(t, err)
-	err = json.Unmarshal(dedottedJSONResponse, &expectedJSONBody)
-	assert.Nil(t, err)
-
-	actualJSONBody = DeDotJSON(actualJSONBody).([]interface{})
-
-	assert.Equal(t, expectedJSONBody, actualJSONBody)
+func TestDeDotJSON(t *testing.T) {
+	var tests = []struct {
+		input  []byte
+		output []byte
+		valuer func() interface{}
+	}{
+		{
+			input: []byte(`[
+				{"key_with_dot.1":"value1_1"},
+				{"key_without_dot_2":"value1_2"},
+				{"key_with_multiple.dots.3": {"key_with_dot.2":"value2_1"}}
+			]
+			`),
+			output: []byte(`[
+				{"key_with_dot_1":"value1_1"},
+				{"key_without_dot_2":"value1_2"},
+				{"key_with_multiple_dots_3": {"key_with_dot_2":"value2_1"}}
+			]
+			`),
+			valuer: func() interface{} { return []interface{}{} },
+		},
+		{
+			input: []byte(`{
+				"key_without_dot_l1": {
+					"key_with_dot.l2": 1,
+					"key.with.multiple.dots_l2": 2,
+					"key_without_dot_l2": {
+						"key_with_dot.l3": 3,
+						"key.with.multiple.dots_l3": 4
+					}
+				}
+			}
+			`),
+			output: []byte(`{
+				"key_without_dot_l1": {
+					"key_with_dot_l2": 1,
+					"key_with_multiple_dots_l2": 2,
+					"key_without_dot_l2": {
+						"key_with_dot_l3": 3,
+						"key_with_multiple_dots_l3": 4
+					}
+				}
+			}
+			`),
+			valuer: func() interface{} { return map[string]interface{}{} },
+		},
+	}
+	for _, test := range tests {
+		input, output := test.valuer(), test.valuer()
+		assert.Nil(t, json.Unmarshal(test.input, &input))
+		assert.Nil(t, json.Unmarshal(test.output, &output))
+		assert.Equal(t, output, DeDotJSON(input))
+		if _, ok := test.valuer().(map[string]interface{}); ok {
+			assert.Equal(t, MapStr(output.(map[string]interface{})), DeDotJSON(MapStr(input.(map[string]interface{}))))
+		}
+	}
 }

libbeat/common/kubernetes/metadata.go

@@ -54,11 +54,11 @@ func (g *metaGenerator) PodMetadata(pod *Pod) common.MapStr {
 	}
 
 	if len(labelMap) != 0 {
-		meta["labels"] = labelMap
+		meta["labels"] = common.DeDotJSON(labelMap)
 	}
 
 	if len(annotationsMap) != 0 {
-		meta["annotations"] = annotationsMap
+		meta["annotations"] = common.DeDotJSON(annotationsMap)
 	}
 
 	return meta

libbeat/common/kubernetes/metadata_test.go

@@ -0,0 +1,29 @@
+package kubernetes
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/libbeat/common"
+)
+
+func TestPodMetadataDeDot(t *testing.T) {
+	tests := []struct {
+		pod  *Pod
+		meta common.MapStr
+	}{
+		{
+			pod: &Pod{
+				Metadata: ObjectMeta{
+					Labels: map[string]string{"a.key": "a.value"},
+				},
+			},
+			meta: common.MapStr{"labels": common.MapStr{"a_key": "a.value"}},
+		},
+	}
+
+	for _, test := range tests {
+		assert.Equal(t, NewMetaGenerator(nil, nil, nil).PodMetadata(test.pod)["labels"], test.meta["labels"])
+	}
+}

metricbeat/module/kubernetes/event/event.go

@@ -105,7 +105,7 @@ func generateMapStrFromEvent(eve *kubernetes.Event) common.MapStr {
 	if len(eve.Metadata.Labels) != 0 {
 		labels := make(common.MapStr, len(eve.Metadata.Labels))
 		for k, v := range eve.Metadata.Labels {
-			labels[k] = v
+			labels[common.DeDot(k)] = v
 		}
 
 		eventMeta["labels"] = labels
@@ -114,7 +114,7 @@ func generateMapStrFromEvent(eve *kubernetes.Event) common.MapStr {
 	if len(eve.Metadata.Annotations) != 0 {
 		annotations := make(common.MapStr, len(eve.Metadata.Annotations))
 		for k, v := range eve.Metadata.Annotations {
-			annotations[k] = v
+			annotations[common.DeDot(k)] = v
 		}
 
 		eventMeta["annotations"] = annotations