master: use etcd as operate queue

_utils/terror_gen/errors_release.txt

@@ -326,6 +326,16 @@ ErrMasterGenEmbedEtcdConfigFail,[code=38037:class=dm-master:scope=internal:level
 ErrMasterStartEmbedEtcdFail,[code=38038:class=dm-master:scope=internal:level=high],"fail to start embed etcd"
 ErrMasterParseURLFail,[code=38039:class=dm-master:scope=internal:level=high],"fail to parse URL %s"
 ErrMasterJoinEmbedEtcdFail,[code=38040:class=dm-master:scope=internal:level=high],"fail to join embed etcd: %s"
+ErrMasterMarshalOperate,[code=38041:class=dm-master:scope=internal:level=high],"fail to marshal operate: %v"
+ErrMasterUnmarshalOperate,[code=38042:class=dm-master:scope=internal:level=high],"fail to unmarshal operate, operate's key: %s"
+ErrMasterUnmarshalOperateRequest,[code=38043:class=dm-master:scope=internal:level=high],"fail to unmarshal operate request: %v"
+ErrMasterMarshalResponse,[code=38044:class=dm-master:scope=internal:level=high],"fail to marshal response"
+ErrMasterWatchEtcd,[code=38045:class=dm-master:scope=internal:level=medium],"fail to watch etcd's key: %s"
+ErrMasterWriteEtcd,[code=38046:class=dm-master:scope=internal:level=high],"fail to write data to etcd"
+ErrMasterCreateEtcdClient,[code=38047:class=dm-master:scope=internal:level=high],"fail to create etcd client"
+ErrMasterMarshalRequest,[code=38048:class=dm-master:scope=internal:level=high],"fail to marshal request: %v"
+ErrMasterUnmarshalResponse,[code=38049:class=dm-master:scope=internal:level=high],"fail to unmarshal response"
+ErrMasterOperateNotHandle,[code=38050:class=dm-master:scope=internal:level=medium],"operate %v is not handled, maybe no DM-master leader"
 ErrWorkerParseFlagSet,[code=40001:class=dm-worker:scope=internal:level=medium],"parse dm-worker config flag set"
 ErrWorkerInvalidFlag,[code=40002:class=dm-worker:scope=internal:level=medium],"'%s' is an invalid flag"
 ErrWorkerDecodeConfigFromFile,[code=40003:class=dm-worker:scope=internal:level=medium],"toml decode file"

dm/ctl/ctl.go

@@ -76,6 +76,7 @@ func NewRootCmd() *cobra.Command {
 		master.NewUpdateRelayCmd(),
 		master.NewPurgeRelayCmd(),
 		master.NewMigrateRelayCmd(),
+		master.NewShowLeaderCmd(),
 	)
 	return cmd
 }

dm/ctl/master/show_leader.go

@@ -0,0 +1,50 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package master
+
+import (
+	"context"
+
+	"github.com/pingcap/errors"
+	"github.com/spf13/cobra"
+
+	"github.com/pingcap/dm/dm/ctl/common"
+	"github.com/pingcap/dm/dm/pb"
+)
+
+// NewShowLeaderCmd creates a ShowMasterLeader command
+func NewShowLeaderCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "show-master-leader",
+		Short: "show the DM-master's leader information",
+		Run:   showLeaderFunc,
+	}
+	return cmd
+}
+
+// showLeaderFunc does show master leader request
+func showLeaderFunc(cmd *cobra.Command, _ []string) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// show master leader
+	cli := common.MasterClient()
+	resp, err := cli.ShowMasterLeader(ctx, &pb.ShowMasterLeaderRequest{})
+	if err != nil {
+		common.PrintLines("fail to show master leader:\n%v", errors.ErrorStack(err))
+		return
+	}
+
+	common.PrettyPrintResponse(resp)
+}

dm/master/election.go

@@ -47,3 +47,9 @@ func (s *Server) electionNotify(ctx context.Context) {
 		}
 	}
 }
+
+// getLeaderMasterInfo gets the DM-master leader's ID, this ID can set by Config.Name, or use hostname by default.
+func (s *Server) getLeaderMasterInfo(ctx context.Context) (string, error) {
+	_, leaderID, err := s.election.LeaderInfo(ctx)
+	return leaderID, err
+}

dm/master/etcd.go

@@ -22,6 +22,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/pingcap/tidb-tools/pkg/etcd"
+	"github.com/pingcap/tidb-tools/pkg/utils"
 	"go.etcd.io/etcd/embed"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
@@ -36,6 +38,10 @@ const (
 	etcdStartTimeout = time.Minute
 	// privateDirMode grants owner to make/remove files inside the directory.
 	privateDirMode os.FileMode = 0700
+
+	defaultOperatePath = "/dm-operate"
+
+	defaultEtcdTimeout = time.Duration(10 * time.Second)
 )
 
 // startEtcd starts an embedded etcd server.
@@ -198,3 +204,18 @@ func isDataExist(d string) bool {
 	}
 	return len(names) != 0
 }
+
+// getEtcdClientForOperate returns an etcd client
+func getEtcdClientForOperate(addr string) (*etcd.Client, error) {
+	ectdEndpoints, err := utils.ParseHostPortAddr(addr)
+	if err != nil {
+		return nil, terror.ErrMasterCreateEtcdClient.Delegate(err)
+	}
+
+	etcdClient, err := etcd.NewClientFromCfg(ectdEndpoints, defaultEtcdTimeout, "", nil)
+	if err != nil {
+		return nil, terror.ErrMasterCreateEtcdClient.Delegate(err)
+	}
+
+	return etcdClient, nil
+}