etcd中raft协议的消息(二) —— 选举相关的消息(MsgHup、MsgPreVote/MsgPreVoteResp、MsgVote/MsgVoteResp）

野性酷女 2022-03-21 06:12 150阅读 0赞

# 一、MsgHup消息 #

在raft协议中我们看到，Leader节点推动心跳计时器，而Follower节点会推动选举计时器。源码主要在etcd的github.com/etcd-io/etcd/tree/master/raft/node.go和raft.go中，raft结构体中有一个electionElapsed字段，该字段是选举计时器的指针，逻辑时钟每推进一次，该字段的值递增1，逻辑时钟是有上层模块来推进的。当electionElapsed选举计时器超时时，则该Follower节点会发送MsgHup消息。

##     1.选举计时器逻辑时钟的推动 ##

选举计时器及超时时间在raft结构体中为以下两个字段：

electionElapsed int     //选举计时器的指针，其单位是逻辑时钟的刻度，逻辑时钟每推进一次，该字段值就会增加1	
    randomizedElectionTimeout int     //随机生成的选举计时器超时时间

选举计时器是tickElection()函数来推动的，Node节点每收到一次逻辑时钟推进，就会调用一次该函数来推动逻辑时钟，并判断逻辑时钟是否超时，如果超时则发送MsgHup类型的消息。

//推动选举计时器，如果超时则发送MsgHub消息
        func (r *raft) tickElection() {
    	r.electionElapsed++
    
    	if r.promotable() && r.pastElectionTimeout() {
    		r.electionElapsed = 0
    		r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
            }
        }

从以上方法我们看到，当选举计时器超时，节点会调用Step方法发送MsgHup消息。

##      2.MsgHup消息流程 ##

![20190131101238166.png][]

从tickElection方法中可以看到，当选举计时器超时时，会将raft节点的electionElapsed心跳计时器置零，并且调用Step方法发送该MsgHup消息。Step方法处理各种消息类型的消息，我们这里只看对MsgHup消息的处理。

如果当前节点是Leader，则忽略MsgHup类型的消息

如果当前节点不是Leader，

获取raftLog中已提交但未应用的Entry记录

检测是否有未应用的EntryConfChange记录，如果有就放弃发起选举的机会

检测当前集群是否开启了PreVote模式，如果开启了则发起PreElection预选举，没有开启则发起Election选举

调用campaign()发起选举

func (r *raft) Step(m pb.Message) error {
    
       // ......  其他类型消息处理略
       switch m.Type {	//根据Message的Type进行分类处理
    	case pb.MsgHup:	//这里针对MsgHup类型进行处理（Flower转成PreCandidate发送的消息）
    		if r.state != StateLeader {	//只有非Leader状态的节点才会处理MsgHup消息
    			//获取raftLog中已提交但未应用的Entry记录
    			ents, err := r.raftLog.slice(r.raftLog.applied+1, r.raftLog.committed+1, noLimit)
    			if err != nil {
    				r.logger.Panicf("unexpected error getting unapplied entries (%v)", err)
    			}
    			//检测是否有未应用的EntryConfChange记录，如果有就放弃发起选举的机会
    			if n := numOfPendingConf(ents); n != 0 && r.raftLog.committed > r.raftLog.applied {
    				r.logger.Warningf("%x cannot campaign at term %d since there are still %d pending configuration changes to apply", r.id, r.Term, n)
    				return nil
    			}
    
    			r.logger.Infof("%x is starting a new election at term %d", r.id, r.Term)
    			if r.preVote { //检测当前集群是否开启了PreVote模式，如果开启了
    				//调用raft.campaign()方法切换当前节点的角色，发起PreVote
    				r.campaign(campaignPreElection)
    			} else {
    				r.campaign(campaignElection)
    			}
    		} else { //如果当前节点已经是Leader状态，则仅仅输出一条Debug日志
    			r.logger.Debugf("%x ignoring MsgHup because already leader", r.id)
    		}
    
    }

接下来我们看一下campaign方法

作用：发起选举或者预选举

预选举 CampaignType为 campaignPreElection

1.将当前节点切换成PreCandidate状态

2.将voteMsg设置为 pb.MsgPreVote

3.任期号自增1

选举

1.将当前节点切换成Candidate状态，该方法会增加raft.Term字段的值和重置相关状态

2.将voteMsg设置为 pb.MsgVote

3.确认term，为当前Term值+1

统计当前节点收到的选票，并统计其得票数是否超过半数，这次检测主要是为单节点设置的，如果大于等于半数以上的投票，如果是PreElection则发起正式的选举，如果是正式的选举则该节点变成Leader。

如果不是单节点情况，则向集群中其他节点发送voteMsg消息。

func (r *raft) campaign(t CampaignType) {
    	//在该方法的最后，会发送一条消息，这里term和voteMsg分别用于确定该消息的Term值和类别
    	var term uint64
    	var voteMsg pb.MessageType
    	if t == campaignPreElection { //切换的目标状态是PreCandidate
    		//将当前节点切换成PreCandidate状态
    		r.becomePreCandidate()
    		//确定最后发送的消息是MsgPreVote类型
    		voteMsg = pb.MsgPreVote
    		// PreVote RPCs are sent for the next term before we've incremented r.Term.
    		//确定最后发送消息的Term值，并未增加raft.Term字段的值
    		term = r.Term + 1
    	} else { //切换目标状态是Candidate
    		//将当前节点切换成Candidate状态，该方法会增加raft.Term字段的值
    		r.becomeCandidate()
    		voteMsg = pb.MsgVote	//确定最后发送消息的类型是MsgPreVote类型
    		term = r.Term		//确定最后发送消息的Term值
    	}
    
    	//统计当前节点收到的选票，并统计其得票数是否超过半数，这次检测主要是为单节点设置的
    	if r.quorum() == r.poll(r.id, voteRespMsgType(voteMsg), true) {
    		// We won the election after voting for ourselves (which must mean that
    		// this is a single-node cluster). Advance to the next state.
    		//当得到足够多的选票时，则将PreCandidate状态的节点切换Candidate状态
    		//Candidate状态的节点则切换成Leader状态
    		if t == campaignPreElection {
    			r.campaign(campaignElection)
    		} else {
    			r.becomeLeader()
    		}
    		return
    	}
    	for id := range r.prs { //状态切换完成之后，当前节点会向集群中其他所有节点发送指定类型（voteMsg）的消息
    		if id == r.id { //跳过当前节点自身
    			continue
    		}
    		r.logger.Infof("%x [logterm: %d, index: %d] sent %s request to %x at term %d",
    			r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), voteMsg, id, r.Term)
    
    		var ctx []byte
    		//在进行Leader节点转移时，MsgPreVote或MsgVote消息会在Context字段中设置该特殊值
    		if t == campaignTransfer {
    			ctx = []byte(t)
    		}
    		//发送指定类型的消息，其中Index和LogTerm分别是当前节点的raftLog中
    		//最后一条消息的Index值和Term值
    		r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
    	}
    }

# 二、MsgPreVote/MsgPreVoteResp和MsgVote/MsgVoteResp消息 #

当Follower的选举计时器超时时，会把当前状态切换成 StatePreCandidate（预选举）或StateCandidate（候选人），并向集群中其他节点发送MsgPreVote或MsgVote消息。当集群中其他节点收到(预)选举消息时，会先进行一些检验，符合相关条件会投同意，否则会投拒绝票，然后发送给该候选节点，发送的消息类型为MsgPreVoteResp或MsgVoteResp类型。如果预选举阶段(StatePreCandidate)成功收到超过半数以上的同意票，那么该节点会认为选举成功，会发起新一轮的正式选举(节点状态切换成SateCandidate(候选人)，发送的消息类型为MsgVote)。如果是正式选举成功收到超过半数以上的同意票，那个该节点会成为新的Leader。如果没有收到超过半数的票则会将节点状态切换成Follower。

注：是否有预选举阶段是根据初始化配置的参数,该字段保存在raft结构体的preVote字段中

预选举流程如下图

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NraDIwMTVqYXZh_size_16_color_FFFFFF_t_70][]

选举阶段流程如下图

![20190131102420192.png][]

当Follower发起选举向集群中其他节点发送MsgPreVote和MsgVote消息时，也会将自己的状态State切换成候选人（StatePreCandidate或StateCandidate)状态。发送MsgPreVote或MsgVote消息前面已经讲过，接下来我们主要看一下其他节点收到MsgPreVote或MsgVote消息时是如何处理的。

###   1.其他节点收到MsgVote/MsgPreVote类型的消息 ###

处理MsgVote/MsgPreVote类型的消息流程在raft的Step方法中，看到当满足一些条件，节点才会投赞成票，负责会投拒绝票，投票发送的消息的类型为MsgVoteResp或MsgPreVoteResp。

当前节点在参与预选时，会综合下面几个条件决定是否投票

1.当前节点是否已经投票

2.MsgPreVote消息发送者的任期号是否更大

3.当前节点投票给了对方节点

4.MsgPreVote消息发送者的raftLog中是否包含当前节点的全部Entry记录

func (r *raft)Step(m pb.Message) error {
    		//其他类型的消息略
            ......
    		case pb.MsgVote, pb.MsgPreVote://投票，预投票消息处理
    
    		if r.isLearner {
    			// TODO: learner may need to vote, in case of node down when confchange.
    			r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] ignored %s from %x [logterm: %d, index: %d] at term %d: learner can not vote",
    				r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term)
    			return nil
    		}
    		// We can vote if this is a repeat of a vote we've already cast...
    		canVote := r.Vote == m.From ||
    		// ...we haven't voted and we don't think there's a leader yet in this term...
    			(r.Vote == None && r.lead == None) ||
    		// ...or this is a PreVote for a future term...
    			(m.Type == pb.MsgPreVote && m.Term > r.Term)
    		// ...and we believe the candidate is up to date.
    		if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
    			r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] cast %s for %x [logterm: %d, index: %d] at term %d",
    				r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term)
    			// When responding to Msg{Pre,}Vote messages we include the term
    			// from the message, not the local term. To see why consider the
    			// case where a single node was previously partitioned away and
    			// it's local term is now of date. If we include the local term
    			// (recall that for pre-votes we don't update the local term), the
    			// (pre-)campaigning node on the other end will proceed to ignore
    			// the message (it ignores all out of date messages).
    			// The term in the original message and current local term are the
    			// same in the case of regular votes, but different for pre-votes.
    			//将票投给MsgPreVote消息的发送节点（包含选举和预选举）
    			r.send(pb.Message{To: m.From, Term: m.Term, Type: voteRespMsgType(m.Type)})
    			if m.Type == pb.MsgVote { //如果是MsgVote处理
    				// Only record real votes.
    				r.electionElapsed = 0
    				r.Vote = m.From
    			}
    		} else { //不满足上述条件，当前节点会返回拒投票（Reject:true）
    			r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected %s from %x [logterm: %d, index: %d] at term %d",
    				r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term)
    			r.send(pb.Message{To: m.From, Term: r.Term, Type: voteRespMsgType(m.Type), Reject: true})
    		}
       }

###  2.候选节点会等待集群中其他节点投票返回的消息 ###

参与选举的候选节点发送预选举(MsgPreVote)或选举(MsgVote)消息后，会等待集群中其他节点返回的消息，如果是MsgPreVote消息，则返回MsgPreVoteResp消息，如果是MsgVote消息，则返回MsgVoteResp消息。处理流程主要在raft中的stepCandidate中。

从该方法中可以看到，处于候选人状态的节点当收到MsgProp、MsgApp、MsgHeartbeat、MsgSnap类型的消息会切换成Follower角色。当收到MsgPreVoteResp或MsgVoteResp类型的消息会统计，如果超过半数则切换成Leader状态。如果无法获取半数以上的选票，则切换成Follower状态。

func stepCandidate(r *raft, m pb.Message) error {
    	// Only handle vote responses corresponding to our candidacy (while in
    	// StateCandidate, we may get stale MsgPreVoteResp messages in this term from
    	// our pre-candidate state).
    	var myVoteRespType pb.MessageType
    	if r.state == StatePreCandidate {  //根据当前节点的状态决定其能够处理的选举响应消息的类型
    		myVoteRespType = pb.MsgPreVoteResp
    	} else {
    		myVoteRespType = pb.MsgVoteResp
    	}
    	switch m.Type {
            //......
            //其他类型消息省略
    	case myVoteRespType:
    		gr := r.poll(m.From, m.Type, !m.Reject)   //记录并统计某个节点投票结果
    		r.logger.Infof("%x [quorum:%d] has received %d %s votes and %d vote rejections", r.id, r.quorum(), gr, m.Type, len(r.votes)-gr)
    		switch r.quorum() {    //得票数为len(r.prs)/2 + 1,即超过半数
    		case gr:   //得票数超过半数
    			if r.state == StatePreCandidate {
    				r.campaign(campaignElection)  //当PreCandidate节点在预选中收到半数以上的选票之后，会发起正式的选举
    			} else {
    				r.becomeLeader()
    				r.bcastAppend()   //向集群中其他节点广播MsgApp消息
    			}
    		case len(r.votes) - gr:  //无法获取半数以上的选票，当前节点切换成跟随者Follower状态，等待下一轮的选举
    			// pb.MsgPreVoteResp contains future term of pre-candidate
    			// m.Term > r.Term; reuse r.Term
    			r.becomeFollower(r.Term, None)
    		}
    	case pb.MsgTimeoutNow:
    		r.logger.Debugf("%x [term %d state %v] ignored MsgTimeoutNow from %x", r.id, r.Term, r.state, m.From)
    	}
    	return nil
    }

[20190131101238166.png]: /images/20220321/1bb44981999d428aa5969496b1c3c71a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NraDIwMTVqYXZh_size_16_color_FFFFFF_t_70]: /images/20220321/1713ad888938475e8faa0c50368c4b8f.png
[20190131102420192.png]: /images/20220321/7b7db8218e864e87b512c5a93a1a9a32.png