Wrap K8s client in E2E with retry logic

test/e2e/adminapi_cluster_getlogs.go

@@ -46,20 +46,22 @@ func testGetPodLogsOK(ctx context.Context, containerName, podName, namespace str
 
 	By("creating a test pod in openshift-azure-operator namespace with some known logs")
 	pod := mockPod(containerName, podName, namespace, expectedLog)
-	pod, err := clients.Kubernetes.CoreV1().Pods(namespace).Create(ctx, pod, metav1.CreateOptions{})
-	Expect(err).NotTo(HaveOccurred())
+	pod = CreateK8sObjectWithRetry(
+		ctx, clients.Kubernetes.CoreV1().Pods(namespace).Create, pod, metav1.CreateOptions{},
+	)
 
 	defer func() {
 		By("deleting the test pod")
-		err = clients.Kubernetes.CoreV1().Pods(namespace).Delete(ctx, pod.Name, metav1.DeleteOptions{})
-		Expect(err).NotTo(HaveOccurred())
+		DeleteK8sObjectWithRetry(
+			ctx, clients.Kubernetes.CoreV1().Pods(namespace).Delete, pod.Name, metav1.DeleteOptions{},
+		)
 	}()
 
 	By("waiting for the pod to successfully terminate")
 	Eventually(func(g Gomega, ctx context.Context) {
-		pod, err = clients.Kubernetes.CoreV1().Pods(namespace).Get(ctx, pod.Name, metav1.GetOptions{})
-		g.Expect(err).NotTo(HaveOccurred())
-
+		pod = GetK8sObjectWithRetry(
+			ctx, clients.Kubernetes.CoreV1().Pods(namespace).Get, pod.Name, metav1.GetOptions{},
+		)
 		g.Expect(pod.Status.Phase).To(Equal(corev1.PodSucceeded))
 	}).WithContext(ctx).WithTimeout(DefaultEventuallyTimeout).Should(Succeed())
 

test/e2e/adminapi_csr.go

@@ -37,16 +37,17 @@ var _ = Describe("[Admin API] CertificateSigningRequest action", func() {
 		By("creating mock CSRs via Kubernetes API")
 		for i := 0; i < csrCount; i++ {
 			csr := mockCSR(prefix+strconv.Itoa(i), namespace, csrData)
-			_, err := clients.Kubernetes.CertificatesV1().CertificateSigningRequests().Create(ctx, csr, metav1.CreateOptions{})
-			Expect(err).NotTo(HaveOccurred())
+
+			createCall := clients.Kubernetes.CertificatesV1().CertificateSigningRequests().Create
+			CreateK8sObjectWithRetry(ctx, createCall, csr, metav1.CreateOptions{})
 		}
 	})
 
 	AfterEach(func(ctx context.Context) {
 		By("deleting the mock CSRs via Kubernetes API")
 		for i := 0; i < csrCount; i++ {
-			err := clients.Kubernetes.CertificatesV1().CertificateSigningRequests().Delete(ctx, prefix+strconv.Itoa(i), metav1.DeleteOptions{})
-			Expect(err).NotTo(HaveOccurred())
+			deleteCall := clients.Kubernetes.CertificatesV1().CertificateSigningRequests().Delete
+			DeleteK8sObjectWithRetry(ctx, deleteCall, prefix+strconv.Itoa(i), metav1.DeleteOptions{})
 		}
 	})
 
@@ -70,8 +71,8 @@ func testCSRApproveOK(ctx context.Context, objName, namespace string) {
 	Expect(resp.StatusCode).To(Equal(http.StatusOK))
 
 	By("checking that the CSR was approved via Kubernetes API")
-	testcsr, err := clients.Kubernetes.CertificatesV1().CertificateSigningRequests().Get(ctx, objName, metav1.GetOptions{})
-	Expect(err).NotTo(HaveOccurred())
+	getCall := clients.Kubernetes.CertificatesV1().CertificateSigningRequests().Get
+	testcsr := GetK8sObjectWithRetry(ctx, getCall, objName, metav1.GetOptions{})
 
 	approved := false
 	for _, condition := range testcsr.Status.Conditions {
@@ -93,8 +94,8 @@ func testCSRMassApproveOK(ctx context.Context, namePrefix, namespace string, csr
 
 	By("checking that all CSRs were approved via Kubernetes API")
 	for i := 1; i < csrCount; i++ {
-		testcsr, err := clients.Kubernetes.CertificatesV1().CertificateSigningRequests().Get(ctx, namePrefix+strconv.Itoa(i), metav1.GetOptions{})
-		Expect(err).NotTo(HaveOccurred())
+		getCall := clients.Kubernetes.CertificatesV1().CertificateSigningRequests().Get
+		testcsr := GetK8sObjectWithRetry(ctx, getCall, namePrefix+strconv.Itoa(i), metav1.GetOptions{})
 
 		approved := false
 		for _, condition := range testcsr.Status.Conditions {

test/e2e/adminapi_delete_managedresource.go

@@ -46,8 +46,8 @@ var _ = Describe("[Admin API] Delete managed resource action", func() {
 		var pipAddressID string
 
 		By("creating a test service of type loadbalancer")
-		_, err := clients.Kubernetes.CoreV1().Services("default").Create(ctx, &loadBalancerService, metav1.CreateOptions{})
-		Expect(err).NotTo(HaveOccurred())
+		creationCall := clients.Kubernetes.CoreV1().Services("default").Create
+		CreateK8sObjectWithRetry(ctx, creationCall, &loadBalancerService, metav1.CreateOptions{})
 
 		defer func() {
 			By("cleaning up the k8s loadbalancer service")
@@ -66,8 +66,8 @@ var _ = Describe("[Admin API] Delete managed resource action", func() {
 
 		// wait for ingress IP to be assigned as this indicate the service is ready
 		Eventually(func(g Gomega, ctx context.Context) {
-			service, err = clients.Kubernetes.CoreV1().Services("default").Get(ctx, "test", metav1.GetOptions{})
-			g.Expect(err).NotTo(HaveOccurred())
+			getCall := clients.Kubernetes.CoreV1().Services("default").Get
+			service = GetK8sObjectWithRetry(ctx, getCall, "test", metav1.GetOptions{})
 			g.Expect(service.Status.LoadBalancer.Ingress).To(HaveLen(1))
 		}).WithContext(ctx).WithTimeout(DefaultEventuallyTimeout).Should(Succeed())
 

test/e2e/adminapi_drainnode.go

@@ -22,10 +22,13 @@ var _ = Describe("[Admin API] Cordon and Drain node actions", func() {
 
 	It("should be able to cordon, drain, and uncordon nodes", func(ctx context.Context) {
 		By("selecting a worker node in the cluster")
-		nodes, err := clients.Kubernetes.CoreV1().Nodes().List(ctx, metav1.ListOptions{
-			LabelSelector: "node-role.kubernetes.io/worker",
-		})
-		Expect(err).NotTo(HaveOccurred())
+		nodes := ListK8sObjectWithRetry(
+			ctx,
+			clients.Kubernetes.CoreV1().Nodes().List,
+			metav1.ListOptions{
+				LabelSelector: "node-role.kubernetes.io/worker",
+			})
+
 		Expect(nodes.Items).ShouldNot(BeEmpty())
 		node := nodes.Items[0]
 		nodeName := node.Name
@@ -48,7 +51,7 @@ var _ = Describe("[Admin API] Cordon and Drain node actions", func() {
 
 		defer func() {
 			By("uncordoning the node via Kubernetes API")
-			err = drain.RunCordonOrUncordon(drainer, &node, false)
+			err := drain.RunCordonOrUncordon(drainer, &node, false)
 			Expect(err).NotTo(HaveOccurred())
 		}()
 
@@ -69,8 +72,8 @@ func testCordonNodeOK(ctx context.Context, nodeName string) {
 	Expect(resp.StatusCode).To(Equal(http.StatusOK))
 
 	By("checking that node was cordoned via Kubernetes API")
-	node, err := clients.Kubernetes.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
-	Expect(err).NotTo(HaveOccurred())
+	node := GetK8sObjectWithRetry(ctx, clients.Kubernetes.CoreV1().Nodes().Get, nodeName, metav1.GetOptions{})
+
 	Expect(node.Name).To(Equal(nodeName))
 	Expect(node.Spec.Unschedulable).Should(BeTrue())
 }
@@ -86,8 +89,7 @@ func testUncordonNodeOK(ctx context.Context, nodeName string) {
 	Expect(resp.StatusCode).To(Equal(http.StatusOK))
 
 	By("checking that node was uncordoned via Kubernetes API")
-	node, err := clients.Kubernetes.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
-	Expect(err).NotTo(HaveOccurred())
+	node := GetK8sObjectWithRetry(ctx, clients.Kubernetes.CoreV1().Nodes().Get, nodeName, metav1.GetOptions{})
 	Expect(node.Name).To(Equal(nodeName))
 	Expect(node.Spec.Unschedulable).Should(BeFalse())
 }

test/e2e/adminapi_kubernetesobjects.go

@@ -67,15 +67,15 @@ var _ = Describe("[Admin API] Kubernetes objects action", func() {
 
 			It("should not be able to create, get, list, update, or delete objects", func(ctx context.Context) {
 				By("creating a test customer namespace via Kubernetes API")
-				_, err := clients.Kubernetes.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{
-					ObjectMeta: metav1.ObjectMeta{Name: namespace},
-				}, metav1.CreateOptions{})
-				Expect(err).NotTo(HaveOccurred())
+				createNamespaceCall := clients.Kubernetes.CoreV1().Namespaces().Create
+				namespaceToCreate := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
+
+				CreateK8sObjectWithRetry(ctx, createNamespaceCall, namespaceToCreate, metav1.CreateOptions{})
 
 				defer func() {
 					By("deleting the test customer namespace via Kubernetes API")
-					err := clients.Kubernetes.CoreV1().Namespaces().Delete(ctx, namespace, metav1.DeleteOptions{})
-					Expect(err).NotTo(HaveOccurred())
+					deleteCall := clients.Kubernetes.CoreV1().Namespaces().Delete
+					DeleteK8sObjectWithRetry(ctx, deleteCall, namespace, metav1.DeleteOptions{})
 
 					// To avoid flakes, we need it to be completely deleted before we can use it again
 					// in a separate run or in a separate It block
@@ -89,10 +89,10 @@ var _ = Describe("[Admin API] Kubernetes objects action", func() {
 				testConfigMapCreateOrUpdateForbidden(ctx, "creating", objName, namespace)
 
 				By("creating an object via Kubernetes API")
-				_, err = clients.Kubernetes.CoreV1().ConfigMaps(namespace).Create(ctx, &corev1.ConfigMap{
-					ObjectMeta: metav1.ObjectMeta{Name: objName},
-				}, metav1.CreateOptions{})
-				Expect(err).NotTo(HaveOccurred())
+				createCMCall := clients.Kubernetes.CoreV1().ConfigMaps(namespace).Create
+				configMapToCreate := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: objName}}
+
+				CreateK8sObjectWithRetry(ctx, createCMCall, configMapToCreate, metav1.CreateOptions{})
 
 				testConfigMapGetForbidden(ctx, objName, namespace)
 				testConfigMapListForbidden(ctx, objName, namespace)
@@ -105,15 +105,15 @@ var _ = Describe("[Admin API] Kubernetes objects action", func() {
 
 			It("should be able to list or get objects", func(ctx context.Context) {
 				By("creating a test customer namespace via Kubernetes API")
-				_, err := clients.Kubernetes.CoreV1().Namespaces().Create(ctx, &corev1.Namespace{
-					ObjectMeta: metav1.ObjectMeta{Name: namespace},
-				}, metav1.CreateOptions{})
-				Expect(err).NotTo(HaveOccurred())
+				createNamespaceCall := clients.Kubernetes.CoreV1().Namespaces().Create
+				namespaceToCreate := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: namespace}}
+
+				CreateK8sObjectWithRetry(ctx, createNamespaceCall, namespaceToCreate, metav1.CreateOptions{})
 
 				defer func() {
 					By("deleting the test customer namespace via Kubernetes API")
-					err := clients.Kubernetes.CoreV1().Namespaces().Delete(ctx, namespace, metav1.DeleteOptions{})
-					Expect(err).NotTo(HaveOccurred())
+					deleteCall := clients.Kubernetes.CoreV1().Namespaces().Delete
+					DeleteK8sObjectWithRetry(ctx, deleteCall, namespace, metav1.DeleteOptions{})
 
 					// To avoid flakes, we need it to be completely deleted before we can use it again
 					// in a separate run or in a separate It block
@@ -125,10 +125,10 @@ var _ = Describe("[Admin API] Kubernetes objects action", func() {
 				}()
 
 				By("creating an object via Kubernetes API")
-				_, err = clients.Kubernetes.CoreV1().ConfigMaps(namespace).Create(ctx, &corev1.ConfigMap{
-					ObjectMeta: metav1.ObjectMeta{Name: objName},
-				}, metav1.CreateOptions{})
-				Expect(err).NotTo(HaveOccurred())
+				createCMCall := clients.Kubernetes.CoreV1().ConfigMaps(namespace).Create
+				configMapToCreate := &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: objName}}
+
+				CreateK8sObjectWithRetry(ctx, createCMCall, configMapToCreate, metav1.CreateOptions{})
 
 				testConfigMapGetOK(ctx, objName, namespace, true)
 				testConfigMapListOK(ctx, objName, namespace, true)
@@ -204,8 +204,10 @@ func testConfigMapCreateOK(ctx context.Context, objName, namespace string) {
 	Expect(resp.StatusCode).To(Equal(http.StatusOK))
 
 	By("checking that the object was created via Kubernetes API")
-	cm, err := clients.Kubernetes.CoreV1().ConfigMaps(namespace).Get(ctx, objName, metav1.GetOptions{})
-	Expect(err).NotTo(HaveOccurred())
+	getCall := clients.Kubernetes.CoreV1().ConfigMaps(namespace).Get
+
+	cm := GetK8sObjectWithRetry(ctx, getCall, objName, metav1.GetOptions{})
+
 	Expect(obj.Namespace).To(Equal(cm.Namespace))
 	Expect(obj.Name).To(Equal(cm.Name))
 	Expect(obj.Data).To(Equal(cm.Data))
@@ -225,9 +227,11 @@ func testConfigMapGetOK(ctx context.Context, objName, namespace string, unrestri
 	Expect(err).NotTo(HaveOccurred())
 	Expect(resp.StatusCode).To(Equal(http.StatusOK))
 
-	By("comparing it to the actual object retrived via Kubernetes API")
-	cm, err := clients.Kubernetes.CoreV1().ConfigMaps(namespace).Get(ctx, objName, metav1.GetOptions{})
-	Expect(err).NotTo(HaveOccurred())
+	By("comparing it to the actual object retrieved via Kubernetes API")
+	getCall := clients.Kubernetes.CoreV1().ConfigMaps(namespace).Get
+
+	cm := GetK8sObjectWithRetry(ctx, getCall, objName, metav1.GetOptions{})
+
 	Expect(obj.Namespace).To(Equal(cm.Namespace))
 	Expect(obj.Name).To(Equal(cm.Name))
 	Expect(obj.Data).To(Equal(cm.Data))
@@ -246,7 +250,7 @@ func testConfigMapListOK(ctx context.Context, objName, namespace string, unrestr
 	Expect(err).NotTo(HaveOccurred())
 	Expect(resp.StatusCode).To(Equal(http.StatusOK))
 
-	By("comparing names from the list action response with names retrived via Kubernetes API")
+	By("comparing names from the list action response with names retrieved via Kubernetes API")
 	var names []string
 	for _, o := range obj.Items {
 		names = append(names, o.Name)
@@ -264,8 +268,10 @@ func testConfigMapUpdateOK(ctx context.Context, objName, namespace string) {
 	Expect(resp.StatusCode).To(Equal(http.StatusOK))
 
 	By("checking that the object changed via Kubernetes API")
-	cm, err := clients.Kubernetes.CoreV1().ConfigMaps(namespace).Get(ctx, objName, metav1.GetOptions{})
-	Expect(err).NotTo(HaveOccurred())
+	getCall := clients.Kubernetes.CoreV1().ConfigMaps(namespace).Get
+
+	cm := GetK8sObjectWithRetry(ctx, getCall, objName, metav1.GetOptions{})
+
 	Expect(cm.Namespace).To(Equal(namespace))
 	Expect(cm.Name).To(Equal(objName))
 	Expect(cm.Data).To(Equal(map[string]string{"key": "new_value"}))
@@ -367,8 +373,10 @@ func testPodCreateOK(ctx context.Context, containerName, objName, namespace stri
 	Expect(resp.StatusCode).To(Equal(http.StatusOK))
 
 	By("checking that the pod was created via Kubernetes API")
-	pod, err := clients.Kubernetes.CoreV1().Pods(namespace).Get(ctx, objName, metav1.GetOptions{})
-	Expect(err).NotTo(HaveOccurred())
+	getCall := clients.Kubernetes.CoreV1().Pods(namespace).Get
+
+	pod := GetK8sObjectWithRetry(ctx, getCall, objName, metav1.GetOptions{})
+
 	Expect(obj.Namespace).To(Equal(pod.Namespace))
 	Expect(obj.Name).To(Equal(pod.Name))
 	Expect(obj.Spec.Containers[0].Name).To(Equal(pod.Spec.Containers[0].Name))

test/e2e/adminapi_redeployvm.go

@@ -17,6 +17,7 @@ import (
 
 	mgmtcompute "github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2020-06-01/compute"
 	corev1 "k8s.io/api/core/v1"
+	kerrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/Azure/ARO-RP/pkg/util/ready"
@@ -63,8 +64,8 @@ var _ = Describe("[Admin API] VM redeploy action", func() {
 		By("waiting for the redeployed node to eventually become Ready in OpenShift")
 		// wait 1 minute - this will guarantee we pass the minimum (default) threshold of Node heartbeats (40 seconds)
 		Eventually(func(g Gomega, ctx context.Context) {
-			node, err := clients.Kubernetes.CoreV1().Nodes().Get(ctx, *vm.Name, metav1.GetOptions{})
-			g.Expect(err).NotTo(HaveOccurred())
+			getCall := clients.Kubernetes.CoreV1().Nodes().Get
+			node := GetK8sObjectWithRetry(ctx, getCall, *vm.Name, metav1.GetOptions{})
 
 			g.Expect(ready.NodeIsReady(node)).To(BeTrue())
 		}).WithContext(ctx).WithTimeout(10 * time.Minute).WithPolling(time.Minute).Should(Succeed())
@@ -79,10 +80,10 @@ var _ = Describe("[Admin API] VM redeploy action", func() {
 func getNodeUptime(g Gomega, ctx context.Context, node string) (time.Time, error) {
 	// container kernel = node kernel = `uptime` in a Pod reflects the Node as well
 	namespace := "default"
-	name := fmt.Sprintf("%s-uptime-%d", node, GinkgoParallelProcess())
+	podName := fmt.Sprintf("%s-uptime-%d", node, GinkgoParallelProcess())
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
-			Name:      name,
+			Name:      podName,
 			Namespace: namespace,
 		},
 		Spec: corev1.PodSpec{
@@ -108,25 +109,30 @@ func getNodeUptime(g Gomega, ctx context.Context, node string) (time.Time, error
 		return time.Time{}, err
 	}
 
-	// Defer Delete
 	defer func() {
-		By("deleting uptime pod")
-		err := clients.Kubernetes.CoreV1().Pods(namespace).Delete(ctx, name, metav1.DeleteOptions{})
-		if err != nil {
-			log.Error("Could not delete test Pod")
-		}
+		By("deleting the uptime pod via Kubernetes API")
+		deleteCall := clients.Kubernetes.CoreV1().Pods(namespace).Delete
+		DeleteK8sObjectWithRetry(ctx, deleteCall, podName, metav1.DeleteOptions{})
+
+		// To avoid flakes, we need it to be completely deleted before we can use it again
+		// in a separate run or in a separate It block
+		By("waiting for uptime pod to be deleted")
+		Eventually(func(g Gomega, ctx context.Context) {
+			_, err := clients.Kubernetes.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{})
+			g.Expect(kerrors.IsNotFound(err)).To(BeTrue(), "expect uptime pod to be deleted")
+		}).WithContext(ctx).WithTimeout(DefaultEventuallyTimeout).Should(Succeed())
 	}()
 
 	By("waiting for uptime pod to move into the Succeeded phase")
 	g.Eventually(func(g Gomega, ctx context.Context) {
-		p, err := clients.Kubernetes.CoreV1().Pods(namespace).Get(ctx, name, metav1.GetOptions{})
-		g.Expect(err).NotTo(HaveOccurred())
+		getCall := clients.Kubernetes.CoreV1().Pods(namespace).Get
+		pod := GetK8sObjectWithRetry(ctx, getCall, podName, metav1.GetOptions{})
 
-		g.Expect(p.Status.Phase).To(Equal(corev1.PodSucceeded))
+		g.Expect(pod.Status.Phase).To(Equal(corev1.PodSucceeded))
 	}).WithContext(ctx).WithTimeout(DefaultEventuallyTimeout).Should(Succeed())
 
 	By("getting logs")
-	req := clients.Kubernetes.CoreV1().Pods(namespace).GetLogs(name, &corev1.PodLogOptions{})
+	req := clients.Kubernetes.CoreV1().Pods(namespace).GetLogs(podName, &corev1.PodLogOptions{})
 	stream, err := req.Stream(ctx)
 	if err != nil {
 		return time.Time{}, err