缘起

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

raft分布式一致性算法

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

目标

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

子目标（Day 8）

  • 简化rpc连接管理器tRaftClientService的实现
  • 剥离IRaftLSM的内部支持接口到iRaftStateContext接口
  • 完成Candidate状态的处理逻辑

设计

  • IRaftLSM：raft有限状态机接口
  • iRaftStateContext：提供状态模式下的上下文支持
  • tCandidateState：Candidate（候选人）状态的实现。基于事件驱动的逻辑编排，基于读写分离的字段管理。
  • tRaftClient：管理到指定raft节点的rpc连接
  • tRaftClientService：管理当前节点到其他raft节点的rpc连接

IRaftLSM.go

raft有限状态机接口

  1. package lsm
  3. import (
  4.     "learning/gooop/etcd/raft/rpc"
  5. )
  7. // IRaftLSM raft有限状态自动机
  8. type IRaftLSM interface {
  9.     rpc.IRaftRPC
  10.     iRaftStateContext
  12.     State() IRaftState
  13. }

iRaftStateContext.go

提供状态模式下的上下文支持

  1. package lsm
  3. import (
  4.     "learning/gooop/etcd/raft/config"
  5.     "learning/gooop/etcd/raft/rpc/client"
  6.     "learning/gooop/etcd/raft/store"
  7. )
  9. type iRaftStateContext interface {
  10.     Config() config.IRaftConfig
  11.     Store() store.ILogStore
  12.     HandleStateChanged(state IRaftState)
  13.     RaftClientService() client.IRaftClientService
  14. }

tCandidateState.go

Candidate（候选人）状态的实现。基于事件驱动的逻辑编排，基于读写分离的字段管理。

  1. package lsm
  3. import (
  4.     "learning/gooop/etcd/raft/roles"
  5.     "learning/gooop/etcd/raft/rpc"
  6.     "learning/gooop/etcd/raft/timeout"
  7.     "sync"
  8.     "time"
  9. )
  11. // tCandidateState presents a candidate node
  12. type tCandidateState struct {
  13.     tEventDrivenModel
  15.     context    iRaftStateContext
  16.     mInitOnce  sync.Once
  17.     mStartOnce sync.Once
  19.     // update: init / ceAskingForVote
  20.     mTerm int64
  22.     // update: ceInit / ceAskingForVote / ceVoteToCandidate
  23.     mVotedTerm int64
  25.     // update: ceInit / ceAskingForVote / ceVoteToCandidate
  26.     mVotedCandidateID string
  28.     // update: ceInit / ceAskingForVote / ceVoteToCandidate
  29.     mVotedTimestamp int64
  31.     // update: ceInit / ceAskingForVote / ceReceiveTicket / ceDisposing
  32.     mTicketCount map[string]bool
  33.     mTicketMutex *sync.Mutex
  35.     // update: ceInit / ceDisposing
  36.     mDisposedFlag bool
  37. }
  39. // trigger: init()
  40. // args: empty
  41. const ceInit = "candidate.init"
  43. // trigger: Start()
  44. // args: empty
  45. const ceStart = "candidate.Start"
  47. // trigger: whenAskingForVoteThenWatchElectionTimeout()
  48. // args: empty
  49. const ceElectionTimeout = "candidate.ElectionTimeout"
  51. // trigger: Heartbeat() / AppendLog() / CommitLog()
  52. // args: empty
  53. const ceLeaderAnnounced = "candidate.LeaderAnnounced"
  55. // trigger: RequestVote()
  56. // args: *rpc.RequestVoteCmd
  57. const ceVoteToCandidate = "candidate.VoteToCandidate"
  59. // trigger: whenLeaderAnnouncedThenSwitchToFollower()
  60. // args: empty
  61. const ceDisposing = "candidate.Disposing"
  63. // trigger: beginAskForVote()
  64. // args: empty
  65. const ceAskingForVote = "candidate.AskingForVote"
  67. // trigger: handleRequestVoteOK()
  68. // args: empty
  69. const ceReceiveTicket = "candidate.ReceiveTicket"
  71. // trigger: whenReceiveTicketThenCheckTicketCount
  72. // args: empty
  73. const ceWinningTheVote = "candidate.ceWinningTheVote"
  75. func newCandidateState(ctx iRaftStateContext, term int64) IRaftState {
  76.     it := new(tCandidateState)
  77.     it.init(ctx, term)
  78.     return it
  79. }
  81. func (me *tCandidateState) init(ctx iRaftStateContext, term int64) {
  82.     me.mInitOnce.Do(func() {
  83.         me.context = ctx
  84.         me.mTerm = term
  85.         me.initEventHandlers()
  87.         me.raise(ceInit)
  88.     })
  89. }
  91. func (me *tCandidateState) initEventHandlers() {
  92.     // write only logic
  93.     me.hookEventsForTerm()
  94.     me.hookEventsForVotedTerm()
  95.     me.hookEventsForVotedCandidateID()
  96.     me.hookEventsForVotedTimestamp()
  97.     me.hookEventsForTicketCount()
  98.     me.hookEventsForDisposedFlag()
  100.     // read only logic
  101.     me.hook(ceStart,
  102.         me.whenStartThenAskForVote)
  103.     me.hook(ceAskingForVote,
  104.         me.whenAskingForVoteThenWatchElectionTimeout)
  105.     me.hook(ceReceiveTicket,
  106.         me.whenReceiveTicketThenCheckTicketCount)
  107.     me.hook(ceElectionTimeout,
  108.         me.whenElectionTimeoutThenAskForVoteAgain)
  109.     me.hook(ceWinningTheVote,
  110.         me.whenWinningTheVoteThenSwitchToLeader)
  111.     me.hook(ceLeaderAnnounced,
  112.         me.whenLeaderAnnouncedThenSwitchToFollower)
  115. }
  117. // hookEventsForTerm maintains field: mTerm
  118. // update: ceElectionTimeout
  119. func (me *tCandidateState) hookEventsForTerm() {
  120.     me.hook(ceAskingForVote, func(e string, args ...interface{}) {
  121.         me.mTerm++
  122.     })
  123. }
  125. // hookEventsForVotedTerm maintains field: mVotedTerm
  126. // update: ceInit / ceElectionTimeout / ceVoteToCandidate
  127. func (me *tCandidateState) hookEventsForVotedTerm() {
  128.     me.hook(ceInit, func(e string, args ...interface{}) {
  129.         // initially, vote to itself
  130.         me.mVotedTerm = me.mTerm
  131.     })
  133.     me.hook(ceAskingForVote, func(e string, args ...interface{}) {
  134.         // when timeout, reset to itself
  135.         me.mVotedTerm = me.mTerm
  136.     })
  138.     me.hook(ceVoteToCandidate, func(e string, args ...interface{}) {
  139.         // after vote to candidate
  140.         cmd := args[0].(*rpc.RequestVoteCmd)
  141.         me.mVotedTerm = cmd.Term
  142.     })
  143. }
  145. // hookEventsForVotedCandidateID maintains field: mVotedCandidateID
  146. // update: ceInit / ceElectionTimeout / ceVoteToCandidate
  147. func (me *tCandidateState) hookEventsForVotedCandidateID() {
  148.     me.hook(ceInit, func(e string, args ...interface{}) {
  149.         // initially, vote to itself
  150.         me.mVotedCandidateID = me.context.Config().ID()
  151.     })
  153.     me.hook(ceAskingForVote, func(e string, args ...interface{}) {
  154.         me.mVotedCandidateID = me.context.Config().ID()
  155.     })
  157.     me.hook(ceVoteToCandidate, func(e string, args ...interface{}) {
  158.         // after vote to candidate
  159.         cmd := args[0].(*rpc.RequestVoteCmd)
  160.         me.mVotedCandidateID = cmd.CandidateID
  161.     })
  162. }
  164. func (me *tCandidateState) hookEventsForVotedTimestamp() {
  165.     me.hook(ceInit, func(e string, args ...interface{}) {
  166.         // initially, vote to itself
  167.         me.mVotedTimestamp = time.Now().UnixNano()
  168.     })
  170.     me.hook(ceAskingForVote, func(e string, args ...interface{}) {
  171.         me.mVotedTimestamp = time.Now().UnixNano()
  172.     })
  174.     me.hook(ceVoteToCandidate, func(e string, args ...interface{}) {
  175.         // after vote to candidate
  176.         me.mVotedTimestamp = time.Now().UnixNano()
  177.     })
  178. }
  180. func (me *tCandidateState) hookEventsForTicketCount() {
  181.     me.hook(ceInit, func(e string, args ...interface{}) {
  182.         me.mTicketMutex = new(sync.Mutex)
  183.         me.mTicketCount = make(map[string]bool, 0)
  184.         me.mTicketCount[me.context.Config().ID()] = true
  185.     })
  187.     me.hook(ceAskingForVote, func(e string, args ...interface{}) {
  188.         me.mTicketMutex.Lock()
  189.         defer me.mTicketMutex.Unlock()
  191.         me.mTicketCount = make(map[string]bool, 0)
  192.         me.mTicketCount[me.context.Config().ID()] = true
  193.     })
  195.     me.hook(ceReceiveTicket, func(e string, args ...interface{}) {
  196.         peerID := args[0].(string)
  198.         me.mTicketMutex.Lock()
  199.         defer me.mTicketMutex.Unlock()
  200.         me.mTicketCount[peerID] = true
  201.     })
  203.     me.hook(ceDisposing, func(e string, args ...interface{}) {
  204.         me.mTicketMutex.Lock()
  205.         defer me.mTicketMutex.Unlock()
  206.         me.mTicketCount = make(map[string]bool, 0)
  207.     })
  208. }
  211. func (me *tCandidateState) hookEventsForDisposedFlag() {
  212.     me.hook(ceInit, func(e string, args ...interface{}) {
  213.         me.mDisposedFlag = false
  214.     })
  216.     me.hook(ceDisposing, func(e string, args ...interface{}) {
  217.         me.mDisposedFlag = true
  218.     })
  219. }
  221. func (me *tCandidateState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
  222.     // check term
  223.     if cmd.Term <= me.mTerm {
  224.         // bad leader
  225.         ret.Code = rpc.HBTermMismatch
  226.         return nil
  227.     }
  229.     // new leader
  230.     me.raise(ceLeaderAnnounced)
  232.     // return ok
  233.     ret.Code = rpc.HBOk
  234.     return nil
  235. }
  237. func (me *tCandidateState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
  238.     // check term
  239.     if cmd.Term <= me.mTerm {
  240.         // bad leader
  241.         ret.Code = rpc.ALTermMismatch
  242.         return nil
  243.     }
  245.     // new leader
  246.     me.raise(ceLeaderAnnounced)
  248.     // ignore and return
  249.     ret.Code = rpc.ALInternalError
  250.     return nil
  251. }
  253. func (me *tCandidateState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
  254.     // ignore and return
  255.     ret.Code = rpc.CLInternalError
  256.     return nil
  257. }
  259. func (me *tCandidateState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
  260.     // check voted term
  261.     if cmd.Term < me.mVotedTerm {
  262.         ret.Code = rpc.RVTermMismatch
  263.         return nil
  264.     }
  266.     if cmd.Term == me.mVotedTerm {
  267.         if me.mVotedCandidateID != "" && me.mVotedCandidateID != cmd.CandidateID {
  268.             // already vote another
  269.             ret.Code = rpc.RVVotedAnother
  270.             return nil
  271.         } else {
  272.             // already voted
  273.             ret.Code = rpc.RVOk
  274.             return nil
  275.         }
  276.     }
  278.     if cmd.Term > me.mVotedTerm {
  279.         // new term, check log
  280.         if cmd.LastLogIndex >= me.context.Store().LastCommittedIndex() {
  281.             // good log
  282.             me.raise(ceVoteToCandidate, cmd)
  283.             ret.Code = rpc.RVOk
  285.         } else {
  286.             // bad log
  287.             ret.Code = rpc.RVLogMismatch
  288.         }
  290.         return nil
  291.     }
  293.     // should not reaches here
  294.     ret.Code = rpc.RVTermMismatch
  295.     return nil
  296. }
  298. func (me *tCandidateState) Role() roles.RaftRole {
  299.     return roles.Candidate
  300. }
  302. func (me *tCandidateState) Start() {
  303.     me.mStartOnce.Do(func() {
  304.         me.raise(feStart)
  305.     })
  306. }
  308. func (me *tCandidateState) whenLeaderAnnouncedThenSwitchToFollower(_ string, _ ...interface{}) {
  309.     me.raise(ceDisposing)
  310.     me.context.HandleStateChanged(newFollowerState(me.context))
  311. }
  313. func (me *tCandidateState) whenElectionTimeoutThenAskForVoteAgain(_ string, _ ...interface{}) {
  314.     me.beginAskForVote()
  315. }
  317. func (me *tCandidateState) whenStartThenAskForVote(_ string, _ ...interface{}) {
  318.     me.beginAskForVote()
  319. }
  321. func (me *tCandidateState) beginAskForVote() {
  322.     // raise ceAskingForVote
  323.     me.raise(ceAskingForVote)
  325.     // for each node, call node.RequestVote
  326.     cmd := new(rpc.RequestVoteCmd)
  327.     cmd.CandidateID = me.context.Config().ID()
  328.     cmd.Term = me.mTerm
  330.     store := me.context.Store()
  331.     cmd.LastLogIndex = store.LastCommittedIndex()
  332.     cmd.LastLogTerm = store.LastCommittedTerm()
  334.     term := me.mTerm
  335.     for _,node := range me.context.Config().Nodes() {
  336.         if node.ID() == me.context.Config().ID() {
  337.             continue
  338.         }
  340.         peerID := node.ID()
  341.         go func() {
  342.             ret := new(rpc.RequestVoteRet)
  343.             err := me.context.RaftClientService().Using(peerID, func(client rpc.IRaftRPC) error {
  344.                 return client.RequestVote(cmd, ret)
  345.             })
  347.             if err == nil && ret.Code == rpc.RVOk {
  348.                 me.handleRequestVoteOK(peerID, term)
  349.             }
  350.         }()
  351.     }
  352. }
  355. func (me *tCandidateState) whenAskingForVoteThenWatchElectionTimeout(_ string, _ ...interface{}) {
  356.     term := me.mTerm
  357.     go func() {
  358.         time.Sleep(timeout.RandElectionTimeout())
  360.         if me.mDisposedFlag || me.mTerm != term {
  361.             return
  362.         }
  364.         tc := me.getTicketCount()
  365.         if tc < len(me.context.Config().Nodes())/2 + 1 {
  366.             me.raise(ceElectionTimeout)
  367.         }
  368.     }()
  369. }
  372. func (me *tCandidateState) handleRequestVoteOK(peerID string, term int64) {
  373.     if me.mDisposedFlag || me.mTerm != term {
  374.         return
  375.     }
  377.     me.raise(ceReceiveTicket, peerID)
  378. }
  381. func (me *tCandidateState) whenReceiveTicketThenCheckTicketCount(_ string, _ ...interface{}) {
  382.     tc := me.getTicketCount()
  383.     if tc >= len(me.context.Config().Nodes())/2 + 1 {
  384.         // win the vote
  385.         me.raise(ceWinningTheVote)
  386.     }
  387. }
  389. func (me *tCandidateState) getTicketCount() int {
  390.     me.mTicketMutex.Lock()
  391.     defer me.mTicketMutex.Unlock()
  392.     return len(me.mTicketCount)
  393. }
  395. func (me *tCandidateState) whenWinningTheVoteThenSwitchToLeader(_ string, _ ...interface{}) {
  396.     me.raise(ceDisposing)
  397.     me.context.HandleStateChanged(newLeaderState(me.context, me.mTerm))
  398. }

tRaftClient.go

管理到指定raft节点的rpc连接

  1. package client
  3. import (
  4.     "learning/gooop/etcd/raft/config"
  5.     rrpc "learning/gooop/etcd/raft/rpc"
  6.     "net/rpc"
  7. )
  9. type tRaftClient struct {
  10.     cfg   config.IRaftNodeConfig
  11.     conn  *rpc.Client
  12.     state iClientState
  13. }
  15. func newRaftClient(cfg config.IRaftNodeConfig, conn *rpc.Client) IRaftClient {
  16.     it := new(tRaftClient)
  17.     it.init(cfg, conn)
  18.     return it
  19. }
  21. func (me *tRaftClient) init(cfg config.IRaftNodeConfig, conn *rpc.Client) {
  22.     me.cfg = cfg
  23.     me.conn = conn
  25.     if conn == nil {
  26.         me.state = newBrokenState(me)
  27.     } else {
  28.         me.state = newConnectedState(me)
  29.     }
  30.     me.state.Start()
  31. }
  33. func (me *tRaftClient) Config() config.IRaftNodeConfig {
  34.     return me.cfg
  35. }
  37. func (me *tRaftClient) GetConn() *rpc.Client {
  38.     return me.conn
  39. }
  41. func (me *tRaftClient) SetConn(conn *rpc.Client) {
  42.     me.conn = conn
  43. }
  45. func (me *tRaftClient) HandleStateChanged(state iClientState) {
  46.     me.state = state
  47.     state.Start()
  48. }
  50. func (me *tRaftClient) Heartbeat(cmd *rrpc.HeartbeatCmd, ret *rrpc.HeartbeatRet) error {
  51.     return me.state.Heartbeat(cmd, ret)
  52. }
  54. func (me *tRaftClient) AppendLog(cmd *rrpc.AppendLogCmd, ret *rrpc.AppendLogRet) error {
  55.     return me.state.AppendLog(cmd, ret)
  56. }
  58. func (me *tRaftClient) CommitLog(cmd *rrpc.CommitLogCmd, ret *rrpc.CommitLogRet) error {
  59.     return me.state.CommitLog(cmd, ret)
  60. }
  62. func (me *tRaftClient) RequestVote(cmd *rrpc.RequestVoteCmd, ret *rrpc.RequestVoteRet) error {
  63.     return me.state.RequestVote(cmd, ret)
  64. }
  66. func (me *tRaftClient) Ping(cmd *PingCmd, ret *PingRet) error {
  67.     return me.state.Ping(cmd, ret)
  68. }

tRaftClientService.go

管理当前节点到其他raft节点的rpc连接

  1. package client
  3. import (
  4.     "errors"
  5.     "learning/gooop/etcd/raft/config"
  6.     "learning/gooop/etcd/raft/rpc"
  7.     netrpc "net/rpc"
  8. )
  10. type tRaftClientService struct {
  11.     cfg     config.IRaftConfig
  12.     clients map[string]IRaftClient
  13. }
  15. func NewRaftClientService(cfg config.IRaftConfig) IRaftClientService {
  16.     it := new(tRaftClientService)
  17.     it.init(cfg)
  18.     return it
  19. }
  21. func (me *tRaftClientService) init(cfg config.IRaftConfig) {
  22.     me.cfg = cfg
  23.     me.clients = make(map[string]IRaftClient)
  25.     for _,nc := range me.cfg.Nodes() {
  26.         me.clients[nc.ID()] = me.createRaftClient(nc)
  27.     }
  28. }
  31. func (me  *tRaftClientService) createRaftClient(nodeCfg config.IRaftNodeConfig) IRaftClient {
  32.     // dial to peer
  33.     conn, err := netrpc.Dial("tcp", nodeCfg.Endpoint())
  34.     if err != nil {
  35.         return newRaftClient(nodeCfg, nil)
  36.     } else {
  37.         return newRaftClient(nodeCfg, conn)
  38.     }
  39. }
  41. func (me *tRaftClientService) Using(peerID string, action func(client rpc.IRaftRPC) error) error {
  42.     it, ok := me.clients[peerID]
  43.     if ok {
  44.         return action(it)
  45.     } else {
  46.         return gErrorUnknownRaftPeer
  47.     }
  48. }
  51. var gErrorUnknownRaftPeer = errors.New("unknown raft peer")

（未完待续）