refactor: make qos test cases parallel (#2957)

Split test cases to make them run in parallel.
Reduce running time from 9 minutes to 4 minutes when iperf test is enabled.

.github/workflows/build-x86-image.yaml

@@ -1939,7 +1939,7 @@ jobs:
       - build-vpc-nat-gateway
       - build-e2e-binaries
     runs-on: ubuntu-22.04
-    timeout-minutes: 10
+    timeout-minutes: 15
     steps:
       - uses: actions/checkout@v3
 

.github/workflows/scheduled-e2e.yaml

@@ -1293,7 +1293,7 @@ jobs:
   iptables-vpc-nat-gw-conformance-e2e:
     name: Iptables VPC NAT Gateway E2E
     runs-on: ubuntu-22.04
-    timeout-minutes: 10
+    timeout-minutes: 15
     strategy:
       fail-fast: false
       matrix:

pkg/controller/gc.go

@@ -89,7 +89,7 @@ func (c *Controller) gcVpcNatGateway() error {
 				return err
 			}
 		}
-		gwStsNames = append(gwStsNames, genNatGwStsName(gw.Name))
+		gwStsNames = append(gwStsNames, util.GenNatGwStsName(gw.Name))
 	}
 
 	sel, _ := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{MatchLabels: map[string]string{util.VpcNatGatewayLabel: "true"}})

pkg/controller/vpc_nat_gateway.go

@@ -54,10 +54,6 @@ const (
 	getIptablesVersion = "get-iptables-version"
 )
 
-func genNatGwStsName(name string) string {
-	return fmt.Sprintf("vpc-nat-gw-%s", name)
-}
-
 func (c *Controller) resyncVpcNatGwConfig() {
 	cm, err := c.configMapsLister.ConfigMaps(c.config.PodNamespace).Get(util.VpcNatGatewayConfig)
 	if err != nil && !k8serrors.IsNotFound(err) {
@@ -208,7 +204,7 @@ func (c *Controller) processNextWorkItem(processName string, queue workqueue.Rat
 func (c *Controller) handleDelVpcNatGw(key string) error {
 	c.vpcNatGwKeyMutex.LockKey(key)
 	defer func() { _ = c.vpcNatGwKeyMutex.UnlockKey(key) }()
-	name := genNatGwStsName(key)
+	name := util.GenNatGwStsName(key)
 	klog.Infof("delete vpc nat gw %s", name)
 	if err := c.config.KubeClient.AppsV1().StatefulSets(c.config.PodNamespace).Delete(context.Background(),
 		name, metav1.DeleteOptions{}); err != nil {
@@ -272,7 +268,7 @@ func (c *Controller) handleAddOrUpdateVpcNatGw(key string) error {
 	needToCreate := false
 	needToUpdate := false
 	oldSts, err := c.config.KubeClient.AppsV1().StatefulSets(c.config.PodNamespace).
-		Get(context.Background(), genNatGwStsName(gw.Name), metav1.GetOptions{})
+		Get(context.Background(), util.GenNatGwStsName(gw.Name), metav1.GetOptions{})
 
 	if err != nil {
 		if k8serrors.IsNotFound(err) {
@@ -739,7 +735,7 @@ func (c *Controller) execNatGwRules(pod *corev1.Pod, operation string, rules []s
 
 func (c *Controller) genNatGwStatefulSet(gw *kubeovnv1.VpcNatGateway, oldSts *v1.StatefulSet) (newSts *v1.StatefulSet) {
 	replicas := int32(1)
-	name := genNatGwStsName(gw.Name)
+	name := util.GenNatGwStsName(gw.Name)
 	allowPrivilegeEscalation := true
 	privileged := true
 	labels := map[string]string{
@@ -827,7 +823,7 @@ func (c *Controller) cleanUpVpcNatGw() error {
 
 func (c *Controller) getNatGwPod(name string) (*corev1.Pod, error) {
 	sel, _ := metav1.LabelSelectorAsSelector(&metav1.LabelSelector{
-		MatchLabels: map[string]string{"app": genNatGwStsName(name), util.VpcNatGatewayLabel: "true"},
+		MatchLabels: map[string]string{"app": util.GenNatGwStsName(name), util.VpcNatGatewayLabel: "true"},
 	})
 
 	pods, err := c.podsLister.Pods(c.config.PodNamespace).List(sel)

pkg/util/vpc_nat_gateway.go

@@ -0,0 +1,11 @@
+package util
+
+import "fmt"
+
+func GenNatGwStsName(name string) string {
+	return fmt.Sprintf("vpc-nat-gw-%s", name)
+}
+
+func GenNatGwPodName(name string) string {
+	return fmt.Sprintf("vpc-nat-gw-%s-0", name)
+}

test/e2e/framework/vpc-nat-gw.go

@@ -11,6 +11,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
+	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/kubernetes/test/e2e/framework"
 
 	"github.com/onsi/gomega"
@@ -47,9 +48,11 @@ func (c *VpcNatGatewayClient) Create(vpcNatGw *apiv1.VpcNatGateway) *apiv1.VpcNa
 }
 
 // CreateSync creates a new vpc nat gw according to the framework specifications, and waits for it to be ready.
-func (c *VpcNatGatewayClient) CreateSync(vpcNatGw *apiv1.VpcNatGateway) *apiv1.VpcNatGateway {
+func (c *VpcNatGatewayClient) CreateSync(vpcNatGw *apiv1.VpcNatGateway, clietSet clientset.Interface) *apiv1.VpcNatGateway {
 	vpcNatGw = c.Create(vpcNatGw)
-	ExpectTrue(c.WaitToBeReady(vpcNatGw.Name, timeout))
+	// When multiple VPC NAT gateways are being created, it may require more time to wait.
+	timeout := 4 * time.Minute
+	ExpectTrue(c.WaitGwPodReady(vpcNatGw.Name, timeout, clietSet))
 	// Get the newest vpc nat gw after it becomes ready
 	return c.Get(vpcNatGw.Name).DeepCopy()
 }
@@ -126,6 +129,27 @@ func (c *VpcNatGatewayClient) WaitToBeReady(name string, timeout time.Duration)
 	return false
 }
 
+// WaitGwPodReady returns whether the vpc nat gw pod is ready within timeout.
+func (c *VpcNatGatewayClient) WaitGwPodReady(name string, timeout time.Duration, clietSet clientset.Interface) bool {
+	podName := util.GenNatGwPodName(name)
+	for start := time.Now(); time.Since(start) < timeout; time.Sleep(poll) {
+		pod, err := clietSet.CoreV1().Pods("kube-system").Get(context.Background(), podName, metav1.GetOptions{})
+		if err != nil {
+			if apierrors.IsNotFound(err) {
+				Logf("natgw %s is not ready err: %s", name, err)
+				continue
+			}
+			framework.ExpectNoError(err, "failed to get pod %v", podName)
+		}
+		if len(pod.Annotations) != 0 && pod.Annotations[util.VpcNatGatewayInitAnnotation] == "true" {
+			Logf("natgw %s is ready ", name)
+			return true
+		}
+		Logf("natgw %s is not ready ", name)
+	}
+	return false
+}
+
 // WaitToBeUpdated returns whether the vpc nat gw is updated within timeout.
 func (c *VpcNatGatewayClient) WaitToBeUpdated(vpcNatGw *apiv1.VpcNatGateway, timeout time.Duration) bool {
 	Logf("Waiting up to %v for vpc nat gw %s to be updated", timeout, vpcNatGw.Name)