缘起

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

raft分布式一致性算法

  1. 分布式存储系统通常会通过维护多个副本来进行容错，
  2. 以提高系统的可用性。
  3. 这就引出了分布式存储系统的核心问题——如何保证多个副本的一致性？
  5. Raft算法把问题分解成了领袖选举（leader election）、
  6. 日志复制（log replication）、安全性（safety）
  7. 和成员关系变化（membership changes）这几个子问题。
  9. Raft算法的基本操作只需2种RPC即可完成。
  10. RequestVote RPC是在选举过程中通过旧的Leader触发的，
  11. AppendEntries RPC是领导人触发的，目的是向其他节点复制日志条目和发送心跳（heartbeat）。

目标

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

子目标（Day 5）

  • 重新设计RPC接口，将原有浓缩的两个接口分解为更易于理解和实现的四个接口( 尽信书则不如无书 -_-|| )
  • 根据新RPC接口重写Follower状态的实现

设计

  • IRaftRPC: 将原有浓缩的两个接口分解为更易于理解和实现的四个接口
  • IRaftLSM: 添加部分包内支持接口
  • iEventDrivenModel：抽取并实现事件驱动型的逻辑编排
  • ILogStore：改造适配新分解的RPC接口
  • tBoltDBStore：基于boltdb实现日志暂存，提交和应用
  • tFollowerState：根据新分解的RPC接口，重写Follower状态的实现（未完成）

IRaftRPC.go

将原有浓缩的两个接口分解为更易于理解和实现的四个接口。尽信书则不如无书-_-||

  1. package rpc
  3. import "learning/gooop/etcd/raft/model"
  5. type IRaftRPC interface {
  6.     // leader to follower
  7.     Heartbeat(cmd *HeartbeatCmd, ret *HeartbeatRet) error
  9.     // leader to follower
  10.     AppendLog(cmd *AppendLogCmd, ret *AppendLogRet) error
  12.     // leader to follower
  13.     CommitLog(cmd *CommitLogCmd, ret *CommitLogRet) error
  15.     // candidate to follower
  16.     RequestVote(cmd *RequestVoteCmd, ret *RequestVoteRet) error
  17. }
  20. type HeartbeatCmd struct {
  21.     LeaderID string
  22.     Term int64
  23. }
  25. type HeartbeatRet struct {
  26.     Code HBCode
  27.     Term int64
  28. }
  30. type HBCode int
  31. const (
  32.     HBOk HBCode = iota
  33.     HBTermMismatch HBCode = iota
  34. )
  36. type RequestVoteCmd struct {
  37.     CandidateID  string
  38.     Term         int64
  39.     LastLogIndex int64
  40.     LastLogTerm int64
  41. }
  43. type RequestVoteRet struct {
  44.     Code RVCode
  45.     Term        int64
  46. }
  48. type RVCode int
  49. const (
  50.     RVOk RVCode = iota
  51.     RVLogMismatch RVCode = iota
  52.     RVTermMismatch RVCode = iota
  53.     RVVotedAnother RVCode = iota
  54. )
  56. type AppendLogCmd struct {
  57.     LeaderID     string
  58.     Term         int64
  59.     Entry *model.LogEntry
  60. }
  62. type AppendLogRet struct {
  63.     Code ALCode
  64.     Term    int64
  65.     PrevLogIndex int64
  66.     PrevLogTerm int64
  67. }
  69. type ALCode int
  70. const (
  71.     ALOk ALCode = iota
  72.     ALTermMismatch ALCode = iota
  73.     ALIndexMismatch ALCode = iota
  74.     ALInternalError ALCode = iota
  75. )
  77. type CommitLogCmd struct {
  78.     LeaderID     string
  79.     Term         int64
  80.     Index int64
  81. }
  83. type CommitLogRet struct {
  84.     Code CLCode
  85. }
  87. type CLCode int
  88. const (
  89.     CLOk CLCode = iota
  90.     CLLogNotFound CLCode = iota
  91.     CLInternalError CLCode = iota
  92. )

IRaftLSM.go

添加部分包内支持接口

  1. package lsm
  3. import (
  4.     "learning/gooop/etcd/raft/config"
  5.     "learning/gooop/etcd/raft/rpc"
  6.     "learning/gooop/etcd/raft/store"
  7. )
  9. // IRaftLSM raft有限状态自动机
  10. type IRaftLSM interface {
  11.     rpc.IRaftRPC
  14.     State() IRaftState
  16.     config() config.IRaftConfig
  17.     store() store.ILogStore
  18.     handleStateChanged(state IRaftState)
  19. }

iEventDrivenModel.go

抽取并实现事件驱动型的逻辑编排

  1. package lsm
  3. type tEventHandleFunc func(e string, args... interface{})
  5. type iEventDrivenModel interface {
  6.     hook(e string, handleFunc tEventHandleFunc)
  7.     raise(e string, args... interface{})
  8. }
  10. type tEventDrivenModel struct {
  11.     items map[string][]tEventHandleFunc
  12. }
  14. func (me *tEventDrivenModel) hook(e string, handler tEventHandleFunc) {
  15.     arr, ok := me.items[e]
  16.     if ok {
  17.         me.items[e] = append(arr, handler)
  18.     } else {
  19.         me.items[e] = []tEventHandleFunc{handler }
  20.     }
  21. }
  24. func (me *tEventDrivenModel) raise(e string, args... interface{}) {
  25.     if handlers, ok := me.items[e];ok {
  26.         for _,it := range handlers {
  27.             it(e, args...)
  28.         }
  29.     }
  30. }

ILogStore.go

改造适配新分解的RPC接口

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

tBoltDBStore.go

基于boltdb实现日志暂存，提交和应用

  1. package store
  3. import (
  4.     "bytes"
  5.     "encoding/binary"
  6.     "errors"
  7.     "github.com/boltdb/bolt"
  8.     "learning/gooop/etcd/raft/model"
  9. )
  11. type tBoltDBStore struct {
  12.     file  string
  13.     lastAppendedTerm  int64
  14.     lastAppendedIndex int64
  15.     lastCommittedTerm  int64
  16.     lastCommittedIndex int64
  18.     db bolt.DB
  19. }
  23. func NewBoltStore(file string) (error, ILogStore) {
  24.     db, err := bolt.Open(file, 0600, nil)
  25.     if err != nil {
  26.         return err, nil
  27.     }
  29.     store := new(tBoltDBStore)
  30.     err = db.Update(func(tx *bolt.Tx) error {
  31.         b, e := tx.CreateBucketIfNotExists(gMetaBucket)
  32.         if e != nil {
  33.             return e
  34.         }
  36.         v := b.Get(gKeyCommittedTerm)
  37.         if v == nil {
  38.             e = b.Put(gKeyCommittedTerm, int64ToBytes(gDefaultTerm))
  39.             if e != nil {
  40.                 return e
  41.             }
  42.             store.lastCommittedTerm = gDefaultTerm
  44.         } else {
  45.             store.lastCommittedTerm = bytesToInt64(v)
  46.         }
  48.         v = b.Get(gKeyCommittedIndex)
  49.         if v == nil {
  50.             e = b.Put(gKeyCommittedIndex, int64ToBytes(gDefaultIndex))
  51.             if e != nil {
  52.                 return e
  53.             }
  54.             store.lastCommittedIndex = gDefaultIndex
  56.         } else {
  57.             store.lastCommittedIndex = bytesToInt64(v)
  58.         }
  60.         b, e = tx.CreateBucketIfNotExists(gDataBucket)
  61.         if e != nil {
  62.             return e
  63.         }
  65.         e = tx.DeleteBucket(gUnstableBucket)
  66.         if e != nil {
  67.             return e
  68.         }
  69.         _, e = tx.CreateBucket(gUnstableBucket)
  70.         if e != nil {
  71.             return e
  72.         }
  74.         _, e = tx.CreateBucketIfNotExists(gCommittedBucket)
  75.         if e != nil {
  76.             return e
  77.         }
  79.         return nil
  80.     })
  82.     if err != nil {
  83.         return err, nil
  84.     }
  86.     return nil, store
  87. }
  89. func int64ToBytes(i int64) []byte {
  90.     buf := bytes.NewBuffer(make([]byte, 8))
  91.     _ = binary.Write(buf, binary.BigEndian, i)
  92.     return buf.Bytes()
  93. }
  95. func bytesToInt64(data []byte) int64 {
  96.     var i int64
  97.     buf := bytes.NewBuffer(data)
  98.     _ = binary.Read(buf, binary.BigEndian, &i)
  99.     return i
  100. }
  103. func (me *tBoltDBStore) LastCommittedTerm() int64 {
  104.     return me.lastCommittedTerm
  105. }
  107. func (me *tBoltDBStore) LastCommittedIndex() int64 {
  108.     return me.lastCommittedIndex
  109. }
  111. func (me *tBoltDBStore) LastAppendedTerm() int64 {
  112.     return me.lastAppendedTerm
  113. }
  115. func (me *tBoltDBStore) LastAppendedIndex() int64 {
  116.     return me.lastAppendedIndex
  117. }
  119. func (me *tBoltDBStore) Append(entry *model.LogEntry) error {
  120.     cmd := gCmdFactory.OfTag(entry.Tag)
  121.     cmd.Unmarshal(entry.Command)
  123.     e, entryData := entry.Marshal()
  124.     if e != nil {
  125.         return e
  126.     }
  128.     return me.db.Update(func(tx *bolt.Tx) error {
  129.         // save log to unstable
  130.         b := tx.Bucket(gUnstableBucket)
  131.         e = b.Put(int64ToBytes(entry.Index), entryData)
  132.         if e != nil {
  133.             return e
  134.         }
  136.         return nil
  137.     })
  138. }
  140. func (me *tBoltDBStore) Commit(index int64) error {
  141.     return me.db.Update(func(tx *bolt.Tx) error {
  142.         // read unstable log
  143.         ub := tx.Bucket(gUnstableBucket)
  144.         k := int64ToBytes(index)
  145.         data := ub.Get(k)
  146.         if data == nil {
  147.             return gErrorCommitLogNotFound
  148.         }
  150.         entry := new(model.LogEntry)
  151.         e := entry.Unmarshal(data)
  152.         if e != nil {
  153.             return e
  154.         }
  156.         // apply cmd
  157.         cmd := gCmdFactory.OfTag(entry.Tag)
  158.         cmd.Unmarshal(entry.Command)
  159.         e = cmd.Apply(tx)
  160.         if e != nil {
  161.             return e
  162.         }
  164.         // save to committed log
  165.         cb := tx.Bucket(gCommittedBucket)
  166.         e = cb.Put(k, data)
  167.         if e != nil {
  168.             return e
  169.         }
  171.         // update committed.index, committed.term
  172.         mb := tx.Bucket(gMetaBucket)
  173.         e = mb.Put(gKeyCommittedIndex, int64ToBytes(index))
  174.         if e != nil {
  175.             return e
  176.         }
  178.         e = mb.Put(gKeyCommittedTerm, int64ToBytes(entry.Term))
  179.         if e != nil {
  180.             return e
  181.         }
  183.         // del unstable.index
  184.         e = ub.Delete(k)
  185.         if e != nil {
  186.             return e
  187.         }
  189.         me.lastCommittedIndex = entry.Index
  190.         me.lastCommittedTerm = entry.Term
  191.         return nil
  192.     })
  193. }
  196. func (me *tBoltDBStore) GetLog(index int64) (error, *model.LogEntry) {
  197.     ret := []*model.LogEntry{ nil }
  198.     e :=  me.db.View(func(tx *bolt.Tx) error {
  199.         k := int64ToBytes(index)
  200.         v := tx.Bucket(gCommittedBucket).Get(k)
  202.         if v == nil {
  203.             return nil
  204.         }
  206.         entry := new(model.LogEntry)
  207.         e := entry.Unmarshal(v)
  208.         if e != nil {
  209.             return e
  210.         }
  212.         ret[0] = entry
  213.         return nil
  214.     })
  216.     return e, ret[0]
  217. }
  219. var gMetaBucket = []byte("meta")
  220. var gUnstableBucket = []byte("unstable")
  221. var gCommittedBucket = []byte("committed")
  222. var gDataBucket = []byte("data")
  224. var gKeyCommittedIndex = []byte("committed.index")
  225. var gKeyCommittedTerm = []byte("committed.term")
  227. var gDefaultTerm int64 = 0
  228. var gDefaultIndex int64 = 0
  230. var gErrorCommitLogNotFound = errors.New("committing log not found")

tFollowerState.go

根据新分解的RPC接口，重写Follower状态的实现（未完成）

  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. // tFollowerState presents a follower node
  12. type tFollowerState struct {
  13.     tEventDrivenModel
  15.     context IRaftLSM
  17.     mInitOnce  sync.Once
  18.     mStartOnce sync.Once
  19.     mDisposeOnce sync.Once
  21.     // updated when init, set term == store.lastCommittedTerm
  22.     // updated when leader.heartbeat
  23.     mTerm int64
  25.     // updated when leader.heartbeat
  26.     mLeaderHeartbeatClock int64
  28.     mVotedLeaderID string
  29.     mVotedTimestamp int64
  30. }
  32. const feStart string = "follower.Start"
  33. const feLeaderHeartbeatTimeout string = "follower.LeaderHeartbeatTimeout"
  35. func newFollowerState(ctx IRaftLSM) IRaftState {
  36.     it := new(tFollowerState)
  37.     it.init(ctx)
  38.     return it
  39. }
  41. func (me *tFollowerState) init(ctx IRaftLSM) {
  42.     me.mInitOnce.Do(func() {
  43.         me.context = ctx
  44.         me.mTerm = ctx.store().LastCommittedTerm()
  45.         me.mLeaderHeartbeatClock = 0
  46.         me.initEventHandlers()
  47.     })
  48. }
  51. func (me *tFollowerState) initEventHandlers() {
  52.     me.hook(feStart,
  53.         me.whenStartThenBeginWatchLeaderTimeout)
  54.     me.hook(feLeaderHeartbeatTimeout,
  55.         me.whenLeaderHeartbeatTimeoutThenSwitchToCandidateState)
  56. }
  58. func (me *tFollowerState) Start() {
  59.     me.mStartOnce.Do(func() {
  60.         me.raise(feStart)
  61.     })
  62. }
  64. func (me *tFollowerState) whenStartThenBeginWatchLeaderTimeout(e string, args ...interface{}) {
  65.     go func() {
  66.         iCheckingTimeoutInterval := timeout.HeartbeatTimeout / 3
  67.         iHeartbeatTimeoutNanos := int64(timeout.HeartbeatTimeout / time.Nanosecond)
  68.         for range time.Tick(iCheckingTimeoutInterval) {
  69.             now := time.Now().UnixNano()
  70.             if now - me.mLeaderHeartbeatClock >= iHeartbeatTimeoutNanos {
  71.                 me.raise(feLeaderHeartbeatTimeout)
  72.                 return
  73.             }
  74.         }
  75.     }()
  76. }
  78. func (me *tFollowerState) whenLeaderHeartbeatTimeoutThenSwitchToCandidateState(_ string, args ...interface{}) {
  79.     panic("implements me")
  80. }
  82. func (me *tFollowerState) Role() roles.RaftRole {
  83.     return roles.Follower
  84. }
  87. // Heartbeat leader to follower
  88. func (me *tFollowerState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
  89.     if cmd.Term < me.mTerm {
  90.         // invalid leader
  91.         ret.Code = rpc.HBTermMismatch
  92.         ret.Term = me.mTerm
  93.         return nil
  95.     } else if cmd.Term > me.mTerm {
  96.         // new leader
  97.         me.mTerm = cmd.Term
  98.     }
  100.     // update heartbeat clock and return
  101.     me.mLeaderHeartbeatClock = time.Now().UnixNano()
  102.     ret.Code = rpc.HBOk
  103.     return nil
  104. }
  106. // AppendLog leader to follower
  107. func (me *tFollowerState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
  108.     ret.Term = me.mTerm
  110.     if cmd.Term < me.mTerm {
  111.         // invalid leader
  112.         ret.Code = rpc.ALTermMismatch
  113.         return nil
  114.     }
  116.     store := me.context.store()
  117.     entry := cmd.Entry
  119.     // check log: expecting appending action follows previous committing action
  120.     if entry.PrevIndex != store.LastCommittedIndex() || entry.PrevTerm != store.LastCommittedTerm() {
  121.         // check log
  122.         e, log := store.GetLog(entry.Index)
  123.         if e != nil {
  124.             ret.Code = rpc.ALInternalError
  125.             return nil
  126.         }
  128.         if log == nil || log.PrevIndex != entry.PrevIndex || log.PrevTerm != entry.PrevTerm {
  129.             // bad log
  130.             ret.Code = rpc.ALIndexMismatch
  131.             ret.PrevLogIndex = store.LastCommittedIndex()
  132.             ret.PrevLogTerm = store.LastCommittedTerm()
  133.             return nil
  134.         }
  136.         // good log, but old, just ignore it
  137.         ret.Code = rpc.ALOk
  138.         return nil
  139.     }
  141.     // good log
  142.     e := store.Append(entry)
  143.     if e != nil {
  144.         ret.Code = rpc.ALInternalError
  145.         return nil
  146.     } else {
  147.         ret.Code = rpc.ALOk
  148.         return nil
  149.     }
  150. }
  152. // CommitLog leader to follower
  153. func (me *tFollowerState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
  154.     store := me.context.store()
  155.     if cmd.Index != store.LastAppendedIndex() || cmd.Term != store.LastAppendedTerm() {
  156.         // bad index
  157.         ret.Code = rpc.CLLogNotFound
  158.         return nil
  159.     }
  161.     e := store.Commit(cmd.Index)
  162.     if e != nil {
  163.         ret.Code = rpc.CLInternalError
  164.         return nil
  165.     }
  167.     ret.Code = rpc.CLOk
  168.     return nil
  169. }
  171. // RequestVote candidate to follower
  172. func (me *tFollowerState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
  173.     panic("implements me")
  174. }

（未完待续）