server: set timeout for MoveLeader (#1538)

Signed-off-by: disksing <i@disksing.com>

server/leader.go

@@ -30,7 +30,10 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-// IsLeader returns whether server is leader or not.
+// The timeout to wait transfer etcd leader to complete.
+const moveLeaderTimeout = 5 * time.Second
+
+// IsLeader returns whether the server is leader or not.
 func (s *Server) IsLeader() bool {
 	// If server is not started. Both leaderID and ID could be 0.
 	return !s.isClosed() && s.GetLeaderID() == s.ID()
@@ -142,7 +145,7 @@ func (s *Server) etcdLeaderLoop() {
 				break
 			}
 			if myPriority > leaderPriority {
-				err := s.etcd.Server.MoveLeader(ctx, etcdLeader, s.ID())
+				err := s.MoveEtcdLeader(ctx, etcdLeader, s.ID())
 				if err != nil {
 					log.Errorf("failed to transfer etcd leader: %v", err)
 				} else {
@@ -156,6 +159,13 @@ func (s *Server) etcdLeaderLoop() {
 	}
 }
 
+// MoveEtcdLeader tries to transfer etcd leader.
+func (s *Server) MoveEtcdLeader(ctx context.Context, old, new uint64) error {
+	moveCtx, cancel := context.WithTimeout(ctx, moveLeaderTimeout)
+	defer cancel()
+	return errors.WithStack(s.etcd.Server.MoveLeader(moveCtx, old, new))
+}
+
 // getLeader gets server leader from etcd.
 func getLeader(c *clientv3.Client, leaderPath string) (*pdpb.Member, int64, error) {
 	leader := &pdpb.Member{}
@@ -358,7 +368,7 @@ func (s *Server) ResignLeader(nextLeader string) error {
 	}
 	nextLeaderID := leaderIDs[rand.Intn(len(leaderIDs))]
 	log.Infof("%s ready to resign leader, next leader: %v", s.Name(), nextLeaderID)
-	err = s.etcd.Server.MoveLeader(s.serverLoopCtx, s.ID(), nextLeaderID)
+	err = s.MoveEtcdLeader(s.serverLoopCtx, s.ID(), nextLeaderID)
 	return errors.WithStack(err)
 }