Add replicaset and job metaGenerator based on watcher

kubernetes/metadata/job.go

@@ -0,0 +1,93 @@
+// 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 metadata
+
+import (
+	k8s "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/elastic/elastic-agent-autodiscover/kubernetes"
+	"github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/mapstr"
+)
+
+type job struct {
+	store    cache.Store
+	resource *Resource
+}
+
+// NewJobMetadataGenerator creates a metagen for namespace resources
+func NewJobMetadataGenerator(cfg *config.C, jobs cache.Store, client k8s.Interface) MetaGen {
+	return &job{
+		resource: NewResourceMetadataGenerator(cfg, client),
+		store:    jobs,
+	}
+}
+
+// Generate generates pod metadata from a resource object
+// Metadata map is in the following form:
+//
+//	{
+//		  "kubernetes": {},
+//	   "some.ecs.field": "asdf"
+//	}
+//
+// All Kubernetes fields that need to be stored under kuberentes. prefix are populetad by
+// GenerateK8s method while fields that are part of ECS are generated by GenerateECS method
+func (jb *job) Generate(obj kubernetes.Resource, opts ...FieldOptions) mapstr.M {
+	ecsFields := jb.GenerateECS(obj)
+	meta := mapstr.M{
+		"kubernetes": jb.GenerateK8s(obj, opts...),
+	}
+	meta.DeepUpdate(ecsFields)
+	return meta
+}
+
+// GenerateECS generates namespace ECS metadata from a resource object
+func (jb *job) GenerateECS(obj kubernetes.Resource) mapstr.M {
+	return jb.resource.GenerateECS(obj)
+}
+
+// GenerateK8s generates namespace metadata from a resource object
+func (jb *job) GenerateK8s(obj kubernetes.Resource, opts ...FieldOptions) mapstr.M {
+	_, ok := obj.(*kubernetes.Job)
+	if !ok {
+		return nil
+	}
+
+	meta := jb.resource.GenerateK8s("job", obj, opts...)
+	return meta
+}
+
+// GenerateFromName generates pod metadata from a namespace name
+func (jb *job) GenerateFromName(name string, opts ...FieldOptions) mapstr.M {
+	if jb.store == nil {
+		return nil
+	}
+
+	if obj, ok, _ := jb.store.GetByKey(name); ok {
+		jobObj, ok := obj.(*kubernetes.Job)
+		if !ok {
+			return nil
+		}
+
+		return jb.GenerateK8s(jobObj, opts...)
+	}
+
+	return nil
+}

kubernetes/metadata/metadata.go

@@ -93,16 +93,32 @@ func GetPodMetaGen(
 	podWatcher kubernetes.Watcher,
 	nodeWatcher kubernetes.Watcher,
 	namespaceWatcher kubernetes.Watcher,
+	replicasetWatcher kubernetes.Watcher,
+	jobWatcher kubernetes.Watcher,
 	metaConf *AddResourceMetadataConfig) MetaGen {
 
-	var nodeMetaGen, namespaceMetaGen MetaGen
+	var nodeMetaGen, namespaceMetaGen, rsMetaGen, jobMetaGen MetaGen
 	if nodeWatcher != nil && metaConf.Node.Enabled() {
 		nodeMetaGen = NewNodeMetadataGenerator(metaConf.Node, nodeWatcher.Store(), nodeWatcher.Client())
 	}
 	if namespaceWatcher != nil && metaConf.Namespace.Enabled() {
 		namespaceMetaGen = NewNamespaceMetadataGenerator(metaConf.Namespace, namespaceWatcher.Store(), namespaceWatcher.Client())
 	}
-	metaGen := NewPodMetadataGenerator(cfg, podWatcher.Store(), podWatcher.Client(), nodeMetaGen, namespaceMetaGen, metaConf)
+	if replicasetWatcher != nil && metaConf.Deployment {
+		rsMetaGen = NewReplicasetMetadataGenerator(cfg, replicasetWatcher.Store(), replicasetWatcher.Client())
+	}
+	if jobWatcher != nil && metaConf.CronJob {
+		jobMetaGen = NewJobMetadataGenerator(cfg, jobWatcher.Store(), jobWatcher.Client())
+	}
+	metaGen := NewPodMetadataGenerator(
+		cfg,
+		podWatcher.Store(),
+		podWatcher.Client(),
+		nodeMetaGen,
+		namespaceMetaGen,
+		rsMetaGen,
+		jobMetaGen,
+		metaConf)
 	return metaGen
 }
 

kubernetes/metadata/pod.go

@@ -18,9 +18,6 @@
 package metadata
 
 import (
-	"context"
-
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	k8s "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 
@@ -33,6 +30,8 @@ type pod struct {
 	store               cache.Store
 	client              k8s.Interface
 	node                MetaGen
+	replicaset          MetaGen
+	job                 MetaGen
 	resource            *Resource
 	addResourceMetadata *AddResourceMetadataConfig
 }
@@ -44,12 +43,16 @@ func NewPodMetadataGenerator(
 	client k8s.Interface,
 	node MetaGen,
 	namespace MetaGen,
+	replicaset MetaGen,
+	job MetaGen,
 	addResourceMetadata *AddResourceMetadataConfig) MetaGen {
 
 	return &pod{
 		resource:            NewNamespaceAwareResourceMetadataGenerator(cfg, client, namespace),
 		store:               pods,
 		node:                node,
+		replicaset:          replicaset,
+		job:                 job,
 		client:              client,
 		addResourceMetadata: addResourceMetadata,
 	}
@@ -88,25 +91,32 @@ func (p *pod) GenerateK8s(obj kubernetes.Resource, opts ...FieldOptions) mapstr.
 
 	out := p.resource.GenerateK8s("pod", obj, opts...)
 
-	// check if Pod is handled by a ReplicaSet which is controlled by a Deployment
-	// same happens with CronJob vs Job. The hierarchy there is CronJob->Job->Pod
+	// check if Pod is handled by a ReplicaSet which is controlled by a Deployment.
+	// The hierarchy there is Deployment->ReplicaSet->Pod.
 	if p.addResourceMetadata.Deployment {
-		rsName, _ := out.GetValue("replicaset.name")
-		if rsName, ok := rsName.(string); ok {
-			dep := p.getRSDeployment(rsName, po.GetNamespace())
-			if dep != "" {
-				_, _ = out.Put("deployment.name", dep)
+		if p.replicaset != nil {
+			rsName, _ := out.GetValue("replicaset.name")
+			if rsName, ok := rsName.(string); ok {
+				meta := p.replicaset.GenerateFromName(po.Namespace + "/" + rsName)
+				deploymentName, _ := meta.GetValue("deployment.name")
+				if deploymentName != "" {
+					_, _ = out.Put("deployment.name", deploymentName)
+				}
 			}
 		}
 	}
 
-	// check if Pod is handled by a Job which is controlled by a CronJob
+	// check if Pod is handled by a Job which is controlled by a CronJob.
+	// The hierarchy there is CronJob->Job->Pod
 	if p.addResourceMetadata.CronJob {
-		jobName, _ := out.GetValue("job.name")
-		if jobName, ok := jobName.(string); ok {
-			dep := p.getCronjobOfJob(jobName, po.GetNamespace())
-			if dep != "" {
-				_, _ = out.Put("cronjob.name", dep)
+		if p.job != nil {
+			jobName, _ := out.GetValue("job.name")
+			if jobName, ok := jobName.(string); ok {
+				meta := p.job.GenerateFromName(po.Namespace + "/" + jobName)
+				cronjobName, _ := meta.GetValue("cronjob.name")
+				if cronjobName != "" {
+					_, _ = out.Put("cronjob.name", cronjobName)
+				}
 			}
 		}
 	}
@@ -146,45 +156,3 @@ func (p *pod) GenerateFromName(name string, opts ...FieldOptions) mapstr.M {
 
 	return nil
 }
-
-// getRSDeployment return the name of the Deployment object that
-// owns the ReplicaSet with the given name under the given Namespace
-func (p *pod) getRSDeployment(rsName string, ns string) string {
-	if p.client == nil {
-		return ""
-	}
-	rs, err := p.client.AppsV1().ReplicaSets(ns).Get(context.TODO(), rsName, metav1.GetOptions{})
-	if err != nil {
-		return ""
-	}
-	for _, ref := range rs.GetOwnerReferences() {
-		if ref.Controller != nil && *ref.Controller {
-			switch ref.Kind {
-			case "Deployment":
-				return ref.Name
-			}
-		}
-	}
-	return ""
-}
-
-// getCronjobOfJob return the name of the Cronjob object that
-// owns the Job with the given name under the given Namespace
-func (p *pod) getCronjobOfJob(jobName string, ns string) string {
-	if p.client == nil {
-		return ""
-	}
-	cronjob, err := p.client.BatchV1().Jobs(ns).Get(context.TODO(), jobName, metav1.GetOptions{})
-	if err != nil {
-		return ""
-	}
-	for _, ref := range cronjob.GetOwnerReferences() {
-		if ref.Controller != nil && *ref.Controller {
-			switch ref.Kind {
-			case "CronJob":
-				return ref.Name
-			}
-		}
-	}
-	return ""
-}

kubernetes/metadata/pod_test.go

@@ -429,7 +429,7 @@ func TestPod_Generate(t *testing.T) {
 	})
 	assert.NoError(t, err)
 
-	metagen := NewPodMetadataGenerator(config, nil, client, nil, nil, addResourceMetadata)
+	metagen := NewPodMetadataGenerator(config, nil, client, nil, nil, nil, nil, addResourceMetadata)
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {
 			assert.Equal(t, test.output, metagen.Generate(test.input))
@@ -549,7 +549,7 @@ func TestPod_GenerateFromName(t *testing.T) {
 		pods := cache.NewStore(cache.MetaNamespaceKeyFunc)
 		err = pods.Add(test.input)
 		require.NoError(t, err)
-		metagen := NewPodMetadataGenerator(config, pods, client, nil, nil, addResourceMetadata)
+		metagen := NewPodMetadataGenerator(config, pods, client, nil, nil, nil, nil, addResourceMetadata)
 
 		accessor, err := meta.Accessor(test.input)
 		require.NoError(t, err)
@@ -671,7 +671,7 @@ func TestPod_GenerateWithNodeNamespace(t *testing.T) {
 		require.NoError(t, err)
 		nsMeta := NewNamespaceMetadataGenerator(config, namespaces, client)
 
-		metagen := NewPodMetadataGenerator(config, pods, client, nodeMeta, nsMeta, addResourceMetadata)
+		metagen := NewPodMetadataGenerator(config, pods, client, nodeMeta, nsMeta, nil, nil, addResourceMetadata)
 		t.Run(test.name, func(t *testing.T) {
 			assert.Equal(t, test.output, metagen.Generate(test.input))
 		})
@@ -832,7 +832,7 @@ func TestPod_GenerateWithNodeNamespaceWithAddResourceConfig(t *testing.T) {
 		require.NoError(t, err)
 		nsMeta := NewNamespaceMetadataGenerator(namespaceConfig, namespaces, client)
 
-		metagen := NewPodMetadataGenerator(c, pods, client, nodeMeta, nsMeta, &metaConfig)
+		metagen := NewPodMetadataGenerator(c, pods, client, nodeMeta, nsMeta, nil, nil, &metaConfig)
 		t.Run(test.name, func(t *testing.T) {
 			assert.Equal(t, test.output, metagen.Generate(test.input))
 		})

kubernetes/metadata/replicaset.go

@@ -0,0 +1,95 @@
+// 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 metadata
+
+import (
+	k8s "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+
+	"github.com/elastic/elastic-agent-autodiscover/kubernetes"
+	"github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/mapstr"
+)
+
+const resourceType = "replicaset"
+
+type replicaset struct {
+	store    cache.Store
+	resource *Resource
+}
+
+// NewReplicasetMetadataGenerator creates a metagen for namespace resources
+func NewReplicasetMetadataGenerator(cfg *config.C, replicasets cache.Store, client k8s.Interface) MetaGen {
+	return &replicaset{
+		resource: NewResourceMetadataGenerator(cfg, client),
+		store:    replicasets,
+	}
+}
+
+// Generate generates pod metadata from a resource object
+// Metadata map is in the following form:
+//
+//	{
+//		  "kubernetes": {},
+//	   "some.ecs.field": "asdf"
+//	}
+//
+// All Kubernetes fields that need to be stored under kuberentes. prefix are populetad by
+// GenerateK8s method while fields that are part of ECS are generated by GenerateECS method
+func (rs *replicaset) Generate(obj kubernetes.Resource, opts ...FieldOptions) mapstr.M {
+	ecsFields := rs.GenerateECS(obj)
+	meta := mapstr.M{
+		"kubernetes": rs.GenerateK8s(obj, opts...),
+	}
+	meta.DeepUpdate(ecsFields)
+	return meta
+}
+
+// GenerateECS generates namespace ECS metadata from a resource object
+func (rs *replicaset) GenerateECS(obj kubernetes.Resource) mapstr.M {
+	return rs.resource.GenerateECS(obj)
+}
+
+// GenerateK8s generates namespace metadata from a resource object
+func (rs *replicaset) GenerateK8s(obj kubernetes.Resource, opts ...FieldOptions) mapstr.M {
+	_, ok := obj.(*kubernetes.ReplicaSet)
+	if !ok {
+		return nil
+	}
+
+	meta := rs.resource.GenerateK8s(resourceType, obj, opts...)
+	return meta
+}
+
+// GenerateFromName generates pod metadata from a namespace name
+func (rs *replicaset) GenerateFromName(name string, opts ...FieldOptions) mapstr.M {
+	if rs.store == nil {
+		return nil
+	}
+
+	if obj, ok, _ := rs.store.GetByKey(name); ok {
+		replicaSet, ok := obj.(*kubernetes.ReplicaSet)
+		if !ok {
+			return nil
+		}
+
+		return rs.GenerateK8s(replicaSet, opts...)
+	}
+
+	return nil
+}