fix CI error for session affinity

test/e2e/framework/networking_utils.go

@@ -220,11 +220,16 @@ func (config *NetworkingTestConfig) DialFromContainer(protocol, containerIP, tar
 	Failf("Failed to find expected endpoints:\nTries %d\nCommand %v\nretrieved %v\nexpected %v\n", maxTries, cmd, eps, expectedEps)
 }
 
-func (config *NetworkingTestConfig) GetEndpointsFromTestContainer(protocol, targetIP string, targetPort, maxTries, minTries int) (sets.String, error) {
-	return config.GetEndpointsFromContainer(protocol, config.TestContainerPod.Status.PodIP, targetIP, TestContainerHttpPort, targetPort, maxTries, minTries)
+func (config *NetworkingTestConfig) GetEndpointsFromTestContainer(protocol, targetIP string, targetPort, tries int) (sets.String, error) {
+	return config.GetEndpointsFromContainer(protocol, config.TestContainerPod.Status.PodIP, targetIP, TestContainerHttpPort, targetPort, tries)
 }
 
-func (config *NetworkingTestConfig) GetEndpointsFromContainer(protocol, containerIP, targetIP string, containerHttpPort, targetPort, maxTries, minTries int) (sets.String, error) {
+// GetEndpointsFromContainer executes a curl via kubectl exec in a test container,
+// which might then translate to a tcp or udp request based on the protocol argument
+// in the url.
+// - tries is the maximum number of curl attempts. If this many attempts pass and
+//   we don't see any endpoints, the test fails.
+func (config *NetworkingTestConfig) GetEndpointsFromContainer(protocol, containerIP, targetIP string, containerHttpPort, targetPort, tries int) (sets.String, error) {
 	cmd := fmt.Sprintf("curl -q -s 'http://%s:%d/dial?request=hostName&protocol=%s&host=%s&port=%d&tries=1'",
 		containerIP,
 		containerHttpPort,
@@ -234,7 +239,7 @@ func (config *NetworkingTestConfig) GetEndpointsFromContainer(protocol, containe
 
 	eps := sets.NewString()
 
-	for i := 0; i < maxTries; i++ {
+	for i := 0; i < tries; i++ {
 		stdout, stderr, err := config.f.ExecShellInPodWithFullOutput(config.HostTestContainerPod.Name, cmd)
 		if err != nil {
 			// A failure to kubectl exec counts as a try, not a hard fail.
@@ -242,6 +247,7 @@ func (config *NetworkingTestConfig) GetEndpointsFromContainer(protocol, containe
 			// we confirm unreachability.
 			Logf("Failed to execute %q: %v, stdout: %q, stderr: %q", cmd, err, stdout, stderr)
 		} else {
+			Logf("maxTries: %d, in try: %d, stdout: %v, stderr: %v", tries, i, stdout, stderr)
 			var output map[string][]string
 			if err := json.Unmarshal([]byte(stdout), &output); err != nil {
 				Logf("WARNING: Failed to unmarshal curl response. Cmd %v run in %v, output: %s, err: %v",
@@ -255,10 +261,15 @@ func (config *NetworkingTestConfig) GetEndpointsFromContainer(protocol, containe
 					eps.Insert(trimmed)
 				}
 			}
+			// Return immediately when we successfully fetch endpoints
+			if len(eps) > 0 {
 				return eps, nil
 			}
+			// TODO: get rid of this delay #36281
+			time.Sleep(hitEndpointRetryDelay)
+		}
 	}
-	return nil, fmt.Errorf("Failed to get endpoints:\nTries %d\nCommand %v\n", minTries, cmd)
+	return nil, fmt.Errorf("error getting endpoints:\nTries %d\nCommand %v\n", tries, cmd)
 }
 
 // DialFromNode executes a tcp or udp request based on protocol via kubectl exec

test/e2e/network/networking.go

@@ -19,7 +19,6 @@ package network
 import (
 	"fmt"
 	"net/http"
-	"time"
 
 	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
@@ -208,15 +207,21 @@ var _ = SIGDescribe("Networking", func() {
 			updateSessionAffinity := func(svc *v1.Service) {
 				svc.Spec.SessionAffinity = v1.ServiceAffinityClientIP
 			}
-			framework.UpdateService(f.ClientSet, config.NodePortService.Namespace, config.NodePortService.Name, updateSessionAffinity)
-			firstEndpoints, err := config.GetEndpointsFromTestContainer("http", config.ClusterIP, framework.ClusterHttpPort, config.MaxTries, 0)
+			_, err := framework.UpdateService(f.ClientSet, config.NodePortService.Namespace, config.NodePortService.Name, updateSessionAffinity)
 			if err != nil {
-				framework.Failf("Unable to get endpoints from test containers: %v", err)
+				framework.Failf("Failed to update service session affinity, error: %v", err)
 			}
+
+			// Fetch first endpoints when visiting service
+			firstEndpoints, err := config.GetEndpointsFromTestContainer("http", config.ClusterIP, framework.ClusterHttpPort, config.MaxTries)
+			if err != nil {
+				framework.Failf("Unable to get endpoints from test container: %v", err)
+			}
+			// Check if first endpoints are equal to endpoints which are fetched later
 			for i := 0; i < framework.SessionAffinityChecks; i++ {
-				eps, err := config.GetEndpointsFromTestContainer("http", config.ClusterIP, framework.ClusterHttpPort, config.MaxTries, 0)
+				eps, err := config.GetEndpointsFromTestContainer("http", config.ClusterIP, framework.ClusterHttpPort, config.MaxTries)
 				if err != nil {
-					framework.Failf("Unable to get endpoints from test containers: %v", err)
+					framework.Failf("Unable to get endpoints from test container: %v", err)
 				}
 				if !eps.Equal(firstEndpoints) {
 					framework.Failf("Expect endpoints: %v, got: %v", firstEndpoints, eps)
@@ -225,19 +230,24 @@ var _ = SIGDescribe("Networking", func() {
 		})
 
 		It("should function for client IP based session affinity: udp", func() {
-			startTime := time.Now()
 			config := framework.NewNetworkingTestConfig(f)
-			By(fmt.Sprintf("dialing(udp) %v --> %v:%v (config.clusterIP)", config.TestContainerPod.Name, config.ClusterIP, framework.ClusterHttpPort))
+			By(fmt.Sprintf("dialing(udp) %v --> %v:%v (config.clusterIP)", config.TestContainerPod.Name, config.ClusterIP, framework.ClusterUdpPort))
 			updateSessionAffinity := func(svc *v1.Service) {
 				svc.Spec.SessionAffinity = v1.ServiceAffinityClientIP
 			}
-			framework.UpdateService(f.ClientSet, config.NodePortService.Namespace, config.NodePortService.Name, updateSessionAffinity)
-			firstEndpoints, err := config.GetEndpointsFromTestContainer("udp", config.ClusterIP, framework.ClusterUdpPort, config.MaxTries, 0)
+			_, err := framework.UpdateService(f.ClientSet, config.NodePortService.Namespace, config.NodePortService.Name, updateSessionAffinity)
+			if err != nil {
+				framework.Failf("Failed to update service session affinity, error: %v", err)
+			}
+
+			// Fetch first endpoints when visiting service
+			firstEndpoints, err := config.GetEndpointsFromTestContainer("udp", config.ClusterIP, framework.ClusterUdpPort, config.MaxTries)
 			if err != nil {
 				framework.Failf("Unable to get endpoints from test containers: %v", err)
 			}
+			// Check if first endpoints are equal to endpoints which are fetched later
 			for i := 0; i < framework.SessionAffinityChecks; i++ {
-				eps, err := config.GetEndpointsFromTestContainer("http", config.ClusterIP, framework.ClusterUdpPort, config.MaxTries, 0)
+				eps, err := config.GetEndpointsFromTestContainer("udp", config.ClusterIP, framework.ClusterUdpPort, config.MaxTries)
 				if err != nil {
 					framework.Failf("Unable to get endpoints from test containers: %v", err)
 				}
@@ -245,7 +255,6 @@ var _ = SIGDescribe("Networking", func() {
 					framework.Failf("Expect endpoints: %v, got: %v", firstEndpoints, eps)
 				}
 			}
-			framework.Failf("test session affinity, cost time: %v", time.Now().Sub(startTime))
 		})
 	})
 })