From 6b487fb1991b73d3da626e9afc8b535790131f00 Mon Sep 17 00:00:00 2001
From: Anthony Romano <anthony.romano@coreos.com>
Date: Wed, 13 Jul 2016 16:32:51 -0700
Subject: [PATCH] e2e: wait for every etcd server to publish to cluster

If etcdctl accesses the cluster before all members are published, it
will get an "unsupported protocol scheme" error. To fix, wait for both
the capabilities and published message.

Fixes #5824
---
 e2e/ctl_v2_test.go |  4 ---
 e2e/etcd_test.go   | 84 +++++++++++++++++++++++-----------------------
 2 files changed, 42 insertions(+), 46 deletions(-)

diff --git a/e2e/ctl_v2_test.go b/e2e/ctl_v2_test.go
index 8cf44dc2b8a..382a48013e3 100644
--- a/e2e/ctl_v2_test.go
+++ b/e2e/ctl_v2_test.go
@@ -258,10 +258,6 @@ func TestCtlV2Backup(t *testing.T) { // For https://github.com/coreos/etcd/issue
 	cfg2.forceNewCluster = true
 	epc2 := setupEtcdctlTest(t, &cfg2, false)
 
-	if _, err := epc2.procs[0].proc.Expect("etcdserver: published"); err != nil {
-		t.Fatal(err)
-	}
-
 	// check if backup went through correctly
 	if err := etcdctlGet(epc2, "foo1", "bar", false); err != nil {
 		t.Fatal(err)
diff --git a/e2e/etcd_test.go b/e2e/etcd_test.go
index d020bb3ec9d..c7afd3aa792 100644
--- a/e2e/etcd_test.go
+++ b/e2e/etcd_test.go
@@ -345,18 +345,8 @@ func (cfg *etcdProcessClusterConfig) tlsArgs() (args []string) {
 
 func (epc *etcdProcessCluster) Start() (err error) {
 	readyC := make(chan error, epc.cfg.clusterSize+epc.cfg.proxySize)
-	readyStr := "enabled capabilities for version"
 	for i := range epc.procs {
-		go func(etcdp *etcdProcess) {
-			etcdp.donec = make(chan struct{})
-			rs := readyStr
-			if etcdp.cfg.isProxy {
-				rs = "httpproxy: endpoints found"
-			}
-			_, err := etcdp.proc.Expect(rs)
-			readyC <- err
-			close(etcdp.donec)
-		}(epc.procs[i])
+		go func(n int) { readyC <- epc.procs[n].waitReady() }(i)
 	}
 	for range epc.procs {
 		if err := <-readyC; err != nil {
@@ -379,28 +369,6 @@ func (epc *etcdProcessCluster) RestartAll() error {
 	return epc.Start()
 }
 
-func (epr *etcdProcess) Restart() error {
-	proc, err := newEtcdProcess(epr.cfg)
-	if err != nil {
-		epr.Stop()
-		return err
-	}
-	*epr = *proc
-
-	readyStr := "enabled capabilities for version"
-	if proc.cfg.isProxy {
-		readyStr = "httpproxy: endpoints found"
-	}
-
-	if _, err = proc.proc.Expect(readyStr); err != nil {
-		epr.Stop()
-		return err
-	}
-	close(proc.donec)
-
-	return nil
-}
-
 func (epc *etcdProcessCluster) StopAll() (err error) {
 	for _, p := range epc.procs {
 		if p == nil {
@@ -418,24 +386,56 @@ func (epc *etcdProcessCluster) StopAll() (err error) {
 	return err
 }
 
-func (epr *etcdProcess) Stop() error {
-	if epr == nil {
-		return nil
+func (epc *etcdProcessCluster) Close() error {
+	err := epc.StopAll()
+	for _, p := range epc.procs {
+		os.RemoveAll(p.cfg.dataDirPath)
 	}
+	return err
+}
 
-	if err := epr.proc.Stop(); err != nil {
+func (ep *etcdProcess) Restart() error {
+	newEp, err := newEtcdProcess(ep.cfg)
+	if err != nil {
+		ep.Stop()
 		return err
 	}
+	*ep = *newEp
+	if err = ep.waitReady(); err != nil {
+		ep.Stop()
+		return err
+	}
+	return nil
+}
 
-	<-epr.donec
+func (ep *etcdProcess) Stop() error {
+	if ep == nil {
+		return nil
+	}
+	if err := ep.proc.Stop(); err != nil {
+		return err
+	}
+	<-ep.donec
 	return nil
 }
 
-func (epc *etcdProcessCluster) Close() error {
-	err := epc.StopAll()
-	for _, p := range epc.procs {
-		os.RemoveAll(p.cfg.dataDirPath)
+func (ep *etcdProcess) waitReady() error {
+	readyStrs := []string{"enabled capabilities for version", "published"}
+	if ep.cfg.isProxy {
+		readyStrs = []string{"httpproxy: endpoints found"}
+	}
+	c := 0
+	matchSet := func(l string) bool {
+		for _, s := range readyStrs {
+			if strings.Contains(l, s) {
+				c++
+				break
+			}
+		}
+		return c == len(readyStrs)
 	}
+	_, err := ep.proc.ExpectFunc(matchSet)
+	close(ep.donec)
 	return err
 }