server: fix the leader cannot election after pd leader lost while etc…

…d leader intact (#6447) (#6460) close #6403, ref #6447 server: fix the leader cannot election after pd leader lost while etcd leader intact Signed-off-by: ti-chi-bot <[email protected]> Signed-off-by: nolouch <[email protected]> Co-authored-by: ShuNing <[email protected]> Co-authored-by: nolouch <[email protected]> Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>
tikv · May 24, 2023 · 91efbda · 91efbda
1 parent c7db3b3
commit 91efbda
Show file tree

Hide file tree

Showing 3 changed files with 77 additions and 0 deletions.
diff --git a/server/member/member.go b/server/member/member.go
@@ -59,6 +59,8 @@ type Member struct {
 	// etcd leader key when the PD node is successfully elected as the PD leader
 	// of the cluster. Every write will use it to check PD leadership.
 	memberValue string
+	// lastLeaderUpdatedTime is the last time when the leader is updated.
+	lastLeaderUpdatedTime atomic.Value
 }
 
 // NewMember create a new Member.
@@ -121,11 +123,13 @@ func (m *Member) GetLeader() *pdpb.Member {
 // setLeader sets the member's PD leader.
 func (m *Member) setLeader(member *pdpb.Member) {
 	m.leader.Store(member)
+	m.lastLeaderUpdatedTime.Store(time.Now())
 }
 
 // unsetLeader unsets the member's PD leader.
 func (m *Member) unsetLeader() {
 	m.leader.Store(&pdpb.Member{})
+	m.lastLeaderUpdatedTime.Store(time.Now())
 }
 
 // EnableLeader sets the member itself to a PD leader.
@@ -143,6 +147,15 @@ func (m *Member) GetLeadership() *election.Leadership {
 	return m.leadership
 }
 
+// GetLastLeaderUpdatedTime returns the last time when the leader is updated.
+func (m *Member) GetLastLeaderUpdatedTime() time.Time {
+	lastLeaderUpdatedTime := m.lastLeaderUpdatedTime.Load()
+	if lastLeaderUpdatedTime == nil {
+		return time.Time{}
+	}
+	return lastLeaderUpdatedTime.(time.Time)
+}
+
 // CampaignLeader is used to campaign a PD member's leadership
 // and make it become a PD leader.
 func (m *Member) CampaignLeader(leaseTimeout int64) error {

diff --git a/server/server.go b/server/server.go
@@ -88,6 +88,9 @@ const (
 	idAllocLabel = "idalloc"
 
 	recoveringMarkPath = "cluster/markers/snapshot-recovering"
+
+	lostPDLeaderMaxTimeoutSecs   = 10
+	lostPDLeaderReElectionFactor = 10
 )
 
 // EtcdStartTimeout the timeout of the startup etcd.
@@ -1388,6 +1391,14 @@ func (s *Server) leaderLoop() {
 		}
 
 		leader, rev, checkAgain := s.member.CheckLeader()
+		// add failpoint to test leader check go to stuck.
+		failpoint.Inject("leaderLoopCheckAgain", func(val failpoint.Value) {
+			memberString := val.(string)
+			memberID, _ := strconv.ParseUint(memberString, 10, 64)
+			if s.member.ID() == memberID {
+				checkAgain = true
+			}
+		})
 		if checkAgain {
 			continue
 		}
@@ -1413,6 +1424,25 @@ func (s *Server) leaderLoop() {
 		// To make sure the etcd leader and PD leader are on the same server.
 		etcdLeader := s.member.GetEtcdLeader()
 		if etcdLeader != s.member.ID() {
+			if s.member.GetLeader() == nil {
+				lastUpdated := s.member.GetLastLeaderUpdatedTime()
+				// use random timeout to avoid leader campaigning storm.
+				randomTimeout := time.Duration(rand.Intn(int(lostPDLeaderMaxTimeoutSecs)))*time.Second + lostPDLeaderMaxTimeoutSecs*time.Second + lostPDLeaderReElectionFactor*s.cfg.ElectionInterval.Duration
+				// add failpoint to test the campaign leader logic.
+				failpoint.Inject("timeoutWaitPDLeader", func() {
+					log.Info("timeoutWaitPDLeader is injected, skip wait other etcd leader be etcd leader")
+					randomTimeout = time.Duration(rand.Intn(10))*time.Millisecond + 100*time.Millisecond
+				})
+				if lastUpdated.Add(randomTimeout).Before(time.Now()) && !lastUpdated.IsZero() && etcdLeader != 0 {
+					log.Info("the pd leader is lost for a long time, try to re-campaign a pd leader with resign etcd leader",
+						zap.Duration("timeout", randomTimeout),
+						zap.Time("last-updated", lastUpdated),
+						zap.String("current-leader-member-id", types.ID(etcdLeader).String()),
+						zap.String("transferee-member-id", types.ID(s.member.ID()).String()),
+					)
+					s.member.MoveEtcdLeader(s.ctx, etcdLeader, s.member.ID())
+				}
+			}
 			log.Info("skip campaigning of pd leader and check later",
 				zap.String("server-name", s.Name()),
 				zap.Uint64("etcd-leader-id", etcdLeader),
@@ -1521,6 +1551,16 @@ func (s *Server) campaignLeader() {
 				log.Info("no longer a leader because lease has expired, pd leader will step down")
 				return
 			}
+			// add failpoint to test exit leader, failpoint judge the member is the give value, then break
+			failpoint.Inject("exitCampaignLeader", func(val failpoint.Value) {
+				memberString := val.(string)
+				memberID, _ := strconv.ParseUint(memberString, 10, 64)
+				if s.member.ID() == memberID {
+					log.Info("exit PD leader")
+					failpoint.Return()
+				}
+			})
+
 			etcdLeader := s.member.GetEtcdLeader()
 			if etcdLeader != s.member.ID() {
 				log.Info("etcd leader changed, resigns pd leadership", zap.String("old-pd-leader-name", s.Name()))

diff --git a/tests/server/member/member_test.go b/tests/server/member/member_test.go
@@ -248,6 +248,30 @@ func TestLeaderResignWithBlock(t *testing.T) {
 	re.NoError(failpoint.Disable("github.com/tikv/pd/server/raftclusterIsBusy"))
 }
 
+func TestPDLeaderLostWhileEtcdLeaderIntact(t *testing.T) {
+	re := require.New(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	cluster, err := tests.NewTestCluster(ctx, 2)
+	defer cluster.Destroy()
+	re.NoError(err)
+
+	err = cluster.RunInitialServers()
+	re.NoError(err)
+
+	leader1 := cluster.WaitLeader()
+	memberID := cluster.GetServer(leader1).GetLeader().GetMemberId()
+
+	re.NoError(failpoint.Enable("github.com/tikv/pd/server/leaderLoopCheckAgain", fmt.Sprintf("return(\"%d\")", memberID)))
+	re.NoError(failpoint.Enable("github.com/tikv/pd/server/exitCampaignLeader", fmt.Sprintf("return(\"%d\")", memberID)))
+	re.NoError(failpoint.Enable("github.com/tikv/pd/server/timeoutWaitPDLeader", `return(true)`))
+	leader2 := waitLeaderChange(re, cluster, leader1)
+	re.NotEqual(leader1, leader2)
+	re.NoError(failpoint.Disable("github.com/tikv/pd/server/leaderLoopCheckAgain"))
+	re.NoError(failpoint.Disable("github.com/tikv/pd/server/exitCampaignLeader"))
+	re.NoError(failpoint.Disable("github.com/tikv/pd/server/timeoutWaitPDLeader"))
+}
+
 func waitLeaderChange(re *require.Assertions, cluster *tests.TestCluster, old string) string {
 	var leader string
 	testutil.Eventually(re, func() bool {