Merge pull request #42367 from kow3ns/ss-flakes

Automatic merge from submit-queue (batch tested with PRs 42443, 38924, 42367, 42391, 42310)

Fix StatefulSet e2e flake

**What this PR does / why we need it**:
Fixes StatefulSet e2e flake by ensuring that the StatefulSet controller has observed the unreadiness of Pods prior to attempting to exercise scale functionality.
**Which issue this PR fixes** 
fixes #41889
```release-note
NONE
```

test/e2e/framework/statefulset_utils.go

@@ -355,7 +355,8 @@ func (s *StatefulSetTester) SetHealthy(ss *apps.StatefulSet) {
 	}
 }
 
-func (s *StatefulSetTester) waitForStatus(ss *apps.StatefulSet, expectedReplicas int32) {
+// WaitForStatus waits for the ss.Status.Replicas to be equal to expectedReplicas
+func (s *StatefulSetTester) WaitForStatus(ss *apps.StatefulSet, expectedReplicas int32) {
 	Logf("Waiting for statefulset status.replicas updated to %d", expectedReplicas)
 
 	ns, name := ss.Namespace, ss.Name
@@ -365,6 +366,9 @@ func (s *StatefulSetTester) waitForStatus(ss *apps.StatefulSet, expectedReplicas
 			if err != nil {
 				return false, err
 			}
+			if *ssGet.Status.ObservedGeneration < ss.Generation {
+				return false, nil
+			}
 			if ssGet.Status.Replicas != expectedReplicas {
 				Logf("Waiting for stateful set status to become %d, currently %d", expectedReplicas, ssGet.Status.Replicas)
 				return false, nil
@@ -402,7 +406,7 @@ func DeleteAllStatefulSets(c clientset.Interface, ns string) {
 		if err := sst.Scale(&ss, 0); err != nil {
 			errList = append(errList, fmt.Sprintf("%v", err))
 		}
-		sst.waitForStatus(&ss, 0)
+		sst.WaitForStatus(&ss, 0)
 		Logf("Deleting statefulset %v", ss.Name)
 		if err := c.Apps().StatefulSets(ss.Namespace).Delete(ss.Name, nil); err != nil {
 			errList = append(errList, fmt.Sprintf("%v", err))

test/e2e/statefulset.go

@@ -217,6 +217,7 @@ var _ = framework.KubeDescribe("StatefulSet", func() {
 
 			By("Before scale up finished setting 2nd pod to be not ready by breaking readiness probe")
 			sst.BreakProbe(ss, testProbe)
+			sst.WaitForStatus(ss, 0)
 			sst.WaitForRunningAndNotReady(2, ss)
 
 			By("Continue scale operation after the 2nd pod, and scaling down to 1 replica")
@@ -280,6 +281,7 @@ var _ = framework.KubeDescribe("StatefulSet", func() {
 			By("Confirming that stateful set scale up will halt with unhealthy stateful pod")
 			sst.BreakProbe(ss, testProbe)
 			sst.WaitForRunningAndNotReady(*ss.Spec.Replicas, ss)
+			sst.WaitForStatus(ss, 0)
 			sst.UpdateReplicas(ss, 3)
 			sst.ConfirmStatefulPodCount(1, ss, 10*time.Second)
 
@@ -309,6 +311,7 @@ var _ = framework.KubeDescribe("StatefulSet", func() {
 			Expect(err).NotTo(HaveOccurred())
 
 			sst.BreakProbe(ss, testProbe)
+			sst.WaitForStatus(ss, 0)
 			sst.WaitForRunningAndNotReady(3, ss)
 			sst.UpdateReplicas(ss, 0)
 			sst.ConfirmStatefulPodCount(3, ss, 10*time.Second)