Reduce the amount of stored ReplicaSet data

Use a transform function to drop all data except for the owner
references, which we need to find the Deployment name.

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

@@ -9,6 +9,8 @@ import (
  "sync"
  "time"
 
+ v1 "k8s.io/api/apps/v1"
+
  "github.com/elastic/elastic-agent-autodiscover/kubernetes"
  "github.com/elastic/elastic-agent-autodiscover/kubernetes/metadata"
  "github.com/elastic/elastic-agent-autodiscover/utils"
@@ -104,11 +106,12 @@ func NewPodEventer(
  // Deployment -> Replicaset -> Pod
  // CronJob -> job -> Pod
  if metaConf.Deployment {
- replicaSetWatcher, err = kubernetes.NewNamedWatcher("resource_metadata_enricher_rs", client, &kubernetes.ReplicaSet{}, kubernetes.WatchOptions{
+ // use a custom watcher here, so we can provide a transform function and limit the data we're storing
+ replicaSetWatcher, err = NewNamedWatcher("resource_metadata_enricher_rs", client, &kubernetes.ReplicaSet{}, kubernetes.WatchOptions{
  SyncTimeout: cfg.SyncPeriod,
  Namespace: cfg.Namespace,
  HonorReSyncs: true,
- }, nil)
+ }, nil, removeUnnecessaryReplicaSetData)
  if err != nil {
  logger.Errorf("Error creating watcher for %T due to error %+v", &kubernetes.Namespace{}, err)
  }
@@ -539,3 +542,19 @@ 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.(*v1.ReplicaSet)
+ if !ok {
+ return nil, fmt.Errorf("obj is not a ReplicaSet")
+ }
+ transformed := v1.ReplicaSet{}
+ transformed.ObjectMeta = kubernetes.ObjectMeta{
+ Name: old.GetName(),
+ Namespace: old.GetNamespace(),
+ OwnerReferences: old.GetOwnerReferences(),
+ }
+ return transformed, nil
+}

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

@@ -0,0 +1,285 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License 2.0;
+// you may not use this file except in compliance with the Elastic License 2.0.
+
+package kubernetes
+
+import (
+ "context"
+ "fmt"
+ "time"
+
+ "k8s.io/apimachinery/pkg/api/meta"
+ "k8s.io/apimachinery/pkg/runtime"
+ utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+ "k8s.io/apimachinery/pkg/util/wait"
+ "k8s.io/client-go/kubernetes"
+ "k8s.io/client-go/tools/cache"
+ "k8s.io/client-go/util/workqueue"
+
+ autodiscoverK8s "github.com/elastic/elastic-agent-autodiscover/kubernetes"
+ "github.com/elastic/elastic-agent-libs/logp"
+)
+
+const (
+ add = "add"
+ update = "update"
+ delete = "delete"
+)
+
+var (
+ accessor = meta.NewAccessor()
+)
+
+// some type aliases to avoid unnecessarily long declarations
+type (
+ ResourceEventHandler = autodiscoverK8s.ResourceEventHandler
+ Resource = autodiscoverK8s.Resource
+ Namespace = autodiscoverK8s.Namespace
+ Node = autodiscoverK8s.Node
+ WatchOptions = autodiscoverK8s.WatchOptions
+ Watcher = autodiscoverK8s.Watcher
+)
+
+type item struct {
+ object interface{}
+ objectRaw interface{}
+ state string
+}
+
+type watcher struct {
+ client kubernetes.Interface
+ informer cache.SharedInformer
+ store cache.Store
+ queue workqueue.Interface //nolint:staticcheck // TODO: use the typed version
+ ctx context.Context
+ stop context.CancelFunc
+ handler ResourceEventHandler
+ logger *logp.Logger
+ cachedObject runtime.Object
+}
+
+// NOTE: This watcher implementation is identical to the one in autodiscovery, with the single difference
+// that it allows setting a transform function on the informer.
+// This is necessary to avoid storing a lot of unnecessary ReplicaSet data.
+
+// NewWatcher initializes the watcher client to provide a events handler for
+// resource from the cluster (filtered to the given node)
+func NewWatcher(
+ client kubernetes.Interface,
+ resource Resource,
+ opts WatchOptions,
+ indexers cache.Indexers,
+ transformFunc cache.TransformFunc,
+) (Watcher, error) {
+ return NewNamedWatcher("", client, resource, opts, indexers, transformFunc)
+}
+
+// NewNamedWatcher initializes the watcher client to provide an events handler for
+// resource from the cluster (filtered to the given node) and also allows to name the k8s
+// client's workqueue that is used by the watcher. Workqueue name is important for exposing workqueue
+// metrics, if it is empty, its metrics will not be logged by the k8s client.
+func NewNamedWatcher(
+ name string,
+ client kubernetes.Interface,
+ resource Resource,
+ opts WatchOptions,
+ indexers cache.Indexers,
+ transformFunc cache.TransformFunc,
+) (Watcher, error) {
+ var store cache.Store
+ var queue workqueue.Interface //nolint:staticcheck // TODO: use the typed version
+ var cachedObject runtime.Object
+ informer, _, err := autodiscoverK8s.NewInformer(client, resource, opts, indexers)
+ if err != nil {
+ return nil, err
+ }
+
+ store = informer.GetStore()
+ queue = workqueue.NewNamed(name)
+
+ if opts.IsUpdated == nil {
+ opts.IsUpdated = func(o, n interface{}) bool {
+ oldVersion, _ := accessor.ResourceVersion(o.(runtime.Object))
+ newVersion, _ := accessor.ResourceVersion(n.(runtime.Object))
+
+ // Only enqueue changes that have a different resource versions to avoid processing resyncs.
+ return oldVersion != newVersion
+ }
+ }
+
+ ctx, cancel := context.WithCancel(context.TODO())
+ w := &watcher{
+ client: client,
+ informer: informer,
+ store: store,
+ queue: queue,
+ ctx: ctx,
+ cachedObject: cachedObject,
+ stop: cancel,
+ logger: logp.NewLogger("kubernetes"),
+ handler: autodiscoverK8s.NoOpEventHandlerFuncs{},
+ }
+
+ _, err = w.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+ AddFunc: func(o interface{}) {
+ w.enqueue(o, add)
+ },
+ DeleteFunc: func(o interface{}) {
+ w.enqueue(o, delete)
+ },
+ UpdateFunc: func(o, n interface{}) {
+ if opts.IsUpdated(o, n) {
+ w.enqueue(n, update)
+ } else if opts.HonorReSyncs {
+ // HonorReSyncs ensure that at the time when the kubernetes client does a "resync", i.e, a full list of all
+ // objects we make sure that autodiscover processes them. Why is this necessary? An effective control loop works
+ // based on two state changes, a list and a watch. A watch is triggered each time the state of the system changes.
+ // However, there is no guarantee that all events from a watch are processed by the receiver. To ensure that missed events
+ // are properly handled, a period re-list is done to ensure that every state within the system is effectively handled.
+ // In this case, we are making sure that we are enqueueing an "add" event because, an runner that is already in Running
+ // state should just be deduped by autodiscover and not stop/started periodically as would be the case with an update.
+ w.enqueue(n, add)
+ }
+
+ //We check the type of resource and only if it is namespace or node return the cacheObject
+ switch resource.(type) {
+ case *Namespace:
+ w.cacheObject(o)
+ case *Node:
+ w.cacheObject(o)
+ }
+ },
+ })
+ if err != nil {
+ return nil, err
+ }
+
+ if transformFunc != nil {
+ err = informer.SetTransform(transformFunc)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ return w, nil
+}
+
+// AddEventHandler adds a resource handler to process each request that is coming into the watcher
+func (w *watcher) AddEventHandler(h ResourceEventHandler) {
+ w.handler = h
+}
+
+// GetEventHandler returns the watcher's event handler
+func (w *watcher) GetEventHandler() ResourceEventHandler {
+ return w.handler
+}
+
+// Store returns the store object for the resource that is being watched
+func (w *watcher) Store() cache.Store {
+ return w.store
+}
+
+// Client returns the kubernetes client object used by the watcher
+func (w *watcher) Client() kubernetes.Interface {
+ return w.client
+}
+
+// CachedObject returns the old object in cache during the last updated event
+func (w *watcher) CachedObject() runtime.Object {
+ return w.cachedObject
+}
+
+// Start watching pods
+func (w *watcher) Start() error {
+ go w.informer.Run(w.ctx.Done())
+
+ if !cache.WaitForCacheSync(w.ctx.Done(), w.informer.HasSynced) {
+ return fmt.Errorf("kubernetes informer unable to sync cache")
+ }
+
+ w.logger.Debugf("cache sync done")
+
+ // Wrap the process function with wait.Until so that if the controller crashes, it starts up again after a second.
+ go wait.Until(func() {
+ for w.process(w.ctx) {
+ }
+ }, time.Second*1, w.ctx.Done())
+
+ return nil
+}
+
+func (w *watcher) Stop() {
+ w.queue.ShutDown()
+ w.stop()
+}
+
+// enqueue takes the most recent object that was received, figures out the namespace/name of the object
+// and adds it to the work queue for processing.
+func (w *watcher) enqueue(obj interface{}, state string) {
+ // DeletionHandlingMetaNamespaceKeyFunc that we get a key only if the resource's state is not Unknown.
+ key, err := cache.DeletionHandlingMetaNamespaceKeyFunc(obj)
+ if err != nil {
+ return
+ }
+ if deleted, ok := obj.(cache.DeletedFinalStateUnknown); ok {
+ w.logger.Debugf("Enqueued DeletedFinalStateUnknown contained object: %+v", deleted.Obj)
+ obj = deleted.Obj
+ }
+ w.queue.Add(&item{key, obj, state})
+}
+
+// cacheObject updates watcher with the old version of cache objects before change during update events
+func (w *watcher) cacheObject(o interface{}) {
+ if old, ok := o.(runtime.Object); !ok {
+ utilruntime.HandleError(fmt.Errorf("expected object in cache got %#v", o))
+ } else {
+ w.cachedObject = old
+ }
+}
+
+// process gets the top of the work queue and processes the object that is received.
+func (w *watcher) process(_ context.Context) bool {
+ obj, quit := w.queue.Get()
+ if quit {
+ return false
+ }
+ defer w.queue.Done(obj)
+
+ var entry *item
+ var ok bool
+ if entry, ok = obj.(*item); !ok {
+ utilruntime.HandleError(fmt.Errorf("expected *item in workqueue but got %#v", obj))
+ return true
+ }
+
+ key, ok := entry.object.(string)
+ if !ok {
+ return false
+ }
+
+ o, exists, err := w.store.GetByKey(key)
+ if err != nil {
+ utilruntime.HandleError(fmt.Errorf("getting object %#v from cache: %w", obj, err))
+ return true
+ }
+ if !exists {
+ if entry.state == delete {
+ w.logger.Debugf("Object %+v was not found in the store, deleting anyway!", key)
+ // delete anyway in order to clean states
+ w.handler.OnDelete(entry.objectRaw)
+ }
+ return true
+ }
+
+ switch entry.state {
+ case add:
+ w.handler.OnAdd(o)
+ case update:
+ w.handler.OnUpdate(o)
+ case delete:
+ w.handler.OnDelete(o)
+ }
+
+ return true
+}