Merge pull request #14984 from mwielgus/spark_fix

test/e2e/examples.go

@@ -34,10 +34,9 @@ import (
 )
 
 const (
-	podListTimeout          = time.Minute
-	serverStartTimeout      = podStartTimeout + 3*time.Minute
-	dnsReadyTimeout         = time.Minute
-	endpointRegisterTimeout = time.Minute
+	podListTimeout     = time.Minute
+	serverStartTimeout = podStartTimeout + 3*time.Minute
+	dnsReadyTimeout    = time.Minute
 )
 
 const queryDnsPythonTemplate string = `
@@ -149,6 +148,8 @@ var _ = Describe("Examples e2e", func() {
 				_, err := lookForStringInLog(ns, pod.Name, "rabbitmq", "Server startup complete", serverStartTimeout)
 				Expect(err).NotTo(HaveOccurred())
 			})
+			waitForEndpoint(c, ns, "rabbitmq-service")
+
 			By("starting celery")
 			runKubectl("create", "-f", celeryControllerYaml, nsFlag)
 			forEachPod(c, ns, "component", "celery", func(pod api.Pod) {
@@ -192,6 +193,9 @@ var _ = Describe("Examples e2e", func() {
 			_, err = lookForStringInLog(ns, "spark-driver", "spark-driver", "Use kubectl exec", serverStartTimeout)
 			Expect(err).NotTo(HaveOccurred())
 
+			By("waiting for master endpoint")
+			waitForEndpoint(c, ns, "spark-master")
+
 			By("starting workers")
 			runKubectl("create", "-f", workerControllerJson, nsFlag)
 			ScaleRC(c, ns, "spark-worker-controller", 2, true)
@@ -221,6 +225,8 @@ var _ = Describe("Examples e2e", func() {
 			_, err = lookForStringInLog(ns, "cassandra", "cassandra", "Listening for thrift clients", serverStartTimeout)
 			Expect(err).NotTo(HaveOccurred())
 
+			waitForEndpoint(c, ns, "cassandra")
+
 			By("create and scale rc")
 			runKubectl("create", "-f", controllerYaml, nsFlag)
 			err = ScaleRC(c, ns, "cassandra", 2, true)
@@ -263,12 +269,14 @@ var _ = Describe("Examples e2e", func() {
 			By("checking if zookeeper is up and running")
 			_, err = lookForStringInLog(ns, zookeeperPod, "zookeeper", "binding to port", serverStartTimeout)
 			Expect(err).NotTo(HaveOccurred())
+			waitForEndpoint(c, ns, "zookeeper")
 
 			By("starting Nimbus")
 			runKubectl("create", "-f", nimbusPodJson, nsFlag)
 			runKubectl("create", "-f", nimbusServiceJson, nsFlag)
 			err = waitForPodRunningInNamespace(c, "nimbus", ns)
 			Expect(err).NotTo(HaveOccurred())
+			waitForEndpoint(c, ns, "nimbus")
 
 			By("starting workers")
 			runKubectl("create", "-f", workerControllerJson, nsFlag)
@@ -382,6 +390,7 @@ var _ = Describe("Examples e2e", func() {
 				})
 			}
 			checkDbInstances()
+			waitForEndpoint(c, ns, "rethinkdb-driver")
 
 			By("scaling rethinkdb")
 			ScaleRC(c, ns, "rethinkdb-rc", 2, true)
@@ -420,6 +429,8 @@ var _ = Describe("Examples e2e", func() {
 				Expect(err).NotTo(HaveOccurred())
 			})
 
+			waitForEndpoint(c, ns, "hazelcast")
+
 			By("scaling hazelcast")
 			ScaleRC(c, ns, "hazelcast", 2, true)
 			forEachPod(c, ns, "name", "hazelcast", func(pod api.Pod) {

test/e2e/util.go

@@ -85,6 +85,7 @@ const (
 
 	podRespondingTimeout     = 2 * time.Minute
 	serviceRespondingTimeout = 2 * time.Minute
+	endpointRegisterTimeout  = time.Minute
 
 	// How wide to print pod names, by default. Useful for aligning printing to
 	// quickly scan through output.
@@ -740,6 +741,20 @@ func waitForReplicationController(c *client.Client, namespace, name string, exis
 	return nil
 }
 
+func waitForEndpoint(c *client.Client, ns, name string) error {
+	for t := time.Now(); time.Since(t) < endpointRegisterTimeout; time.Sleep(poll) {
+		endpoint, err := c.Endpoints(ns).Get(name)
+		Expect(err).NotTo(HaveOccurred())
+		if len(endpoint.Subsets) == 0 || len(endpoint.Subsets[0].Addresses) == 0 {
+			Logf("Endpoint %s/%s is not ready yet", ns, name)
+			continue
+		} else {
+			return nil
+		}
+	}
+	return fmt.Errorf("Failed to get entpoints for %s/%s", ns, name)
+}
+
 // Context for checking pods responses by issuing GETs to them and verifying if the answer with pod name.
 type podResponseChecker struct {
 	c              *client.Client