专栏首页分布式深入浅出etcd之raft实现
原创

深入浅出etcd之raft实现

导语

etcd是coreOS使用golang开发的分布式,一致性的kv存储系统,因其易用性和高可靠性被广泛运用于服务发现、消息发布和订阅、分布式锁和共享配置等方面,也被认为是zookeeper的强有力的竞争者。作为分布式kv,其底层使用raft算法实现多副本数据的强一致性。etcd作为raft开源实现的标杆,在设计上,将 raft 算法逻辑和持久化、网络、线程等完全抽离出来单独实现,充分解耦,在工程上,实现了诸多性能优化,是 raft 开源实践中较早的工业级的实现,很多后来的 raft 实践者都直接或者间接的参考了 ectd-raft 的设计和实现,例如kubernetes,tiDb等。其广泛的影响力和优雅的golang代码实践也使得ectd成为golang的明星项目。在我们实际的分布式存储系统的项目开发中,raft也被应用于元信息管理和数据存储等多个模块,因此熟悉和理解etcd-raft的实现具有重大意义,本文从raft的基本原理出发,深入浅出地分析了raft在ectd中的具体实现。

raft原理

架构

image

每个节点都包含状态机,日志模块和一致性模块。功能分别是:

  • 状态机:数据一致性指的即是状态机的一致性,从内部服务看表现为状态机中的数据都保持一致
  • log模块:保存了所有的操作记录
  • 一致性模块:一致性模块算法保证写入log命令的一致性,是raft的核心内容。

实现一致性的过程可分为Leader选举(Leader election)日志同步(Log replication),安全性(safty),日志压缩(Log compaction)成员变更(membership change)

leader 选举

竞选过程

  • 节点由Follower变为Candidate,同时设置当前Term。
  • Candidate给自己投票,带上termid 和日志序号,同时向其他节点发送拉票请求
  • 等待结果,成为Leader,follower 或者在选举未成为产生结果的情况下节点状态保持为Candidatae。

选举结果

  • 成功当选收到超过半数的选票时,成为Leader,定时给其他节点发送心跳,并带上任期id,其他节点发现当前的任期id小于接收到leader发送过来的id,则将将状态切换至follower.
  • 选举失败在Candidate状态接收到其他节点发送的心跳信息,且心跳中的任期id大于自己,则变为follower。
  • 未产生结果没有一个Candidate所获得的选票超过半数,未产生leader,则Candidate再进入下一轮投票。为了避免长期没有leader产生,raft采用如下策略避免:
  • 选举超时时间为随机值,第一个超时的节点带着最大的任期id立刻进入新一任的选举
  • 如果存在多个Candidate同时竞选的情况,发送拉票请求也是一段随机延时。

日志同步(Log Replication)

image

Leader选出后接受客户端请求,Leader把请求日志作为日志条目加入到日志中,然后向其他Follower节点复制日志,但超过半数的日志复制成功,则Leader将日志应用到状态机并向客户端返回执行结果,同时Follower也将结果提交。如果存在Follower没有成功复制日志,Leader会无限重试。

日志同步的关键点:

  • 日志由有序编号的日志条目组成,每条日志包含创建的任期和用于执行的命令,日志是保证所有节点数据一致的关键。
  • Leader 负责一致性检查,同时让所有的Follower都和自己保持一致。
  • 在Leader发生切换时,如何保证各节点日志一致。leader为每一个follower维护一个nextIndex,将index和termid信息发送至follower,从缺失的termid和index 为follow 补齐数据,直至和leader完全一致。
  • 只允许主节点提交包含当前term的日志。否则会出现已经commit的日志出现更改的情况

安全性

安全性的原则是一个term只有一个leader,被提交至状态机的数据不能发生更改。保证安全性主要通过限制leader的选举来保证:

  • Candidate在拉票时需要携带本地已持久化的最新的日志信息,如果投票节点发现本地的日志信息比Candidate更新,则拒绝投票。
  • 只允许Leader提交当前Term的日志。
  • 拥有最新的已提交的log entry的Follower才有资格成为Leader。

raft协议实现

raft的golang的开源实现主要包含两个:coreOS的raft实现 , 使用的项目如tidbcockroachdb这两个经典的newsql。另外一个是hashicrop的raft实现,使用的项目如服务发现解决方案consul和时序数据库influxdb。对比二者的实现主要有如下特点:

  • hashicrop的实现完整度高,包含了snapshot,wal,storage等,在集成时只需要关注业务逻辑
  • etcd中的raft模块则是raft协议的轻量级实现,对于上述功能只定义了相关interface,需要业务方去具体实现,优点是增加灵活性,etcdserver就是集成raft算法并实现snapshot,wal,storage这样一个应用程序。

etcd/raft 代码结构

  • 日志持久化
    • storage.go:持久化日志保存模块,以interface的方式定义了实现的方式,并基于内存实现了memoryStorage用于存储日志数据。
    • log.go:raft算法日志模块的逻辑
    • log_unstable.go:raft 算法的日志缓存,日志优先写缓存,待状态稳定后进行持久化
  • 节点
    • node.go: raft集群节点行为的实现,定义了各节点通信方式
    • process.go:从leader的角度,为每个follower维护一个子状态机,根据状态的切换决定leader该发什么消息给Follower.
  • Raft算法
    • raft.go:raft算法的具体逻辑实现,每个节点都有一个raft实例
    • read_only.go: 实现了线性一致读(linearizable read),线性一致读要求读请求读到最新提交的数据。针对raft存在的stale read(多leader场景),此模块通过ReadIndex的方式保证了一致性。

etcd/raft的实现分析

分析raft的实现流程,我们可以从raft的几个核心问题入手:

  • 如何选举leader?
  • 如何实现log的复制?
  • 如何进行leadership的transfer?
  • 如何实现线性一致读?

其中leader的选举、log复制和线性一致读是raft协议的最基本要求,而leadership的转移在工程实践中有重大意义。

核心数据结构

  • struct node node 中主要定义一系列channel,raft的实现就是通过channel 传递消息,当节点启动通过select机制监听上述channel确定相应的状态切换。
// node is the canonical implementation of the Node interface
type node struct {
	propc      chan msgWithResult
	recvc      chan pb.Message
	confc      chan pb.ConfChange
	confstatec chan pb.ConfState
	readyc     chan Ready
	advancec   chan struct{}
	tickc      chan struct{}
	done       chan struct{}
	stop       chan struct{}
	status     chan chan Status

	logger Logger
}
  • interface node定义了node要实现raft算法必须实现的方法
type Node interface {
	Tick() //时钟的实现,选举超时和心跳超时基于此实现
	Campaign(ctx context.Context) error //参与leader竞争
	Propose(ctx context.Context, data []byte) error //在日志中追加数据,需要实现方保证数据追加的成功
	ProposeConfChange(ctx context.Context, cc pb.ConfChange) error // 集群配置变更
	Step(ctx context.Context, msg pb.Message) error //根据消息变更状态机的状态
	//标志某一状态的完成,收到状态变化的节点必须提交变更
	Ready() <-chan Ready
	//进行状态的提交,收到完成标志后,必须提交过后节点才会实际进行状态机的更新。在包含快照的场景,为了避免快照落地带来的长时间阻塞,允许继续接受和提交其他状态,即使之前的快照状态变更并没有完成。
	Advance()
	//进行集群配置变更
	ApplyConfChange(cc pb.ConfChange) *pb.ConfState
	//变更leader
	TransferLeadership(ctx context.Context, lead, transferee uint64)
	//保证线性一致性读,
	ReadIndex(ctx context.Context, rctx []byte) error
	//状态机当前的配置
	Status() Status
	// ReportUnreachable reports the given node is not reachable for the last send.
	//上报节点的不可达
	ReportUnreachable(id uint64)
	//上报快照状态
	ReportSnapshot(id uint64, status SnapshotStatus)
	//停止节点
	Stop()
}

节点的启动和运行

节点初始化raft,读取配置启动各个各个节点,初始化logindex.启动后 以for-loop方式循环运行,用select 机制监听不同的channel 实现对状态变化的监听,并执行相应动作。

//启动
func StartNode(c *Config, peers []Peer) Node {
	r := newRaft(c) //初始化raft算法实例
	r.becomeFollower(1, None)
	//将配置中的节点加入集群
	for _, peer := range peers {
		...
	}
	//初始化logindex
	r.raftLog.committed = r.raftLog.lastIndex()
	for _, peer := range peers {
		//初始化节点状态机(progress)
		r.addNode(peer.ID)
	}
	n := newNode()
	n.logger = c.Logger
	go n.run(r)
	return &n
}

//运行
func (n *node) run(r *raft) {
	...

		select {
		//接收到写消息
		case pm := <-propc:
			...
		//接收到readindex 请求
		case m := <-n.recvc:
			...
		//配置变更
		case cc := <-n.confc:
			...
		//超时时间到,包括心跳超时和选举超时等
		case <-n.tickc:
			...
		//数据ready
		case readyc <- rd:
			...
		//可以进行状态变更和日志提交
		case <-advancec:
			...
		//节点状态信号
		case c := <-n.status:
			...
		//收到停止信号
		case <-n.stop:
			...
		}
	}
}

leader 选举

初始化node为follower,设置任期为1,并初始化tickElection函数,这是实际参与选举的函数,同时也初始化step为stepFollower,这是作为follower的核心信息处理函数,后续选举,日志复制和快照等功能都基于此函数进行:

	r := newRaft(c)
	r.becomeFollower(1, None)

当节点接收leader的heartbeat超时时(每个节点都有随机的超时时间),会触发run函数中的tickc这个channel。发送MsgHup消息,并调用campaign参选, 将自身设置为candidate,并递增currentTerm,向其他节点发送竞选消息。其他节点通过监听propc channel获取其他节点发送的投票消息,并调用Step对消息进行判断,选择是否投票。

其中投票的判断逻辑主要分两步:1.如果投票信息中的任期id 是否 小于自身的id,则直接返回nil。2.通过isUpToDate判断能否投票,通过和本地已存在的最新log比较,首先要有最大任期id,如果任期id相同则要求有最大的logindex。

candidate节点收到其他节点的回复后,判断获取的票数是否超过半数,如果是则设置自身为leader,否则为follower。

func (n *node) run(r *raft) {
    ...
    for {
        select {
            ...
            //触发heartbeat 超时
            case <-n.tickc:
			    r.tick()
            ...
        }
    }
}

//超时触发选举
func (r *raft) tickElection() {
	r.electionElapsed++

	if r.promotable() && r.pastElectionTimeout() {
		r.electionElapsed = 0
		r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
	}
}

//随机超时时间
func (r *raft) pastElectionTimeout() bool {
	return r.electionElapsed >= r.randomizedElectionTimeout
}

func (r *raft) resetRandomizedElectionTimeout() {
	r.randomizedElectionTimeout = r.electionTimeout + globalRand.Intn(r.electionTimeout)
}


//参与选举
func (r *raft) campaign(t CampaignType) {
	var term uint64
	var voteMsg pb.MessageType
	//成为candicate,将任期id加1
	if t == campaignPreElection {
		r.becomePreCandidate()
		voteMsg = pb.MsgPreVote
		term = r.Term + 1
	} else {
		r.becomeCandidate()
		voteMsg = pb.MsgVote
		term = r.Term
	}
	//判断获取的票数是否超过半数,如果是当选为leader
	if r.quorum() == r.poll(r.id, voteRespMsgType(voteMsg), true) {
		if t == campaignPreElection {
			r.campaign(campaignElection)
		} else {
			r.becomeLeader()
		}
		return
	}
	//向其他节点发送竞选消息
	for id := range r.prs {
		if id == r.id {
			continue
		}
		var ctx []byte
		if t == campaignTransfer {
			ctx = []byte(t)
		}
		r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
	}
}

//节点投票过程
func (r *raft) Step(m pb.Message) error {
	...
	//比较任期id
	case m.Term > r.Term:
		if m.Type == pb.MsgVote || m.Type == pb.MsgPreVote {
			force := bytes.Equal(m.Context, []byte(campaignTransfer))
			inLease := r.checkQuorum && r.lead != None && r.electionElapsed < r.electionTimeout
			if !force && inLease {
				return nil
			}
		}
		
	switch m.Type {
	case pb.MsgVote, pb.MsgPreVote:
		...
		//与本地最新的持久化日志比较
		if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
			//发送投票信息
			r.send(pb.Message{To: m.From, Term: m.Term, Type: voteRespMsgType(m.Type)})
			if m.Type == pb.MsgVote {
				// Only record real votes.
				r.electionElapsed = 0
				r.Vote = m.From
			}
		} 
		...
	return nil
}

func (l *raftLog) isUpToDate(lasti, term uint64) bool {
	return term > l.lastTerm() || (term == l.lastTerm() && lasti >= l.lastIndex())
}

//投票结果判断
	case myVoteRespType:
		gr := r.poll(m.From, m.Type, !m.Reject)
		//计算票数是否超过半数
		switch r.quorum() {
		case gr:
			if r.state == StatePreCandidate {
				r.campaign(campaignElection)
			} else {
				r.becomeLeader()
				r.bcastAppend()
			}
		case len(r.votes) - gr:
			r.becomeFollower(r.Term, None)
		}

日志复制

node节点为外界提供了日志提交接口 Propose,在ectd的server对该接口进行了封装。Propose 内部具体调用stepWithWaitOption实现日志消息的传递,并阻塞/非阻塞地等待结果的返回。

func (n *node) Propose(ctx context.Context, data []byte) error {
	return n.stepWait(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
}

func (n *node) stepWithWaitOption(ctx context.Context, m pb.Message, wait bool) error {
    ...
	//提交日志数据至 node的 propc channel 队列
	select {
	case ch <- pm:
		if !wait {
		    //非阻塞直接返回
			return nil
		}
	case <-ctx.Done():
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
	select {
	//等待结果的返回
	case rsp := <-pm.result:
		if rsp != nil {
			return rsp
		}
	case <-ctx.Done():
		return ctx.Err()
	case <-n.done:
		return ErrStopped
	}
	return nil
}

proc消息进入stepFollower处理,因为只有leader才能处理客户端提交的信息,因此将消息的接收者设置为leader后转发。在stepLeader中调用appendEntry将消息追到leader的raftLog之中,但不进行数据的commit。之后调用bcastAppend 将消息广播至其他follower节点。

func stepLeader(r *raft, m pb.Message) error {
	case pb.MsgProp:
		...
		if !r.appendEntry(m.Entries...) {
			return ErrProposalDropped
		}
		r.bcastAppend()
		...
}

follower节点接收到请求后,调用handleAppendEntries判断是否接受leader提交的日志。判断逻辑如下:如果leader提交的logindex小于本地已经提交的logindex则将本地的logindex回复给leader。查找追加的日志和本地log的冲突,如果有冲突,则先找到冲突的位置,用leader的日志从冲突位置开始进行覆盖,日志追加成功后,返回最新的logindex至leader。如何任期信息不一致,则直接拒绝leader的追加请求。

func (r *raft) handleAppendEntries(m pb.Message) {
    //leader提交的logindex小于本地已经提交的logindex
	if m.Index < r.raftLog.committed {
		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
		return
	}
	//追加日志,可能存在冲突的情况,需要找到冲突的位置用leader的日志进行覆盖
	if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
	    //mlastIndex表示最佳成功的最新位置
		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
	} else {
	    //任期信息不一致,拒绝此次追加请求,并把最新的logindex回复给leader,便于进行追加
		r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: m.Index, Reject: true, RejectHint: r.raftLog.lastIndex()})
	}
}

leader接收到follower的请求后,针对拒绝和接收的两个场景有不同的处理逻辑,这也是保证follower一致性的关键环节

  • follower 正常接收append请求 当leader 确认follower已经接收了append请求后,则调用maybeCommit进行提交,在提交过程中确认各个节点的matchindex,排序后取中间值比较,如果中间值都都比本地的commitindex大,就认为超过半数已经认可此次提交,可以进行commit,之后调用sendAppend向所有节点广播消息,follower接收到请求后调用maybeAppend进行日志的提交。值得注意的是,日志的append过程可能由于之前的请求被拒绝,等待snapshot或者消息发送窗口(inflight)已满导致中止,这时需要重新向follower节点发送最新的append请求。
   func stepLeader(r *raft, m pb.Message) error {
	    case pb.MsgAppResp:
		    pr.RecentActive = true

		    if m.Reject {
			...
		} else {
			oldPaused := pr.IsPaused()
			//更新索引信息,更新该follower的match index 和next index.
			if pr.maybeUpdate(m.Index) {
				switch {
				//日志追加成功,状态由复制探测状态变成复制状态,加快日志的追加
				case pr.State == ProgressStateProbe:
					pr.becomeReplicate()
				case pr.State == ProgressStateSnapshot && pr.needSnapshotAbort():
					r.logger.Debugf("%x snapshot aborted, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
					pr.becomeProbe()
				//pr.ins用于限制消息发送的速率,用于统计当前处于发送状态的日志数量
				case pr.State == ProgressStateReplicate:
					pr.ins.freeTo(m.Index)
				}
				//leader进行本地的提交
				if r.maybeCommit() {
					//广播至所有follower 通知进行log的提交
					r.bcastAppend()
				} else if oldPaused {
					//append请求被中止,则重新发送最新的请求
					r.sendAppend(m.From)
				}
				}
			}
		}
    }
    
    func (r *raft) maybeCommit() bool {
		if cap(r.matchBuf) < len(r.prs) {
			r.matchBuf = make(uint64Slice, len(r.prs))
		}
		mis := r.matchBuf[:len(r.prs)]
		idx := 0
		for _, p := range r.prs {
			mis[idx] = p.Match
			idx++
		}
		//排序取取中间值
		sort.Sort(mis)
		mci := mis[len(mis)-r.quorum()]
		return r.raftLog.maybeCommit(mci, r.Term)
	}
	
	func (l *raftLog) maybeCommit(maxIndex, term uint64) bool {
		//match的中间值是否已经大于本地已经commit的matchindex
		if maxIndex > l.committed && l.zeroTermOnErrCompacted(l.term(maxIndex)) == term {
			l.commitTo(maxIndex)
			return true
		}
		return false
	}
  • follower拒绝leader的append请求 在异常情况下,follower会拒绝leader的append请求。其判断逻辑主要位于matchTerm,当leader append请求中的logindex在当前节点已提交的日志中到不到对应的任期,或者任期与leader提交的任期不一致时follower会拒绝当前append请求。leader接收到拒绝请求后会进入探测状态,探测follower最新匹配的位置。
   //follower接收leader的请求
   func (l *raftLog) maybeAppend(index, logTerm, committed uint64, ents ...pb.Entry) (lastnewi uint64, ok bool) {
       if l.matchTerm(index, logTerm) {
   	    ...
       }
       //拒绝leader当前的append请求
       return 0, false
    }
   //对leader提交append请求中的logindex和termid进行判断
   func (l *raftLog) matchTerm(i, term uint64) bool {
       t, err := l.term(i)
       if err != nil {
   	    return false
       }
       return t == term
    }
    
    func stepLeader(r *raft, m pb.Message) error {
       case pb.MsgAppResp:
   	    pr.RecentActive = true
   
   	    if m.Reject {
   		    if pr.maybeDecrTo(m.Index, m.RejectHint) {
   			    //由复制状态进入探测状态,探测follower最新的匹配位置
   			    if pr.State == ProgressStateReplicate {
   				    pr.becomeProbe()
   			    }
   			    r.sendAppend(m.From)
   		    }
    }

下面来分析leader接收到拒绝请求后的处理逻辑。由于各种原因可能导致follower节点的日志与leader不一致,如下图所示:

日志同步

在raft的论文中提出通过遍历index和term的方式保证日志的一致性。具体的实现位于maybeDecrTo,因为follower在拒绝请求时带上了当前最新的logindex,因此在进行日志补推时,直接将next至为follower中最新的logindex 和当前index中的最小值。

func (pr *Progress) maybeDecrTo(rejected, last uint64) bool {
       if pr.State == ProgressStateReplicate {
   	    if rejected <= pr.Match {
   		    return false
   	    }
   	    // directly decrease next to match + 1
   	    //复制状态将pr的next置为当前匹配位置+1
   	    pr.Next = pr.Match + 1
   	    return true
       }
   
       if pr.Next-1 != rejected {
   	    return false
       }
   
    	//如果是探测状态,则将next置为follower中最新的logindex  和当前index中的最小值。
       if pr.Next = min(rejected, last+1); pr.Next < 1 {
   	    pr.Next = 1
       }
       pr.resume()
       return true
    }
      日志推送的具体实现位于maybeSendAppend.func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
	pr := r.getProgress(to)
	if pr.IsPaused() {
		return false
	}
	m := pb.Message{}
	m.To = to

	//发送给follower的最后一条日志对应的任期
	term, errt := r.raftLog.term(pr.Next - 1)
	//需要发送给follower的日志条数
	ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
	if len(ents) == 0 && !sendIfEmpty {
		return false
	}

	if errt != nil || erre != nil { // send snapshot if we failed to get term or entries
			...
	} else {
		m.Type = pb.MsgApp
		m.Index = pr.Next - 1
		m.LogTerm = term
		m.Entries = ents
		//leader 已经提交的最新index
		m.Commit = r.raftLog.committed
		if n := len(m.Entries); n != 0 {
			switch pr.State {
			//在日志复制状态,乐观地增加next, 加快日志的推送速度
			case ProgressStateReplicate:
				last := m.Entries[n-1].Index
				pr.optimisticUpdate(last)
				pr.ins.add(last)
			case ProgressStateProbe:
				pr.pause()
			default:
				r.logger.Panicf("%x is sending append in unhandled state %s", r.id, pr.State)
			}
		}
	}
	r.send(m)
	return true
}

至此raft集群的日志复制基本已经完成,但是仅限于raft协议层面,日志和快照目前还是保存在Ready结构中,并放入了readyc队列,等待上游的模块处理。之前提到过etcd-raft 只是协议层的实现,提供了WAL,snapshot和storage等模块的扩展接口,应用层需要实现上述接口最终实现的数据的落地。

func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
		...
		//日志数据
		rd := Ready{
			Entries:          r.raftLog.unstableEntries(),
			CommittedEntries: r.raftLog.nextEnts(),
			Messages:         r.msgs,
		}
		...
	}

leadership transfer

leadership transfer 指的是leader身份的转换,raft提供接口允许客户端进行leader切换,此功能可用来做负载均衡,让客户端有机会结合实际的机器和负载情况去选择最优的leader;同时也是multi-raft实现的基础。下面具体分析transfer的实现。

raft协议提供了transferLeaderShip方法供应用层使用用于触发leader的转换,transferLeaderShip会发送MsgTransferLeader类型消息至recvc消息队列中(channel)。当follower收到TransferLeader消息后不处理将消息转发至leader进行处理。

 //etcd/raft/raft.go
 func (n *node) TransferLeadership(ctx context.Context, lead, transferee uint64) {
	    select {
    	//通过recvc发送MsgTransferLeader消息至集群中节点
	    case n.recvc <- pb.Message{Type: pb.MsgTransferLeader, From: transferee, To: lead}:
	    case <-n.done:
	    case <-ctx.Done():
	    }
    }

leader收到transfer消息后,如果发现当前正在进行leader切换或者不发生leader变换则直接放弃。一个节点要成为leader的要求是有最新的日志数据。如果有则立即发送MsgTimeoutNow消息,transfee收到消息后立即调用campaign方法进行选择,而不是像正常leader选举时需要等待超时,而且也不需要采用预投票的方式,之后的选举流程与正常选举过程一致。如果transfee没有最新的日志数据,则leader进行日志的同步,当同步完成收到回复且正处在leader transfer的过程中,发送MsgTimeoutNow,之后与上述流程一致。

 //etcd/raft/raft.go
 func stepLeader(r *raft, m pb.Message) error {	
		switch m.Type {
		...
		case pb.MsgTransferLeader:
			if pr.IsLearner {
				r.logger.Debugf("%x is learner. Ignored transferring leadership", r.id)
				return nil
			}
			leadTransferee := m.From
			lastLeadTransferee := r.leadTransferee
			//上一次transfer正在进行
			if lastLeadTransferee != None {
				if lastLeadTransferee == leadTransferee {
					r.logger.Infof("%x [term %d] transfer leadership to %x is in progress, ignores request to same node %x",
						r.id, r.Term, leadTransferee, leadTransferee)
					return nil
				}
				r.abortLeaderTransfer()
				r.logger.Infof("%x [term %d] abort previous transferring leadership to %x", r.id, r.Term, lastLeadTransferee)
			}
			//transfee和当前leader相同
			if leadTransferee == r.id {
				r.logger.Debugf("%x is already leader. Ignored transferring leadership to self", r.id)
				return nil
			}
			// Transfer leadership to third party.
			// Transfer leadership should be finished in one electionTimeout, so reset r.electionElapsed.
			r.electionElapsed = 0
			r.leadTransferee = leadTransferee
			if pr.Match == r.raftLog.lastIndex() {
				//transfee的日志已经是最新和leader保持一致了,则立刻发送MsgTimeoutNow,触发选举
				r.sendTimeoutNow(leadTransferee)
				r.logger.Infof("%x sends MsgTimeoutNow to %x immediately as %x already has up-to-date log", r.id, leadTransferee, leadTransferee)
			} else {
				//日志非最新进行日志的同步
				r.sendAppend(leadTransferee)
			}
		}
		return nil
	}

线性一致读

线性一致性读是分布式系统的基本要求,在raft中leader和follower都可以接受读请求,但在以下场景下可能出现数据的不一致:

  • Leader和Follower复制期间的状态不一致
  • 因为网络分区导致多个leader的存在,不同leader间的状态不一致,即脑裂(split-brain)现象。如果请求分别被新旧leader处理,所得的结果也不一致

为解决raft的线性一致读问题,etcd-raft提供了两种实现方案:

  • ReadIndex(ReadOnlySafe)。其原理是接收到客户端请求后,向集群发起ReadIndex请求来读取commitedIndex,Leader收到请求后向节点发送心跳,当收到大多数节点的确认自己仍是leader后,回复ReadIndex请求并告知最新的commitedIndex。ReadIndex是etcd-raft的默认方案。
  • Lease read方案(ReadOnlyLeaseBased)。其原理是通过维护leader的租期,确认leader的唯一性,不需要通过心跳来进行leader的确认。其风险在于需要全局一直的时钟来保证lease机制的准确性。etcd-raft不推荐采用此方案,pingcap开源的分布式数据库tidb中的pd 模块在实现TSO(Timestamp Oracle)的前提下,采用此方案。
ReadIndex实现分析

在raft初始化的过程中完成了linearizable read的配置,包括需要采用的方案。

   func newRaft(c *Config) *raft {
   	...
   	}
   	r := &raft{
   		id:                        c.ID,
   		...
   		//初始化readOnly配置
   		readOnly:                  newReadOnly(c.ReadOnlyOption),
   		disableProposalForwarding: c.DisableProposalForwarding,
   	}
   }
   
   const (
   	//ReadIndex方案
   	ReadOnlySafe ReadOnlyOption = iota
   	//leaseRead方案
   	ReadOnlyLeaseBased
   )

阻塞的recvc channel收到ReadIndex请求后,将请求加入队列,初始化ReadIndex状态。之后发送广播心跳。

   	func stepLeader(r *raft, m pb.Message) error {
   	switch m.Type {
   	...
   	case pb.MsgReadIndex:
   			switch r.readOnly.option {
   			case ReadOnlySafe:
   				//加入请求队列
   				r.readOnly.addRequest(r.raftLog.committed, m)
   				//广播心跳消息
   				r.bcastHeartbeatWithCtx(m.Entries[0].Data)
   			}
   		} else {
   			r.readStates = append(r.readStates, ReadState{Index: r.raftLog.committed, RequestCtx: m.Entries[0].Data})
   		}
   	}
   }
   
   func (ro *readOnly) addRequest(index uint64, m pb.Message) {
   	ctx := string(m.Entries[0].Data)
   	if _, ok := ro.pendingReadIndex[ctx]; ok {
   		return
   	}
   	//index是当前集群的committedIndex,acks 用来收集节点心跳回复包
   	ro.pendingReadIndex[ctx] = &readIndexStatus{index: index, req: m, acks: make(map[uint64]struct{})}
   	ro.readIndexQueue = append(ro.readIndexQueue, ctx)
   }

当leader收到心跳回复后,对心跳进行统计,如果是本地请求直接将消息追加到readstatus中,最终会由newReady函数将消息发送到readyc channel,监听ready channel的客户端会最终回复请求。

   	func stepLeader(r *raft, m pb.Message) error {
   	case pb.MsgHeartbeatResp:
   		...
   		}
   		//统计回复结果,如果未超过半数则直接返回
   		ackCount := r.readOnly.recvAck(m)
   		if ackCount < r.quorum() {
   			return nil
   		}
   
   		rss := r.readOnly.advance(m)
   		for _, rs := range rss {
   			req := rs.req
   			//如果是本地的请求
   			if req.From == None || req.From == r.id { // from local member
   				r.readStates = append(r.readStates, ReadState{Index: rs.index, RequestCtx: req.Entries[0].Data})
   			} else {
   				//如果是来自follower的请求,将结果返回给follower
   				r.send(pb.Message{To: req.From, Type: pb.MsgReadIndexResp, Index: rs.index, Entries: req.Entries})
   			}
   		}
   	}
   	
   	func newReady(r *raft, prevSoftSt *SoftState, prevHardSt pb.HardState) Ready {
   		rd := Ready{
   			Entries:          r.raftLog.unstableEntries(),
   			CommittedEntries: r.raftLog.nextEnts(),
   			Messages:         r.msgs,
   		}
   		...
   		//readIndex消息追加
   		if len(r.readStates) != 0 {
   			rd.ReadStates = r.readStates
   		}
   		rd.MustSync = MustSync(r.hardState(), prevHardSt, len(rd.Entries))
   		return rd
   	}
   	
   	func (n *node) run(r *raft) {
   		....
   		for {
   			if advancec != nil {
   				readyc = nil
   			} else {
   				//消息加入readyc队列
   				rd = newReady(r, prevSoftSt, prevHardSt)
   				if rd.containsUpdates() {
   					readyc = n.readyc
   				} else {
   					readyc = nil
   				}
   			}
   			....
   	}	

如果是follower接收到ReadIndex请求,直接将消息转发至leader,leader按上述流程处理,follower接收到消息后采用上述类似机制加入readyc队列,异步回复客户端。

   func stepFollower(r *raft, m pb.Message) error {	
   	...
   	case pb.MsgReadIndex:
   		if r.lead == None {
   			r.logger.Infof("%x no leader at term %d; dropping index reading msg", r.id, r.Term)
   			return nil
   		}
   		//将ReadIndex请求转发给leader
   		m.To = r.lead
   		r.send(m)
   	case pb.MsgReadIndexResp:
   		if len(m.Entries) != 1 {
   			r.logger.Errorf("%x invalid format of MsgReadIndexResp from %x, entries count: %d", r.id, m.From, len(m.Entries))
   			return nil
   		}
   		//收到leader回复后将消息加入readStatus
   		r.readStates = append(r.readStates, ReadState{Index: m.Index, RequestCtx: m.Entries[0].Data})
   		...
   }

总结

本文从raft算法的基本原理出发,简单的分析了leader选举和日志复制的实现过程。之后从工程实践的角度出发分析了etcd-raft的代码实现,重点剖析了leader选举,日志复制,leadership transfer和线性一致读的核心流程。而raft算法博大精深,etcd也是工业级的完整实现,除了本文介绍的几个核心环节外,leader的预选举(prevote)、节点成员变更、配置变更和日志的批量追加等也是raft的关键环节,因篇幅所限就不再一一介绍。

参考

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 从0到1,手把手教你入门 etcd

    etcd 是 CoreOS 团队于 2013 年 6月发起的开源项目,它的目标是构建一个高可用的分布式键值(key-value)数据库。etcd 内部采用raf...

    民工哥
  • k8s etcd 的实现原理

    etcd 的官方将它定位成一个可信赖的分布式键值存储服务,它能够为整个分布式集群存储一些关键数据,协助分布式集群的正常运转。

    AlbertZhang
  • 今日推荐:awesome-architecture

    但是这条路还是有很多人走,而且也留下了相应的封神之法,今天推荐的就是一个相当详细的架构师框架学习图。内容很充实,看目录的时候,滚动条滚了很多次!学习起来肯定也不...

    仇诺伊
  • ETCD:从应用场景到实现原理的全方位解读 转

    随着CoreOS和Kubernetes等项目在开源社区日益火热,它们项目中都用到的etcd组件作为一个高可用强一致性的服务发现存储仓库,渐渐为开发人员所关注。在...

    henrylee2cn
  • ETCD 简介 + 使用

    随着CoreOS和Kubernetes等项目在开源社区日益火热,它们项目中都用到的etcd组件作为一个高可用强一致性的服务发现存储仓库,渐渐为开发人员所关注。在...

    菲宇
  • 三年之久的 etcd3 数据不一致 bug 分析

    “ etcd 作为 Kubernetes 集群的元数据存储,是被业界广泛使用的强一致性 KV 存储,但近日被挖掘出一个存在 3 年之久的数据不一致 bug—...

    腾讯云原生
  • etcd原理及实践

    原文地址:https://www.cnblogs.com/panpanwelcome/p/8242418.html

    二狗不要跑
  • etcd使用入门

    etcd是CoreOS团队于2013年6月发起的开源项目,它的目标是构建一个高可用的分布式键值(key-value)数据库。etcd内部采用raft协议作为一致...

    菲宇
  • 让我们一起揭开Etcd背后Raft算法的面纱

    最近在使用K8S过程中,一直用到了一个Key-Value数据库Etcd,每当看到有介绍Etcd的教程时,介绍不多,大多都是独立于K8S集群之外,保存状态数据。再...

    用户5166556
  • 01 . etcd简介原理,应用场景及部署,简单使用

    https://developer.aliyun.com/article/765312

    常见_youmen
  • 斗鱼直播云原生实践之注册中心篇

    孔令圳,斗鱼首席架构师,全面负责斗鱼全站技术架构体系规划和建设,10 余年中大型互联网产品架构经验,擅长高并发、高可用场景下的架构与方案设计。 于竞,斗鱼技术...

    腾讯云原生
  • golang-etcd系列(一)--初识

    etcd 是一个 golang 编写的分布式、高可用的一致性键值存储系统,是目前容器编排领域火热的 Kubernetes(k8s) 内置的服务发现与节点一致性中...

    astraw99
  • etcd:用于服务发现的键值存储系统

    etcd是一个高可用的键值存储系统,主要用于共享配置和服务发现。etcd是由CoreOS开发并维护的,灵感来自于 ZooKeeper 和 Doozer,它使用G...

    张善友
  • Kubernetes之etcd数据库

    etcd是CoreOS团队于2013年6月发起的开源项目,它的目标是构建一个高可用的分布式键值(key-value)数据库。etcd内部采用raft协议作为一致...

    聂伟星
  • Ubuntu 搭建etcd

    etcd是一个高可用的分布式键值(key-value)数据库。etcd内部采用raft协议作为一致性算法,etcd基于Go语言实现。

    py3study
  • TiDB 5.0 跨中心部署能力初探 | Joint Consensus 助力 TiDB 5.0 无畏调度

    TiDB 5.0 已于上周正式发布,在这个大版本更新中提升 TiDB 集群的跨中心部署能力是一个重要的着力点,在共识算法这一层,最激动人心莫过于 Joint C...

    PingCAP
  • 服务发现之 Etcd VS Consul

    在分布式微服务架构中,一个应用可能由一组职责单一化的服务组成。这时候就需要一个注册服务的机制,注册某个服务或者某个节点是可用的,还需要一个发现服务的机制来找到哪...

    spilledyear
  • ETCD入门指南

    之前对etcd不是很了解,于是下定决心学习一下。随手把过程记录了一下,希望对大家有帮助。

    没有故事的陈师傅
  • 关于 etcd 的一些谣言

    这是一个被广为流传的误解,众所周知 etcd 使用 Raft 协议来解决数据一致性问题。一个 Raft Group 只能有一个 Leader 存在,如果一旦发生...

    poslua

扫码关注云+社区

领取腾讯云代金券