Merge pull request kubernetes#92163 from knight42/fix/large-cluster-network-e2e

fix(e2e::network): eliminate the dependency of external ips

Author: k8s-ci-robot

test/e2e/framework/network/utils.go

@@ -160,6 +160,12 @@ type NetworkingTestConfig struct {
 	Namespace string
 }
 
+// NetexecDialResponse represents the response returned by the `netexec` subcommand of `agnhost`
+type NetexecDialResponse struct {
+	Responses []string `json:"responses"`
+	Errors    []string `json:"errors"`
+}
+
 // DialFromEndpointContainer executes a curl via kubectl exec in an endpoint container.
 func (config *NetworkingTestConfig) DialFromEndpointContainer(protocol, targetIP string, targetPort, maxTries, minTries int, expectedEps sets.String) {
 	config.DialFromContainer(protocol, echoHostname, config.EndpointPods[0].Status.PodIP, targetIP, EndpointHTTPPort, targetPort, maxTries, minTries, expectedEps)
@@ -212,6 +218,18 @@ func (config *NetworkingTestConfig) EndpointHostnames() sets.String {
 	return expectedEps
 }
 
+func makeCURLDialCommand(ipPort, dialCmd, protocol, targetIP string, targetPort int) string {
+	// The current versions of curl included in CentOS and RHEL distros
+	// misinterpret square brackets around IPv6 as globbing, so use the -g
+	// argument to disable globbing to handle the IPv6 case.
+	return fmt.Sprintf("curl -g -q -s 'http://%s/dial?request=%s&protocol=%s&host=%s&port=%d&tries=1'",
+		ipPort,
+		dialCmd,
+		protocol,
+		targetIP,
+		targetPort)
+}
+
 // DialFromContainer 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.
@@ -232,38 +250,23 @@ func (config *NetworkingTestConfig) EndpointHostnames() sets.String {
 // pod and confirm it doesn't show up as an endpoint.
 func (config *NetworkingTestConfig) DialFromContainer(protocol, dialCommand, containerIP, targetIP string, containerHTTPPort, targetPort, maxTries, minTries int, expectedResponses sets.String) {
 	ipPort := net.JoinHostPort(containerIP, strconv.Itoa(containerHTTPPort))
-	// The current versions of curl included in CentOS and RHEL distros
-	// misinterpret square brackets around IPv6 as globbing, so use the -g
-	// argument to disable globbing to handle the IPv6 case.
-	cmd := fmt.Sprintf("curl -g -q -s 'http://%s/dial?request=%s&protocol=%s&host=%s&port=%d&tries=1'",
-		ipPort,
-		dialCommand,
-		protocol,
-		targetIP,
-		targetPort)
+	cmd := makeCURLDialCommand(ipPort, dialCommand, protocol, targetIP, targetPort)
 
 	responses := sets.NewString()
 
 	for i := 0; i < maxTries; i++ {
-		stdout, stderr, err := config.f.ExecShellInPodWithFullOutput(config.TestContainerPod.Name, cmd)
+		resp, err := config.GetResponseFromContainer(protocol, dialCommand, containerIP, targetIP, containerHTTPPort, targetPort)
 		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.
-			framework.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 {
-				framework.Logf("WARNING: Failed to unmarshal curl response. Cmd %v run in %v, output: %s, err: %v",
-					cmd, config.TestContainerPod.Name, stdout, err)
-				continue
-			}
-
-			for _, response := range output["responses"] {
-				trimmed := strings.TrimSpace(response)
-				if trimmed != "" {
-					responses.Insert(trimmed)
-				}
+			framework.Logf("GetResponseFromContainer: %s", err)
+			continue
+		}
+		for _, response := range resp.Responses {
+			trimmed := strings.TrimSpace(response)
+			if trimmed != "" {
+				responses.Insert(trimmed)
 			}
 		}
 		framework.Logf("Waiting for responses: %v", expectedResponses.Difference(responses))
@@ -294,14 +297,7 @@ func (config *NetworkingTestConfig) GetEndpointsFromTestContainer(protocol, targ
 //   we don't see any endpoints, the test fails.
 func (config *NetworkingTestConfig) GetEndpointsFromContainer(protocol, containerIP, targetIP string, containerHTTPPort, targetPort, tries int) (sets.String, error) {
 	ipPort := net.JoinHostPort(containerIP, strconv.Itoa(containerHTTPPort))
-	// The current versions of curl included in CentOS and RHEL distros
-	// misinterpret square brackets around IPv6 as globbing, so use the -g
-	// argument to disable globbing to handle the IPv6 case.
-	cmd := fmt.Sprintf("curl -g -q -s 'http://%s/dial?request=hostName&protocol=%s&host=%s&port=%d&tries=1'",
-		ipPort,
-		protocol,
-		targetIP,
-		targetPort)
+	cmd := makeCURLDialCommand(ipPort, "hostName", protocol, targetIP, targetPort)
 
 	eps := sets.NewString()
 
@@ -314,14 +310,14 @@ func (config *NetworkingTestConfig) GetEndpointsFromContainer(protocol, containe
 			framework.Logf("Failed to execute %q: %v, stdout: %q, stderr: %q", cmd, err, stdout, stderr)
 		} else {
 			framework.Logf("Tries: %d, in try: %d, stdout: %v, stderr: %v, command run in: %#v", tries, i, stdout, stderr, config.TestContainerPod)
-			var output map[string][]string
+			var output NetexecDialResponse
 			if err := json.Unmarshal([]byte(stdout), &output); err != nil {
 				framework.Logf("WARNING: Failed to unmarshal curl response. Cmd %v run in %v, output: %s, err: %v",
 					cmd, config.TestContainerPod.Name, stdout, err)
 				continue
 			}
 
-			for _, hostName := range output["responses"] {
+			for _, hostName := range output.Responses {
 				trimmed := strings.TrimSpace(hostName)
 				if trimmed != "" {
 					eps.Insert(trimmed)
@@ -334,6 +330,50 @@ func (config *NetworkingTestConfig) GetEndpointsFromContainer(protocol, containe
 	return eps, nil
 }
 
+// GetResponseFromContainer executes a curl via kubectl exec in a container.
+func (config *NetworkingTestConfig) GetResponseFromContainer(protocol, dialCommand, containerIP, targetIP string, containerHTTPPort, targetPort int) (NetexecDialResponse, error) {
+	ipPort := net.JoinHostPort(containerIP, strconv.Itoa(containerHTTPPort))
+	cmd := makeCURLDialCommand(ipPort, dialCommand, protocol, targetIP, targetPort)
+
+	stdout, stderr, err := config.f.ExecShellInPodWithFullOutput(config.TestContainerPod.Name, cmd)
+	if err != nil {
+		return NetexecDialResponse{}, fmt.Errorf("failed to execute %q: %v, stdout: %q, stderr: %q", cmd, err, stdout, stderr)
+	}
+
+	var output NetexecDialResponse
+	if err := json.Unmarshal([]byte(stdout), &output); err != nil {
+		return NetexecDialResponse{}, fmt.Errorf("failed to unmarshal curl response. Cmd %v run in %v, output: %s, err: %v",
+			cmd, config.TestContainerPod.Name, stdout, err)
+	}
+	return output, nil
+}
+
+// GetResponseFromTestContainer executes a curl via kubectl exec in a test container.
+func (config *NetworkingTestConfig) GetResponseFromTestContainer(protocol, dialCommand, targetIP string, targetPort int) (NetexecDialResponse, error) {
+	return config.GetResponseFromContainer(protocol, dialCommand, config.TestContainerPod.Status.PodIP, targetIP, testContainerHTTPPort, targetPort)
+}
+
+// GetHTTPCodeFromTestContainer executes a curl via kubectl exec in a test container and returns the status code.
+func (config *NetworkingTestConfig) GetHTTPCodeFromTestContainer(path, targetIP string, targetPort int) (int, error) {
+	cmd := fmt.Sprintf("curl -g -q -s -o /dev/null -w %%{http_code} http://%s:%d%s",
+		targetIP,
+		targetPort,
+		path)
+	stdout, stderr, err := config.f.ExecShellInPodWithFullOutput(config.TestContainerPod.Name, cmd)
+	// We only care about the status code reported by curl,
+	// and want to return any other errors, such as cannot execute command in the Pod.
+	// If curl failed to connect to host, it would exit with code 7, which makes `ExecShellInPodWithFullOutput`
+	// return a non-nil error and output "000" to stdout.
+	if err != nil && len(stdout) == 0 {
+		return 0, fmt.Errorf("failed to execute %q: %v, stderr: %q", cmd, err, stderr)
+	}
+	code, err := strconv.Atoi(stdout)
+	if err != nil {
+		return 0, fmt.Errorf("failed to parse status code returned by healthz endpoint: %w, code: %s", err, stdout)
+	}
+	return code, nil
+}
+
 // 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/framework/service/jig.go

@@ -260,21 +260,42 @@ func (j *TestJig) CreateLoadBalancerService(timeout time.Duration, tweak func(sv
 // GetEndpointNodes returns a map of nodenames:external-ip on which the
 // endpoints of the Service are running.
 func (j *TestJig) GetEndpointNodes() (map[string][]string, error) {
-	nodes, err := e2enode.GetBoundedReadySchedulableNodes(j.Client, MaxNodesForEndpointsTests)
+	return j.GetEndpointNodesWithIP(v1.NodeExternalIP)
+}
+
+// GetEndpointNodesWithIP returns a map of nodenames:<ip of given type> on which the
+// endpoints of the Service are running.
+func (j *TestJig) GetEndpointNodesWithIP(addressType v1.NodeAddressType) (map[string][]string, error) {
+	nodes, err := j.ListNodesWithEndpoint()
 	if err != nil {
 		return nil, err
 	}
-	epNodes, err := j.GetEndpointNodeNames()
+	nodeMap := map[string][]string{}
+	for _, node := range nodes {
+		nodeMap[node.Name] = e2enode.GetAddresses(&node, addressType)
+	}
+	return nodeMap, nil
+}
+
+// ListNodesWithEndpoint returns a list of nodes on which the
+// endpoints of the given Service are running.
+func (j *TestJig) ListNodesWithEndpoint() ([]v1.Node, error) {
+	nodeNames, err := j.GetEndpointNodeNames()
 	if err != nil {
 		return nil, err
 	}
-	nodeMap := map[string][]string{}
-	for _, n := range nodes.Items {
-		if epNodes.Has(n.Name) {
-			nodeMap[n.Name] = e2enode.GetAddresses(&n, v1.NodeExternalIP)
+	ctx := context.TODO()
+	allNodes, err := j.Client.CoreV1().Nodes().List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return nil, err
+	}
+	epNodes := make([]v1.Node, 0, nodeNames.Len())
+	for _, node := range allNodes.Items {
+		if nodeNames.Has(node.Name) {
+			epNodes = append(epNodes, node)
 		}
 	}
-	return nodeMap, nil
+	return epNodes, nil
 }
 
 // GetEndpointNodeNames returns a string set of node names on which the