Align k8s metadata configurations in Kubernetes module: add addResour…

…ceMetadata config (#29133)

* use addResourceMetadata config instead of hardcoded bool

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>

* revert config for local dev

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>

* update doc; use common function to create all watchers; add nodeWatcher and nsWatcher to enricher

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>

* revert ek_stack

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>

* add stop watchers; adjust test for dedoting labels; fix overriding config that leads to 'failed to parse field' error

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>

* adjust log messages; check that watcher is not nil

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>

* fix error message format; rename getPodMetadataWatchers -> getResourceMetadataWatchers; return resource watcher if nodeWatcher/nsWatcher failed to be created

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>

* add changelog

Signed-off-by: Tetiana Kravchenko <tetiana.kravchenko@elastic.co>

CHANGELOG.next.asciidoc

@@ -371,6 +371,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Register additional name for `storage` metricset in the azure module. {pull}28447[28447]
 - Update reference to gosigar pacakge for filesystem windows fix. {pull}28909[28909]
 - Override `Host()` on statsd MetricSet {pull}29103[29103]
+- Add `add_resource_metadata` configuration to Kubernetes module. {pull}29133[29133]
 
 *Packetbeat*
 

libbeat/autodiscover/providers/kubernetes/pod.go

@@ -100,9 +100,6 @@ func NewPodEventer(uuid uuid.UUID, cfg *common.Config, client k8s.Interface, pub
 		options.Namespace = config.Namespace
 	}
 	metaConf := config.AddResourceMetadata
-	if metaConf == nil {
-		metaConf = metadata.GetDefaultResourceMetadataConfig()
-	}
 	nodeWatcher, err := kubernetes.NewNamedWatcher("node", client, &kubernetes.Node{}, options, nil)
 	if err != nil {
 		logger.Errorf("couldn't create watcher for %T due to error %+v", &kubernetes.Node{}, err)

libbeat/autodiscover/providers/kubernetes/pod_test.go

@@ -1907,7 +1907,7 @@ func TestPod_EmitEvent(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			metaGen := metadata.NewPodMetadataGenerator(common.NewConfig(), nil, client, nil, nil, true)
+			metaGen := metadata.NewPodMetadataGenerator(common.NewConfig(), nil, client, nil, nil, nil)
 			p := &Provider{
 				config:    defaultConfig(),
 				bus:       bus.New(logp.NewLogger("bus"), "test"),

libbeat/common/kubernetes/metadata/metadata.go

@@ -96,8 +96,7 @@ func GetPodMetaGen(
 	if namespaceWatcher != nil && metaConf.Namespace.Enabled() {
 		namespaceMetaGen = NewNamespaceMetadataGenerator(metaConf.Namespace, namespaceWatcher.Store(), namespaceWatcher.Client())
 	}
-	metaGen := NewPodMetadataGenerator(cfg, podWatcher.Store(), podWatcher.Client(), nodeMetaGen, namespaceMetaGen, metaConf.Deployment)
-
+	metaGen := NewPodMetadataGenerator(cfg, podWatcher.Store(), podWatcher.Client(), nodeMetaGen, namespaceMetaGen, metaConf)
 	return metaGen
 }
 

libbeat/common/kubernetes/metadata/namespace_test.go

@@ -51,9 +51,11 @@ func TestNamespace_Generate(t *testing.T) {
 					UID:  types.UID(uid),
 					Labels: map[string]string{
 						"foo": "bar",
+						"key": "value",
 					},
 					Annotations: map[string]string{
 						"spam": "baz",
+						"key":  "value",
 					},
 				},
 				TypeMeta: metav1.TypeMeta{
@@ -75,6 +77,7 @@ func TestNamespace_Generate(t *testing.T) {
 	}
 
 	cfg, err := common.NewConfigFrom(Config{
+		IncludeLabels:      []string{"foo"},
 		IncludeAnnotations: []string{"spam"},
 	})
 	if err != nil {

libbeat/common/kubernetes/metadata/pod.go

@@ -29,12 +29,12 @@ import (
 )
 
 type pod struct {
-	store         cache.Store
-	client        k8s.Interface
-	node          MetaGen
-	namespace     MetaGen
-	resource      *Resource
-	addDeployment bool
+	store               cache.Store
+	client              k8s.Interface
+	node                MetaGen
+	namespace           MetaGen
+	resource            *Resource
+	addResourceMetadata *AddResourceMetadataConfig
 }
 
 // NewPodMetadataGenerator creates a metagen for pod resources
@@ -44,14 +44,19 @@ func NewPodMetadataGenerator(
 	client k8s.Interface,
 	node MetaGen,
 	namespace MetaGen,
-	addDeploymentMeta bool) MetaGen {
+	addResourceMetadata *AddResourceMetadataConfig) MetaGen {
+
+	if addResourceMetadata == nil {
+		addResourceMetadata = GetDefaultResourceMetadataConfig()
+	}
+
 	return &pod{
-		resource:      NewResourceMetadataGenerator(cfg, client),
-		store:         pods,
-		node:          node,
-		namespace:     namespace,
-		client:        client,
-		addDeployment: addDeploymentMeta,
+		resource:            NewResourceMetadataGenerator(cfg, client),
+		store:               pods,
+		node:                node,
+		namespace:           namespace,
+		client:              client,
+		addResourceMetadata: addResourceMetadata,
 	}
 }
 
@@ -87,7 +92,7 @@ func (p *pod) GenerateK8s(obj kubernetes.Resource, opts ...FieldOptions) common.
 	out := p.resource.GenerateK8s("pod", obj, opts...)
 
 	// check if Pod is handled by a ReplicaSet which is controlled by a Deployment
-	if p.addDeployment {
+	if p.addResourceMetadata.Deployment {
 		rsName, _ := out.GetValue("replicaset.name")
 		if rsName, ok := rsName.(string); ok {
 			dep := p.getRSDeployment(rsName, po.GetNamespace())

libbeat/common/kubernetes/metadata/pod_test.go

@@ -374,7 +374,7 @@ func TestPod_Generate(t *testing.T) {
 	})
 	assert.NoError(t, err)
 
-	metagen := NewPodMetadataGenerator(config, nil, client, nil, nil, true)
+	metagen := NewPodMetadataGenerator(config, nil, client, nil, nil, nil)
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			assert.Equal(t, test.output, metagen.Generate(test.input))
@@ -496,7 +496,7 @@ func TestPod_GenerateFromName(t *testing.T) {
 		assert.NoError(t, err)
 		pods := cache.NewStore(cache.MetaNamespaceKeyFunc)
 		pods.Add(test.input)
-		metagen := NewPodMetadataGenerator(config, pods, client, nil, nil, true)
+		metagen := NewPodMetadataGenerator(config, pods, client, nil, nil, nil)
 
 		accessor, err := meta.Accessor(test.input)
 		require.NoError(t, err)
@@ -618,7 +618,156 @@ func TestPod_GenerateWithNodeNamespace(t *testing.T) {
 		namespaces.Add(test.namespace)
 		nsMeta := NewNamespaceMetadataGenerator(config, namespaces, client)
 
-		metagen := NewPodMetadataGenerator(config, pods, client, nodeMeta, nsMeta, true)
+		metagen := NewPodMetadataGenerator(config, pods, client, nodeMeta, nsMeta, nil)
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.output, metagen.Generate(test.input))
+		})
+	}
+}
+
+func TestPod_GenerateWithNodeNamespaceWithAddResourceConfig(t *testing.T) {
+	client := k8sfake.NewSimpleClientset()
+	uid := "005f3b90-4b9d-12f8-acf0-31020a840133"
+	namespace := "default"
+	name := "obj"
+	boolean := true
+
+	tests := []struct {
+		input     kubernetes.Resource
+		node      kubernetes.Resource
+		namespace kubernetes.Resource
+		output    common.MapStr
+		name      string
+	}{
+		{
+			name: "test simple object",
+			input: &v1.Pod{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      name,
+					UID:       types.UID(uid),
+					Namespace: namespace,
+					Labels: map[string]string{
+						"app.kubernetes.io/component": "exporter",
+					},
+					Annotations: map[string]string{
+						"app": "production",
+					},
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							APIVersion: "apps",
+							Kind:       "ReplicaSet",
+							Name:       "nginx-rs",
+							UID:        "005f3b90-4b9d-12f8-acf0-31020a8409087",
+							Controller: &boolean,
+						},
+					},
+				},
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Pod",
+					APIVersion: "v1",
+				},
+
+				Spec: v1.PodSpec{
+					NodeName: "testnode",
+				},
+				Status: v1.PodStatus{PodIP: "127.0.0.5"},
+			},
+			node: &v1.Node{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "testnode",
+					UID:  types.UID(uid),
+					Labels: map[string]string{
+						"nodekey":  "nodevalue",
+						"nodekey2": "nodevalue2",
+					},
+					Annotations: map[string]string{},
+				},
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Node",
+					APIVersion: "v1",
+				},
+				Status: v1.NodeStatus{
+					Addresses: []v1.NodeAddress{{Type: v1.NodeHostName, Address: "node1"}},
+				},
+			},
+			namespace: &v1.Namespace{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: namespace,
+					UID:  types.UID(uid),
+					Labels: map[string]string{
+						"app.kubernetes.io/name": "kube-state-metrics",
+						"nskey2":                 "nsvalue2",
+					},
+					Annotations: map[string]string{},
+				},
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Namespace",
+					APIVersion: "v1",
+				},
+			},
+			output: common.MapStr{"kubernetes": common.MapStr{
+				"pod": common.MapStr{
+					"name": "obj",
+					"uid":  uid,
+					"ip":   "127.0.0.5",
+				},
+				"namespace":     "default",
+				"namespace_uid": uid,
+				"namespace_labels": common.MapStr{
+					"app_kubernetes_io/name": "kube-state-metrics",
+				},
+				"node": common.MapStr{
+					"name": "testnode",
+					"uid":  uid,
+					"labels": common.MapStr{
+						"nodekey2": "nodevalue2",
+					},
+					"hostname": "node1",
+				},
+				"labels": common.MapStr{
+					"app_kubernetes_io/component": "exporter",
+				},
+				"annotations": common.MapStr{
+					"app": "production",
+				},
+				"replicaset": common.MapStr{
+					"name": "nginx-rs",
+				},
+			}},
+		},
+	}
+
+	for _, test := range tests {
+		config, err := common.NewConfigFrom(map[string]interface{}{
+			"include_annotations": []string{"app"},
+		})
+
+		assert.NoError(t, err)
+
+		namespaceConfig, _ := common.NewConfigFrom(map[string]interface{}{
+			"include_labels": []string{"app.kubernetes.io/name"},
+		})
+		nodeConfig, _ := common.NewConfigFrom(map[string]interface{}{
+			"include_labels": []string{"nodekey2"},
+		})
+		metaConfig := AddResourceMetadataConfig{
+			Namespace:  namespaceConfig,
+			Node:       nodeConfig,
+			Deployment: false,
+		}
+
+		pods := cache.NewStore(cache.MetaNamespaceKeyFunc)
+		pods.Add(test.input)
+
+		nodes := cache.NewStore(cache.MetaNamespaceKeyFunc)
+		nodes.Add(test.node)
+		nodeMeta := NewNodeMetadataGenerator(nodeConfig, nodes, client)
+
+		namespaces := cache.NewStore(cache.MetaNamespaceKeyFunc)
+		namespaces.Add(test.namespace)
+		nsMeta := NewNamespaceMetadataGenerator(namespaceConfig, namespaces, client)
+
+		metagen := NewPodMetadataGenerator(config, pods, client, nodeMeta, nsMeta, &metaConfig)
 		t.Run(test.name, func(t *testing.T) {
 			assert.Equal(t, test.output, metagen.Generate(test.input))
 		})

libbeat/common/kubernetes/metadata/resource.go

@@ -85,7 +85,7 @@ func (r *Resource) GenerateK8s(kind string, obj kubernetes.Resource, options ...
 		return nil
 	}
 
-	labelMap := common.MapStr{}
+	var labelMap common.MapStr
 	if len(r.config.IncludeLabels) == 0 {
 		labelMap = GenerateMap(accessor.GetLabels(), r.config.LabelsDedot)
 	} else {

libbeat/processors/add_kubernetes_metadata/indexers_test.go

@@ -32,7 +32,7 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common/kubernetes"
 )
 
-var metagen = metadata.NewPodMetadataGenerator(common.NewConfig(), nil, nil, nil, nil, true)
+var metagen = metadata.NewPodMetadataGenerator(common.NewConfig(), nil, nil, nil, nil, nil)
 
 func TestPodIndexer(t *testing.T) {
 	var testConfig = common.NewConfig()
@@ -90,7 +90,7 @@ func TestPodIndexer(t *testing.T) {
 func TestPodUIDIndexer(t *testing.T) {
 	var testConfig = common.NewConfig()
 
-	metaGenWithPodUID := metadata.NewPodMetadataGenerator(common.NewConfig(), nil, nil, nil, nil, true)
+	metaGenWithPodUID := metadata.NewPodMetadataGenerator(common.NewConfig(), nil, nil, nil, nil, nil)
 
 	podUIDIndexer, err := NewPodUIDIndexer(*testConfig, metaGenWithPodUID)
 	assert.NoError(t, err)
@@ -301,7 +301,7 @@ func TestFilteredGenMeta(t *testing.T) {
 	})
 	assert.NoError(t, err)
 
-	filteredGen := metadata.NewPodMetadataGenerator(config, nil, nil, nil, nil, true)
+	filteredGen := metadata.NewPodMetadataGenerator(config, nil, nil, nil, nil, nil)
 
 	podIndexer, err = NewPodNameIndexer(*testConfig, filteredGen)
 	assert.NoError(t, err)
@@ -338,7 +338,7 @@ func TestFilteredGenMetaExclusion(t *testing.T) {
 	})
 	assert.NoError(t, err)
 
-	filteredGen := metadata.NewPodMetadataGenerator(config, nil, nil, nil, nil, true)
+	filteredGen := metadata.NewPodMetadataGenerator(config, nil, nil, nil, nil, nil)
 
 	podIndexer, err := NewPodNameIndexer(*testConfig, filteredGen)
 	assert.NoError(t, err)

libbeat/processors/add_kubernetes_metadata/kubernetes.go

@@ -191,9 +191,6 @@ func (k *kubernetesAnnotator) init(config kubeAnnotatorConfig, cfg *common.Confi
 		}
 
 		metaConf := config.AddResourceMetadata
-		if metaConf == nil {
-			metaConf = metadata.GetDefaultResourceMetadataConfig()
-		}
 
 		options := kubernetes.WatchOptions{
 			SyncTimeout: config.SyncPeriod,

metricbeat/docs/modules/kubernetes.asciidoc

@@ -222,6 +222,16 @@ metricbeat.modules:
   # If kube_config is not set, KUBECONFIG environment variable will be checked
   # and if not present it will fall back to InCluster
   #kube_config: ~/.kube/config
+  # To configure additionally node and namespace metadata, added to pod, service and container resource types,
+  # `add_resource_metadata` can be defined.
+  # By default all labels will be included while annotations are not added by default.
+  # add_resource_metadata:
+  #   namespace:
+  #     include_labels: ["namespacelabel1"]
+  #   node:
+  #     include_labels: ["nodelabel2"]
+  #     include_annotations: ["nodeannotation1"]
+  #   deployment: false
   # Kubernetes client QPS and burst can be configured additionally
   #kube_client_options:
   #  qps: 5
@@ -256,6 +266,16 @@ metricbeat.modules:
   # If kube_config is not set, KUBECONFIG environment variable will be checked
   # and if not present it will fall back to InCluster
   #kube_config: ~/.kube/config
+  # To configure additionally node and namespace metadata, added to pod, service and container resource types,
+  # `add_resource_metadata` can be defined.
+  # By default all labels will be included while annotations are not added by default.
+  # add_resource_metadata:
+  #   namespace:
+  #     include_labels: ["namespacelabel1"]
+  #   node:
+  #     include_labels: ["nodelabel2"]
+  #     include_annotations: ["nodeannotation1"]
+  #   deployment: false
   # Kubernetes client QPS and burst can be configured additionally
   #kube_client_options:
   #  qps: 5