Refactor shutdown e2e tests

test/e2e/framework/framework.go

@@ -19,7 +19,6 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-	"os/exec"
 	"strings"
 	"time"
 
@@ -34,6 +33,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	networking "k8s.io/api/networking/v1beta1"
 	apiextcs "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/util/intstr"
@@ -502,14 +502,12 @@ func UpdateDeployment(kubeClientSet kubernetes.Interface, namespace string, name
 		return err
 	}
 
-	rolloutStatsCmd := fmt.Sprintf("%v --namespace %s rollout status deployment/%s -w --timeout 5m", KubectlPath, namespace, deployment.Name)
-
 	if updateFunc != nil {
 		if err := updateFunc(deployment); err != nil {
 			return err
 		}
 
-		err = exec.Command("bash", "-c", rolloutStatsCmd).Run()
+		err = waitForDeploymentRollout(kubeClientSet, deployment)
 		if err != nil {
 			return err
 		}
@@ -522,7 +520,7 @@ func UpdateDeployment(kubeClientSet kubernetes.Interface, namespace string, name
 			return errors.Wrapf(err, "scaling the number of replicas to %v", replicas)
 		}
 
-		err = exec.Command("/bin/bash", "-c", rolloutStatsCmd).Run()
+		err = waitForDeploymentRollout(kubeClientSet, deployment)
 		if err != nil {
 			return err
 		}
@@ -538,6 +536,29 @@ func UpdateDeployment(kubeClientSet kubernetes.Interface, namespace string, name
 	return nil
 }
 
+func waitForDeploymentRollout(kubeClientSet kubernetes.Interface, resource *appsv1.Deployment) error {
+	return wait.Poll(Poll, 5*time.Minute, func() (bool, error) {
+		d, err := kubeClientSet.AppsV1().Deployments(resource.Namespace).Get(context.TODO(), resource.Name, metav1.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			return false, nil
+		}
+
+		if err != nil {
+			return false, nil
+		}
+
+		if d.DeletionTimestamp != nil {
+			return false, fmt.Errorf("deployment %q is being deleted", resource.Name)
+		}
+
+		if d.Generation <= d.Status.ObservedGeneration && d.Status.UpdatedReplicas == d.Status.Replicas && d.Status.UnavailableReplicas == 0 {
+			return true, nil
+		}
+
+		return false, nil
+	})
+}
+
 // UpdateIngress runs the given updateFunc on the ingress
 func UpdateIngress(kubeClientSet kubernetes.Interface, namespace string, name string, updateFunc func(d *networking.Ingress) error) error {
 	ingress, err := kubeClientSet.NetworkingV1beta1().Ingresses(namespace).Get(context.TODO(), name, metav1.GetOptions{})

test/e2e/framework/k8s.go

@@ -166,7 +166,7 @@ func waitForPodsDeleted(kubeClientSet kubernetes.Interface, timeout time.Duratio
 	})
 }
 
-// WaitForEndpoints waits for a given amount of time until an endpoint contains.
+// WaitForEndpoints waits for a given amount of time until the number of endpoints = expectedEndpoints.
 func WaitForEndpoints(kubeClientSet kubernetes.Interface, timeout time.Duration, name, ns string, expectedEndpoints int) error {
 	if expectedEndpoints == 0 {
 		return nil
@@ -180,16 +180,7 @@ func WaitForEndpoints(kubeClientSet kubernetes.Interface, timeout time.Duration,
 
 		assert.Nil(ginkgo.GinkgoT(), err, "getting endpoints")
 
-		if len(endpoint.Subsets) == 0 || len(endpoint.Subsets[0].Addresses) == 0 {
-			return false, nil
-		}
-
-		r := 0
-		for _, es := range endpoint.Subsets {
-			r += len(es.Addresses)
-		}
-
-		if r == expectedEndpoints {
+		if countReadyEndpoints(endpoint) == expectedEndpoints {
 			return true, nil
 		}
 
@@ -197,6 +188,19 @@ func WaitForEndpoints(kubeClientSet kubernetes.Interface, timeout time.Duration,
 	})
 }
 
+func countReadyEndpoints(e *core.Endpoints) int {
+	if e == nil || e.Subsets == nil {
+		return 0
+	}
+
+	num := 0
+	for _, sub := range e.Subsets {
+		num += len(sub.Addresses)
+	}
+
+	return num
+}
+
 // podRunningReady checks whether pod p's phase is running and it has a ready
 // condition of status true.
 func podRunningReady(p *core.Pod) (bool, error) {