Reduce the amount of stored ReplicaSet data

internal/pkg/composable/providers/kubernetes/metagen.go

@@ -0,0 +1,123 @@
+package kubernetes
+
+import (
+	"github.com/elastic/elastic-agent-autodiscover/kubernetes"
+	"github.com/elastic/elastic-agent-autodiscover/kubernetes/metadata"
+	"github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/mapstr"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8s "k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/cache"
+)
+
+type (
+	AddResourceMetadataConfig = metadata.AddResourceMetadataConfig
+	MetaGen                   = metadata.MetaGen
+	FieldOptions              = metadata.FieldOptions
+)
+
+// GetPodMetaGen is a wrapper function that creates a metaGen for pod resource and has embeeded
+// nodeMetaGen and namespaceMetaGen
+func GetPodMetaGen(
+	cfg *config.C,
+	podWatcher kubernetes.Watcher,
+	nodeWatcher kubernetes.Watcher,
+	namespaceWatcher kubernetes.Watcher,
+	replicasetWatcher kubernetes.Watcher,
+	jobWatcher kubernetes.Watcher,
+	metaConf *AddResourceMetadataConfig) MetaGen {
+
+	var nodeMetaGen, namespaceMetaGen, rsMetaGen, jobMetaGen MetaGen
+	if nodeWatcher != nil && metaConf.Node.Enabled() {
+		nodeMetaGen = metadata.NewNodeMetadataGenerator(metaConf.Node, nodeWatcher.Store(), nodeWatcher.Client())
+	}
+	if namespaceWatcher != nil && metaConf.Namespace.Enabled() {
+		namespaceMetaGen = metadata.NewNamespaceMetadataGenerator(metaConf.Namespace, namespaceWatcher.Store(), namespaceWatcher.Client())
+	}
+	if replicasetWatcher != nil && metaConf.Deployment {
+		// use our own implementation of this generator, which can avoid tracking the full ReplicaSet resource
+		// TODO: Remove this after upstreaming the change to the autodiscovery lib
+		rsMetaGen = NewReplicasetMetadataGenerator(cfg, replicasetWatcher.Store(), replicasetWatcher.Client())
+	}
+	if jobWatcher != nil && metaConf.CronJob {
+		jobMetaGen = metadata.NewJobMetadataGenerator(cfg, jobWatcher.Store(), jobWatcher.Client())
+	}
+	metaGen := metadata.NewPodMetadataGenerator(
+		cfg,
+		podWatcher.Store(),
+		podWatcher.Client(),
+		nodeMetaGen,
+		namespaceMetaGen,
+		rsMetaGen,
+		jobMetaGen,
+		metaConf)
+	return metaGen
+}
+
+const resourceType = "replicaset"
+
+type replicaset struct {
+	store    cache.Store
+	resource *metadata.Resource
+}
+
+// NewReplicasetMetadataGenerator creates a metagen for replicaset resources
+func NewReplicasetMetadataGenerator(cfg *config.C, replicasets cache.Store, client k8s.Interface) MetaGen {
+	return &replicaset{
+		resource: metadata.NewResourceMetadataGenerator(cfg, client),
+		store:    replicasets,
+	}
+}
+
+// Generate generates replicaset 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 replicaset ECS metadata from a resource object
+func (rs *replicaset) GenerateECS(obj kubernetes.Resource) mapstr.M {
+	return rs.resource.GenerateECS(obj)
+}
+
+// GenerateK8s generates replicaset metadata from a resource object
+func (rs *replicaset) GenerateK8s(obj kubernetes.Resource, opts ...FieldOptions) mapstr.M {
+	_, ok := obj.(metav1.Object) // one of the changes from upstream autodiscovery
+	if !ok {
+		return nil
+	}
+
+	meta := rs.resource.GenerateK8s(resourceType, obj, opts...)
+	return meta
+}
+
+// GenerateFromName generates replicaset metadata from a replicaset 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.Resource) // one of the changes from upstream autodiscovery
+		if !ok {
+			return nil
+		}
+
+		return rs.GenerateK8s(replicaSet, opts...)
+	}
+
+	return nil
+}

internal/pkg/composable/providers/kubernetes/pod.go

@@ -6,6 +6,8 @@ package kubernetes
 
 import (
 	"fmt"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sync"
 	"time"
 
@@ -18,6 +20,7 @@ import (
 	"github.com/elastic/elastic-agent-libs/safemapstr"
 
 	k8s "k8s.io/client-go/kubernetes"
+	clientgometa "k8s.io/client-go/metadata"
 
 	"github.com/elastic/elastic-agent/internal/pkg/agent/errors"
 	"github.com/elastic/elastic-agent/internal/pkg/composable"
@@ -104,11 +107,21 @@ func NewPodEventer(
 	// Deployment -> Replicaset -> Pod
 	// CronJob -> job -> Pod
 	if metaConf.Deployment {
-		replicaSetWatcher, err = kubernetes.NewNamedWatcher("resource_metadata_enricher_rs", client, &kubernetes.ReplicaSet{}, kubernetes.WatchOptions{
-			SyncTimeout:  cfg.SyncPeriod,
-			Namespace:    cfg.Namespace,
-			HonorReSyncs: true,
-		}, nil)
+		metadataClient, err := GetKubernetesMetadataClient(cfg.KubeConfig, cfg.KubeClientOptions)
+		if err != nil {
+			logger.Errorf("Error creating metadata client for %T due to error %+v", &kubernetes.Namespace{}, err)
+		}
+		// use a custom watcher here, so we can provide a transform function and limit the data we're storing
+		replicaSetWatcher, err = NewNamedMetaWatcher(
+			"resource_metadata_enricher_rs",
+			client,
+			metadataClient,
+			schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "replicasets"},
+			kubernetes.WatchOptions{
+				SyncTimeout:  cfg.SyncPeriod,
+				Namespace:    cfg.Namespace,
+				HonorReSyncs: true,
+			}, nil, removeUnnecessaryReplicaSetData)
 		if err != nil {
 			logger.Errorf("Error creating watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
 		}
@@ -128,7 +141,7 @@ func NewPodEventer(
 	if err != nil {
 		return nil, errors.New(err, "failed to unpack configuration")
 	}
-	metaGen := metadata.GetPodMetaGen(rawConfig, watcher, nodeWatcher, namespaceWatcher, replicaSetWatcher, jobWatcher, metaConf)
+	metaGen := GetPodMetaGen(rawConfig, watcher, nodeWatcher, namespaceWatcher, replicaSetWatcher, jobWatcher, metaConf)
 
 	p := &pod{
 		logger:            logger,
@@ -539,3 +552,43 @@ func hintsCheck(annotations mapstr.M, container string, prefix string, validate
 	}
 	return hints, incorrecthints
 }
+
+// removeUnnecessaryReplicaSetData removes all data from a ReplicaSet resource, except what we need to compute
+// Pod metadata. Which is just the name and owner references.
+func removeUnnecessaryReplicaSetData(obj interface{}) (interface{}, error) {
+	old, ok := obj.(*metav1.PartialObjectMetadata)
+	if !ok {
+		return nil, fmt.Errorf("obj is not a ReplicaSet")
+	}
+	transformed := &metav1.PartialObjectMetadata{
+		ObjectMeta: kubernetes.ObjectMeta{
+			Name:            old.GetName(),
+			Namespace:       old.GetNamespace(),
+			OwnerReferences: old.GetOwnerReferences(),
+			ResourceVersion: old.GetResourceVersion(),
+		},
+	}
+	return transformed, nil
+}
+
+// GetKubernetesMetadataClient returns a kubernetes metadata client. If inCluster is true, it returns an
+// in cluster configuration based on the secrets mounted in the Pod. If kubeConfig is passed,
+// it parses the config file to get the config required to build a client.
+func GetKubernetesMetadataClient(kubeconfig string, opt kubernetes.KubeClientOptions) (clientgometa.Interface, error) {
+	if kubeconfig == "" {
+		kubeconfig = kubernetes.GetKubeConfigEnvironmentVariable()
+	}
+
+	cfg, err := kubernetes.BuildConfig(kubeconfig)
+	if err != nil {
+		return nil, fmt.Errorf("unable to build kube config due to error: %w", err)
+	}
+	cfg.QPS = opt.QPS
+	cfg.Burst = opt.Burst
+	client, err := clientgometa.NewForConfig(cfg)
+	if err != nil {
+		return nil, fmt.Errorf("unable to build kubernetes clientset: %w", err)
+	}
+
+	return client, nil
+}