Add a mutation cache to the ComputeDomainManager in the daemon pod

Signed-off-by: Kevin Klues <kklues@nvidia.com>

Author: klueska

cmd/compute-domain-daemon/computedomain.go

@@ -34,6 +34,7 @@ import (
 
 const (
 	informerResyncPeriod = 10 * time.Minute
+	mutationCacheTTL     = time.Hour
 )
 
 // GetComputeDomainFunc is a function type for getting a ComputeDomain by UID.
@@ -54,7 +55,8 @@ type ComputeDomainManager struct {
 	previousNodes    []*nvapi.ComputeDomainNode
 	updatedNodesChan chan []*nvapi.ComputeDomainNode
 
-	podManager *PodManager
+	podManager    *PodManager
+	mutationCache cache.MutationCache
 }
 
 // NewComputeDomainManager creates a new ComputeDomainManager instance.
@@ -101,6 +103,15 @@ func (m *ComputeDomainManager) Start(ctx context.Context) (rerr error) {
 		return fmt.Errorf("error adding indexer for ComputeDomain UID: %w", err)
 	}
 
+	// Create mutation cache to track our own updates
+	m.mutationCache = cache.NewIntegerResourceVersionMutationCache(
+		klog.Background(),
+		m.informer.GetStore(),
+		m.informer.GetIndexer(),
+		mutationCacheTTL,
+		true,
+	)
+
 	_, err = m.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj any) {
 			m.config.workQueue.Enqueue(obj, m.onAddOrUpdate)
@@ -157,23 +168,19 @@ func (m *ComputeDomainManager) Stop() error {
 	return nil
 }
 
-// Get gets the ComputeDomain by UID from the informer cache.
+// Get gets the ComputeDomain by UID from the mutation cache.
 func (m *ComputeDomainManager) Get(uid string) (*nvapi.ComputeDomain, error) {
-	objs, err := m.informer.GetIndexer().ByIndex("uid", uid)
+	cds, err := getByComputeDomainUID[*nvapi.ComputeDomain](m.mutationCache, uid)
 	if err != nil {
 		return nil, fmt.Errorf("error retrieving ComputeDomain by UID: %w", err)
 	}
-	if len(objs) == 0 {
+	if len(cds) == 0 {
 		return nil, nil
 	}
-	if len(objs) != 1 {
+	if len(cds) != 1 {
 		return nil, fmt.Errorf("multiple ComputeDomains with the same UID")
 	}
-	cd, ok := objs[0].(*nvapi.ComputeDomain)
-	if !ok {
-		return nil, fmt.Errorf("error casting to ComputeDomain")
-	}
-	return cd, nil
+	return cds[0], nil
 }
 
 // onAddOrUpdate handles the addition or update of a ComputeDomain.
@@ -266,6 +273,9 @@ func (m *ComputeDomainManager) UpdateComputeDomainNodeInfo(ctx context.Context,
 		return fmt.Errorf("error updating nodes in ComputeDomain status: %w", err)
 	}
 
+	// Add the updated ComputeDomain to the mutation cache
+	m.mutationCache.Mutation(newCD)
+
 	return nil
 }
 

cmd/compute-domain-daemon/indexers.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
 )
 
 func uidIndexer[T metav1.ObjectMetaAccessor](obj any) ([]string, error) {
@@ -29,3 +30,19 @@ func uidIndexer[T metav1.ObjectMetaAccessor](obj any) ([]string, error) {
 	}
 	return []string{string(d.GetObjectMeta().GetUID())}, nil
 }
+
+// getByComputeDomainUID retrieves objects by UID using the mutation cache.
+func getByComputeDomainUID[T any](mutationCache cache.MutationCache, uid string) ([]T, error) {
+	objs, err := mutationCache.ByIndex("uid", uid)
+	if err != nil {
+		return nil, fmt.Errorf("error retrieving objects by UID: %w", err)
+	}
+
+	var result []T
+	for _, obj := range objs {
+		if t, ok := obj.(T); ok {
+			result = append(result, t)
+		}
+	}
+	return result, nil
+}