缘起

最近阅读 [云原生分布式存储基石：etcd深入解析] (杜军 , 2019.1)
本系列笔记拟采用golang练习之

raft分布式一致性算法

  1. 分布式存储系统通常会通过维护多个副本来进行容错，
  2. 以提高系统的可用性。
  3. 这就引出了分布式存储系统的核心问题——如何保证多个副本的一致性？
  5. Raft算法把问题分解成了四个子问题：
  6. 1. 领袖选举（leader election）、
  7. 2. 日志复制（log replication）、
  8. 3. 安全性（safety）
  9. 4. 成员关系变化（membership changes）
  10. 这几个子问题。
  12. 源码gitee地址：
  13. https://gitee.com/ioly/learning.gooop

目标

  • 根据raft协议，实现高可用分布式强一致的kv存储

子目标（Day 10）

  • 添加put/get/del kv键值对的rpc接口
  • 继续完善Leader状态的raft协议响应

设计

  • rpc/IKVStoreRPC: kv操作的rpc接口
  • store/IKVStore: kv操作的持久化接口
  • stoer/ILogStore: 从IKVStore继承，以支持kv持久化
  • lsm/IRaftState: 继承rpc.IKVStoreRPC接口，以支持kv操作
  • lsm/tLeaderState: 初步实现Leader状态的raft协议处理，事件驱动的逻辑编排，读写分离的字段管理。

rpc/IKVStoreRPC.go

kv操作的rpc接口

  1. package rpc
  3. type IKVStoreRPC interface {
  4.     ExecuteKVCmd(cmd *KVCmd, ret *KVRet) error
  5. }
  7. type KVCmd struct {
  8.     OPCode KVOPCode
  9.     Key []byte
  10.     Content []byte
  11. }
  13. type KVOPCode int
  14. const (
  15.     KVGet KVOPCode = iota
  16.     KVPut KVOPCode = iota
  17.     KVDel KVOPCode = iota
  18. )
  20. type KVRet struct {
  21.     Code KVRetCode
  22.     Key []byte
  23.     Content []byte
  24. }
  26. type KVRetCode int
  27. const (
  28.     KVOk KVRetCode = iota
  29.     KVKeyNotFound KVRetCode = iota
  30.     KVInternalError KVRetCode = iota
  31. )

store/IKVStore.go

kv操作的持久化接口

  1. package store
  3. type IKVStore interface {
  4.     Get(key []byte) (error, []byte)
  5.     Put(key []byte, content []byte) error
  6.     Del(key []byte) error
  7. }

stoer/ILogStore.go

从IKVStore继承，以支持kv持久化

  1. package store
  3. import (
  4.     "learning/gooop/etcd/raft/model"
  5. )
  7. type ILogStore interface {
  8.     IKVStore
  10.     LastAppendedTerm() int64
  11.     LastAppendedIndex() int64
  12.     LastCommittedTerm() int64
  13.     LastCommittedIndex() int64
  15.     Append(entry *model.LogEntry) error
  16.     Commit(index int64) error
  17.     GetLog(index int64) (error, *model.LogEntry)
  18. }

lsm/IRaftState.go

继承rpc.IKVStoreRPC接口，以支持kv操作

  1. package lsm
  3. import (
  4.     "learning/gooop/etcd/raft/roles"
  5.     "learning/gooop/etcd/raft/rpc"
  6. )
  8. type IRaftState interface {
  9.     rpc.IRaftRPC
  10.     rpc.IKVStoreRPC
  12.     Role() roles.RaftRole
  13.     Start()
  14. }

lsm/tLeaderState.go

初步实现Leader状态的raft协议处理，事件驱动的逻辑编排，读写分离的字段管理。

  1. package lsm
  3. import (
  4.     "errors"
  5.     "learning/gooop/etcd/raft/config"
  6.     "learning/gooop/etcd/raft/model"
  7.     "learning/gooop/etcd/raft/roles"
  8.     "learning/gooop/etcd/raft/rpc"
  9.     "learning/gooop/etcd/raft/store"
  10.     "learning/gooop/etcd/raft/timeout"
  11.     "sync"
  12.     "time"
  13. )
  15. // tLeaderState presents a leader node
  16. type tLeaderState struct {
  17.     tEventDrivenModel
  19.     context    iRaftStateContext
  20.     mInitOnce  sync.Once
  21.     mStartOnce sync.Once
  23.     // update: leInit / leLeaderHeartbeat
  24.     mTerm int64
  26.     // update: leInit / leDisposing
  27.     mDisposedFlag bool
  29.     // update: leVoteToCandidate
  30.     mVotedTerm int64
  31.     mVotedCandidateID string
  32.     mVotedTimestamp int64
  33. }
  35. // trigger: init()
  36. // args: empty
  37. const leInit = "leader.init"
  39. // trigger: Start()
  40. // args: empty
  41. const leStart = "leader.Start"
  43. // trigger: whenNewLeaderAnnouncedThenSwitchToFollower
  44. // args: empty
  45. const leDiposing = "leader.Disposing"
  47. // trigger : Heartbeat() / AppendLog()
  48. // args: term int64
  49. const leNewLeaderAnnounced = "leader.NewLeaderAnnounced"
  52. // trigger: RequestVote()
  53. // args: *rpc.RequestVoteCmd
  54. const leBeforeRequestVote = "leader.BeforeRequestVote"
  56. // trigger:
  57. // args: *rpc.RequestVoteCmd
  58. const leVoteToCandidate = "leader.VoteToCandidate"
  60. // trigger: handleHeartbeat()
  61. // args: term int64
  62. const leHeartbeatRejected = "leader.HeartbeatRejected"
  65. func newLeaderState(ctx iRaftStateContext, term int64) IRaftState {
  66.     it := new(tLeaderState)
  67.     it.init(ctx, term)
  68.     return it
  69. }
  71. func (me *tLeaderState) init(ctx iRaftStateContext, term int64) {
  72.     me.mInitOnce.Do(func() {
  73.         me.context = ctx
  74.         me.mTerm = term
  76.         me.initEventHandlers()
  77.         me.raise(leInit)
  78.     })
  79. }
  81. func (me *tLeaderState) initEventHandlers() {
  82.     // write only logic
  83.     me.hookEventsForDisposedFlag()
  84.     me.hookEventsForVotedTerm()
  86.     // read only logic
  87.     me.hook(leStart,
  88.         me.whenStartThenBeginHeartbeatToOthers)
  89.     me.hook(leNewLeaderAnnounced,
  90.         me.whenNewLeaderAnnouncedThenSwitchToFollower)
  91.     me.hook(leHeartbeatRejected,
  92.         me.whenHeartbeatRejectedThenSwitchToFollower)
  93. }
  96. func (me *tLeaderState) hookEventsForDisposedFlag() {
  97.     me.hook(leInit, func(e string, args ...interface{}) {
  98.         me.mDisposedFlag = false
  99.     })
  101.     me.hook(leDiposing, func(e string, args ...interface{}) {
  102.         me.mDisposedFlag = true
  103.     })
  104. }
  106. func (me *tLeaderState) hookEventsForVotedTerm() {
  107.     me.hook(leBeforeRequestVote, func(e string, args ...interface{}) {
  108.         // check last vote timeout
  109.         if me.mVotedTerm == 0 {
  110.             return
  111.         }
  113.         if time.Duration(time.Now().UnixNano() - me.mVotedTimestamp)*time.Nanosecond >= timeout.ElectionTimeout {
  114.             me.mVotedTerm = 0
  115.             me.mVotedTimestamp = 0
  116.             me.mVotedCandidateID = ""
  117.         }
  118.     })
  120.     me.hook(leVoteToCandidate, func(e string, args ...interface{}) {
  121.         // after vote to candidate
  122.         cmd := args[0].(*rpc.RequestVoteCmd)
  123.         me.mVotedTerm = cmd.Term
  124.         me.mVotedCandidateID = cmd.CandidateID
  125.         me.mVotedTimestamp = time.Now().UnixNano()
  126.     })
  127. }
  131. func (me *tLeaderState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
  132.     // check term
  133.     if cmd.Term <= me.mTerm {
  134.         ret.Code = rpc.HBTermMismatch
  135.         return nil
  136.     }
  138.     // new leader
  139.     me.raise(leNewLeaderAnnounced, cmd.Term)
  141.     // return ok
  142.     ret.Code = rpc.HBOk
  143.     return nil
  144. }
  146. func (me *tLeaderState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
  147.     // check term
  148.     if cmd.Term <= me.mTerm {
  149.         ret.Code = rpc.ALTermMismatch
  150.         return nil
  151.     }
  153.     // new leader
  154.     me.raise(leNewLeaderAnnounced, cmd.Term)
  156.     // return ok
  157.     ret.Code = rpc.ALInternalError
  158.     return nil
  159. }
  161. func (me *tLeaderState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
  162.     // just ignore
  163.     ret.Code = rpc.CLInternalError
  164.     return nil
  165. }
  167. func (me *tLeaderState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
  168.     me.raise(leBeforeRequestVote, cmd)
  170.     // check voted term
  171.     if cmd.Term < me.mVotedTerm {
  172.         ret.Code = rpc.RVTermMismatch
  173.         return nil
  174.     }
  176.     if cmd.Term == me.mVotedTerm {
  177.         if me.mVotedCandidateID != "" && me.mVotedCandidateID != cmd.CandidateID {
  178.             // already vote another
  179.             ret.Code = rpc.RVVotedAnother
  180.             return nil
  181.         } else {
  182.             // already voted
  183.             ret.Code = rpc.RVOk
  184.             return nil
  185.         }
  186.     }
  188.     if cmd.Term > me.mVotedTerm {
  189.         // new term, check log
  190.         if cmd.LastLogIndex >= me.context.Store().LastCommittedIndex() {
  191.             // good log
  192.             me.raise(leVoteToCandidate, cmd)
  193.             ret.Code = rpc.RVOk
  195.         } else {
  196.             // bad log
  197.             ret.Code = rpc.RVLogMismatch
  198.         }
  200.         return nil
  201.     }
  203.     // should not reach here
  204.     ret.Code = rpc.RVTermMismatch
  205.     return nil
  206. }
  208. func (me *tLeaderState) Role() roles.RaftRole {
  209.     return roles.Leader
  210. }
  212. func (me *tLeaderState) Start() {
  213.     me.mStartOnce.Do(func() {
  214.         me.raise(leStart)
  215.     })
  216. }
  218. func (me *tLeaderState) whenStartThenBeginHeartbeatToOthers(_ string, _ ...interface{}) {
  219.     go func() {
  220.         for !me.mDisposedFlag {
  221.             _ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error {
  222.                 return me.handleHeartbeat(client)
  223.             })
  224.             time.Sleep(timeout.HeartbeatInterval)
  225.         }
  226.     }()
  227. }
  230. func (me *tLeaderState) boardcast(action func(config.IRaftNodeConfig, rpc.IRaftRPC) error) error {
  231.     for _,it := range me.context.Config().Nodes() {
  232.         if it.ID() == me.context.Config().ID() {
  233.             continue
  234.         }
  236.         e := me.context.RaftClientService().Using(it.ID(), func(client rpc.IRaftRPC) error {
  237.             return action(it, client)
  238.         })
  239.         if e != nil {
  240.             return e
  241.         }
  242.     }
  244.     return nil
  245. }
  248. func (me *tLeaderState) handleHeartbeat(client rpc.IRaftRPC) error {
  249.     cmd := new(rpc.HeartbeatCmd)
  250.     cmd.Term = me.mTerm
  251.     cmd.LeaderID = me.context.Config().ID()
  253.     ret := new(rpc.HeartbeatRet)
  254.     e := client.Heartbeat(cmd, ret)
  255.     if e != nil {
  256.         return e
  257.     }
  259.     switch ret.Code {
  260.     case rpc.HBTermMismatch:
  261.         me.raise(leHeartbeatRejected, ret.Term)
  262.         break
  263.     }
  265.     return nil
  266. }
  268. func (me *tLeaderState) whenNewLeaderAnnouncedThenSwitchToFollower(_ string, args ...interface{}) {
  269.     me.raise(leDiposing)
  271.     term := args[0].(int64)
  272.     me.context.HandleStateChanged(newFollowerState(me.context, term))
  273. }
  275. func (me *tLeaderState) whenHeartbeatRejectedThenSwitchToFollower(_ string, args ...interface{}) {
  276.     me.raise(leDiposing)
  278.     term := args[0].(int64)
  279.     me.context.HandleStateChanged(newFollowerState(me.context, term))
  280. }
  283. func (me *tLeaderState) ExecuteKVCmd(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
  284.     switch cmd.OPCode {
  285.     case rpc.KVGet:
  286.         return me.handleKVGet(cmd, ret)
  288.     case rpc.KVPut:
  289.         return me.handleKVPut(cmd, ret)
  291.     case rpc.KVDel:
  292.         return me.handleKVDel(cmd, ret)
  293.     }
  295.     return nil
  296. }
  298. func (me *tLeaderState) handleKVGet(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
  299.     e, v := me.context.Store().Get(cmd.Key)
  300.     if e != nil {
  301.         ret.Code = rpc.KVInternalError
  302.         return e
  303.     }
  305.     ret.Code = rpc.KVOk
  306.     ret.Content = v
  307.     return nil
  308. }
  310. func (me *tLeaderState) handleKVPut(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
  311.     kvcmd := new(store.PutCmd)
  312.     kvcmd.Key = cmd.Key
  313.     kvcmd.Value = cmd.Content
  315.     // create/append/commit log
  316.     e := me.broadcastKVCmd(kvcmd, ret)
  317.     if e != nil {
  318.         return e
  319.     }
  321.     // apply cmd
  322.     return me.context.Store().Put(cmd.Key, cmd.Content)
  323. }
  325. func (me *tLeaderState) handleKVDel(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
  326.     kvcmd := new(store.DelCmd)
  327.     kvcmd.Key = cmd.Key
  329.     // create/append/commit log
  330.     e := me.broadcastKVCmd(kvcmd, ret)
  331.     if e != nil {
  332.         return e
  333.     }
  335.     // apply cmd
  336.     return me.context.Store().Put(cmd.Key, cmd.Content)
  337. }
  339. func (me *tLeaderState) broadcastKVCmd(cmd store.IKVCmd, ret *rpc.KVRet) error {
  340.     // create log
  341.     st := me.context.Store()
  342.     log := new(model.LogEntry)
  343.     log.Term = me.mTerm
  344.     log.Index = st.LastCommittedIndex() + 1
  345.     log.PrevTerm = st.LastCommittedTerm()
  346.     log.PrevIndex = st.LastCommittedIndex()
  347.     log.Command = cmd.Marshal()
  349.     // append log
  350.     e := st.Append(log)
  351.     if e != nil {
  352.         ret.Code = rpc.KVInternalError
  353.         return e
  354.     }
  356.     // ask other nodes to append log
  357.     alcmd := new(rpc.AppendLogCmd)
  358.     alcmd.Term = me.mTerm
  359.     alcmd.LeaderID = me.context.Config().ID()
  360.     alcmd.Entry = log
  362.     sumOk := []int{ 0 }
  363.     _ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error {
  364.         alret := new(rpc.AppendLogRet)
  365.         e := client.AppendLog(alcmd, alret)
  366.         if e != nil {
  367.             return e
  368.         }
  370.         switch alret.Code {
  371.         case rpc.ALOk:
  372.             sumOk[0]++
  373.             break
  375.         case rpc.ALTermMismatch:
  376.             // todo: fixme
  377.             break
  379.         case rpc.ALIndexMismatch:
  380.             // todo: fixme
  381.             break
  382.         }
  384.         return nil
  385.     })
  387.     // wait for most nodes
  388.     if sumOk[0] >= len(me.context.Config().Nodes()) / 2 {
  389.         // commit log
  390.         clcmd := new(rpc.CommitLogCmd)
  391.         clcmd.LeaderID = me.context.Config().ID()
  392.         clcmd.Term = me.mTerm
  393.         clcmd.Index = log.Index
  395.         _ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error {
  396.             ret := new(rpc.CommitLogRet)
  397.             e := client.CommitLog(clcmd, ret)
  398.             if e != nil {
  399.                 return e
  400.             }
  402.             switch ret.Code {
  403.             case rpc.CLInternalError:
  404.                 // todo: fixme
  405.                 break
  407.             case rpc.CLLogNotFound:
  408.                 // todo: fixme
  409.                 break
  411.             case rpc.CLOk:
  412.                 return nil
  413.             }
  415.             return nil
  416.         })
  418.         // ok
  419.         return nil
  421.     } else {
  422.         return gErrorCannotReachAgreement
  423.     }
  424. }
  426. var gErrorCannotReachAgreement = errors.New("cannot reach agreement")

（未完待续）