e2e test session affinity

test/e2e/framework/networking_utils.go

@@ -62,6 +62,8 @@ const (
 	testTries = 30
 	// Maximum number of pods in a test, to make test work in large clusters.
 	maxNetProxyPodsCount = 10
+	// Number of checks to hit a given set of endpoints when enable session affinity.
+	SessionAffinityChecks = 10
 )
 
 // NewNetworkingTestConfig creates and sets up a new test config helper.
@@ -217,6 +219,47 @@ func (config *NetworkingTestConfig) DialFromContainer(protocol, containerIP, tar
 	Failf("Failed to find expected endpoints:\nTries %d\nCommand %v\nretrieved %v\nexpected %v\n", minTries, 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) GetEndpointsFromContainer(protocol, containerIP, targetIP string, containerHttpPort, targetPort, maxTries, minTries 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,
+		protocol,
+		targetIP,
+		targetPort)
+
+	eps := sets.NewString()
+
+	for i := 0; i < maxTries; 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.
+			// Also note that we will keep failing for maxTries in tests where
+			// we confirm unreachability.
+			Logf("Failed to execute %q: %v, stdout: %q, stderr: %q", cmd, err, stdout, stderr)
+		} else {
+			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",
+					cmd, config.HostTestContainerPod.Name, stdout, err)
+				continue
+			}
+
+			for _, hostName := range output["responses"] {
+				trimmed := strings.TrimSpace(hostName)
+				if trimmed != "" {
+					eps.Insert(trimmed)
+				}
+			}
+			return eps, nil
+		}
+	}
+	return nil, fmt.Errorf("Failed to get endpoints:\nTries %d\nCommand %v\n", minTries, cmd)
+}
+
 // DialFromNode executes a tcp or udp request based on protocol via kubectl exec
 // in a test container running with host networking.
 // - minTries is the minimum number of curl attempts required before declaring

test/e2e/network/networking.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"net/http"
 
+	"k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/kubernetes/pkg/master/ports"
 	"k8s.io/kubernetes/test/e2e/framework"
@@ -192,6 +193,49 @@ var _ = SIGDescribe("Networking", func() {
 			By(fmt.Sprintf("dialing(udp) %v (node) --> %v:%v (nodeIP)", config.NodeIP, config.NodeIP, config.NodeUdpPort))
 			config.DialFromNode("udp", config.NodeIP, config.NodeUdpPort, config.MaxTries, config.MaxTries, sets.NewString())
 		})
-		// TODO: Test sessionAffinity #31712
+
+		It("should function for client IP based session affinity: http", func() {
+			config := framework.NewNetworkingTestConfig(f)
+			By(fmt.Sprintf("dialing(http) %v --> %v:%v (config.clusterIP)", config.TestContainerPod.Name, config.ClusterIP, framework.ClusterHttpPort))
+			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)
+			if err != nil {
+				framework.Failf("Unable to get endpoints from test containers: %v", err)
+			}
+			for i := 0; i < framework.SessionAffinityChecks; i++ {
+				eps, err := config.GetEndpointsFromTestContainer("http", config.ClusterIP, framework.ClusterHttpPort, config.MaxTries, 0)
+				if err != nil {
+					framework.Failf("Unable to get endpoints from test containers: %v", err)
+				}
+				if !eps.Equal(firstEndpoints) {
+					framework.Failf("Expect endpoints: %v, got: %v", firstEndpoints, eps)
+				}
+			}
+		})
+
+		It("should function for client IP based session affinity: udp", func() {
+			config := framework.NewNetworkingTestConfig(f)
+			By(fmt.Sprintf("dialing(udp) %v --> %v:%v (config.clusterIP)", config.TestContainerPod.Name, config.ClusterIP, framework.ClusterHttpPort))
+			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)
+			if err != nil {
+				framework.Failf("Unable to get endpoints from test containers: %v", err)
+			}
+			for i := 0; i < framework.SessionAffinityChecks; i++ {
+				eps, err := config.GetEndpointsFromTestContainer("http", config.ClusterIP, framework.ClusterUdpPort, config.MaxTries, 0)
+				if err != nil {
+					framework.Failf("Unable to get endpoints from test containers: %v", err)
+				}
+				if !eps.Equal(firstEndpoints) {
+					framework.Failf("Expect endpoints: %v, got: %v", firstEndpoints, eps)
+				}
+			}
+		})
 	})
 })