etcd中raft协议的消息(七)——快照复制消息(MsgSnap消息)

妖狐艹你老母 2022-03-21 07:12 258阅读 0赞

# MsgSnap消息 #

通过前面介绍的raft.sendAppend()方法可知，在Leader节点尝试向集群中的Follower节点发送MsgApp消息时，如果查找不到待发送的Entry记录（即该Follower节点对应的Progress.Next指定的Entry记录），则会尝试通过MsgSnap消息将快照数据发送到Follower节点，Follower节点之后会通过快照数据回复其自身状态，从而可以与Leader节点进行正常的Entry记录复制。例如当Follower节点宕机时间较长，就会出现上述发送MsgSnap消息的场景。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NraDIwMTVqYXZh_size_16_color_FFFFFF_t_70][]

Leader发送MsgSnap消息的主要流程如下：

1.上述两次raftLog查找出现异常时(获取不到需要发送的Entry记录)，就会形成MsgSnap消息，将快照数据发送到指定节点。

2.向该节点发送MsgSnap类型的消息

3.将目标Follower节点对应的Progress切换成ProgressStateSnapshot状态

func (r *raft) sendAppend(to uint64) {
    	pr := r.getProgress(to)  //获取目标节点的Progress
    	if pr.IsPaused() { //检测当前节点是否可以向目标节点发送消息
    		return
    	}
    	m := pb.Message{} //创建待发送的消息
    	m.To = to	//设置目标节点的ID
    
    	term, errt := r.raftLog.term(pr.Next - 1)  //pr.Next是下个待复制Entry的索引位置，获取Next索引对应的记录的Term值
    	ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)	//获取需要发送的Entry记录（ents）
    
    	if errt != nil || erre != nil { // 上述两次raftLog查找出现异常时，就会形成MsgSnap消息，将快照数据发送到指定节点。
    	
    		if !pr.RecentActive { //如果该节点已经不存活，则退出(RecentActive为true表示该节点存活）
    			r.logger.Debugf("ignore sending snapshot to %x since it is not recently active", to)
    			return
    		}
    
    		m.Type = pb.MsgSnap  //将消息设置成MsgSnap，为后续发送快照做准备
    		snapshot, err := r.raftLog.snapshot()  //获取快照数据
    		if err != nil { //异常检测，如果获取快照数据异常，则终止整个程序
    			if err == ErrSnapshotTemporarilyUnavailable {
    				r.logger.Debugf("%x failed to send snapshot to %x because snapshot is temporarily unavailable", r.id, to)
    				return
    			}
    			panic(err) // TODO(bdarnell)
    		}
    		if IsEmptySnap(snapshot) {
    			panic("need non-empty snapshot")
    		}
    		m.Snapshot = snapshot  //设置MsgSnap消息的Snapshot字段
    		sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term  //获取快照的相关信息
    		r.logger.Debugf("%x [firstindex: %d, commit: %d] sent snapshot[index: %d, term: %d] to %x [%s]",
    			r.id, r.raftLog.firstIndex(), r.raftLog.committed, sindex, sterm, to, pr)
    		pr.becomeSnapshot(sindex)//将目标Follower节点对应的Progress切换成ProgressStateSnapshot状态,其中会用Progress.PendingSnapshot字段记录快照数据信息
    		r.logger.Debugf("%x paused sending replication messages to %x [%s]", r.id, to, pr)
    	} else {
    			//发送MsgApp消息略
    			}
    		}
    	}
    	//发送前面创建的MsgApp消息，raft.send()会设置MsgApp消息的Term值，并将其追加到raft.msgs中等待发送。
    	r.send(m)
    }

### Follower处理MsgSnap消息 ###

看到Follower节点接收到MsgSnap消息，会将选举计时器置为0防止发生选举，并调用handleSnapshot函数重建当前节点的raftLog日志。

func stepFollower(r *raft, m pb.Message) error {
    	switch m.Type {
             ......
    	case pb.MsgSnap:
    		r.electionElapsed = 0  //重置electionElapsed,防止发生选举
    		r.lead = m.From		//设置Leader的id
    		r.handleSnapshot(m)	//通过MsgSnap消息中的快照数据，重建当前节点的raftLog
            }
    }

1.获取快照的元数据索引值(Index)和任期号(Term)

2.调用restore函数去重建和恢复raftLog

3.如果重建成功则返回MsgAppResp消息，Index为当前日志最新的索引值

4.如果重建失败则返回MsgAppResp消息，Index为当前日志已提交位置

func (r *raft) handleSnapshot(m pb.Message) {
    	sindex, sterm := m.Snapshot.Metadata.Index, m.Snapshot.Metadata.Term  //获取快照的元数据
    	if r.restore(m.Snapshot) {		//返回值表示是否通过快照数据进行重建
    		r.logger.Infof("%x [commit: %d] restored snapshot [index: %d, term: %d]",
    			r.id, r.raftLog.committed, sindex, sterm)
    		/*
    		向Leader节点返回MsgAppResp消息(Reject始终为false)。该MsgAppResp消息作为MsgSnap消息的响应，与前面介绍的MsgApp消息并无差别，
    		*/
    		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.lastIndex()})
    	} else {
    		r.logger.Infof("%x [commit: %d] ignored snapshot [index: %d, term: %d]",
    			r.id, r.raftLog.committed, sindex, sterm)
    		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
    	}
    }

1.如果快照中的Index小于当前raftLog已提交位置则返回false

2.根据快照数据的元数据查找匹配的Entry记录，如果存在，则表示当前节点已经拥有了该快照中的全部数据，所以无需后续的重建

3.如果快照中的Learners中包含当前节点id，则返回false。因为normal节点不能变成learner

4.通过raftLog.unstable记录该快照数据，同时重置相关字段

5.清空raft.prs字段，并根据快照的元数据进行重建

func (r *raft) restore(s pb.Snapshot) bool {
    	//快照元数据的索引小于等于当前raftLog的已提交的位置返回false
    	if s.Metadata.Index <= r.raftLog.committed {
    		return false
    	}
    	//根据快照数据的元数据查找匹配的Entry记录，如果存在，则表示当前节点已经拥有了该快照中的全部数据，所以无需后续的重建
    	if r.raftLog.matchTerm(s.Metadata.Index, s.Metadata.Term) {
    		r.logger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] fast-forwarded commit to snapshot [index: %d, term: %d]",
    			r.id, r.raftLog.committed, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term)
    		r.raftLog.commitTo(s.Metadata.Index)
    		return false
    	}
    
    	// The normal peer can't become learner.
    	if !r.isLearner {
    		for _, id := range s.Metadata.ConfState.Learners {
    			if id == r.id {
    				r.logger.Errorf("%x can't become learner when restores snapshot [index: %d, term: %d]", r.id, s.Metadata.Index, s.Metadata.Term)
    				return false
    			}
    		}
    	}
    
    	r.logger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] starts to restore snapshot [index: %d, term: %d]",
    		r.id, r.raftLog.committed, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term)
    
    	//通过raftLog.unstable记录该快照数据，同时重置相关字段
    	r.raftLog.restore(s)
    	//清空raft.prs字段，并根据快照的元数据进行重建
    	r.prs = make(map[uint64]*Progress)
    	r.learnerPrs = make(map[uint64]*Progress)
    	r.restoreNode(s.Metadata.ConfState.Nodes, false)
    	r.restoreNode(s.Metadata.ConfState.Learners, true)
    	return true
    }

# MsgSnapStatus消息和MsgUnreachable消息 #

这里还有两个本地消息与MsgSnap消息相关，它们分别是MsgSnapStatus消息和MsgUnreachable消息。如果Leader发送MsgSnap消息出现异常，则会调用Raft接口的ReportUnreachable()函数和ReportSnapshot()函数发送MsgSnapStatus消息和MsgUnreachable消息。下面看一下Leader对这两类消息的处理。

func stepLeader(r *raft, m pb.Message) error {
    
    switch m.Type {
    
    	case pb.MsgSnapStatus:
    		//检验节点对应的状态是否是ProgressStateSnapshot
    		if pr.State != ProgressStateSnapshot {
    			return nil
    		}
    		if !m.Reject {  //之前发送的MsgSnap消息时出现异常
    			pr.becomeProbe()
    			r.logger.Debugf("%x snapshot succeeded, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
    		} else {
    			//发送MsgSnap消息失败，这里会清空对应的Progress.PendingSnapshot字段
    			pr.snapshotFailure()
    			pr.becomeProbe()
    			r.logger.Debugf("%x snapshot failed, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
    		}
    		// If snapshot finish, wait for the msgAppResp from the remote node before sending
    		// out the next msgApp.
    		// If snapshot failure, wait for a heartbeat interval before next try
    		/*
    		无论MsgSnap消息发送是否失败，都会将对应的Progress切换成ProgressStateProbe状态，之后单条发送消息进行试探
    
    		暂停Leader节点向Follower节点继续发送消息，如果发送MsgSnap消息成功了，则待Leader节点收到相应的响应消息(MsgAppResp消息),即可继续发送后续的
    		MsgApp消息，Leader节点对MsgAppResp消息的处理已经介绍过了，如果发送MsgSnap消息失败了，
    		则Leader节点会等到收到MsgHeartbeatResp消息时，才会重新开始发送后续消息
    		*/
    		pr.pause()
    	case pb.MsgUnreachable:
    		// During optimistic replication, if the remote becomes unreachable,
    		// there is huge probability that a MsgApp is lost.
    		//当Follower节点变得不可达，如果继续发送MsgApp消息，则会丢失大量消息
    		if pr.State == ProgressStateReplicate {
    			pr.becomeProbe()  //将Follower节点对应的Progress实例切换成ProgressStateProbe状态
    		}
    		r.logger.Debugf("%x failed to send message to %x because it is unreachable [%s]", r.id, m.From, pr)
      }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NraDIwMTVqYXZh_size_16_color_FFFFFF_t_70]: /images/20220321/5118a72b6cfc48fca3712707da65510b.png

发表评论取消回复

表情：

评论列表（有 0 条评论，258人围观）

还没有评论，来说两句吧...

相关阅读

相关消息队列中的事务消息

大家好，我是易安！今天我们谈一谈消息队列中的事务消息这个话题。一说起事务，你可能自然会联想到数据库。我们日常使用事务的场景，绝大部分都是在操作数据库的时候。像MySQL、O

Dear 丶/ 2024年03月22日 13:37/ 0 赞/ 20 阅读

相关从etcd看Raft协议

从etcd看Raft协议首先，什么是etcd？看官方的定义：A highly-available key value store for shared config

柔情只为你懂/ 2022年07月29日 11:57/ 0 赞/ 204 阅读

相关 etcd中raft协议的消息(七)——快照复制消息(MsgSnap消息)

MsgSnap消息通过前面介绍的raft.sendAppend()方法可知，在Leader节点尝试向集群中的Follower节点发送MsgApp消息时，如

妖狐艹你老母/ 2022年03月21日 07:12/ 0 赞/ 259 阅读

相关 etcd中raft协议的消息(六)——客户端的写请求相关的消息(MsgProp消息)

在etcd-raft模块中，客户端发往集群中的写请求是通过MsgProp消息表示的。Raft集群中只有Leader节点能够响应客户端的写入请求。

妖狐艹你老母/ 2022年03月21日 07:12/ 0 赞/ 235 阅读

相关 etcd中raft协议的消息(五)—— 客户端只读相关的消息(MsgReadIndex和MsgReadIndexResp消息)

readOnly模式在看MsgReadIndex类型的消息之前需要先对readOnly的模式有所了解，raft结构体中的readOnly作用是批量处理只读

太过爱你忘了你带给我的痛/ 2022年03月21日 07:11/ 0 赞/ 152 阅读

相关 etcd中raft协议的消息(三) ——MspApp和MsgAppResp消息

MsgApp消息上小结我们看到节点发起选举的相关消息，当候选人通过选举成为新的Leader后，首先会调用becomeLeader方法初始化相关状态，然后

叁歲伎倆/ 2022年03月21日 06:26/ 0 赞/ 259 阅读

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

一、MsgHup消息在raft协议中我们看到，Leader节点推动心跳计时器，而Follower节点会推动选举计时器。源码主要在etcd的github.com/e

野性酷女/ 2022年03月21日 06:12/ 0 赞/ 150 阅读

相关 etcd中raft协议的消息(一) —— 简述raft中的所有类型消息

etcd中的Raft算法简介 Raft 是一种为了管理复制日志的一致性算法。一致性算法是从复制状态机的背景下提出的（参考英文原文引用37）。在这种方法中，一组服务器

心已赠人/ 2022年03月21日 06:08/ 0 赞/ 394 阅读

相关 etcd中raft协议的消息(四) —— 心跳相关的消息(MsgBeat、MsgHeartbeat、MsgHeartbeatResp和MsgCheckQuorum）

MsgBeat和MsgHeartbeat消息 Leader推动心跳计时器(heartbeatElapsed)，而Follower推动选举计时器(electio

不念不忘少年蓝@/ 2022年03月20日 21:12/ 0 赞/ 191 阅读

相关基本功：消息协议

消息是信息交换的主体，简单的讲，就是两个进程约定一个协议格式。消息表示指的是序列化后的消息字节流在直观上的表现形式，它看起来是对人类友好还是对计算机友好。文本形式对人类友好，二

青旅半醒/ 2022年02月25日 12:44/ 0 赞/ 251 阅读