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

@@ -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)

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
 }