Merge pull request #39313 from MrHohn/e2e-service-util

Automatic merge from submit-queue (batch tested with PRs 39280, 37350, 39389, 39390, 39313)

Moves e2e service util functions into service_util.go and cleans up

Basically moves codes into a central place for service util functions.

Some other codes are touched mostly only due to this migration. Also put a bunch of network reachability utils functions into network_utils.go. They seem somehow redundant, may consider combine they later.

@bowei @freehan

test/e2e/BUILD

@@ -142,6 +142,7 @@ go_library(
         "//pkg/client/transport:go_default_library",
         "//pkg/client/unversioned/clientcmd:go_default_library",
         "//pkg/client/unversioned/clientcmd/api:go_default_library",
+        "//pkg/cloudprovider:go_default_library",
         "//pkg/cloudprovider/providers/aws:go_default_library",
         "//pkg/cloudprovider/providers/gce:go_default_library",
         "//pkg/controller:go_default_library",

test/e2e/cluster_upgrade.go

@@ -146,20 +146,20 @@ func testService(f *framework.Framework, sem *chaosmonkey.Semaphore, testDuringD
 	// Setup
 	serviceName := "service-test"
 
-	jig := NewServiceTestJig(f.ClientSet, serviceName)
-	// nodeIP := pickNodeIP(jig.Client) // for later
+	jig := framework.NewServiceTestJig(f.ClientSet, serviceName)
+	// nodeIP := framework.PickNodeIP(jig.Client) // for later
 
 	By("creating a TCP service " + serviceName + " with type=LoadBalancer in namespace " + f.Namespace.Name)
 	// TODO it's weird that we have to do this and then wait WaitForLoadBalancer which changes
 	// tcpService.
 	tcpService := jig.CreateTCPServiceOrFail(f.Namespace.Name, func(s *v1.Service) {
 		s.Spec.Type = v1.ServiceTypeLoadBalancer
 	})
-	tcpService = jig.WaitForLoadBalancerOrFail(f.Namespace.Name, tcpService.Name, loadBalancerCreateTimeoutDefault)
+	tcpService = jig.WaitForLoadBalancerOrFail(f.Namespace.Name, tcpService.Name, framework.LoadBalancerCreateTimeoutDefault)
 	jig.SanityCheckService(tcpService, v1.ServiceTypeLoadBalancer)
 
 	// Get info to hit it with
-	tcpIngressIP := getIngressPoint(&tcpService.Status.LoadBalancer.Ingress[0])
+	tcpIngressIP := framework.GetIngressPoint(&tcpService.Status.LoadBalancer.Ingress[0])
 	svcPort := int(tcpService.Spec.Ports[0].Port)
 
 	By("creating pod to be part of service " + serviceName)
@@ -169,7 +169,7 @@ func testService(f *framework.Framework, sem *chaosmonkey.Semaphore, testDuringD
 
 	// Hit it once before considering ourselves ready
 	By("hitting the pod through the service's LoadBalancer")
-	jig.TestReachableHTTP(tcpIngressIP, svcPort, loadBalancerLagTimeoutDefault)
+	jig.TestReachableHTTP(tcpIngressIP, svcPort, framework.LoadBalancerLagTimeoutDefault)
 
 	sem.Ready()
 
@@ -187,7 +187,7 @@ func testService(f *framework.Framework, sem *chaosmonkey.Semaphore, testDuringD
 
 	// Sanity check and hit it once more
 	By("hitting the pod through the service's LoadBalancer")
-	jig.TestReachableHTTP(tcpIngressIP, svcPort, loadBalancerLagTimeoutDefault)
+	jig.TestReachableHTTP(tcpIngressIP, svcPort, framework.LoadBalancerLagTimeoutDefault)
 	jig.SanityCheckService(tcpService, v1.ServiceTypeLoadBalancer)
 }
 

test/e2e/daemon_restart.go

@@ -302,7 +302,7 @@ var _ = framework.KubeDescribe("DaemonRestart [Disruptive]", func() {
 
 	It("Kubelet should not restart containers across restart", func() {
 
-		nodeIPs, err := getNodePublicIps(f.ClientSet)
+		nodeIPs, err := framework.GetNodePublicIps(f.ClientSet)
 		framework.ExpectNoError(err)
 		preRestarts, badNodes := getContainerRestarts(f.ClientSet, ns, labelSelector)
 		if preRestarts != 0 {

test/e2e/disruption.go

@@ -35,8 +35,11 @@ import (
 // schedulingTimeout is longer specifically because sometimes we need to wait
 // awhile to guarantee that we've been patient waiting for something ordinary
 // to happen: a pod to get scheduled and move into Ready
-const schedulingTimeout = 10 * time.Minute
-const bigClusterSize = 7
+const (
+	bigClusterSize    = 7
+	schedulingTimeout = 10 * time.Minute
+	timeout           = 60 * time.Second
+)
 
 var _ = framework.KubeDescribe("DisruptionController", func() {
 	f := framework.NewDefaultFramework("disruption")

test/e2e/dns.go

@@ -464,7 +464,7 @@ var _ = framework.KubeDescribe("DNS", func() {
 
 		// Test changing the externalName field
 		By("changing the externalName to bar.example.com")
-		_, err = updateService(f.ClientSet, f.Namespace.Name, serviceName, func(s *v1.Service) {
+		_, err = framework.UpdateService(f.ClientSet, f.Namespace.Name, serviceName, func(s *v1.Service) {
 			s.Spec.ExternalName = "bar.example.com"
 		})
 		Expect(err).NotTo(HaveOccurred())
@@ -481,7 +481,7 @@ var _ = framework.KubeDescribe("DNS", func() {
 
 		// Test changing type from ExternalName to ClusterIP
 		By("changing the service to type=ClusterIP")
-		_, err = updateService(f.ClientSet, f.Namespace.Name, serviceName, func(s *v1.Service) {
+		_, err = framework.UpdateService(f.ClientSet, f.Namespace.Name, serviceName, func(s *v1.Service) {
 			s.Spec.Type = v1.ServiceTypeClusterIP
 			s.Spec.ClusterIP = "127.1.2.3"
 			s.Spec.Ports = []v1.ServicePort{

test/e2e/federated-ingress.go

@@ -408,9 +408,9 @@ func updateIngressOrFail(clientset *fedclientset.Clientset, namespace string) (n
 
 func (j *federationTestJig) waitForFederatedIngress() {
 	// Wait for the loadbalancer IP.
-	address, err := waitForFederatedIngressAddress(j.client, j.ing.Namespace, j.ing.Name, lbPollTimeout)
+	address, err := waitForFederatedIngressAddress(j.client, j.ing.Namespace, j.ing.Name, framework.LoadBalancerPollTimeout)
 	if err != nil {
-		framework.Failf("Ingress failed to acquire an IP address within %v", lbPollTimeout)
+		framework.Failf("Ingress failed to acquire an IP address within %v", framework.LoadBalancerPollTimeout)
 	}
 	j.address = address
 	framework.Logf("Found address %v for ingress %v", j.address, j.ing.Name)
@@ -422,7 +422,7 @@ func (j *federationTestJig) waitForFederatedIngress() {
 		for _, p := range rules.IngressRuleValue.HTTP.Paths {
 			route := fmt.Sprintf("%v://%v%v", proto, address, p.Path)
 			framework.Logf("Testing route %v host %v with simple GET", route, rules.Host)
-			framework.ExpectNoError(pollURL(route, rules.Host, lbPollTimeout, lbPollInterval, timeoutClient, false))
+			framework.ExpectNoError(pollURL(route, rules.Host, framework.LoadBalancerPollTimeout, framework.LoadBalancerPollInterval, timeoutClient, false))
 		}
 	}
 }

test/e2e/firewall.go

@@ -18,7 +18,6 @@ package e2e
 
 import (
 	"fmt"
-	"time"
 
 	"k8s.io/kubernetes/pkg/api/v1"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
@@ -31,14 +30,6 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-const (
-	firewallTimeoutDefault = 3 * time.Minute
-	firewallTestTcpTimeout = time.Duration(1 * time.Second)
-	// Set ports outside of 30000-32767, 80 and 8080 to avoid being whitelisted by the e2e cluster
-	firewallTestHttpPort = int32(29999)
-	firewallTestUdpPort  = int32(29998)
-)
-
 var _ = framework.KubeDescribe("Firewall rule", func() {
 	var firewall_test_name = "firewall-test"
 	f := framework.NewDefaultFramework(firewall_test_name)
@@ -61,18 +52,18 @@ var _ = framework.KubeDescribe("Firewall rule", func() {
 		firewallTestSourceRanges := []string{"0.0.0.0/1", "128.0.0.0/1"}
 		serviceName := "firewall-test-loadbalancer"
 
-		jig := NewServiceTestJig(cs, serviceName)
-		nodesNames := jig.GetNodesNames(maxNodesForEndpointsTests)
+		jig := framework.NewServiceTestJig(cs, serviceName)
+		nodesNames := jig.GetNodesNames(framework.MaxNodesForEndpointsTests)
 		if len(nodesNames) <= 0 {
 			framework.Failf("Expect at least 1 node, got: %v", nodesNames)
 		}
 		nodesSet := sets.NewString(nodesNames...)
 
 		// OnlyLocal service is needed to examine which exact nodes the requests are being forwarded to by the Load Balancer on GCE
 		By("Creating a LoadBalancer type service with onlyLocal annotation")
-		svc := jig.createOnlyLocalLoadBalancerService(ns, serviceName,
-			loadBalancerCreateTimeoutDefault, false, func(svc *v1.Service) {
-				svc.Spec.Ports = []v1.ServicePort{{Protocol: "TCP", Port: firewallTestHttpPort}}
+		svc := jig.CreateOnlyLocalLoadBalancerService(ns, serviceName,
+			framework.LoadBalancerCreateTimeoutDefault, false, func(svc *v1.Service) {
+				svc.Spec.Ports = []v1.ServicePort{{Protocol: "TCP", Port: framework.FirewallTestHttpPort}}
 				svc.Spec.LoadBalancerSourceRanges = firewallTestSourceRanges
 			})
 		defer func() {
@@ -91,10 +82,10 @@ var _ = framework.KubeDescribe("Firewall rule", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(framework.VerifyFirewallRule(fw, expFw, cloudConfig.Network, false)).NotTo(HaveOccurred())
 
-		By(fmt.Sprintf("Creating netexec pods on at most %v nodes", maxNodesForEndpointsTests))
+		By(fmt.Sprintf("Creating netexec pods on at most %v nodes", framework.MaxNodesForEndpointsTests))
 		for i, nodeName := range nodesNames {
 			podName := fmt.Sprintf("netexec%v", i)
-			jig.LaunchNetexecPodOnNode(f, nodeName, podName, firewallTestHttpPort, firewallTestUdpPort, true)
+			jig.LaunchNetexecPodOnNode(f, nodeName, podName, framework.FirewallTestHttpPort, framework.FirewallTestUdpPort, true)
 			defer func() {
 				framework.Logf("Cleaning up the netexec pod: %v", podName)
 				Expect(cs.Core().Pods(ns).Delete(podName, nil)).NotTo(HaveOccurred())
@@ -103,7 +94,7 @@ var _ = framework.KubeDescribe("Firewall rule", func() {
 
 		// Send requests from outside of the cluster because internal traffic is whitelisted
 		By("Accessing the external service ip from outside, all non-master nodes should be reached")
-		Expect(testHitNodesFromOutside(svcExternalIP, firewallTestHttpPort, firewallTimeoutDefault, nodesSet)).NotTo(HaveOccurred())
+		Expect(framework.TestHitNodesFromOutside(svcExternalIP, framework.FirewallTestHttpPort, framework.FirewallTimeoutDefault, nodesSet)).NotTo(HaveOccurred())
 
 		// Check if there are overlapping tags on the firewall that extend beyond just the vms in our cluster
 		// by removing the tag on one vm and make sure it doesn't get any traffic. This is an imperfect
@@ -117,11 +108,11 @@ var _ = framework.KubeDescribe("Firewall rule", func() {
 			nodesSet.Insert(nodesNames[0])
 			framework.SetInstanceTags(cloudConfig, nodesNames[0], removedTags)
 			// Make sure traffic is recovered before exit
-			Expect(testHitNodesFromOutside(svcExternalIP, firewallTestHttpPort, firewallTimeoutDefault, nodesSet)).NotTo(HaveOccurred())
+			Expect(framework.TestHitNodesFromOutside(svcExternalIP, framework.FirewallTestHttpPort, framework.FirewallTimeoutDefault, nodesSet)).NotTo(HaveOccurred())
 		}()
 
 		By("Accessing serivce through the external ip and examine got no response from the node without tags")
-		Expect(testHitNodesFromOutsideWithCount(svcExternalIP, firewallTestHttpPort, firewallTimeoutDefault, nodesSet, 15)).NotTo(HaveOccurred())
+		Expect(framework.TestHitNodesFromOutsideWithCount(svcExternalIP, framework.FirewallTestHttpPort, framework.FirewallTimeoutDefault, nodesSet, 15)).NotTo(HaveOccurred())
 	})
 
 	It("should have correct firewall rules for e2e cluster", func() {
@@ -147,15 +138,15 @@ var _ = framework.KubeDescribe("Firewall rule", func() {
 		nodeAddrs := framework.NodeAddresses(nodes, v1.NodeExternalIP)
 		Expect(len(nodeAddrs)).NotTo(BeZero())
 		masterAddr := framework.GetMasterAddress(cs)
-		flag, _ := testNotReachableHTTPTimeout(masterAddr, ports.ControllerManagerPort, firewallTestTcpTimeout)
+		flag, _ := framework.TestNotReachableHTTPTimeout(masterAddr, ports.ControllerManagerPort, framework.FirewallTestTcpTimeout)
 		Expect(flag).To(BeTrue())
-		flag, _ = testNotReachableHTTPTimeout(masterAddr, ports.SchedulerPort, firewallTestTcpTimeout)
+		flag, _ = framework.TestNotReachableHTTPTimeout(masterAddr, ports.SchedulerPort, framework.FirewallTestTcpTimeout)
 		Expect(flag).To(BeTrue())
-		flag, _ = testNotReachableHTTPTimeout(nodeAddrs[0], ports.KubeletPort, firewallTestTcpTimeout)
+		flag, _ = framework.TestNotReachableHTTPTimeout(nodeAddrs[0], ports.KubeletPort, framework.FirewallTestTcpTimeout)
 		Expect(flag).To(BeTrue())
-		flag, _ = testNotReachableHTTPTimeout(nodeAddrs[0], ports.KubeletReadOnlyPort, firewallTestTcpTimeout)
+		flag, _ = framework.TestNotReachableHTTPTimeout(nodeAddrs[0], ports.KubeletReadOnlyPort, framework.FirewallTestTcpTimeout)
 		Expect(flag).To(BeTrue())
-		flag, _ = testNotReachableHTTPTimeout(nodeAddrs[0], ports.ProxyStatusPort, firewallTestTcpTimeout)
+		flag, _ = framework.TestNotReachableHTTPTimeout(nodeAddrs[0], ports.ProxyStatusPort, framework.FirewallTestTcpTimeout)
 		Expect(flag).To(BeTrue())
 	})
 })

test/e2e/framework/BUILD

@@ -26,6 +26,7 @@ go_library(
         "perf_util.go",
         "pods.go",
         "resource_usage_gatherer.go",
+        "service_util.go",
         "test_context.go",
         "util.go",
     ],
@@ -36,6 +37,7 @@ go_library(
         "//pkg/api:go_default_library",
         "//pkg/api/errors:go_default_library",
         "//pkg/api/v1:go_default_library",
+        "//pkg/api/v1/service:go_default_library",
         "//pkg/api/validation:go_default_library",
         "//pkg/apimachinery/registered:go_default_library",
         "//pkg/apis/apps/v1beta1:go_default_library",
@@ -82,6 +84,7 @@ go_library(
         "//pkg/util/exec:go_default_library",
         "//pkg/util/intstr:go_default_library",
         "//pkg/util/labels:go_default_library",
+        "//pkg/util/net:go_default_library",
         "//pkg/util/rand:go_default_library",
         "//pkg/util/runtime:go_default_library",
         "//pkg/util/sets:go_default_library",

test/e2e/framework/firewall_util.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"time"
 
 	"k8s.io/kubernetes/pkg/api/v1"
 	clientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
@@ -31,6 +32,14 @@ import (
 	compute "google.golang.org/api/compute/v1"
 )
 
+const (
+	FirewallTimeoutDefault = 3 * time.Minute
+	FirewallTestTcpTimeout = time.Duration(1 * time.Second)
+	// Set ports outside of 30000-32767, 80 and 8080 to avoid being whitelisted by the e2e cluster
+	FirewallTestHttpPort = int32(29999)
+	FirewallTestUdpPort  = int32(29998)
+)
+
 // MakeFirewallNameForLBService return the expected firewall name for a LB service.
 // This should match the formatting of makeFirewallName() in pkg/cloudprovider/providers/gce/gce.go
 func MakeFirewallNameForLBService(name string) string {