Raft协议实现之etcd(三)：日志同步
前言
Etcd集群通过Raft协议的选举机制产生一个Leader，这样客户端在对etcd存储的数据进行更新时，就可以直接将请求发给Leader。一旦回复客户端成功，则表示数据更新完成，即使因为故障导致Leader后续无法响应，客户端也可以从新选出的Leader上读到正确的数据。在etcd的实现中，通过两点来保证这一点，通过Raft协议的日志复制保证数据更新请求同步到超过半数节点后提交，通过KV存储提供带事务的数据更新。所以如果返回成功的结果给客户端，则数据一定已经写入Leader的KV存储中。
KV存储实现不属于Raft范畴，Raft只保证日志复制的一致性，这篇文章通过etcd的源码解析来看下raft日志复制的实现，以及etcd如何实现日志的持久存储。Raft对于日志复制的一致性保证请参考之前的原理解析文章（传送门）。
存储设计
Etcd中跟存储部分相关的模块主要有3块，Raft状态机中存储的日志条目、持久化到文件的日志条目以及后端的KV存储。
Raft状态机存储
回顾下第一篇中讲到的Etcd整体架构，raft模块只负责算法实现，所以所有收到的日志条目都是存在内存中。数据结构如下：
上图中，所有日志条目都是存储在一个raftLog的结构中.
type raftLog struct {
// 自从上次快照后已经持久化的日志
storage Storage
// 还未持久化的日志
unstable unstable
// 集群中已提交的日志index
committed uint64
// 本节点已经应用到状态机的日志index
applied uint64
...
...
}
- raftLog中通过两个字段来存储日志，storage存储了已经持久化到磁盘的日志和最近一次快照的信息，也就是上图中已经写到了WAL中的数据。这部分日志即使在节点重启的情况下也不会丢失，重启时etcd会从wal中读取出这部分数据写到raft的内存中。
为啥是上次快照之后的呢？因为raft节点的内存毕竟是有限的，etcd中会定期对KV做快照，快照结束之后，storage就只需要存储快照的信息和在快照之后接收的日志就可以了，这在raft协议中也有定义。 - unstable结构中存储了尚未持久化的日志条目和快照，当日志持久化之后就会从unstable中移到storage中。
- raft协议的committed和applied属性也存在raftLog中，因为根据raft协议的规定，这两个属性也是需要持久化存储的。
Storage
raft状态机Storage接口定义如下：
type Storage interface {
// 已经持久化的HardState和ConfState信息
InitialState() (pb.HardState, pb.ConfState, error)
// 返回日志条目
Entries(lo, hi, maxSize uint64) ([]pb.Entry, error)
// 当前的选举周期
Term(i uint64) (uint64, error)
// 最后一条日志的index
LastIndex() (uint64, error)
// 第一条日志的index
FirstIndex() (uint64, error)
// 返回最近一次Snapshot
Snapshot() (pb.Snapshot, error)
}
Storage接口定义了所有Raft协议中要求的需要持久化的信息接口，比如HardState中的term、commitIndex，以及日志条目等。
etcd中对于该接口的默认实现是MemoryStorage，从名字可以看出来数据是存在内存中的，看起来这么做跟raft的要求不符。这是因为MemoryStorage中存的日志和状态信息都在WAL中，所以这里只需要内存就够了，重启的时候，etcd会从WAL中恢复数据写道Storage中。MemoryStorage的定义如下：
type MemoryStorage struct {
// 读写锁
sync.Mutex
// term，commitIndex，vote封装在HardState中
hardState pb.HardState
//最近一次Snapshot
snapshot pb.Snapshot
//snapshot之后的日志条目，第一条日志条目的index为snapshot.Metadata.Index
ents []pb.Entry
}
unstable
Raft模块已经收到还没有持久化到WAL的日志条目存在unstable中
type unstable struct {
// 从leader收到的snapshot
snapshot *pb.Snapshot
// 新收到还未持久化的日志条目
entries []pb.Entry
//第一条日志的偏移量
offset uint64
}
日志持久化存储
Raft模块日志数据持久化通过WAL实现，WAL通过追加写的方式来将数据写入磁盘以提高性能。Etcd在如下几种情况下会在WAL追加一条记录：
- 节点启动时记录节点和集群信息，对应的记录类型是metadataType；
- 收到新的日志条目，对应的记录类型是entryType；
- 状态变化时，比如新的选举周期，commitIndex变化等，对应的记录类型是stateType；
- 做数据快照时，对应的记录类型是snapshotType；
- 生成新的wal文件时，wal文件达到一定大小时，etcd就会生成一个新的文件，新的文件第一条记录会记录上一个文件的crc，以备数据校验。对应的记录类型是crcType
type WAL struct {
lg *zap.Logger
// wal文件的存储目录
dir string
dirFile *os.File
// wal文件构建后会写的第一个metadata记录
metadata []byte
// wal文件构建后会写的第一个state记录
state raftpb.HardState
// wal开始的snapshot，代表读取wal时从这个snapshot的记录之后开始
start walpb.Snapshot
//wal记录的反序列化器
decoder *decoder
...
//底层数据文件列表
locks []*fileutil.LockedFile
//
}
WAL底层对应着磁盘上一系列文件，当收到需要持久化的日志条目时就会追加到文件的末尾，文件达到一定大小时，WAL会主动创建一个新的磁盘文件，防止单个WAL文件过大。
etcd 会定期对数据做快照，快照时会在WAL中追加一条记录。在etcd节点重启恢复时，会查找wal中最后一次快照的记录，将快照后的日志条目重新给到raft模块恢复内存数据。
KV数据库存储
Etcd最终生效的数据存在KV数据库中，并对后端存储抽象了一个Backend接口，Backend的实现需要支持事务和多版本管理。Backend接口的定义如下：
type Backend interface {
// 开启读事务.
ReadTx() ReadTx
//开启写事务
BatchTx() BatchTx
// 开启并发读事务，互相之间不阻塞
ConcurrentReadTx() ReadTx
// 对db做快照
Snapshot() Snapshot
Hash(ignores map[IgnoreKey]struct{}) (uint32, error)
// DB占用的物理磁盘大小，空间可以预分配，所以不是实际数据大小
Size() int64
// 实际使用的磁盘空间
SizeInUse() int64
// 返回当前读事务个数
OpenReadTxN() int64
// 数据文件整理，会回收已删除key和已更新的key旧版本占用的磁盘
Defrag() error
ForceCommit()
Close() error
}
该接口的默认实现如下：
type backend struct {
// 已经占用的磁盘大小
size int64
// 实际使用的大小
sizeInUse int64
// 已提交事务数
commits int64
// 当前开启的读事务数
openReadTxN int64
// 读写锁
mu sync.RWMutex
//底层存储为boltDB
db *bolt.DB
// 批量写提交间隔
batchInterval time.Duration
// 批量写最大事务数
batchLimit int
// 写事务缓冲队列
batchTx *batchTxBuffered
// 写事务
readTx *readTx
stopc chan struct{}
donec chan struct{}
lg *zap.Logger
}
从上面的实现中可以看出，etcd的默认底层存储使用的是boltDB。为了提高读写效率，etcd会维护一个写事务的缓存队列，当队列大小达到一定数或者离上次已经过了一定的时间后，才会真正将数据写到磁盘上。
存储总结
数据从客户端提交到Etcd后，会经过3个存储的地方。首先会进入Raft算法模块，raft将日志保存在内存中，然后通知etcd持久化。为了提高效率，etcd会将数据写到WAL中，因为wal底层文件只追加不更新和删除，所以完成这一步数据就不会丢了。之后etcd的leader节点将日志分发到集群中，当收到超过半数节点响应后，就会提交数据，将数据存入后端KV存储中。
日志同步
了解了etcd中的存储设计，可以更好的理解一条数据变更请求的整个流转过程，下面通过源码看一下。
请求处理
当客户端提交一条数据变更请求时，比如put hello 为 world的写请求，v3版本中会调用EtcdServer的Put()方法，最终都会调用到processInternalRaftRequestOnce()。
func (s *EtcdServer) processInternalRaftRequestOnce(ctx context.Context, r pb.InternalRaftRequest) (*applyResult, error) {
//判断已提交未apply的记录是否超过限制
ai := s.getAppliedIndex()
ci := s.getCommittedIndex()
if ci > ai+maxGapBetweenApplyAndCommitIndex {
return nil, ErrTooManyRequests
}
//生成一个requestID
r.Header = &pb.RequestHeader{
ID: s.reqIDGen.Next(),
}
authInfo, err := s.AuthInfoFromCtx(ctx)
if err != nil {
return nil, err
}
if authInfo != nil {
r.Header.Username = authInfo.Username
r.Header.AuthRevision = authInfo.Revision
}
//反序列化请求数据
data, err := r.Marshal()
if err != nil {
return nil, err
}
if len(data) > int(s.Cfg.MaxRequestBytes) {
return nil, ErrRequestTooLarge
}
id := r.ID
if id == 0 {
id = r.Header.ID
}
//注册一个channel，等待处理完成
ch := s.w.Register(id)
//设置请求超时
cctx, cancel := context.WithTimeout(ctx, s.Cfg.ReqTimeout())
defer cancel()
start := time.Now()
// 调用raft模块的Propose处理请求
err = s.r.Propose(cctx, data)
if err != nil {
proposalsFailed.Inc()
s.w.Trigger(id, nil) // GC wait
return nil, err
}
proposalsPending.Inc()
defer proposalsPending.Dec()
select {
// 等待收到apply结果返回给客户端
case x := <-ch:
return x.(*applyResult), nil
case <-cctx.Done():
proposalsFailed.Inc()
s.w.Trigger(id, nil) // GC wait
return nil, s.parseProposeCtxErr(cctx.Err(), start)
case <-s.done:
return nil, ErrStopped
}
}
上面的方法中，etcd对请求做了基本的校验之后，会通过调用Propose()方法提交给Raft处理，然后等待反馈。在etcd实现中，会一直到数据apply到状态机之后，才会返回结果给客户端。在Propose()方法中，raft会将请求封装成一个MsgProp消息并调用Step函数。
func (rn *RawNode) Propose(data []byte) error {
return rn.raft.Step(pb.Message{
Type: pb.MsgProp,
From: rn.raft.id,
Entries: []pb.Entry{
{Data: data},
}})
}
etcd中只允许Leader处理数据变更请求，所以如果是Follower收到客户端的命令，会直接转给leader处理，然后等待Leader的反馈后将结果返回给客户端。所以，这里只需要看Leader的处理逻辑，上面的Step()函数最终调用的是raft模块的stepLeader(*raft, pb.Message) 函数。
对于为什么进到stepLeader方法，前一篇文章里面已经讲过了，印象不深的话可以回看一下
func stepLeader(r *raft, m pb.Message) error {
// These message types do not require any progress for m.From.
switch m.Type {
case pb.MsgBeat:
...
case pb.MsgCheckQuorum:
...
case pb.MsgProp:
if len(m.Entries) == 0 {
r.logger.Panicf("%x stepped empty MsgProp", r.id)
}
if r.prs.Progress[r.id] == nil {
// 判断当前节点是不是已经被从集群中移除了
return ErrProposalDropped
}
if r.leadTransferee != None {
// 如果正在进行leader切换，拒绝写入
return ErrProposalDropped
}
for i := range m.Entries {
//判断是否有配置变更的日志，有的话做一些特殊处理
}
//将日志追加到raft状态机中
if !r.appendEntry(m.Entries...) {
return ErrProposalDropped
}
// 发送日志给集群其它节点
r.bcastAppend()
return nil
case pb.MsgReadIndex:
...
return nil
}
...
...
return nil
}
Raft协议是一个基于日志复制的协议，所以客户端数据变更请求会封装成一条日志条目。上面的逻辑中首先做了一些基本的校验，通过后将Message中的日志条目追加到raft的日志列表中，追加成功后就会将日志广播给所有Follower。
Raft日志新增
上面讲存储的时候讲到，raft算法实现模块只是将日志存在内存中，所以appendEntry的逻辑也很简单。
func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) {
//1. 获取raft节点当前最后一条日志条目的index
li := r.raftLog.lastIndex()
//2. 给新的日志条目设置term和index
for i := range es {
es[i].Term = r.Term
es[i].Index = li + 1 + uint64(i)
}
// 3. 判断未提交的日志条目是不是超过限制，是的话拒绝并返回失败
if !r.increaseUncommittedSize(es) {
return false
}
// 4. 将日志条目追加到raftLog中
li = r.raftLog.append(es...)
// 5. 检查并更新日志进度
r.prs.Progress[r.id].MaybeUpdate(li)
// 6. 判断是否做一次commit
r.maybeCommit()
return true
}
- 获取raft当前日志中最后一条日志条目的index
- raft的日志条目index是单调递增的
- etcd限制了leader上最多有多少未提交的条目，防止因为leader和follower之间出现网络问题时，导致条目一直累积。
- 将日志条目追加到raftLog内存队列中，并且返回最大一条日志的index，对于leader追加日志的情况，这里返回的li肯定等于方法第1行中获取的li
- raft的leader节点保存了所有节点的日志同步进度，这里面也包括它自己
- 这里忽略maybeCommit()结果，直接返回true，开始广播日志。
同步给Follower
Leader节点将日志条目存到raftLog的内存中后，调用bcastAppend()方法触发一次广播操作，同步日志给Follower。
func (r *raft) bcastAppend() {
//遍历所有节点，给除自己外的节点发送日志Append消息
r.prs.Visit(func(id uint64, _ *tracker.Progress) {
if id == r.id {
return
}
r.sendAppend(id)
})
}
func (r *raft) sendAppend(to uint64) {
r.maybeSendAppend(to, true)
}
func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
//1. 获取对端节点当前同步进度
pr := r.prs.Progress[to]
if pr.IsPaused() {
return false
}
m := pb.Message{}
m.To = to
//2. 注意这里带的term是本次发送给follower的第一条日志条目的term
term, errt := r.raftLog.term(pr.Next - 1)
ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
if len(ents) == 0 && !sendIfEmpty {
return false
}
if errt != nil || erre != nil {
//3. 如果获取term或日志失败，说明follower落后太多，raftLog内存中日志已经做过快照后被删除了
if !pr.RecentActive {
r.logger.Debugf("ignore sending snapshot to %x since it is not recently active", to)
return false
}
//4. 改为发送Snapshot消息
m.Type = pb.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 false
}
panic(err) // TODO(bdarnell)
}
if IsEmptySnap(snapshot) {
panic("need non-empty snapshot")
}
m.Snapshot = snapshot
sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term
pr.BecomeSnapshot(sindex)
} else {
//5. 发送Append消息
m.Type = pb.MsgApp
m.Index = pr.Next - 1
m.LogTerm = term
m.Entries = ents
//6. 每次发送日志或心跳都会带上最新的commitIndex
m.Commit = r.raftLog.committed
if n := len(m.Entries); n != 0 {
...
...
}
}
//7. 发送消息
r.send(m)
return true
}
上面的逻辑中，leader在收到新的更新日志后，会遍历集群中所有follower节点，触发一次日志同步。
- 按照raft协议规定，leader需要缓存当前所有Follower的日志同步进度
- 根据日志进度去取日志条目的时候发现，follower日志落后太多，这通常出现在新节点刚加入或者网络连接出现故障的情况下。那么在这种情况下，leader改为发送最近一次快照给Follower，从而提高同步效率
- 正常情况下会发送新的日志给Follower，消息类型为MsgApp，最终调用r.send(m)提交消息。
日志写WAL
在上一篇讲心跳消息发送的时候已经讲过，EtcdServer中会有一个goroutine监听raft的channel是不是有新的Ready数据过来，收到后就会将里面的msgs发送给接收端。这个MsgApp类型的消息也是一样提交的，这里就不在重复了。
日志发送给Follower的同时，Leader会将日志落盘，即写到WAL中，这是通过调用WAL.Save()方法实现的。
func (w *WAL) Save(st raftpb.HardState, ents []raftpb.Entry) error {
//获取wal的写锁
w.mu.Lock()
defer w.mu.Unlock()
// HardState变化或者新的日志条目则需要写wal
if raft.IsEmptyHardState(st) && len(ents) == 0 {
return nil
}
mustSync := raft.MustSync(st, w.state, len(ents))
// 写日志条目
for i := range ents {
if err := w.saveEntry(&ents[i]); err != nil {
return err
}
}
// 写state变化
if err := w.saveState(&st); err != nil {
return err
}
// 判断文件大小是否超过最大值
curOff, err := w.tail().Seek(0, io.SeekCurrent)
if err != nil {
return err
}
if curOff < SegmentSizeBytes {
if mustSync {
return w.sync()
}
return nil
}
// 文件切分
return w.cut()
}
WAL文件结构上面已经讲过了，对于新增日志的情况，wal中新增entryType的记录。
Follower日志处理
Leader节点处理完命令后，发送日志和持久化操作都是异步进行的，但是这不代表客户端已经收到回复。Raft协议要求在返回客户端成功的时候，日志一定已经提交了，所以Leader需要等待超过半数的Follower节点处理完日志并反馈，下面先看一下Follower的日志处理。
日志消息到达Follower后，也是由EtcdServer.Process()方法来处理，最终会进到Raft模块的stepFollower()函数中。
func stepFollower(r *raft, m pb.Message) error {
switch m.Type {
...
case pb.MsgApp:
// 重置心跳计数
r.electionElapsed = 0
// 设置Leader
r.lead = m.From
// 处理日志条目
r.handleAppendEntries(m)
...
}
...
}
Follower收到消息后首先跟心跳消息的处理逻辑一样，重置心跳计数和leader，然后再处理日志条目。
func (r *raft) handleAppendEntries(m pb.Message) {
// 判断是否是过时的消息
if m.Index < r.raftLog.committed {
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
return
}
if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
// 处理成功，发送MsgAppResp给Leader
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
} else {
// 日志的index和Follower的lastIndex不匹配，返回reject消息
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: m.Index, Reject: true, RejectHint: r.raftLog.lastIndex()})
}
}
调用raftLog存储日志，并返回结果给Leader。这里follower失败可能有2种情况造成的，一种是日志条目中带的term和follower的term不一致，还有一种是日志列表中最小的index大于follower的最大的日志index。
上面的maybeAppend() 方法只会将日志存储到RaftLog维护的内存队列中，日志的持久化是异步进行的，这个和Leader节点的存储WAL逻辑基本相同。有一点区别就是follower节点正式发送MsgAppResp消息会在wal保存成功后，而leader节点是先发送消息，后保存的wal。
提交(Commit)
Leader节点在向Follower广播日志后，就一直在等待follower的MsgAppResp消息，收到后还是会进到stepLeader函数。
func stepLeader(r *raft, m pb.Message) error {
...
...
pr := r.prs.Progress[m.From]
switch m.Type {
case pb.MsgAppResp:
pr.RecentActive = true
if m.Reject {
//如果收到的是reject消息，则根据follower反馈的index重新发送日志
if pr.MaybeDecrTo(m.Index, m.RejectHint) {
if pr.State == tracker.StateReplicate {
pr.BecomeProbe()
}
r.sendAppend(m.From)
}
} else {
oldPaused := pr.IsPaused()
//更新缓存的日志同步进度
if pr.MaybeUpdate(m.Index) {
switch {
case pr.State == tracker.StateProbe:
pr.BecomeReplicate()
case pr.State == tracker.StateSnapshot && pr.Match >= pr.PendingSnapshot:
pr.BecomeProbe()
pr.BecomeReplicate()
case pr.State == tracker.StateReplicate:
pr.Inflights.FreeLE(m.Index)
}
//如果进度有更新，判断并更新commitIndex
if r.maybeCommit() {
//commitIndex有变化则立即发送日志
r.bcastAppend()
} else if oldPaused {
r.sendAppend(m.From)
}
// 循环发送所有剩余的日志给follower
for r.maybeSendAppend(m.From, false) {
}
// 是否正在进行leader转移
if m.From == r.leadTransferee && pr.Match == r.raftLog.lastIndex() {
r.logger.Infof("%x sent MsgTimeoutNow to %x after received MsgAppResp", r.id, m.From)
r.sendTimeoutNow(m.From)
}
}
}
...
...
return nil
}
func (r *raft) maybeCommit() bool {
//获取最大的超过半数确认的index
mci := r.prs.Committed()
//更新commitIndex
return r.raftLog.maybeCommit(mci, r.Term)
}
收到Follower的回复以后，如果是reject的，leader会根据返回的index重新发送日志。如果是成功的消息，则更新缓存中的日志同步进度，并判断超过半数确认的index是否有变化。有变化则通知raftLog更新commitIndex。到此为止，客户端的这条数据更新命令，就正式提交了。下面就看一下，数据是怎样写到DB中的。
数据更新(Apply)
前面已经讲过，EtcdServer在启动时会启动一个goroutine监听raft模块是否有Ready消息过来。当上一步的commitIndex发生变化后，Ready中的HardState就会有值了。Etcd会获取ready结构中的committedEntries，提交给Apply模块应用到后端存储中。
func (r *raftNode) start(rh *raftReadyHandler) {
internalTimeout := time.Second
go func() {
defer r.onStop()
islead := false
for {
...
case rd := <-r.Ready():
if rd.SoftState != nil {
...
...
}
if len(rd.ReadStates) != 0 {
...
...
}
// 生成apply请求
notifyc := make(chan struct{}, 1)
ap := apply{
entries: rd.CommittedEntries,
snapshot: rd.Snapshot,
notifyc: notifyc,
}
// 更新etcdServer缓存的commitIndex为最新值
updateCommittedIndex(&ap, rh)
// 将已提交日志应用到状态机
select {
case r.applyc <- ap:
case <-r.stopped:
return
}
if islead {
// 如果有新的日志条目
r.transport.Send(r.processMessages(rd.Messages))
}
// 如果有snapshot
if !raft.IsEmptySnap(rd.Snapshot) {
...
...
}
//将hardState和日志条目保存到WAL中
if err := r.storage.Save(rd.HardState, rd.Entries); err != nil {
...
...
}
if !raft.IsEmptyHardState(rd.HardState) {
proposalsCommitted.Set(float64(rd.HardState.Commit))
}
if !raft.IsEmptySnap(rd.Snapshot) {
...
...
}
r.raftStorage.Append(rd.Entries)
if !islead {
...
...
} else {
notifyc <- struct{}{}
}
//更新raft模块的applied index和将日志从unstable转到stable中
r.Advance()
case <-r.stopped:
return
}
}
}()
}
这里需要注意的是，在将已提交日志条目应用到状态机的操作是异步完成的，在Apply完成后，会将结果写到客户端调用进来时注册的channel中。这样一次完整的写操作就完成了。
总结
etcd整个写操作涉及到raft协议处理，日志同步，持久化以及更新KV的流程。这里面首先要理解etcd的存储结构，就能对整个流程有清晰的理解。
作者:空挡
原文链接:https://www.jianshu.com/p/518f75c5bf55