Fix k8s metadata issue (elastic#16834)

(cherry picked from commit 034ee55)

CHANGELOG.next.asciidoc

@@ -40,6 +40,27 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 
 *Affecting all Beats*
 
+- Fix `add_cloud_metadata` to better support modifying sub-fields with other processors. {pull}13808[13808]
+- TLS or Beats that accept connections over TLS and validate client certificates. {pull}14146[14146]
+- Fix panics that could result from invalid TLS certificates. This can affect Beats that connect over TLS, or Beats that accept connections over TLS and validate client certificates. {pull}14146[14146]
+- Fix panic in the Logstash output when trying to send events to closed connection. {pull}15568[15568]
+- Fix missing output in dockerlogbeat {pull}15719[15719]
+- Fix logging target settings being ignored when Beats are started via systemd or docker. {issue}12024[12024] {pull}15422[15442]
+- Do not load dashboards where not available. {pull}15802[15802]
+- Fix issue where TLS settings would be ignored when a forward proxy was in use. {pull}15516{15516}
+- Update replicaset group to apps/v1 {pull}15854[15802]
+- Fix issue where default go logger is not discarded when either * or stdout is selected. {issue}10251[10251] {pull}15708[15708]
+- Upgrade go-ucfg to latest v0.8.1. {pull}15937{15937}
+- Fix index names for indexing not always guaranteed to be lower case. {pull}16081[16081]
+- Add `ssl.ca_sha256` option to the supported TLS option, this allow to check that a specific certificate is used as part of the verified chain. {issue}15717[15717]
+- Fix loading processors from annotation hints. {pull}16348[16348]
+- Fix an issue that could cause redundant configuration reloads. {pull}16440[16440]
+- Fix k8s pods labels broken schema. {pull}16480[16480]
+- Fix k8s pods annotations broken schema. {pull}16554[16554]
+- Upgrade go-ucfg to latest v0.8.3. {pull}16450{16450}
+- Fix `NewContainerMetadataEnricher` to use default config for kubernetes module. {pull}16857[16857]
+- Improve some logging messages for add_kubernetes_metadata processor {pull}16866[16866]
+- Fix k8s metadata issue regarding node labels not shown up on root level of metadata. {pull}16834[16834]
 
 *Auditbeat*
 

libbeat/common/kubernetes/metadata/metadata.go

@@ -18,6 +18,8 @@
 package metadata
 
 import (
+	"strings"
+
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/common/kubernetes"
 	"github.com/elastic/beats/libbeat/common/safemapstr"
@@ -40,3 +42,11 @@ func WithFields(key string, value interface{}) FieldOptions {
 		safemapstr.Put(meta, key, value)
 	}
 }
+
+// WithLabels FieldOption allows adding labels under sub-resource(kind)
+// example if kind=namespace namespace.labels key will be added
+func WithLabels(kind string) FieldOptions {
+	return func(meta common.MapStr) {
+		safemapstr.Put(meta, strings.ToLower(kind)+".labels", meta["labels"])
+	}
+}

libbeat/common/kubernetes/metadata/namespace.go

@@ -67,9 +67,8 @@ func (n *namespace) GenerateFromName(name string, opts ...FieldOptions) common.M
 		}
 
 		return n.Generate(no, opts...)
-	} else {
-		return nil
 	}
+	return nil
 }
 
 func flattenMetadata(in common.MapStr) common.MapStr {
@@ -83,7 +82,6 @@ func flattenMetadata(in common.MapStr) common.MapStr {
 	if !ok {
 		return nil
 	}
-
 	for k, v := range fields {
 		if k == "name" {
 			out[resource] = v
@@ -92,5 +90,15 @@ func flattenMetadata(in common.MapStr) common.MapStr {
 		}
 	}
 
+	rawLabels, err := in.GetValue("labels")
+	if err != nil {
+		return out
+	}
+	labels, ok := rawLabels.(common.MapStr)
+	if !ok {
+		return out
+	}
+	out[resource+"_labels"] = labels
+
 	return out
 }

libbeat/common/kubernetes/metadata/namespace_test.go

@@ -58,17 +58,18 @@ func TestNamespace_Generate(t *testing.T) {
 				},
 			},
 			// Use this for 8.0
-			/*output: common.MapStr{
-				"namespace": common.MapStr{
-					"name": "obj",
-					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
+			/*
+				output: common.MapStr{
+					"namespace": common.MapStr{
+						"name": name,
+						"uid":  uid,
+						"labels": common.MapStr{
+							"foo": "bar",
+						},
 					},
-				},
-			},*/
+				},*/
 			output: common.MapStr{
-				"namespace":     "obj",
+				"namespace":     name,
 				"namespace_uid": uid,
 				"namespace_labels": common.MapStr{
 					"foo": "bar",
@@ -114,15 +115,15 @@ func TestNamespace_GenerateFromName(t *testing.T) {
 			/*
 				output: common.MapStr{
 					"namespace": common.MapStr{
-						"name": "obj",
+						"name": name,
 						"uid":  uid,
 						"labels": common.MapStr{
 							"foo": "bar",
 						},
 					},
 				},*/
 			output: common.MapStr{
-				"namespace":     "obj",
+				"namespace":     name,
 				"namespace_uid": uid,
 				"namespace_labels": common.MapStr{
 					"foo": "bar",

libbeat/common/kubernetes/metadata/node_test.go

@@ -61,9 +61,9 @@ func TestNode_Generate(t *testing.T) {
 				"node": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
+				},
+				"labels": common.MapStr{
+					"foo": "bar",
 				},
 			},
 		},
@@ -106,9 +106,9 @@ func TestNode_GenerateFromName(t *testing.T) {
 				"node": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
+				},
+				"labels": common.MapStr{
+					"foo": "bar",
 				},
 			},
 		},

libbeat/common/kubernetes/metadata/pod.go

@@ -49,11 +49,9 @@ func (p *pod) Generate(obj kubernetes.Resource, opts ...FieldOptions) common.Map
 	}
 
 	out := p.resource.Generate("pod", obj, opts...)
-	// TODO: remove this call when moving to 8.0
-	out = p.exportPodLabelsAndAnnotations(out)
 
 	if p.node != nil {
-		meta := p.node.GenerateFromName(po.Spec.NodeName)
+		meta := p.node.GenerateFromName(po.Spec.NodeName, WithLabels("node"))
 		if meta != nil {
 			out.Put("node", meta["node"])
 		} else {
@@ -91,21 +89,3 @@ func (p *pod) GenerateFromName(name string, opts ...FieldOptions) common.MapStr
 
 	return nil
 }
-
-func (p *pod) exportPodLabelsAndAnnotations(in common.MapStr) common.MapStr {
-	labels, err := in.GetValue("pod.labels")
-	if err != nil {
-		return in
-	}
-	in.Put("labels", labels)
-	in.Delete("pod.labels")
-
-	annotations, err := in.GetValue("pod.annotations")
-	if err != nil {
-		return in
-	}
-	in.Put("annotations", annotations)
-	in.Delete("pod.annotations")
-
-	return in
-}

libbeat/common/kubernetes/metadata/resource.go

@@ -98,11 +98,11 @@ func (r *Resource) Generate(kind string, obj kubernetes.Resource, options ...Fie
 	}
 
 	if len(labelMap) != 0 {
-		safemapstr.Put(meta, strings.ToLower(kind)+".labels", labelMap)
+		safemapstr.Put(meta, "labels", labelMap)
 	}
 
 	if len(annotationsMap) != 0 {
-		safemapstr.Put(meta, strings.ToLower(kind)+".annotations", annotationsMap)
+		safemapstr.Put(meta, "annotations", annotationsMap)
 	}
 
 	for _, option := range options {

libbeat/common/kubernetes/metadata/resource_test.go

@@ -60,9 +60,9 @@ func TestResource_Generate(t *testing.T) {
 				"pod": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
+				},
+				"labels": common.MapStr{
+					"foo": "bar",
 				},
 				"namespace": "default",
 			},
@@ -97,9 +97,9 @@ func TestResource_Generate(t *testing.T) {
 				"pod": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
+				},
+				"labels": common.MapStr{
+					"foo": "bar",
 				},
 				"namespace": "default",
 				"deployment": common.MapStr{

libbeat/common/kubernetes/metadata/service_test.go

@@ -64,9 +64,9 @@ func TestService_Generate(t *testing.T) {
 				"service": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
+				},
+				"labels": common.MapStr{
+					"foo": "bar",
 				},
 				"namespace": "default",
 			},
@@ -101,9 +101,9 @@ func TestService_Generate(t *testing.T) {
 				"service": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
+				},
+				"labels": common.MapStr{
+					"foo": "bar",
 				},
 				"namespace": "default",
 				"deployment": common.MapStr{
@@ -153,9 +153,9 @@ func TestService_GenerateFromName(t *testing.T) {
 				"service": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
+				},
+				"labels": common.MapStr{
+					"foo": "bar",
 				},
 				"namespace": "default",
 			},
@@ -190,9 +190,9 @@ func TestService_GenerateFromName(t *testing.T) {
 				"service": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
+				},
+				"labels": common.MapStr{
+					"foo": "bar",
 				},
 				"namespace": "default",
 				"deployment": common.MapStr{
@@ -262,10 +262,19 @@ func TestService_GenerateWithNamespace(t *testing.T) {
 				"service": common.MapStr{
 					"name": "obj",
 					"uid":  uid,
-					"labels": common.MapStr{
-						"foo": "bar",
-					},
 				},
+				"labels": common.MapStr{
+					"foo": "bar",
+				},
+				// Use this for 8.0
+				/*
+					"namespace": common.MapStr{
+						"name": "default",
+						"uid":  uid,
+						"labels": common.MapStr{
+							"nskey": "nsvalue",
+						},
+				},*/
 				"namespace":     "default",
 				"namespace_uid": uid,
 				"namespace_labels": common.MapStr{