*: upgrade etcd to master (#1101)

* update etcd to master

Gopkg.toml

@@ -5,15 +5,15 @@
 
 [[constraint]]
   name = "github.com/coreos/etcd"
-  version = "=3.3.4"
+  branch = "master"
 
 [[override]]
   name = "github.com/coreos/bbolt"
   version = "=v1.3.1-coreos.6"
 
 [[override]]
   name = "github.com/grpc-ecosystem/grpc-gateway"
-  version = "~1.3"
+  version = "~1.4"
 
 [[constraint]]
   name = "google.golang.org/grpc"

pd-client/client_test.go

@@ -29,6 +29,7 @@ import (
 )
 
 func TestClient(t *testing.T) {
+	server.EnableZap = true
 	TestingT(t)
 }
 

pd-client/leader_change_test.go

@@ -38,7 +38,6 @@ func (s *testLeaderChangeSuite) prepareClusterN(c *C, n int) (svrs map[string]*s
 
 	for i := 0; i < n; i++ {
 		cfg := cfgs[i]
-
 		go func() {
 			svr, err := server.CreateServer(cfg, api.NewHandler)
 			c.Assert(err, IsNil)

pkg/etcdutil/etcdutil.go

@@ -55,7 +55,7 @@ func CheckClusterID(localClusterID types.ID, um types.URLsMap, tlsConfig *tls.Co
 		trp := &http.Transport{
 			TLSClientConfig: tlsConfig,
 		}
-		remoteCluster, gerr := etcdserver.GetClusterFromRemotePeers([]string{u}, trp)
+		remoteCluster, gerr := etcdserver.GetClusterFromRemotePeers(nil, []string{u}, trp)
 		trp.CloseIdleConnections()
 		if gerr != nil {
 			// Do not return error, because other members may be not ready.

pkg/etcdutil/etcdutil_test.go

@@ -42,6 +42,8 @@ func newTestSingleConfig() *embed.Config {
 	cfg.Name = "test_etcd"
 	cfg.Dir, _ = ioutil.TempDir("/tmp", "test_etcd")
 	cfg.WalDir = ""
+	cfg.Logger = "zap"
+	cfg.LogOutputs = []string{"stdout"}
 
 	pu, _ := url.Parse(tempurl.Alloc())
 	cfg.LPUrls = []url.URL{*pu}

server/api/server_test.go

@@ -50,6 +50,7 @@ var (
 )
 
 func TestAPIServer(t *testing.T) {
+	server.EnableZap = true
 	TestingT(t)
 }
 

server/api/store_ns_test.go

@@ -61,12 +61,12 @@ func (s *testStoreNsSuite) SetUpSuite(c *C) {
 
 	cfg := server.NewTestSingleConfig()
 	cfg.NamespaceClassifier = "table"
-	srv, err := server.CreateServer(cfg, NewHandler)
+	svr, err := server.CreateServer(cfg, NewHandler)
 	c.Assert(err, IsNil)
-	c.Assert(srv.Run(), IsNil)
-	s.svr = srv
+	c.Assert(svr.Run(), IsNil)
+	s.svr = svr
 	s.cleanup = func() {
-		srv.Close()
+		svr.Close()
 		cleanServer(cfg)
 	}
 

server/join.go

@@ -15,14 +15,15 @@ package server
 
 import (
 	"fmt"
+	"os"
 	"path"
 	"strings"
 
 	"github.com/coreos/etcd/clientv3"
 	"github.com/coreos/etcd/embed"
-	"github.com/coreos/etcd/wal"
 	"github.com/juju/errors"
 	"github.com/pingcap/pd/pkg/etcdutil"
+	log "github.com/sirupsen/logrus"
 )
 
 // PrepareJoinCluster sends MemberAdd command to PD cluster,
@@ -74,7 +75,7 @@ func PrepareJoinCluster(cfg *Config) error {
 
 	// Cases with data directory.
 	initialCluster := ""
-	if wal.Exist(path.Join(cfg.DataDir, "member")) {
+	if isDataExist(path.Join(cfg.DataDir, "member")) {
 		cfg.InitialCluster = initialCluster
 		cfg.InitialClusterState = embed.ClusterStateFlagExisting
 		return nil
@@ -139,3 +140,19 @@ func PrepareJoinCluster(cfg *Config) error {
 	cfg.InitialClusterState = embed.ClusterStateFlagExisting
 	return nil
 }
+
+func isDataExist(d string) bool {
+	dir, err := os.Open(d)
+	if err != nil {
+		log.Error("failed to open:", err)
+		return false
+	}
+	defer dir.Close()
+
+	names, err := dir.Readdirnames(-1)
+	if err != nil {
+		log.Error("failed to list:", err)
+		return false
+	}
+	return len(names) != 0
+}

server/leader_test.go

@@ -33,16 +33,14 @@ type testGetLeaderSuite struct {
 func (s *testGetLeaderSuite) SetUpSuite(c *C) {
 	cfg := NewTestSingleConfig()
 
-	// Send requests before server has started.
 	s.wg.Add(1)
 	s.done = make(chan bool)
-	go s.sendRequest(c, cfg.ClientUrls)
-	time.Sleep(100 * time.Millisecond)
-
 	svr, err := CreateServer(cfg, nil)
 	c.Assert(err, IsNil)
-
 	err = svr.Run()
+	// Send requests after server has started.
+	go s.sendRequest(c, cfg.ClientUrls)
+	time.Sleep(100 * time.Millisecond)
 	c.Assert(err, IsNil)
 
 	s.svr = svr

server/server.go

@@ -49,6 +49,9 @@ const (
 	pdClusterIDPath = "/pd/cluster_id"
 )
 
+// EnableZap enable the zap logger in embed etcd.
+var EnableZap = false
+
 // Server is the pd server.
 type Server struct {
 	// Server state.
@@ -114,7 +117,13 @@ func CreateServer(cfg *Config, apiRegister func(*Server) http.Handler) (*Server,
 	}
 	etcdCfg.ServiceRegister = func(gs *grpc.Server) { pdpb.RegisterPDServer(gs, s) }
 	s.etcdCfg = etcdCfg
-
+	if EnableZap {
+		// The etcd master version has removed embed.Config.SetupLogging.
+		// Now logger is set up automatically based on embed.Config.Logger, embed.Config.LogOutputs, embed.Config.Debug fields.
+		// Use zap logger in the test, otherwise will panic. Reference: https://github.com/coreos/etcd/blob/master/embed/config_logging.go#L45
+		s.etcdCfg.Logger = "zap"
+		s.etcdCfg.LogOutputs = []string{"stdout"}
+	}
 	return s, nil
 }
 

server/server_test.go

@@ -26,6 +26,7 @@ import (
 )
 
 func TestServer(t *testing.T) {
+	EnableZap = true
 	TestingT(t)
 }