CD daemon: pod mngr: store UpdateStatus return value in mutation cache

This makes sure that fast incremental mutations on
the same CD object performed during shutdown are done
conflict-free (i.e., in actual, incremental fashion
using intermediate state returned by the API server).

Without this patch:

I1007 16:49:01.678050       1 podmanager.go:196] Successfully updated node gb-nvl-043-compute06 status to NotReady
E1007 16:49:01.681345       1 computedomain.go:161] Failed to remove node from ComputeDomain during shutdown: [...] \
				"the object has been modified" [...]

With this patch:

I1007 16:59:55.350436       1 podmanager.go:200] Successfully updated node gb-nvl-043-compute07 status to NotReady
I1007 16:59:55.353551       1 computedomain.go:402] Successfully removed node with IP 192.168.34.153 from ComputeDomain default/imex-channel-injection

Signed-off-by: Dr. Jan-Philip Gehrcke <[email protected]>

Author: jgehrcke

cmd/compute-domain-daemon/computedomain.go

@@ -111,7 +111,7 @@ func (m *ComputeDomainManager) Start(ctx context.Context) (rerr error) {
 		true,
 	)
 
-	m.podManager = NewPodManager(m.config, m.Get, &m.mutationCache)
+	m.podManager = NewPodManager(m.config, m.Get, m.mutationCache)
 
 	_, err = m.informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj any) {
@@ -192,8 +192,9 @@ func (m *ComputeDomainManager) onAddOrUpdate(ctx context.Context, obj any) error
 		return fmt.Errorf("failed to cast to ComputeDomain")
 	}
 
-	// Get the latest ComputeDomain object from the informer cache since we
-	// plan to update it later and always *must* have the latest version.
+	// Get the latest ComputeDomain object from the mutation cache (backed by
+	// the informer cache) since we plan to update it later and always *must*
+	// have the latest version.
 	cd, err := m.Get(string(o.GetUID()))
 	if err != nil {
 		return fmt.Errorf("error getting latest ComputeDomain: %w", err)
@@ -216,8 +217,9 @@ func (m *ComputeDomainManager) onAddOrUpdate(ctx context.Context, obj any) error
 	return nil
 }
 
-// UpdateComputeDomainNodeInfo updates the Nodes field in the ComputeDomain
-// with info about the ComputeDomain daemon running on this node.
+// UpdateComputeDomainNodeInfo updates the Nodes field in the ComputeDomain with
+// info about the ComputeDomain daemon running on this node. Upon success, it
+// reflects the mutation in `m.mutationCache`.
 func (m *ComputeDomainManager) UpdateComputeDomainNodeInfo(ctx context.Context, cd *nvapi.ComputeDomain) (rerr error) {
 	var nodeInfo *nvapi.ComputeDomainNode
 
@@ -269,12 +271,12 @@ func (m *ComputeDomainManager) UpdateComputeDomainNodeInfo(ctx context.Context,
 		newCD.Status.Status = nvapi.ComputeDomainStatusNotReady
 	}
 
-	// Update the status
-	if _, err := m.config.clientsets.Nvidia.ResourceV1beta1().ComputeDomains(newCD.Namespace).UpdateStatus(ctx, newCD, metav1.UpdateOptions{}); err != nil {
-		return fmt.Errorf("error updating nodes in ComputeDomain status: %w", err)
+	// Update status and (upon success) store the latest version of the object
+	// (as returned by the API server) in the mutation cache.
+	newCD, err := m.config.clientsets.Nvidia.ResourceV1beta1().ComputeDomains(newCD.Namespace).UpdateStatus(ctx, newCD, metav1.UpdateOptions{})
+	if err != nil {
+		return fmt.Errorf("error updating ComputeDomain status: %w", err)
 	}
-
-	// Add the updated ComputeDomain to the mutation cache
 	m.mutationCache.Mutation(newCD)
 
 	return nil
@@ -361,15 +363,13 @@ func (m *ComputeDomainManager) GetNodesUpdateChan() chan []*nvapi.ComputeDomainN
 
 // removeNodeFromComputeDomain removes the current node's entry from the ComputeDomain status.
 func (m *ComputeDomainManager) removeNodeFromComputeDomain(ctx context.Context) error {
-	objs := m.informer.GetIndexer().List()
-	if len(objs) == 0 {
-		klog.Infof("No ComputeDomain objects found in informer cache during cleanup")
-		return nil
+	cd, err := m.Get(m.config.computeDomainUUID)
+	if err != nil {
+		return fmt.Errorf("error getting ComputeDomain from mutation cache: %w", err)
 	}
-
-	cd, ok := objs[0].(*nvapi.ComputeDomain)
-	if !ok {
-		return fmt.Errorf("failed to cast object to ComputeDomain")
+	if cd == nil {
+		klog.Infof("No ComputeDomain object found in mutation cache during cleanup")
+		return nil
 	}
 
 	newCD := cd.DeepCopy()
@@ -397,6 +397,9 @@ func (m *ComputeDomainManager) removeNodeFromComputeDomain(ctx context.Context)
 		return fmt.Errorf("error removing node from ComputeDomain status: %w", err)
 	}
 
+	// Add the updated ComputeDomain to the mutation cache
+	m.mutationCache.Mutation(newCD)
+
 	klog.Infof("Successfully removed node with IP %s from ComputeDomain %s/%s", m.config.podIP, newCD.Namespace, newCD.Name)
 	return nil
 }

cmd/compute-domain-daemon/podmanager.go

@@ -33,17 +33,17 @@ import (
 
 // PodManager watches for changes to its own pod and updates the CD status accordingly.
 type PodManager struct {
-	config           *ManagerConfig
-	getComputeDomain GetComputeDomainFunc
-	waitGroup        sync.WaitGroup
-	cancelContext    context.CancelFunc
-	podInformer      cache.SharedIndexInformer
-	informerFactory  informers.SharedInformerFactory
-	mutationCache    cache.MutationCache
+	config                     *ManagerConfig
+	getComputeDomain           GetComputeDomainFunc
+	waitGroup                  sync.WaitGroup
+	cancelContext              context.CancelFunc
+	podInformer                cache.SharedIndexInformer
+	informerFactory            informers.SharedInformerFactory
+	computeDomainMutationCache cache.MutationCache
 }
 
 // NewPodManager creates a new PodManager instance.
-func NewPodManager(config *ManagerConfig, getComputeDomain GetComputeDomainFunc, mutationCache *cache.MutationCache) *PodManager {
+func NewPodManager(config *ManagerConfig, getComputeDomain GetComputeDomainFunc, mutationCache cache.MutationCache) *PodManager {
 	informerFactory := informers.NewSharedInformerFactoryWithOptions(
 		config.clientsets.Core,
 		informerResyncPeriod,
@@ -56,11 +56,11 @@ func NewPodManager(config *ManagerConfig, getComputeDomain GetComputeDomainFunc,
 	podInformer := informerFactory.Core().V1().Pods().Informer()
 
 	p := &PodManager{
-		config:           config,
-		getComputeDomain: getComputeDomain,
-		podInformer:      podInformer,
-		informerFactory:  informerFactory,
-		mutationCache:    *mutationCache,
+		config:                     config,
+		getComputeDomain:           getComputeDomain,
+		podInformer:                podInformer,
+		informerFactory:            informerFactory,
+		computeDomainMutationCache: mutationCache,
 	}
 
 	return p
@@ -113,6 +113,7 @@ func (pm *PodManager) Stop() error {
 	}
 
 	pm.waitGroup.Wait()
+	klog.Infof("Terminating: pod manager")
 	return nil
 }
 
@@ -188,13 +189,15 @@ func (pm *PodManager) updateNodeStatus(ctx context.Context, status string) error
 	// Update the node status to the new value
 	node.Status = status
 
-	// Update the CD status
-	if _, err := pm.config.clientsets.Nvidia.ResourceV1beta1().ComputeDomains(newCD.Namespace).UpdateStatus(ctx, newCD, metav1.UpdateOptions{}); err != nil {
+	// Update the CD status. Store the CD object returned by the API server in
+	// the mutationCache (it now has a newer ResourceVersion than `newCD` and
+	// hence any further mutations should be performed based on that).
+	newCD, err = pm.config.clientsets.Nvidia.ResourceV1beta1().ComputeDomains(newCD.Namespace).UpdateStatus(ctx, newCD, metav1.UpdateOptions{})
+	if err != nil {
 		return fmt.Errorf("error updating node status in ComputeDomain: %w", err)
 	}
+	pm.computeDomainMutationCache.Mutation(newCD)
 
 	klog.Infof("Successfully updated node %s status to %s", pm.config.nodeName, status)
-
-	pm.mutationCache.Mutation(newCD)
 	return nil
 }