Merge pull request #12993 from sbueringer/pr-improve-wait-for-cache

🌱 Improve wait for cache

Author: k8s-ci-robot

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"
@@ -334,6 +335,7 @@ func (r *Reconciler) cleanupDeployment(ctx context.Context, oldMSs []*clusterv1.
 
 	sort.Sort(mdutil.MachineSetsByCreationTimestamp(cleanableMSes))
 
+	machineSetsDeleted := []*clusterv1.MachineSet{}
 	for i := range cleanableMSCount {
 		ms := cleanableMSes[i]
 		if ms.Spec.Replicas == nil {
@@ -351,10 +353,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 errors.Wrapf(err, "failed to delete MachineSet %s (cleanup of old MachineSets)", klog.KObj(ms))
 		}
+		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(fmt.Sprintf("MachineSet %s deleting (cleanup of old MachineSets)", ms.Name), "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

@@ -896,7 +896,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) {
@@ -944,7 +944,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))