学习分享｜Etcd/Raft 原理篇

• Raft是什么？

• Etcd是什么？



• 为什么选择etcd-raft



文章主要讲解 etcd/raft——原理解析篇。

etcd中实现的raft实现本质上是提供了一个基于raft的sdk，作为sdk方式接入到etcd中，和etcd存储系统是解耦的。同时sdk的实现非常巧妙，只实现了基本的功能，包括leader选举、日志处理、状态变更等逻辑。上层的网络传输、数据存储等模块，提供接口由上层应用者来实现：

• 存储层：定义了Storage接口用来管理raft log，用户也可以自行实现该接口并作为参数传入，非常灵活；同时也提供了对该接口一个基于改接口的存储实现MemoryStorage，该实现是基于内存数组实现的非持久化的存储。
• 网络层：节点间的数据通信这部分在实现中没有做任何约束，通过channel来和应用层交互，由应用层用户自定义实现来处理收到的消息。

简单画个图来表达这种设计：

我暂且把这种设计原则称之为“最小原则”，sdk中只实现基本的核心功能，sdk依赖的其他能力只定义好接口，通过参数或者其他的方式进行注入。

（后续的开发中可以借鉴这种思想，不一定大而全的sdk实现就是好用的，可能反而会冗余或者互相依赖等，当然也要分具体的场景具体分析）

一条数据提交到raft集群后，etcd/raft内部保证数据在集群各节点是一致的，其实现流程如下：

整个etcd/raft可以抽象概括为2大模块：

• 日志复制模块：外部提交的数据统称为日志，通过日志复制算法，实现数据的分布式一致性。
• 集群变更选举模块：实现集群选举功能，和集群节点配置的管理，同时对原生raft选主算法做了一些算法和性能优化。

下面分别针对这两大功能结合代码做详细介绍。

2.1. 日志复制模块

• 日志存储结构

type raftLog struct {  // storage存储了从最后一次snapshot到现在的所有可靠的（stable）日志（Entry),即保存到snapshot之后提交的数据，    // 同时Storage定义为接口，由上层使用者实现和维护的，raft需要访问的时候直接读取无需访问上层持久化的存储。  storage Storage  // 用于保存还没有持久化的数据和快照，与storage形成了鲜明的对比，使用者没有通知raft日志持久化完毕前，这些日志都还不可靠    // 当使用者持久化完毕后，这些日志就会从unstable删除，最终都会保存到storage中  unstable unstable  // committed保存是写入持久化存储storage中的最高index，    // 这里提交索引是集群的一个状态，而不是某一节点的状态，它是由leader统计出来的，并广播给所有的节点。  committed uint64  // 已经提交的日志要被使用者应用，applied就是该节点已经被应用的最大索引  // 一条日志首先要提交成功（即committed），才能被applied到状态机中因此以下不等式一直成立：applied <= committed  applied uint64  logger Logger}type unstable struct {  snapshot *pb.Snapshot // 保存还没有持久化的快照数据  entries []pb.Entry    // 还未持久化的日志提交粒度的数据  offset  uint64        // offset保存的为entries数组中的数据的起始index  logger Logger}

对于unstable，按照代码逻辑推断(并不是100%确定)：快照和entrise数据不会同时存在，快照只会在启动时进行快照数据恢复时存在，当应用层使用快照数据进行恢复之后，raft切换为可以接收日志数据的状态，后续的日志数据都会写到entrise数组中了，而两者的分界线就是offset变量。

一条提交的日志会首先写入到unstable中，因为unstable为非未持久化数据的缓冲区，因此这其中的数据可能会发生回滚（rollback）现象，具体实现为

func (u *unstable) truncateAndAppend(ents []pb.Entry) {  // 先拿到这些数据的第一个索引  after := ents[0].Index  switch {  case after == u.offset+uint64(len(u.entries)):    // 如果正好是紧接着当前数据的，就直接append    u.entries = append(u.entries, ents...)  case after <= u.offset:    // 如果比当前偏移量小，那用新的数据替换当前数据，需要同时更改offset和entries    u.offset = after    u.entries = ents  default:    // 到了这里，说明 u.offset < after < u.offset+uint64(len(u.entries))    // 那么新的entries需要拼接而成    u.entries = append([]pb.Entry{}, u.slice(u.offset, after)...)    u.entries = append(u.entries, ents...)  }}

Storage接口定义的是持久化存储，之所以etcd使用了基于内存的MemoryStorage，是因为etcd在写入MemoryStorage前，需要先写入预写日志（Write Ahead Log，WAL）或快照。而预写日志和快照是保存在稳定存储中的。这样，在每次重启时，etcd可以基于保存在稳定存储中的快照和预写日志恢复MemoryStorage的状态。也就是说，etcd的稳定存储是通过快照、预写日志、MemoryStorage三者共同实现的。

• 日志存储过程

a.首先，`raft`库会首先将日志数据写入`未持久化数据缓冲区`。

b.由于`未持久化数据缓冲区`中有新增的数据，会通过`Ready`结构体通知给应用层。

c.应用层收到`Ready`结构体之后，将其中的数据写入WAL持久化存储，然后更新这块数据到`已持久化数据缓冲区

d.持久化完毕后，应用层通过`Advance`接口通知`Raft`库这些数据已经持久化，于是raft库修改`未持久化数据缓冲区`将客户端刚提交的数据从这个缓冲区中删除。

e.持久化完毕之后，除了通知删除`未持久化数据缓冲区`，还讲数据通过网络同步给集群中其他节点。

对于存储结构raftLog、storage 和unstable等和相关的代码函数细节这里不再展开解析，感兴趣的可以通过最后给的地址去翻阅，整体上这块实现不算复杂。

• 日志复制进度跟踪

在etcd-raft中，tracker是raft代码目录下单独的一个包，核心实现就包括一个ProgressTracker。

了解核心类是ProgressTracker之前，需要先看其变量Progress，根据注释不难理解，Leader节点除了要维护未持久化缓冲区之外，还需要维护一个数据结构，用于保存集群中其他节点的进度，简称Progress，简单描述就是作为集群的leader，需要知道其他节点日志同步的具体情况。

其结构定义解释如下：

type Progress struct {  // Next保存的是下一次leader发送append消息时传送过来的日志索引  // 当选举出新的leader时，首先初始化Next为该leader最后一条日志+1  // 如果向该节点append日志失败，则递减Next回退日志，一直回退到索引匹配为止    // Match保存在该节点上保存的日志的最大索引，初始化为0  // 正常情况下，Next = Match + 1  // 以下情况下不是上面这种情况：  // 1. 切换到Probe状态时，如果上一个状态是Snapshot状态，即正在接收快照，那么Next = max(pr.Match+1, pendingSnapshot+1)  // 2. 当该follower不在Replicate状态时，说明不是正常的接收副本状态。  //    此时当leader与follower同步leader上的日志时，可能出现覆盖的情况，即此时follower上面假设Match为3，但是索引为3的数据会被  //    leader覆盖，此时Next指针可能会一直回溯到与leader上日志匹配的位置，再开始正常同步日志，此时也会出现Next != Match + 1的情况出现  Match, Next uint64    // 三种状态  // ProgressStateProbe：探测状17  // ProgressStateReplicate：副本状态  // ProgressStateSnapshot：快照状态  State ProgressStateType        // 探测状态时才有用，表示探测消息是否已经发送了，如果发送了就不会再发了，避免不必要的IO。  ProbeSent bool    // 如果向该节点发送快照消息，PendingSnapshot用于保存快照消息的索引  // 当PendingSnapshot不为0时，该节点也被标记为暂停状态。  // raft只有在这个正在进行中的快照同步失败以后，才会重传快照消息  PendingSnapshot uint64    // 如果进程最近处于活跃状态则为 true（收到来自跟随者的任意消息都认为是活动状态）。在超时后会重置重置为false  RecentActive bool    // 用于实现滑动窗口，用来做流量控制，后边会展开单独作为一个额外福利介绍  Inflights *Inflights}

总结来说，Progress结构体做的工作：

1.维护follower节点的match、next索引，(0, Next)的日志已经发送给节点了，(0,Match]是节点的已经接收到的日志。

2.维护着follower节点当前的状态3中状态，不同的状态，其会采取不同的行为：

• StateProbe：探测状态。当 follower因异常原因落后Leader节点数据过多，此时会拒绝了最近主同步的append消息，那么就标记该follower会进入Probe状态。leader 会试图继续往前追溯该 follower 的日志从哪里开始丢失的。在probe 状态时，leader 每次最多 append 一条日志，如果收到的回应中带有 RejectHint 信息，则回退 Next 索引，以便下次重试。在初始时，leader 会把所有 follower 的状态设为 probe，因为它并不知道各个 follower 的同步状态，所以需要慢慢试探
• StateReplicate：副本状态。正常接收副本数据的状态，当处于该状态时，leader在发送副本消息之后，就修改该节点的next索引为发送消息的最大索引+1,此时Inflights值也会放大用于快速日志复制。
• StateSnapshot：接收快照状态。当 leader 向某个 follower 发送 append 消息，试图让该 follower 状态跟上 leader 时，发现此时 leader上保存的索引数据已经对不上了，比如leader在index为10之前的数据都已经写入快照中了，但是该 follower 需要的是 10 之前的数据，此时就会切换到该状态下，发送快照给该 follower。当快照数据同步追上之后，并不是直接切换到 Replicate 状态，而是首先切换到 Probe 状态。



3.流量控制Inflights，避免follower节点超载

• Inflights流控实现

Inflights主要用于StateReplicate状态下在日志复制时可以控制数据传输速度，具体大小用户可以在应用层指定。设计上非常的巧妙，其思想类似于“往池子注水和放水”的过程，通过给定池子的大小控制流速。而raft在实现上没有使用queue，而是在一个内存块上采用循环方式模拟queue的特性，这样效率会更高。

2.2. 选主模块

对于选举算法有很多实现，我之前开发中也实现过一个raft选举算法传送门，一般节点包含三种不同的角色candidate、follower、leader。每种角色对不同类型的日志数据需要有不同的处理，这里选主流程不展开，本节的目的主要在于发现etcd/raft中我理解与众不同的地方。

在etcd/raft的实现中，巧妙之处在于针对三种不同的角色，通过修改函数指针的方式在切换了不同角色时的处理，大概意思如下图所示：

也就是说在于其很好地剥离了各模块的职责。在etcd/raft的实现中，raft结构体是一个Raft状态机，其通过Step方法进行状态转移。只要涉及到Raft状态机的状态转移，最终都会通过Step方法完成。可能我也解释的不是很到位，具体可以看下代码实现：

//Step为节点收到应用层发来的消息，就会执行对应逻辑func (r *raft) Step(m pb.Message) error {    //...    switch m.Type {        case pb.MsgHup: //准备选举时触发        //...        case pb.MsgVote, pb.MsgPreVote: //在选举中触发        //...        default:            r.step(r, m)    }}

对其他执行选择的流程，本文不再介绍。

2.3. 变更模块

etcd/raft采用将修改集群配置的命令放在日志条目中来处理，也就是一个配置变更其实是一次日志数据的提交，不过是一种特殊类型的日志这样做的好处是：

• 可以继续沿用原来的AppendEntries命令来同步日志数据，只要把修改集群的命令作为一种特殊的命令就可以了。
• 在这个过程中，可以继续处理客户端请求。

成员删减：操作作为日志的特殊类型，当可以进行commit的情况下，各个节点拿出该消息进行节点内部的成员删减操作。

leader转让：

• 旧leader在接收到转让leader消息之后，会做如下的判断：a. 如果新的leader上的日志，已经跟当前leader上的日志同步了，那么发送timeout消息。b. 否则继续发append消息到新的leader上，目的为了让其能够与旧leader日志同步。
• 当旧leader处于转让leader状态时，将停止接收新的prop消息，这样就避免出现在转让过程中新旧leader一直日志不能同步的情况。
• 当旧leader收到append消息应答时，如果当前处于leader转让状态，那么会判断新的leader日志是否已经与当前leader同步，如果是将发送timeout消息。

• 新的leader当收到timeout消息时，将使用context为campaignTransfer的选举消息发起新一轮选举，当context为该类型时，此时的选举是强制进行的。



• etcd/raft实现为一个单独包，以sdk的方式接入到etcd系统中，外部使用者同样也可以单独使用改sdk；具体如何使用以及其工程实现将会在第二篇分享。
• 实现架构上有最小原则设计可以在后续开发中借用参考。
• 重点介绍了日志复制功能，包括其存储结构、流转方式以及Leader管理其他节点日志复制进度的实现方式。
• 日志复制过程中通过Inflights算法实现流量控制，实现非常巧妙。
• 选举功能实现上也比较巧妙，函数指针的方式通过一个Step函数解决不同角色的自定义功能。
• 集群中节点状态变更、配置变更等都是共用的通过日志复制的传输链路，保证代码实现简洁抽象。

参考文档：

• etcd/raft源码地址：https://github.com/etcd-io/etcd/tree/raft/v3.5.9/raft

文章引用微信公众号"阿里开发者",如有侵权,请联系管理员删除!