Ensure metadata updates don't replace existing pod metrics (#7573)

* Ensure metadata updates don't replace existing pod metrics

Before things change, `add_kubernetes_metadata` would replace everything
under `kubernetes.pod` when enriching events.

This is not what you want, especially if you are using it in conjunction
with the Kubernetes module, where you could smash pod metrics when
enriching.

* Update CHANGELOG.asciidoc

CHANGELOG.asciidoc

@@ -127,6 +127,7 @@ https://github.com/elastic/beats/compare/v6.2.3...master[Check the HEAD diff]
 - Fix Windows service metricset when using a 32-bit binary on a 64-bit OS. {pull}7294[7294]
 - Fix Jolokia attribute mapping when using wildcards and MBean names with multiple properties. {pull}7321[7321]
 - Do not report Metricbeat container host as hostname in Kubernetes deployment. {issue}7199[7199]
+- Ensure metadata updates don't replace existing pod metrics. {pull}7573[7573]
 
 *Packetbeat*
 

libbeat/processors/add_kubernetes_metadata/kubernetes.go

@@ -154,16 +154,9 @@ func (k *kubernetesAnnotator) Run(event *beat.Event) (*beat.Event, error) {
 		return event, nil
 	}
 
-	meta := common.MapStr{}
-	metaIface, ok := event.Fields["kubernetes"]
-	if !ok {
-		event.Fields["kubernetes"] = common.MapStr{}
-	} else {
-		meta = metaIface.(common.MapStr)
-	}
-
-	meta.Update(metadata)
-	event.Fields["kubernetes"] = meta
+	event.Fields.DeepUpdate(common.MapStr{
+		"kubernetes": metadata,
+	})
 
 	return event, nil
 }

libbeat/processors/add_kubernetes_metadata/kubernetes_test.go

@@ -0,0 +1,88 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package add_kubernetes_metadata
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+)
+
+// Test metadata updates don't replace existing pod metrics
+func TestAnnotatorDeepUpdate(t *testing.T) {
+	cfg := common.MustNewConfigFrom(map[string]interface{}{
+		"lookup_fields": []string{"kubernetes.pod.name"},
+	})
+	matcher, err := NewFieldMatcher(*cfg)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	processor := kubernetesAnnotator{
+		cache: newCache(10 * time.Second),
+		matchers: &Matchers{
+			matchers: []Matcher{matcher},
+		},
+	}
+
+	processor.cache.set("foo", common.MapStr{
+		"pod": common.MapStr{
+			"labels": common.MapStr{
+				"dont":     "replace",
+				"original": "fields",
+			},
+		},
+	})
+
+	event, err := processor.Run(&beat.Event{
+		Fields: common.MapStr{
+			"kubernetes": common.MapStr{
+				"pod": common.MapStr{
+					"name": "foo",
+					"id":   "pod_id",
+					"metrics": common.MapStr{
+						"a": 1,
+						"b": 2,
+					},
+				},
+			},
+		},
+	})
+	assert.NoError(t, err)
+
+	assert.Equal(t, common.MapStr{
+		"kubernetes": common.MapStr{
+			"pod": common.MapStr{
+				"name": "foo",
+				"id":   "pod_id",
+				"metrics": common.MapStr{
+					"a": 1,
+					"b": 2,
+				},
+				"labels": common.MapStr{
+					"dont":     "replace",
+					"original": "fields",
+				},
+			},
+		},
+	}, event.Fields)
+}