Merge pull request etcd-io#8414 from heyitsanthony/fix-multi-peer

embed: associate peer serve() listener with corresponding peer

e2e/etcd_config_test.go

@@ -15,7 +15,13 @@
 package e2e
 
 import (
+	"fmt"
+	"io/ioutil"
+	"os"
+	"strings"
 	"testing"
+
+	"github.com/coreos/etcd/pkg/expect"
 )
 
 const exampleConfigFile = "../etcd.conf.yml.sample"
@@ -32,3 +38,49 @@ func TestEtcdExampleConfig(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestEtcdMultiPeer(t *testing.T) {
+	peers, tmpdirs := make([]string, 3), make([]string, 3)
+	for i := range peers {
+		peers[i] = fmt.Sprintf("e%d=http://127.0.0.1:%d", i, etcdProcessBasePort+i)
+		d, err := ioutil.TempDir("", fmt.Sprintf("e%d.etcd", i))
+		if err != nil {
+			t.Fatal(err)
+		}
+		tmpdirs[i] = d
+	}
+	ic := strings.Join(peers, ",")
+
+	procs := make([]*expect.ExpectProcess, len(peers))
+	defer func() {
+		for i := range procs {
+			if procs[i] != nil {
+				procs[i].Stop()
+			}
+			os.RemoveAll(tmpdirs[i])
+		}
+	}()
+	for i := range procs {
+		args := []string{
+			binDir + "/etcd",
+			"--name", fmt.Sprintf("e%d", i),
+			"--listen-client-urls", "http://0.0.0.0:0",
+			"--data-dir", tmpdirs[i],
+			"--advertise-client-urls", "http://0.0.0.0:0",
+			"--listen-peer-urls", fmt.Sprintf("http://127.0.0.1:%d,http://127.0.0.1:%d", etcdProcessBasePort+i, etcdProcessBasePort+len(peers)+i),
+			"--initial-advertise-peer-urls", fmt.Sprintf("http://127.0.0.1:%d", etcdProcessBasePort+i),
+			"--initial-cluster", ic,
+		}
+		p, err := spawnCmd(args)
+		if err != nil {
+			t.Fatal(err)
+		}
+		procs[i] = p
+	}
+
+	for _, p := range procs {
+		if err := waitReadyExpectProc(p, etcdServerReadyLines); err != nil {
+			t.Fatal(err)
+		}
+	}
+}

embed/etcd.go

@@ -154,16 +154,16 @@ func StartEtcd(inCfg *Config) (e *Etcd, err error) {
 
 	// configure peer handlers after rafthttp.Transport started
 	ph := etcdhttp.NewPeerHandler(e.Server)
-	for i := range e.Peers {
+	for _, p := range e.Peers {
 		srv := &http.Server{
 			Handler:     ph,
 			ReadTimeout: 5 * time.Minute,
 			ErrorLog:    defaultLog.New(ioutil.Discard, "", 0), // do not log user error
 		}
-		e.Peers[i].serve = func() error {
-			return srv.Serve(e.Peers[i].Listener)
-		}
-		e.Peers[i].close = func(ctx context.Context) error {
+
+		l := p.Listener
+		p.serve = func() error { return srv.Serve(l) }
+		p.close = func(ctx context.Context) error {
 			// gracefully shutdown http.Server
 			// close open listeners, idle connections
 			// until context cancel or time-out