Merge pull request #10714 from mwielgus/storm

E2E test for examples/storm

examples/storm/README.md

@@ -166,12 +166,11 @@ Clients:
  /192.168.45.0:39568[1](queued=0,recved=14072,sent=14072)
  /192.168.86.1:57591[1](queued=0,recved=34,sent=34)
  /192.168.8.0:50375[1](queued=0,recved=34,sent=34)
- /192.168.45.0:39576[1](queued=0,recved=34,sent=34)
 
 Latency min/avg/max: 0/2/2570
 Received: 23199
 Sent: 23198
-Connections: 5
+Connections: 4
 Outstanding: 0
 Zxid: 0xa39
 Mode: standalone

examples/storm/storm-worker-controller.json

@@ -8,7 +8,7 @@
     }
   },
   "spec": {
-    "replicas": 3,
+    "replicas": 2,
     "selector": {
       "name": "storm-worker"
     },

test/e2e/e2e_test.go

@@ -116,7 +116,7 @@ func TestE2E(t *testing.T) {
 		}
 	}
 
-	// Disable density test unless it's explicitly requested.
+	// Disable skipped tests unless they are explicitly requested.
 	if config.GinkgoConfig.FocusString == "" && config.GinkgoConfig.SkipString == "" {
 		config.GinkgoConfig.SkipString = "Skipped"
 	}

test/e2e/examples.go

@@ -224,6 +224,56 @@ var _ = Describe("Examples e2e", func() {
 			})
 		})
 	})
+
+	Describe("[Skipped][Example]Storm", func() {
+		It("should create and stop Zookeeper, Nimbus and Storm worker servers", func() {
+			mkpath := func(file string) string {
+				return filepath.Join(testContext.RepoRoot, "examples", "storm", file)
+			}
+			zookeeperServiceJson := mkpath("zookeeper-service.json")
+			zookeeperPodJson := mkpath("zookeeper.json")
+			nimbusServiceJson := mkpath("storm-nimbus-service.json")
+			nimbusPodJson := mkpath("storm-nimbus.json")
+			workerControllerJson := mkpath("storm-worker-controller.json")
+			nsFlag := fmt.Sprintf("--namespace=%v", ns)
+			zookeeperPod := "zookeeper"
+
+			By("starting Zookeeper")
+			runKubectl("create", "-f", zookeeperPodJson, nsFlag)
+			runKubectl("create", "-f", zookeeperServiceJson, nsFlag)
+			err := waitForPodRunningInNamespace(c, zookeeperPod, ns)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("checking if zookeeper is up and running")
+			_, err = lookForStringInLog(ns, zookeeperPod, "zookeeper", "binding to port", serverStartTimeout)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("starting Nimbus")
+			runKubectl("create", "-f", nimbusPodJson, nsFlag)
+			runKubectl("create", "-f", nimbusServiceJson, nsFlag)
+			err = waitForPodRunningInNamespace(c, "nimbus", ns)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("starting workers")
+			runKubectl("create", "-f", workerControllerJson, nsFlag)
+			forEachPod(c, ns, "name", "storm-worker", func(pod api.Pod) {
+				//do nothing, just wait for the pod to be running
+			})
+			// TODO: Add logging configuration to nimbus & workers images and then
+			// look for a string instead of sleeping.
+			time.Sleep(20 * time.Second)
+
+			By("checking if there are established connections to Zookeeper")
+			_, err = lookForStringInLog(ns, zookeeperPod, "zookeeper", "Established session", serverStartTimeout)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("checking if Nimbus responds to requests")
+			lookForString("No topologies running.", time.Minute, func() string {
+				return runKubectl("exec", "nimbus", nsFlag, "--", "bin/storm", "list")
+			})
+		})
+	})
+
 })
 
 func makeHttpRequestToService(c *client.Client, ns, service, path string) (string, error) {