optimize e2e framework (#2492)

1. add WaitUntil function;
2. support namespaced client.

test/e2e/framework/daemonset.go

@@ -7,16 +7,38 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	clientset "k8s.io/client-go/kubernetes"
+	v1apps "k8s.io/client-go/kubernetes/typed/apps/v1"
 )
 
-func GetPodsForDaemonSet(cs clientset.Interface, ds *appsv1.DaemonSet) (*corev1.PodList, error) {
+type DaemonSetClient struct {
+	f *Framework
+	v1apps.DaemonSetInterface
+}
+
+func (f *Framework) DaemonSetClient() *DaemonSetClient {
+	return f.DaemonSetClientNS(f.Namespace.Name)
+}
+
+func (f *Framework) DaemonSetClientNS(namespace string) *DaemonSetClient {
+	return &DaemonSetClient{
+		f:                  f,
+		DaemonSetInterface: f.ClientSet.AppsV1().DaemonSets(namespace),
+	}
+}
+
+func (c *DaemonSetClient) Get(name string) *appsv1.DaemonSet {
+	ds, err := c.DaemonSetInterface.Get(context.TODO(), name, metav1.GetOptions{})
+	ExpectNoError(err)
+	return ds
+}
+
+func (c *DaemonSetClient) GetPods(ds *appsv1.DaemonSet) (*corev1.PodList, error) {
 	podSelector, err := metav1.LabelSelectorAsSelector(ds.Spec.Selector)
 	if err != nil {
 		return nil, err
 	}
 	podListOptions := metav1.ListOptions{LabelSelector: podSelector.String()}
-	allPods, err := cs.CoreV1().Pods(ds.Namespace).List(context.TODO(), podListOptions)
+	allPods, err := c.f.ClientSet.CoreV1().Pods(ds.Namespace).List(context.TODO(), podListOptions)
 	if err != nil {
 		return nil, err
 	}
@@ -32,8 +54,8 @@ func GetPodsForDaemonSet(cs clientset.Interface, ds *appsv1.DaemonSet) (*corev1.
 	return ownedPods, nil
 }
 
-func GetPodOnNodeForDaemonSet(cs clientset.Interface, ds *appsv1.DaemonSet, node string) (*corev1.Pod, error) {
-	pods, err := GetPodsForDaemonSet(cs, ds)
+func (c *DaemonSetClient) GetPodOnNode(ds *appsv1.DaemonSet, node string) (*corev1.Pod, error) {
+	pods, err := c.GetPods(ds)
 	if err != nil {
 		return nil, err
 	}

test/e2e/framework/deployment.go

@@ -23,9 +23,13 @@ type DeploymentClient struct {
 }
 
 func (f *Framework) DeploymentClient() *DeploymentClient {
+	return f.DeploymentClientNS(f.Namespace.Name)
+}
+
+func (f *Framework) DeploymentClientNS(namespace string) *DeploymentClient {
 	return &DeploymentClient{
 		f:                   f,
-		DeploymentInterface: f.ClientSet.AppsV1().Deployments(f.Namespace.Name),
+		DeploymentInterface: f.ClientSet.AppsV1().Deployments(namespace),
 	}
 }
 

test/e2e/framework/event.go

@@ -15,14 +15,14 @@ type EventClient struct {
 	typedcorev1.EventInterface
 }
 
-func (f *Framework) EventClient(namespace string) *EventClient {
-	ns := f.Namespace.Name
-	if namespace != "" {
-		ns = namespace
-	}
+func (f *Framework) EventClient() *EventClient {
+	return f.EventClientNS(f.Namespace.Name)
+}
+
+func (f *Framework) EventClientNS(namespace string) *EventClient {
 	return &EventClient{
 		f:              f,
-		EventInterface: f.ClientSet.CoreV1().Events(ns),
+		EventInterface: f.ClientSet.CoreV1().Events(namespace),
 	}
 }
 

test/e2e/framework/kube-ovn.go

@@ -3,7 +3,6 @@ package framework
 import (
 	"context"
 
-	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	clientset "k8s.io/client-go/kubernetes"
 )
@@ -13,11 +12,3 @@ func GetKubeOvnImage(cs clientset.Interface) string {
 	ExpectNoError(err, "getting daemonset %s/%s", KubeOvnNamespace, DaemonSetOvsOvn)
 	return ds.Spec.Template.Spec.Containers[0].Image
 }
-
-func GetOvsPodOnNode(cs clientset.Interface, node string) *corev1.Pod {
-	ds, err := cs.AppsV1().DaemonSets(KubeOvnNamespace).Get(context.TODO(), DaemonSetOvsOvn, metav1.GetOptions{})
-	ExpectNoError(err, "getting daemonset %s/%s", KubeOvnNamespace, DaemonSetOvsOvn)
-	ovsPod, err := GetPodOnNodeForDaemonSet(cs, ds, node)
-	ExpectNoError(err, "getting daemonset %s/%s running on node %s", KubeOvnNamespace, DaemonSetOvsOvn, node)
-	return ovsPod
-}

test/e2e/framework/network-policy.go

@@ -21,9 +21,13 @@ type NetworkPolicyClient struct {
 }
 
 func (f *Framework) NetworkPolicyClient() *NetworkPolicyClient {
+	return f.NetworkPolicyClientNS(f.Namespace.Name)
+}
+
+func (f *Framework) NetworkPolicyClientNS(namespace string) *NetworkPolicyClient {
 	return &NetworkPolicyClient{
 		f:                      f,
-		NetworkPolicyInterface: f.ClientSet.NetworkingV1().NetworkPolicies(f.Namespace.Name),
+		NetworkPolicyInterface: f.ClientSet.NetworkingV1().NetworkPolicies(namespace),
 	}
 }
 

test/e2e/framework/pod.go

@@ -18,7 +18,11 @@ type PodClient struct {
 }
 
 func (f *Framework) PodClient() *PodClient {
-	return &PodClient{e2epod.NewPodClient(f.Framework)}
+	return f.PodClientNS(f.Namespace.Name)
+}
+
+func (f *Framework) PodClientNS(namespace string) *PodClient {
+	return &PodClient{e2epod.PodClientNS(f.Framework, namespace)}
 }
 
 func (c *PodClient) DeleteSync(name string) {

test/e2e/framework/service.go

@@ -3,7 +3,6 @@ package framework
 import (
 	"context"
 	"fmt"
-	"math/big"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
@@ -24,9 +23,13 @@ type ServiceClient struct {
 }
 
 func (f *Framework) ServiceClient() *ServiceClient {
+	return f.ServiceClientNS(f.Namespace.Name)
+}
+
+func (f *Framework) ServiceClientNS(namespace string) *ServiceClient {
 	return &ServiceClient{
 		f:                f,
-		ServiceInterface: f.ClientSet.CoreV1().Services(f.Namespace.Name),
+		ServiceInterface: f.ClientSet.CoreV1().Services(namespace),
 	}
 }
 
@@ -44,16 +47,13 @@ func (c *ServiceClient) Create(service *corev1.Service) *corev1.Service {
 }
 
 // CreateSync creates a new service according to the framework specifications, and waits for it to be updated.
-func (c *ServiceClient) CreateSync(service *corev1.Service) *corev1.Service {
-	s := c.Create(service)
-	ExpectTrue(c.WaitToBeUpdated(s))
-	// Get the newest service
-	return c.Get(s.Name).DeepCopy()
+func (c *ServiceClient) CreateSync(service *corev1.Service, cond func(s *corev1.Service) (bool, error), condDesc string) *corev1.Service {
+	_ = c.Create(service)
+	return c.WaitUntil(service.Name, cond, condDesc, 2*time.Second, timeout)
 }
 
 // Patch patches the service
 func (c *ServiceClient) Patch(original, modified *corev1.Service) *corev1.Service {
-
 	patch, err := util.GenerateMergePatchPayload(original, modified)
 	ExpectNoError(err)
 
@@ -78,6 +78,12 @@ func (c *ServiceClient) Patch(original, modified *corev1.Service) *corev1.Servic
 	return nil
 }
 
+// PatchSync patches the service and waits the service to meet the condition
+func (c *ServiceClient) PatchSync(original, modified *corev1.Service, cond func(s *corev1.Service) (bool, error), condDesc string) *corev1.Service {
+	_ = c.Patch(original, modified)
+	return c.WaitUntil(original.Name, cond, condDesc, 2*time.Second, timeout)
+}
+
 // Delete deletes a service if the service exists
 func (c *ServiceClient) Delete(name string) {
 	err := c.ServiceInterface.Delete(context.TODO(), name, metav1.DeleteOptions{})
@@ -93,18 +99,26 @@ func (c *ServiceClient) DeleteSync(name string) {
 	gomega.Expect(c.WaitToDisappear(name, 2*time.Second, timeout)).To(gomega.Succeed(), "wait for service %q to disappear", name)
 }
 
-// WaitToBeUpdated returns whether the service is updated within timeout.
-func (c *ServiceClient) WaitToBeUpdated(service *corev1.Service) bool {
-	Logf("Waiting up to %v for service %s to be updated", timeout, service.Name)
-	rv, _ := big.NewInt(0).SetString(service.ResourceVersion, 10)
-	for start := time.Now(); time.Since(start) < timeout; time.Sleep(poll) {
-		s := c.Get(service.Name)
-		if current, _ := big.NewInt(0).SetString(s.ResourceVersion, 10); current.Cmp(rv) > 0 {
-			return true
+// WaitUntil waits the given timeout duration for the specified condition to be met.
+func (c *ServiceClient) WaitUntil(name string, cond func(s *corev1.Service) (bool, error), condDesc string, interval, timeout time.Duration) *corev1.Service {
+	var service *corev1.Service
+	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
+		Logf("Waiting for service %s to meet condition %q", name, condDesc)
+		service = c.Get(name).DeepCopy()
+		met, err := cond(service)
+		if err != nil {
+			return false, fmt.Errorf("failed to check condition for service %s: %v", name, err)
 		}
+		return met, nil
+	})
+	if err == nil {
+		return service
 	}
-	Logf("Service %s was not updated within %v", service.Name, timeout)
-	return false
+	if IsTimeout(err) {
+		Failf("timed out while waiting for service %s to meet condition %q", name, condDesc)
+	}
+	Fail(maybeTimeoutError(err, "waiting for service %s to meet condition %q", name, condDesc).Error())
+	return nil
 }
 
 // WaitToDisappear waits the given timeout duration for the specified service to disappear.

test/e2e/framework/statefulset.go

@@ -22,9 +22,13 @@ type StatefulSetClient struct {
 }
 
 func (f *Framework) StatefulSetClient() *StatefulSetClient {
+	return f.StatefulSetClientNS(f.Namespace.Name)
+}
+
+func (f *Framework) StatefulSetClientNS(namespace string) *StatefulSetClient {
 	return &StatefulSetClient{
 		f:                    f,
-		StatefulSetInterface: f.ClientSet.AppsV1().StatefulSets(f.Namespace.Name),
+		StatefulSetInterface: f.ClientSet.AppsV1().StatefulSets(namespace),
 	}
 }
 

test/e2e/framework/subnet.go

@@ -198,37 +198,26 @@ func (c *SubnetClient) WaitToBeUpdated(subnet *apiv1.Subnet, timeout time.Durati
 	return false
 }
 
-// Wait waits the given timeout duration for the specified condition to be met.
-func (c *SubnetClient) Wait(name string, cond func(s *apiv1.Subnet) (bool, error), condDesc string, interval, timeout time.Duration) (*apiv1.Subnet, error) {
-	var lastSubnet *apiv1.Subnet
+// WaitUntil waits the given timeout duration for the specified condition to be met.
+func (c *SubnetClient) WaitUntil(name string, cond func(s *apiv1.Subnet) (bool, error), condDesc string, interval, timeout time.Duration) *apiv1.Subnet {
+	var subnet *apiv1.Subnet
 	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 		Logf("Waiting for subnet %s to meet condition %q", name, condDesc)
-		subnets, err := c.List(context.TODO(), metav1.ListOptions{})
+		subnet = c.Get(name).DeepCopy()
+		met, err := cond(subnet)
 		if err != nil {
-			return handleWaitingAPIError(err, true, "listing subnets")
+			return false, fmt.Errorf("failed to check condition for subnet %s: %v", name, err)
 		}
-		met := false
-		for _, subnet := range subnets.Items {
-			if subnet.Name == name {
-				if met, err = cond(&subnet); err != nil {
-					return false, fmt.Errorf("failed to check condition for subnet %s: %v", name, err)
-				}
-				lastSubnet = subnet.DeepCopy()
-				break
-			}
-		}
-
 		return met, nil
 	})
 	if err == nil {
-		return lastSubnet, nil
+		return subnet
 	}
 	if IsTimeout(err) {
-		return lastSubnet, TimeoutError(fmt.Sprintf("timed out while waiting for subnet %s to meet condition %q", name, condDesc),
-			lastSubnet,
-		)
+		Failf("timed out while waiting for subnet %s to meet condition %q", name, condDesc)
 	}
-	return lastSubnet, maybeTimeoutError(err, "waiting for subnet %s to meet condition %q", name, condDesc)
+	Fail(maybeTimeoutError(err, "waiting for subnet %s to meet condition %q", name, condDesc).Error())
+	return nil
 }
 
 // WaitToDisappear waits the given timeout duration for the specified subnet to disappear.

test/e2e/framework/wait.go

@@ -97,3 +97,13 @@ func shouldRetry(err error) (retry bool, retryAfter time.Duration) {
 
 	return false, 0
 }
+
+// WaitUntil waits the condition to be met
+func WaitUntil(cond func() (bool, error), condDesc string) {
+	if err := wait.PollImmediate(2*time.Second, timeout, cond); err != nil {
+		if IsTimeout(err) {
+			Failf("timed out while waiting for the condition to be met: %s", condDesc)
+		}
+		Fail(maybeTimeoutError(err, "waiting for the condition %q to be met", condDesc).Error())
+	}
+}

test/e2e/kube-ovn/network-policy/network-policy.go

@@ -1,7 +1,6 @@
 package network_policy
 
 import (
-	"context"
 	"fmt"
 	"math/rand"
 	"net"
@@ -32,6 +31,7 @@ var _ = framework.Describe("[group:network-policy]", func() {
 	var podClient *framework.PodClient
 	var subnetClient *framework.SubnetClient
 	var netpolClient *framework.NetworkPolicyClient
+	var daemonSetClient *framework.DaemonSetClient
 	var namespaceName, netpolName, subnetName, podName string
 	var cidr, image string
 
@@ -40,6 +40,7 @@ var _ = framework.Describe("[group:network-policy]", func() {
 		podClient = f.PodClient()
 		subnetClient = f.SubnetClient()
 		netpolClient = f.NetworkPolicyClient()
+		daemonSetClient = f.DaemonSetClientNS(framework.KubeOvnNamespace)
 		namespaceName = f.Namespace.Name
 		netpolName = "netpol-" + framework.RandomSuffix()
 		podName = "pod-" + framework.RandomSuffix()
@@ -90,13 +91,12 @@ var _ = framework.Describe("[group:network-policy]", func() {
 		framework.ExpectNotEmpty(nodeList.Items)
 
 		ginkgo.By("Getting daemonset kube-ovn-cni")
-		ds, err := cs.AppsV1().DaemonSets(framework.KubeOvnNamespace).Get(context.TODO(), "kube-ovn-cni", metav1.GetOptions{})
-		framework.ExpectNoError(err, "failed to to get daemonset")
+		ds := daemonSetClient.Get("kube-ovn-cni")
 
 		ginkgo.By("Getting kube-ovn-cni pods")
 		pods := make([]corev1.Pod, 0, len(nodeList.Items))
 		for _, node := range nodeList.Items {
-			pod, err := framework.GetPodOnNodeForDaemonSet(cs, ds, node.Name)
+			pod, err := daemonSetClient.GetPodOnNode(ds, node.Name)
 			framework.ExpectNoError(err, "failed to get kube-ovn-cni pod running on node %s", node.Name)
 			pods = append(pods, *pod)
 		}

test/e2e/kube-ovn/node/node.go

@@ -173,7 +173,9 @@ var _ = framework.OrderedDescribe("[group:node]", func() {
 		service := framework.MakeService(serviceName, "", nil, podLabels, ports, "")
 		service.Spec.IPFamilyPolicy = new(corev1.IPFamilyPolicy)
 		*service.Spec.IPFamilyPolicy = corev1.IPFamilyPolicyPreferDualStack
-		_ = serviceClient.CreateSync(service)
+		_ = serviceClient.CreateSync(service, func(s *corev1.Service) (bool, error) {
+			return len(s.Spec.ClusterIPs) != 0, nil
+		}, "cluster ips are not empty")
 
 		ginkgo.By("Creating pod " + hostPodName + " with host network")
 		cmd := []string{"sh", "-c", "sleep infinity"}