internal: fix waiting for deployment get ready with new replicas

So far the old function was waiting for the deployment to be ready with
the same replicas count as the old one. It is however possible that the
new deployment version includes a different replicas field hence should
expect to a different replicas count. The old version would fail on such
case where the old count vs the new one are not the same.
This commit fixes this issue by introducing new function that accepts the
expected new replicas count and compare against it.
The old function API still exists but is redirected to use the new
function with same old behavior.
Existing tests are not affected by this.

Signed-off-by: Shereen Haj <shajmakh@redhat.com>
(cherry picked from commit f30a924)

Author: shajmakh

internal/wait/deployment.go

@@ -24,11 +24,14 @@ import (
 	"k8s.io/klog/v2"
 )
 
+// ForDeploymentComplete waits for the deployment to be complete with the number of replicas specified in the deployment spec.
 func (wt Waiter) ForDeploymentComplete(ctx context.Context, dp *appsv1.Deployment) (*appsv1.Deployment, error) {
-	// This function waits for the readiness of the pods under the deployment. The best use of this check is for
-	// completely new deployments. If the deployment exists on the cluster and simply updated, this check is
-	// not enough to guarantee that the deployment is ready with the NEW replica, thus need to cover that by
-	// additional checks as the context requires
+	return wt.ForDeploymentCompleteWithReplicas(ctx, dp, *(dp.Spec.Replicas))
+}
+
+// ForDeploymentCompleteWithReplicas waits for the deployment to be complete and have the specified number of replicas.
+// Use this function when the deployment is expected to have new number of replicas after an update.
+func (wt Waiter) ForDeploymentCompleteWithReplicas(ctx context.Context, dp *appsv1.Deployment, newExpectedReplicas int32) (*appsv1.Deployment, error) {
 	key := ObjectKeyFromObject(dp)
 	updatedDp := &appsv1.Deployment{}
 	immediate := true
@@ -39,7 +42,7 @@ func (wt Waiter) ForDeploymentComplete(ctx context.Context, dp *appsv1.Deploymen
 			return false, err
 		}
 
-		if !IsDeploymentComplete(dp, &updatedDp.Status) {
+		if !IsDeploymentComplete(dp.Generation, &updatedDp.Status, newExpectedReplicas) {
 			klog.Warningf("deployment %s not yet complete", key.String())
 			return false, nil
 		}
@@ -66,13 +69,13 @@ func areDeploymentReplicasAvailable(newStatus *appsv1.DeploymentStatus, replicas
 	return true
 }
 
-func IsDeploymentComplete(dp *appsv1.Deployment, newStatus *appsv1.DeploymentStatus) bool {
-	if newStatus.ObservedGeneration < dp.Generation {
-		klog.InfoS("generation is older than expected to indicate the deployment is complete", "old", dp.Generation, "new", newStatus.ObservedGeneration)
+func IsDeploymentComplete(oldGeneration int64, newStatus *appsv1.DeploymentStatus, expectedReplicas int32) bool {
+	if newStatus.ObservedGeneration < oldGeneration {
+		klog.InfoS("generation is older than expected to indicate the deployment is complete", "old", oldGeneration, "new", newStatus.ObservedGeneration)
 		return false
 	}
 
-	return areDeploymentReplicasAvailable(newStatus, *(dp.Spec.Replicas))
+	return areDeploymentReplicasAvailable(newStatus, expectedReplicas)
 }
 
 func (wt Waiter) ForDeploymentReplicasCreation(ctx context.Context, dp *appsv1.Deployment, expectedReplicas int32) (*appsv1.Deployment, error) {

test/e2e/sched/sched_test.go

@@ -26,20 +26,23 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/sets"
 	klog "k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/k8stopologyawareschedwg/deployer/pkg/deployer/platform"
 	"github.com/k8stopologyawareschedwg/deployer/pkg/manifests"
 
 	nropv1 "github.com/openshift-kni/numaresources-operator/api/v1"
 	"github.com/openshift-kni/numaresources-operator/internal/podlist"
 	"github.com/openshift-kni/numaresources-operator/internal/wait"
 	schedstate "github.com/openshift-kni/numaresources-operator/pkg/numaresourcesscheduler/objectstate/sched"
 	e2eclient "github.com/openshift-kni/numaresources-operator/test/internal/clients"
+	"github.com/openshift-kni/numaresources-operator/test/internal/configuration"
 	e2efixture "github.com/openshift-kni/numaresources-operator/test/internal/fixture"
 	e2eimages "github.com/openshift-kni/numaresources-operator/test/internal/images"
 	"github.com/openshift-kni/numaresources-operator/test/internal/objects"
@@ -80,7 +83,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 			dp, err := podlist.With(e2eclient.Client).DeploymentByOwnerReference(context.TODO(), nroSchedObj.UID)
 			Expect(err).ToNot(HaveOccurred(), "unable to get deployment by owner reference")
 
-			_, err = wait.With(e2eclient.Client).Timeout(5*time.Minute).Interval(10*time.Second).ForDeploymentComplete(context.TODO(), dp)
+			_, err = wait.With(e2eclient.Client).Timeout(5*time.Minute).Interval(10*time.Second).ForDeploymentCompleteWithReplicas(context.TODO(), dp, *dp.Spec.Replicas)
 			Expect(err).ToNot(HaveOccurred())
 		})
 	})
@@ -137,7 +140,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 			dp, err := podlist.With(e2eclient.Client).DeploymentByOwnerReference(context.TODO(), nroSchedObj.GetUID())
 			Expect(err).ToNot(HaveOccurred())
 
-			_, err = wait.With(e2eclient.Client).Interval(30*time.Second).Timeout(2*time.Minute).ForDeploymentComplete(context.TODO(), dp)
+			_, err = wait.With(e2eclient.Client).Interval(30*time.Second).Timeout(2*time.Minute).ForDeploymentCompleteWithReplicas(context.TODO(), dp, *dp.Spec.Replicas)
 			Expect(err).ToNot(HaveOccurred())
 		})
 
@@ -298,7 +301,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 			dp, err := podlist.With(e2eclient.Client).DeploymentByOwnerReference(context.TODO(), nroSchedObj.UID)
 			Expect(err).ToNot(HaveOccurred(), "unable to get deployment by owner reference")
 
-			dp, err = wait.With(e2eclient.Client).Timeout(5*time.Minute).Interval(10*time.Second).ForDeploymentComplete(context.TODO(), dp)
+			dp, err = wait.With(e2eclient.Client).Timeout(5*time.Minute).Interval(10*time.Second).ForDeploymentCompleteWithReplicas(context.TODO(), dp, *dp.Spec.Replicas)
 			Expect(err).ToNot(HaveOccurred())
 
 			podList, err = podlist.With(e2eclient.Client).ByDeployment(context.TODO(), *dp)
@@ -311,6 +314,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 	})
 
 	When("testing deployment TopologySpreadConstraints", func() {
+
 		var (
 			nroSchedKey               client.ObjectKey
 			autoDetectedReplicasCount *int32
@@ -326,14 +330,38 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 			Expect(err).ToNot(HaveOccurred())
 
 			if currentReplicas != nil {
+				By("compute expected autodetected replicas count")
+				var labelSelector map[string]string
+				switch configuration.Plat {
+				case platform.HyperShift:
+					labelSelector = map[string]string{
+						"node-role.kubernetes.io/worker": "",
+					}
+				case platform.OpenShift:
+					labelSelector = map[string]string{
+						"node-role.kubernetes.io/control-plane": "",
+					}
+				default:
+					Fail("unknown platform")
+				}
+
+				nodeList := &corev1.NodeList{}
+				Expect(e2eclient.Client.List(ctx, nodeList, &client.ListOptions{LabelSelector: labels.SelectorFromSet(labelSelector)})).To(Succeed())
+				Expect(nodeList.Items).ToNot(BeEmpty(), "no nodes found with control-plane or worker role")
+				expectedAutodetectedCount := int32(len(nodeList.Items))
+				if expectedAutodetectedCount > 3 {
+					// this is how it is capped in the controller
+					expectedAutodetectedCount = 3
+				}
+
 				e2efixture.By("configure NRS replicas for autodetection: current=%d desired=nil", *currentReplicas)
 				Eventually(func(g Gomega) {
 					g.Expect(e2eclient.Client.Get(ctx, nroSchedKey, nroSchedObj)).To(Succeed())
 					nroSchedObj.Spec.Replicas = nil
 					g.Expect(e2eclient.Client.Update(ctx, nroSchedObj)).To(Succeed())
 				}).WithTimeout(5*time.Minute).WithPolling(10*time.Second).Should(Succeed(), "failed to update NRS with autodetection mode for replicas")
 
-				schedDp, err = wait.With(e2eclient.Client).Timeout(5*time.Minute).Interval(10*time.Second).ForDeploymentComplete(ctx, schedDp)
+				schedDp, err = wait.With(e2eclient.Client).Timeout(5*time.Minute).Interval(10*time.Second).ForDeploymentCompleteWithReplicas(ctx, schedDp, expectedAutodetectedCount)
 				if err != nil {
 					infoUponDeploymentCompleteFailure(ctx, schedDp)
 				}
@@ -346,7 +374,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 						nroSchedObj.Spec.Replicas = currentReplicas
 						g.Expect(e2eclient.Client.Update(ctx, nroSchedObj)).To(Succeed())
 					}).WithTimeout(5*time.Minute).WithPolling(10*time.Second).Should(Succeed(), "failed to update NRS with autodetection mode for replicas")
-					schedDp, err := wait.With(e2eclient.Client).Timeout(5*time.Minute).Interval(10*time.Second).ForDeploymentComplete(ctx, schedDp)
+					schedDp, err := wait.With(e2eclient.Client).Timeout(5*time.Minute).Interval(10*time.Second).ForDeploymentCompleteWithReplicas(ctx, schedDp, *currentReplicas)
 					if err != nil {
 						infoUponDeploymentCompleteFailure(ctx, schedDp)
 					}
@@ -394,7 +422,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 				g.Expect(e2eclient.Client.Update(ctx, nroSchedObj)).To(Succeed())
 			}).WithTimeout(5*time.Minute).WithPolling(10*time.Second).Should(Succeed(), "failed to update NRS with explicit replicas count")
 
-			schedDp, err := wait.With(e2eclient.Client).Timeout(10*time.Minute).Interval(10*time.Second).ForDeploymentComplete(ctx, schedDp)
+			schedDp, err := wait.With(e2eclient.Client).Timeout(10*time.Minute).Interval(10*time.Second).ForDeploymentCompleteWithReplicas(ctx, schedDp, 0)
 			if err != nil {
 				infoUponDeploymentCompleteFailure(ctx, schedDp)
 			}
@@ -420,7 +448,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 				g.Expect(e2eclient.Client.Update(ctx, nroSchedObj)).To(Succeed())
 			}).WithTimeout(5*time.Minute).WithPolling(10*time.Second).Should(Succeed(), "failed to update NRS with autodetection mode for replicas")
 
-			schedDp, err = wait.With(e2eclient.Client).Timeout(10*time.Minute).Interval(10*time.Second).ForDeploymentComplete(ctx, schedDp)
+			schedDp, err = wait.With(e2eclient.Client).Timeout(10*time.Minute).Interval(10*time.Second).ForDeploymentCompleteWithReplicas(ctx, schedDp, *autoDetectedReplicasCount)
 			if err != nil {
 				infoUponDeploymentCompleteFailure(ctx, schedDp)
 			}
@@ -482,7 +510,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 				g.Expect(e2eclient.Client.Update(ctx, nroSchedObj)).To(Succeed())
 			}).WithTimeout(5*time.Minute).WithPolling(10*time.Second).Should(Succeed(), "failed to update NRS with explicit replicas count")
 
-			schedDp, err := wait.With(e2eclient.Client).Timeout(10*time.Minute).Interval(10*time.Second).ForDeploymentComplete(ctx, schedDp)
+			schedDp, err := wait.With(e2eclient.Client).Timeout(10*time.Minute).Interval(10*time.Second).ForDeploymentCompleteWithReplicas(ctx, schedDp, 1)
 			if err != nil {
 				infoUponDeploymentCompleteFailure(ctx, schedDp)
 			}
@@ -504,7 +532,7 @@ var _ = Describe("[Scheduler] scheduler object updates", func() {
 				g.Expect(e2eclient.Client.Update(ctx, nroSchedObj)).To(Succeed())
 			}).WithTimeout(5*time.Minute).WithPolling(10*time.Second).Should(Succeed(), "failed to update NRS with autodetection mode for replicas")
 
-			schedDp, err = wait.With(e2eclient.Client).Timeout(10*time.Minute).Interval(10*time.Second).ForDeploymentComplete(ctx, schedDp)
+			schedDp, err = wait.With(e2eclient.Client).Timeout(10*time.Minute).Interval(10*time.Second).ForDeploymentCompleteWithReplicas(ctx, schedDp, *autoDetectedReplicasCount)
 			if err != nil {
 				infoUponDeploymentCompleteFailure(ctx, schedDp)
 			}

test/e2e/serial/tests/configuration.go

@@ -918,7 +918,7 @@ var _ = Describe("[serial][disruptive] numaresources configuration management",
 				By(fmt.Sprintf("ensuring the deployment %q keep being pending %d/%d", deployment.Name, step+1, maxStep))
 				err = fxt.Client.Get(ctx, client.ObjectKeyFromObject(deployment), &updatedDeployment)
 				Expect(err).ToNot(HaveOccurred())
-				Expect(wait.IsDeploymentComplete(deployment, &updatedDeployment.Status)).To(BeFalse(), "deployment %q become ready", deployment.Name)
+				Expect(wait.IsDeploymentComplete(deployment.Generation, &updatedDeployment.Status, *(deployment.Spec.Replicas))).To(BeFalse(), "deployment %q become ready", deployment.Name)
 			}
 
 			By("checking the deployment pod has failed scheduling and its at the pending status")

test/e2e/serial/tests/scheduler_removal.go

@@ -96,7 +96,7 @@ var _ = Describe("[serial][disruptive][scheduler][schedrst] numaresources schedu
 				err = fxt.Client.Get(context.TODO(), client.ObjectKeyFromObject(dp), updatedDp)
 				Expect(err).ToNot(HaveOccurred())
 
-				Expect(wait.IsDeploymentComplete(dp, &updatedDp.Status)).To(BeTrue(), "deployment %q become unready", dp.Name)
+				Expect(wait.IsDeploymentComplete(dp.Generation, &updatedDp.Status, *dp.Spec.Replicas)).To(BeTrue(), "deployment %q become unready", dp.Name)
 			}
 		})
 
@@ -123,7 +123,7 @@ var _ = Describe("[serial][disruptive][scheduler][schedrst] numaresources schedu
 				err = fxt.Client.Get(context.TODO(), client.ObjectKeyFromObject(dp), updatedDp)
 				Expect(err).ToNot(HaveOccurred())
 
-				Expect(wait.IsDeploymentComplete(dp, &updatedDp.Status)).To(BeFalse(), "deployment %q become ready", dp.Name)
+				Expect(wait.IsDeploymentComplete(dp.Generation, &updatedDp.Status, *dp.Spec.Replicas)).To(BeFalse(), "deployment %q become ready", dp.Name)
 			}
 		})
 	})
@@ -154,7 +154,7 @@ var _ = Describe("[serial][disruptive][scheduler][schedrst] numaresources schedu
 				err = fxt.Client.Get(ctx, client.ObjectKeyFromObject(dp), updatedDp)
 				Expect(err).ToNot(HaveOccurred())
 
-				Expect(wait.IsDeploymentComplete(dp, &updatedDp.Status)).To(BeFalse(), "deployment %q become ready", dp.Name)
+				Expect(wait.IsDeploymentComplete(dp.Generation, &updatedDp.Status, *dp.Spec.Replicas)).To(BeFalse(), "deployment %q become ready", dp.Name)
 			}
 
 			restoreScheduler(fxt, nroSchedObj)

test/e2e/serial/tests/workload_placement_nodelabel.go

@@ -348,7 +348,7 @@ var _ = Describe("[serial][disruptive][scheduler] numaresources workload placeme
 						updatedDp := &appsv1.Deployment{}
 						err := fxt.Client.Get(context.TODO(), client.ObjectKeyFromObject(deployment), updatedDp)
 						Expect(err).ToNot(HaveOccurred())
-						return wait.IsDeploymentComplete(deployment, &updatedDp.Status)
+						return wait.IsDeploymentComplete(deployment.Generation, &updatedDp.Status, *deployment.Spec.Replicas)
 					}).WithTimeout(time.Second*30).WithPolling(time.Second*5).Should(BeTrue(), "deployment %q became unready", deployment.Name)
 				},
 				Entry("[test_id:47597] should be able to schedule pod with affinity property requiredDuringSchedulingIgnoredDuringExecution on the available node with feasible numa zone", createNodeAffinityRequiredDuringSchedulingIgnoredDuringExecution),