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 job resources
+func NewJobMetadataGenerator(cfg *config.C, jobs cache.Store, client k8s.Interface) MetaGen {
+	return &job{
+		resource: NewResourceMetadataGenerator(cfg, client),
+		store:    jobs,
+	}
+}
+
+// Generate generates job 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 job ECS metadata from a resource object
+func (jb *job) GenerateECS(obj kubernetes.Resource) mapstr.M {
+	return jb.resource.GenerateECS(obj)
+}
+
+// GenerateK8s generates job 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 job 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/job_test.go

@@ -0,0 +1,163 @@
+// 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 (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	batchv1 "k8s.io/api/batch/v1"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	k8sfake "k8s.io/client-go/kubernetes/fake"
+	"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"
+)
+
+func TestJob_Generate(t *testing.T) {
+	client := k8sfake.NewSimpleClientset()
+	boolean := true
+	tests := []struct {
+		input  kubernetes.Resource
+		output mapstr.M
+		name   string
+	}{
+		{
+			name: "test simple object with owner",
+			input: &batchv1.Job{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      name,
+					UID:       types.UID(uid),
+					Namespace: defaultNs,
+					Labels: map[string]string{
+						"foo": "bar",
+					},
+					Annotations: map[string]string{},
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							APIVersion: "apps",
+							Kind:       "CronJob",
+							Name:       "nginx-job",
+							UID:        "005f3b90-4b9d-12f8-acf0-31020a840144",
+							Controller: &boolean,
+						},
+					},
+				},
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Job",
+					APIVersion: "v1",
+				},
+			},
+			output: mapstr.M{
+				"kubernetes": mapstr.M{
+					"job": mapstr.M{
+						"name": name,
+						"uid":  uid,
+					},
+					"labels": mapstr.M{
+						"foo": "bar",
+					},
+					"cronjob": mapstr.M{
+						"name": "nginx-job",
+					},
+					"namespace": defaultNs,
+				},
+			},
+		},
+	}
+
+	cfg := config.NewConfig()
+	metagen := NewJobMetadataGenerator(cfg, nil, client)
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.output, metagen.Generate(test.input))
+		})
+	}
+}
+
+func TestJob_GenerateFromName(t *testing.T) {
+	client := k8sfake.NewSimpleClientset()
+	boolean := true
+	tests := []struct {
+		input  kubernetes.Resource
+		output mapstr.M
+		name   string
+	}{
+		{
+			name: "test simple object with owner",
+			input: &batchv1.Job{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      name,
+					UID:       types.UID(uid),
+					Namespace: defaultNs,
+					Labels: map[string]string{
+						"foo": "bar",
+					},
+					Annotations: map[string]string{},
+					OwnerReferences: []metav1.OwnerReference{
+						{
+							APIVersion: "apps",
+							Kind:       "CronJob",
+							Name:       "nginx-job",
+							UID:        "005f3b90-4b9d-12f8-acf0-31020a840144",
+							Controller: &boolean,
+						},
+					},
+				},
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "Job",
+					APIVersion: "v1",
+				},
+			},
+			output: mapstr.M{
+				"job": mapstr.M{
+					"name": name,
+					"uid":  uid,
+				},
+				"labels": mapstr.M{
+					"foo": "bar",
+				},
+				"cronjob": mapstr.M{
+					"name": "nginx-job",
+				},
+				"namespace": defaultNs,
+			},
+		},
+	}
+
+	for _, test := range tests {
+		cfg := config.NewConfig()
+		jobs := cache.NewStore(cache.MetaNamespaceKeyFunc)
+		err := jobs.Add(test.input)
+		require.NoError(t, err)
+		metagen := NewJobMetadataGenerator(cfg, jobs, client)
+
+		accessor, err := meta.Accessor(test.input)
+		require.NoError(t, err)
+
+		t.Run(test.name, func(t *testing.T) {
+			assert.Equal(t, test.output, metagen.GenerateFromName(fmt.Sprint(accessor.GetNamespace(), "/", accessor.GetName())))
+		})
+	}
+}

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
 }