fix: batch probing in e2e NP tests (#7068)

On older machines, the tests `AntreaPolicyExtendedNamespaces`
would fail but not on CI. The failure was due to the 1k+ simultaneous
probes. By limiting the number of concurrent probes, older
machines can handle the load and CI speed is not compromised

Fixes #7067

Signed-off-by: Peter Tran <[email protected]>

Author: petertran-avgo

test/e2e/k8s_util.go

@@ -1032,55 +1032,116 @@ func (k *KubernetesUtils) waitForPodInNamespace(ns string, pod string) ([]string
 	}
 }
 
+type httpServerReadiness struct {
+	*KubernetesUtils
+	pods              []Pod
+	reachability      *Reachability
+	remoteCluster     *KubernetesUtils
+	protocolPortPairs map[utils.AntreaPolicyProtocol][]int32
+}
+
+func (hsr *httpServerReadiness) isReady() bool {
+	hsr.reachability = NewReachability(hsr.pods, Connected)
+	hsr.validate()
+	if _, wrong, _ := hsr.reachability.Summary(); wrong != 0 {
+		return false
+	}
+
+	return true
+}
+
 func (k *KubernetesUtils) waitForHTTPServers(allPods []Pod) error {
-	const maxTries = 10
 	log.Infof("waiting for HTTP servers (ports 80, 81 and 8080:8085) to become ready")
 
-	serversAreReady := func() bool {
-		reachability := NewReachability(allPods, Connected)
-		k.Validate(allPods, reachability, []int32{80, 81, 8080, 8081, 8082, 8083, 8084, 8085}, utils.ProtocolTCP)
-		if _, wrong, _ := reachability.Summary(); wrong != 0 {
-			return false
-		}
+	httpServerReadiness := httpServerReadiness{
+		pods:            allPods,
+		KubernetesUtils: k,
+		protocolPortPairs: map[utils.AntreaPolicyProtocol][]int32{
+			utils.ProtocolTCP:  {80, 81, 8080, 8081, 8082, 8083, 8084, 8085},
+			utils.ProtocolUDP:  {80, 81},
+			utils.ProtocolSCTP: {80, 81},
+		},
+	}
 
-		k.Validate(allPods, reachability, []int32{80, 81}, utils.ProtocolUDP)
-		if _, wrong, _ := reachability.Summary(); wrong != 0 {
-			return false
-		}
+	if httpServerReadiness.isReady() {
+		log.Infof("All HTTP servers are ready")
+		return nil
+	} else {
+		return fmt.Errorf("HTTP servers are not ready")
+	}
+}
+
+// Encapsulate the data needed to perform a probe between pods
+type probeVector struct {
+	fromPod  Pod
+	toPod    Pod
+	port     int32
+	protocol utils.AntreaPolicyProtocol
+}
 
-		k.Validate(allPods, reachability, []int32{80, 81}, utils.ProtocolSCTP)
-		if _, wrong, _ := reachability.Summary(); wrong != 0 {
-			return false
+// Populate the channel with all combinations of probes based on the required ports and protocols
+func (hsr *httpServerReadiness) buildProbeVectors(probes chan<- probeVector) {
+	for protocol, ports := range hsr.protocolPortPairs {
+		for _, fromPod := range hsr.pods {
+			for _, podTo := range hsr.pods {
+				for _, port := range ports {
+					probes <- probeVector{fromPod, podTo, port, protocol}
+				}
+			}
 		}
-		return true
 	}
+	close(probes)
+}
 
-	for i := 0; i < maxTries; i++ {
-		if serversAreReady() {
-			log.Infof("All HTTP servers are ready")
-			return nil
-		}
-		time.Sleep(defaultInterval)
+// Calculate the number of probes created across all port protocol permutations
+func (hsr *httpServerReadiness) numProbes() int {
+	probeCount := 0
+	podCount := len(hsr.pods)
+	podCountSquared := podCount * podCount
+	for protocol := range hsr.protocolPortPairs {
+		ports := hsr.protocolPortPairs[protocol]
+		probeCount += podCountSquared * len(ports)
 	}
-	return fmt.Errorf("after %d tries, HTTP servers are not ready", maxTries)
+	return probeCount
 }
 
-func (k *KubernetesUtils) validateOnePort(allPods []Pod, reachability *Reachability, port int32, protocol utils.AntreaPolicyProtocol) {
-	numProbes := len(allPods) * len(allPods)
-	resultsCh := make(chan *probeResult, numProbes)
-	// TODO: find better metrics, this is only for POC.
-	oneProbe := func(podFrom, podTo Pod, port int32) {
+// Spawn a fixed set of workers to complete probing of the servers
+func (hsr *httpServerReadiness) spawnProberPool(resultsCh chan *probeResult) {
+	numProbes := hsr.numProbes()
+	probes := make(chan probeVector, numProbes)
+	hsr.buildProbeVectors(probes)
+
+	probe := func(vector probeVector) {
+		podFrom := vector.fromPod
+		podTo := vector.toPod
+		port := vector.port
+		protocol := vector.protocol
 		log.Tracef("Probing: %s -> %s", podFrom, podTo)
-		expectedResult := reachability.Expected.Get(podFrom.String(), podTo.String())
-		connectivity, err := k.Probe(podFrom.Namespace(), podFrom.PodName(), podTo.Namespace(), podTo.PodName(), port, protocol, nil, &expectedResult)
+		expectedResult := hsr.reachability.Expected.Get(podFrom.String(), podTo.String())
+		connectivity, err := hsr.Probe(podFrom.Namespace(), podFrom.PodName(), podTo.Namespace(), podTo.PodName(), port, protocol, hsr.remoteCluster, &expectedResult)
 		resultsCh <- &probeResult{podFrom, podTo, connectivity, err}
 	}
-	for _, pod1 := range allPods {
-		for _, pod2 := range allPods {
-			go oneProbe(pod1, pod2, port)
+
+	startProber := func() {
+		for vector := range probes {
+			probe(vector)
 		}
 	}
-	for i := 0; i < numProbes; i++ {
+
+	// Tested value as the upper limit for running locally with minimal impacts to CI speeds
+	proberRateLimit := 150
+	for range min(proberRateLimit, numProbes) {
+		go startProber()
+	}
+}
+
+// Validates two way connectivity between all pods across all protocol and port permutations
+func (hsr *httpServerReadiness) validate() {
+	numProbes := hsr.numProbes()
+	resultsCh := make(chan *probeResult, numProbes)
+	hsr.spawnProberPool(resultsCh)
+
+	for range numProbes {
 		r := <-resultsCh
 		if r.err != nil {
 			log.Errorf("unable to perform probe %s -> %s: %v", r.podFrom, r.podTo, r.err)
@@ -1093,11 +1154,11 @@ func (k *KubernetesUtils) validateOnePort(allPods []Pod, reachability *Reachabil
 		// If the connectivity from podFrom to podTo has been observed and is different
 		// from the connectivity we received, store Error connectivity in reachability
 		// matrix.
-		prevConn := reachability.Observed.Get(r.podFrom.String(), r.podTo.String())
+		prevConn := hsr.reachability.Observed.Get(r.podFrom.String(), r.podTo.String())
 		if prevConn == Unknown {
-			reachability.Observe(r.podFrom, r.podTo, r.connectivity)
+			hsr.reachability.Observe(r.podFrom, r.podTo, r.connectivity)
 		} else if prevConn != r.connectivity {
-			reachability.Observe(r.podFrom, r.podTo, Error)
+			hsr.reachability.Observe(r.podFrom, r.podTo, Error)
 		}
 	}
 }
@@ -1107,41 +1168,28 @@ func (k *KubernetesUtils) validateOnePort(allPods []Pod, reachability *Reachabil
 // be consistent across all provided ports. Otherwise, this connectivity will be
 // treated as Error.
 func (k *KubernetesUtils) Validate(allPods []Pod, reachability *Reachability, ports []int32, protocol utils.AntreaPolicyProtocol) {
-	for _, port := range ports {
-		// we do not run all the probes in parallel as we have experienced that on some
-		// machines, this can cause a fraction of the probes to always fail, despite the
-		// built-in retry (3x) mechanism. Probably because of the large number of probes,
-		// each one being executed in its own goroutine. For example, with 9 Pods and for
-		// ports 80, 81, 8080, 8081, 8082, 8083, 8084 and 8085, we would end up with
-		// potentially 9*9*8 = 648 simultaneous probes.
-		k.validateOnePort(allPods, reachability, port, protocol)
+	httpServerReadiness := httpServerReadiness{
+		pods:            allPods,
+		KubernetesUtils: k,
+		protocolPortPairs: map[utils.AntreaPolicyProtocol][]int32{
+			protocol: ports,
+		},
+		reachability: reachability,
 	}
+	httpServerReadiness.validate()
 }
 
 func (k *KubernetesUtils) ValidateRemoteCluster(remoteCluster *KubernetesUtils, allPods []Pod, reachability *Reachability, port int32, protocol utils.AntreaPolicyProtocol) {
-	numProbes := len(allPods) * len(allPods)
-	resultsCh := make(chan *probeResult, numProbes)
-	oneProbe := func(podFrom, podTo Pod, port int32) {
-		log.Tracef("Probing: %s -> %s", podFrom, podTo)
-		expectedResult := reachability.Expected.Get(podFrom.String(), podTo.String())
-		connectivity, err := k.Probe(podFrom.Namespace(), podFrom.PodName(), podTo.Namespace(), podTo.PodName(), port, protocol, remoteCluster, &expectedResult)
-		resultsCh <- &probeResult{podFrom, podTo, connectivity, err}
-	}
-	for _, pod1 := range allPods {
-		for _, pod2 := range allPods {
-			go oneProbe(pod1, pod2, port)
-		}
-	}
-	for i := 0; i < numProbes; i++ {
-		r := <-resultsCh
-		if r.err != nil {
-			log.Errorf("unable to perform probe %s -> %s in %s: %v", r.podFrom, r.podTo, k.ClusterName, r.err)
-		}
-		prevConn := reachability.Observed.Get(r.podFrom.String(), r.podTo.String())
-		if prevConn == Unknown {
-			reachability.Observe(r.podFrom, r.podTo, r.connectivity)
-		}
+	httpServerReadiness := httpServerReadiness{
+		pods:            allPods,
+		KubernetesUtils: k,
+		protocolPortPairs: map[utils.AntreaPolicyProtocol][]int32{
+			protocol: {port},
+		},
+		reachability:  reachability,
+		remoteCluster: remoteCluster,
 	}
+	httpServerReadiness.validate()
 }
 
 func (k *KubernetesUtils) Bootstrap(namespaces map[string]TestNamespaceMeta, podsPerNamespace []string, createNamespaces bool, nodeNames map[string]string, hostNetworks map[string]bool) (map[string][]string, error) {