Improve wait for cache

Author: sbueringer

internal/controllers/machinedeployment/machinedeployment_controller.go

@@ -354,7 +354,7 @@ func (r *Reconciler) createOrUpdateMachineSetsAndSyncMachineDeploymentRevision(c
 			// Keep trying to get the MachineSet. This will force the cache to update and prevent any future reconciliation of
 			// the MachineDeployment to reconcile with an outdated list of MachineSets which could lead to unwanted creation of
 			// a duplicate MachineSet.
-			if err := clientutil.WaitForCacheToBeUpToDate(ctx, r.Client, "MachineSet creation", ms); err != nil {
+			if err := clientutil.WaitForObjectsToBeAddedToTheCache(ctx, r.Client, "MachineSet creation", ms); err != nil {
 				return err
 			}
 

internal/controllers/machinedeployment/machinedeployment_sync.go

@@ -32,6 +32,7 @@ import (
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/core/v1beta2"
 	"sigs.k8s.io/cluster-api/internal/controllers/machinedeployment/mdutil"
+	clientutil "sigs.k8s.io/cluster-api/internal/util/client"
 	"sigs.k8s.io/cluster-api/util/collections"
 	v1beta1conditions "sigs.k8s.io/cluster-api/util/conditions/deprecated/v1beta1"
 	"sigs.k8s.io/cluster-api/util/patch"
@@ -335,6 +336,7 @@ func (r *Reconciler) cleanupDeployment(ctx context.Context, oldMSs []*clusterv1.
 	sort.Sort(mdutil.MachineSetsByCreationTimestamp(cleanableMSes))
 	log.V(4).Info("Looking to cleanup old machine sets for deployment")
 
+	machineSetsDeleted := []*clusterv1.MachineSet{}
 	for i := range cleanableMSCount {
 		ms := cleanableMSes[i]
 		if ms.Spec.Replicas == nil {
@@ -353,10 +355,12 @@ func (r *Reconciler) cleanupDeployment(ctx context.Context, oldMSs []*clusterv1.
 			r.recorder.Eventf(deployment, corev1.EventTypeWarning, "FailedDelete", "Failed to delete MachineSet %q: %v", ms.Name, err)
 			return err
 		}
+		machineSetsDeleted = append(machineSetsDeleted, ms)
+
 		// Note: We intentionally log after Delete because we want this log line to show up only after DeletionTimestamp has been set.
 		log.Info("Deleting MachineSet (cleanup of old MachineSet)", "MachineSet", klog.KObj(ms))
 		r.recorder.Eventf(deployment, corev1.EventTypeNormal, "SuccessfulDelete", "Deleted MachineSet %q", ms.Name)
 	}
 
-	return nil
+	return clientutil.WaitForObjectsToBeDeletedFromTheCache(ctx, r.Client, "MachineSet deletion (cleanup of old MachineSet)", machineSetsDeleted...)
 }

internal/controllers/machineset/machineset_controller.go

@@ -894,7 +894,7 @@ func (r *Reconciler) createMachines(ctx context.Context, s *scope, machinesToAdd
 	}
 
 	// Wait for cache update to ensure following reconcile gets latest change.
-	return ctrl.Result{}, clientutil.WaitForCacheToBeUpToDate(ctx, r.Client, "Machine creation", machinesAdded...)
+	return ctrl.Result{}, clientutil.WaitForObjectsToBeAddedToTheCache(ctx, r.Client, "Machine creation", machinesAdded...)
 }
 
 func (r *Reconciler) deleteMachines(ctx context.Context, s *scope, machinesToDelete int) (ctrl.Result, error) {
@@ -942,7 +942,7 @@ func (r *Reconciler) deleteMachines(ctx context.Context, s *scope, machinesToDel
 	}
 
 	// Wait for cache update to ensure following reconcile gets latest change.
-	if err := clientutil.WaitForObjectsToBeDeletedFromTheCache(ctx, r.Client, "Machine deletion", machinesDeleted...); err != nil {
+	if err := clientutil.WaitForObjectsToBeDeletedFromTheCache(ctx, r.Client, "Machine deletion (scale down)", machinesDeleted...); err != nil {
 		errs = append(errs, err)
 	}
 	if len(errs) > 0 {

internal/controllers/topology/cluster/reconcile_state.go

@@ -693,7 +693,7 @@ func (r *Reconciler) createMachineDeployment(ctx context.Context, s *scope.Scope
 	// Wait until MachineDeployment is visible in the cache.
 	// Note: We have to do this because otherwise using a cached client in current state could
 	// miss a newly created MachineDeployment (because the cache might be stale).
-	if err := clientutil.WaitForCacheToBeUpToDate(ctx, r.Client, "MachineDeployment creation", md.Object); err != nil {
+	if err := clientutil.WaitForObjectsToBeAddedToTheCache(ctx, r.Client, "MachineDeployment creation", md.Object); err != nil {
 		return err
 	}
 
@@ -1016,7 +1016,7 @@ func (r *Reconciler) createMachinePool(ctx context.Context, s *scope.Scope, mp *
 	// Wait until MachinePool is visible in the cache.
 	// Note: We have to do this because otherwise using a cached client in current state could
 	// miss a newly created MachinePool (because the cache might be stale).
-	return clientutil.WaitForCacheToBeUpToDate(ctx, r.Client, "MachinePool creation", mp.Object)
+	return clientutil.WaitForObjectsToBeAddedToTheCache(ctx, r.Client, "MachinePool creation", mp.Object)
 }
 
 // updateMachinePool updates a MachinePool. Also updates the corresponding objects if necessary.

internal/util/client/client.go

@@ -49,11 +49,20 @@ var (
 // all passed in objects with at least the passed in resourceVersion.
 // This is done by retrieving objects from the cache via the client and then comparing resourceVersions.
 // Note: This func will update the passed in objects while polling.
+// Note: resourceVersion must be set on the passed in objects.
 // Note: The generic parameter enforces that all objects have the same type.
 func WaitForCacheToBeUpToDate[T client.Object](ctx context.Context, c client.Client, action string, objs ...T) error {
 	return waitFor(ctx, c, action, checkIfObjectUpToDate, objs...)
 }
 
+// WaitForObjectsToBeAddedToTheCache waits until the cache is up-to-date in the sense of that the
+// passed in objects exist in the cache.
+// Note: This func will update the passed in objects while polling.
+// Note: The generic parameter enforces that all objects have the same type.
+func WaitForObjectsToBeAddedToTheCache[T client.Object](ctx context.Context, c client.Client, action string, objs ...T) error {
+	return waitFor(ctx, c, action, checkIfObjectAdded, objs...)
+}
+
 // WaitForObjectsToBeDeletedFromTheCache waits until the cache is up-to-date in the sense of that the
 // passed in objects have been either removed from the cache or they have a deletionTimestamp set.
 // Note: This func will update the passed in objects while polling.
@@ -64,22 +73,21 @@ func WaitForObjectsToBeDeletedFromTheCache[T client.Object](ctx context.Context,
 
 // checkIfObjectUpToDate checks if an object is up-to-date and returns an error if it is not.
 func checkIfObjectUpToDate(ctx context.Context, c client.Client, desiredObj desiredObject) (isErrorRetryable bool, err error) {
+	if desiredObj.MinimumResourceVersion == "" {
+		// Unexpected error occurred: resourceVersion not set on passed in object (not retryable).
+		return false, errors.Errorf("%s: cannot compare with invalid resourceVersion: resourceVersion not set",
+			klog.KObj(desiredObj.Object))
+	}
+
 	if err := c.Get(ctx, desiredObj.Key, desiredObj.Object); err != nil {
 		if apierrors.IsNotFound(err) {
-			// Object is not yet in the cache (retryable).
-			return true, err
+			// Done, object was deleted in the meantime.
+			return false, nil
 		}
 		// Unexpected error occurred (not retryable).
 		return false, err
 	}
 
-	if desiredObj.MinimumResourceVersion == "" {
-		// Done, if MinimumResourceVersion is empty, as it is enough if the object exists in the cache.
-		// Note: This can happen when the ServerSidePatchHelper is used to create an object as the ServerSidePatchHelper
-		//       does not update the object after Apply and accordingly resourceVersion remains empty.
-		return false, nil
-	}
-
 	cmp, err := compareResourceVersion(desiredObj.Object.GetResourceVersion(), desiredObj.MinimumResourceVersion)
 	if err != nil {
 		// Unexpected error occurred: invalid resourceVersion (not retryable).
@@ -96,6 +104,20 @@ func checkIfObjectUpToDate(ctx context.Context, c client.Client, desiredObj desi
 	return false, nil
 }
 
+func checkIfObjectAdded(ctx context.Context, c client.Client, desiredObj desiredObject) (isErrorRetryable bool, err error) {
+	if err := c.Get(ctx, desiredObj.Key, desiredObj.Object); err != nil {
+		if apierrors.IsNotFound(err) {
+			// Object is not yet in the cache (retryable).
+			return true, err
+		}
+		// Unexpected error occurred (not retryable).
+		return false, err
+	}
+
+	// Done, object exists in the cache.
+	return false, nil
+}
+
 func checkIfObjectDeleted(ctx context.Context, c client.Client, desiredObj desiredObject) (isErrorRetryable bool, err error) {
 	if err := c.Get(ctx, desiredObj.Key, desiredObj.Object); err != nil {
 		if apierrors.IsNotFound(err) {

internal/util/client/client_test.go

@@ -56,6 +56,23 @@ func Test_WaitForCacheToBeUpToDate(t *testing.T) {
 		{
 			name: "no-op if no objects are passed in",
 		},
+		{
+			name: "error if passed in objects have no resourceVersion set",
+			objs: []client.Object{
+				machine("machine-1", "", nil),
+				machine("machine-2", "", nil),
+			},
+			clientResponses: map[client.ObjectKey][]client.Object{
+				{Namespace: metav1.NamespaceDefault, Name: "machine-1"}: {
+					machine("machine-1", "1", nil),
+				},
+				{Namespace: metav1.NamespaceDefault, Name: "machine-2"}: {
+					machine("machine-2", "2", nil),
+				},
+			},
+			wantErr: "failed to wait for up-to-date Machine objects in the cache after Machine update: " +
+				"default/machine-1: cannot compare with invalid resourceVersion: resourceVersion not set",
+		},
 		{
 			name: "error if passed in objects have invalid resourceVersion",
 			objs: []client.Object{
@@ -70,7 +87,7 @@ func Test_WaitForCacheToBeUpToDate(t *testing.T) {
 					machine("machine-2", "2", nil),
 				},
 			},
-			wantErr: "failed to wait for up-to-date Machine objects in the cache after Machine creation: " +
+			wantErr: "failed to wait for up-to-date Machine objects in the cache after Machine update: " +
 				"default/machine-1: cannot compare with invalid resourceVersion: current: 1, expected to be >= invalidResourceVersion: resource version is not well formed: invalidResourceVersion",
 		},
 		{
@@ -87,35 +104,30 @@ func Test_WaitForCacheToBeUpToDate(t *testing.T) {
 					machine("machine-2", "invalidResourceVersion", nil),
 				},
 			},
-			wantErr: "failed to wait for up-to-date Machine objects in the cache after Machine creation: " +
+			wantErr: "failed to wait for up-to-date Machine objects in the cache after Machine update: " +
 				"default/machine-1: cannot compare with invalid resourceVersion: current: invalidResourceVersion, expected to be >= 1: resource version is not well formed: invalidResourceVersion",
 		},
 		{
-			name: "error if objects never show up in the cache",
+			name: "success if objects are never visible in the cache (deleted before WaitForCacheToBeUpToDate is called)",
 			objs: []client.Object{
 				machine("machine-1", "1", nil),
 				machine("machine-2", "2", nil),
 				machine("machine-3", "3", nil),
 				machine("machine-4", "4", nil),
 			},
 			clientResponses: map[client.ObjectKey][]client.Object{},
-			wantErr: "failed to wait for up-to-date Machine objects in the cache after Machine creation: timed out: [" +
-				"machines.cluster.x-k8s.io \"machine-1\" not found, " +
-				"machines.cluster.x-k8s.io \"machine-2\" not found, " +
-				"machines.cluster.x-k8s.io \"machine-3\" not found, " +
-				"machines.cluster.x-k8s.io \"machine-4\" not found]",
 		},
 		{
 			name: "success if objects are instantly up-to-date",
 			objs: []client.Object{
-				machine("machine-1", "", nil),
+				machine("machine-1", "1", nil),
 				machine("machine-2", "2", nil),
 				machine("machine-3", "3", nil),
 				machine("machine-4", "4", nil),
 			},
 			clientResponses: map[client.ObjectKey][]client.Object{
 				{Namespace: metav1.NamespaceDefault, Name: "machine-1"}: {
-					// For this object it's enough if it shows up, exact resourceVersion doesn't matter.
+					// This object has an even newer resourceVersion.
 					machine("machine-1", "5", nil),
 				},
 				{Namespace: metav1.NamespaceDefault, Name: "machine-2"}: {
@@ -133,14 +145,13 @@ func Test_WaitForCacheToBeUpToDate(t *testing.T) {
 		{
 			name: "success if objects are up-to-date after a few tries",
 			objs: []client.Object{
-				machine("machine-1", "", nil),
+				machine("machine-1", "1", nil),
 				machine("machine-2", "10", nil),
 				machine("machine-3", "11", nil),
 				machine("machine-4", "12", nil),
 			},
 			clientResponses: map[client.ObjectKey][]client.Object{
 				{Namespace: metav1.NamespaceDefault, Name: "machine-1"}: {
-					// For this object it's enough if it shows up, exact resourceVersion doesn't matter.
 					machine("machine-1", "4", nil),
 				},
 				{Namespace: metav1.NamespaceDefault, Name: "machine-2"}: {
@@ -195,7 +206,109 @@ func Test_WaitForCacheToBeUpToDate(t *testing.T) {
 				},
 			})
 
-			err := WaitForCacheToBeUpToDate(t.Context(), fakeClient, "Machine creation", tt.objs...)
+			err := WaitForCacheToBeUpToDate(t.Context(), fakeClient, "Machine update", tt.objs...)
+			if tt.wantErr != "" {
+				g.Expect(err).To(HaveOccurred())
+				g.Expect(err.Error()).To(Equal(tt.wantErr))
+			} else {
+				g.Expect(err).ToNot(HaveOccurred())
+			}
+		})
+	}
+}
+
+func Test_WaitForObjectsToBeAddedToTheCache(t *testing.T) {
+	// Modify timeout to speed up test
+	waitBackoff = wait.Backoff{
+		Duration: 25 * time.Microsecond,
+		Cap:      2 * time.Second,
+		Factor:   1.2,
+		Steps:    5,
+	}
+
+	tests := []struct {
+		name            string
+		objs            []client.Object
+		clientResponses map[client.ObjectKey][]client.Object
+		wantErr         string
+	}{
+		{
+			name: "no-op if no objects are passed in",
+		},
+		{
+			name: "error if objects never show up in the cache",
+			objs: []client.Object{
+				machine("machine-1", "1", nil),
+				machine("machine-2", "2", nil),
+				machine("machine-3", "3", nil),
+				machine("machine-4", "4", nil),
+			},
+			clientResponses: map[client.ObjectKey][]client.Object{},
+			wantErr: "failed to wait for up-to-date Machine objects in the cache after Machine creation: timed out: [" +
+				"machines.cluster.x-k8s.io \"machine-1\" not found, " +
+				"machines.cluster.x-k8s.io \"machine-2\" not found, " +
+				"machines.cluster.x-k8s.io \"machine-3\" not found, " +
+				"machines.cluster.x-k8s.io \"machine-4\" not found]",
+		},
+		{
+			name: "success if objects instantly show up in the cache",
+			objs: []client.Object{
+				machine("machine-1", "1", nil),
+				machine("machine-2", "2", nil),
+				machine("machine-3", "3", nil),
+				machine("machine-4", "4", nil),
+			},
+			clientResponses: map[client.ObjectKey][]client.Object{
+				{Namespace: metav1.NamespaceDefault, Name: "machine-1"}: {
+					// This object has an even newer resourceVersion.
+					machine("machine-1", "5", nil),
+				},
+				{Namespace: metav1.NamespaceDefault, Name: "machine-2"}: {
+					machine("machine-2", "2", nil),
+				},
+				{Namespace: metav1.NamespaceDefault, Name: "machine-3"}: {
+					machine("machine-3", "3", nil),
+				},
+				{Namespace: metav1.NamespaceDefault, Name: "machine-4"}: {
+					// This object has an even newer resourceVersion.
+					machine("machine-4", "6", nil),
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			scheme := runtime.NewScheme()
+			_ = clusterv1.AddToScheme(scheme)
+
+			callCounter := map[client.ObjectKey]int{}
+			fakeClient := interceptor.NewClient(fake.NewClientBuilder().WithScheme(scheme).Build(), interceptor.Funcs{
+				Get: func(ctx context.Context, _ client.WithWatch, key client.ObjectKey, obj client.Object, _ ...client.GetOption) error {
+					if len(tt.clientResponses) == 0 || len(tt.clientResponses[key]) == 0 {
+						return apierrors.NewNotFound(schema.GroupResource{
+							Group:    clusterv1.GroupVersion.Group,
+							Resource: "machines",
+						}, key.Name)
+					}
+
+					currentCall := callCounter[key]
+					currentCall = min(currentCall, len(tt.clientResponses[key])-1)
+
+					// Write back the modified object so callers can access the patched object.
+					if err := scheme.Convert(tt.clientResponses[key][currentCall], obj, ctx); err != nil {
+						return errors.Wrapf(err, "unexpected error: failed to get")
+					}
+
+					callCounter[key]++
+
+					return nil
+				},
+			})
+
+			err := WaitForObjectsToBeAddedToTheCache(t.Context(), fakeClient, "Machine creation", tt.objs...)
 			if tt.wantErr != "" {
 				g.Expect(err).To(HaveOccurred())
 				g.Expect(err.Error()).To(Equal(tt.wantErr))