缘起

最近阅读 [云原生分布式存储基石：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 4）

  • 使用boltdb存储操作日志和kv键值数据
    • unstable存储桶：已收到未提交的日志，重启后清空
    • committed存储桶：已提交的日志
    • data存储桶：kv键值数据
    • meta存储桶：记录末次提交的index和term

设计

  • model/LogEntry: 日志条目
  • ICmd：操作指令接口
  • ICmdFactory：操作指令工厂
  • ILogStore：日志存储接口
  • tCmdBase：指令基类
  • PutCmd：put指令
  • DelCmd：del指令
  • tBoltDBStore：基于boltdb实现日志暂存，提交和应用

LogEntry.go

日志条目

  1. package model
  3. import "encoding/json"
  5. type LogEntry struct {
  6.     Tag       int
  7.     Term      int64
  8.     Index     int64
  9.     PrevTerm  int64
  10.     PrevIndex int64
  11.     Command   []byte
  12. }
  14. func (me *LogEntry) Marshal() (error, []byte) {
  15.     j, e := json.Marshal(me)
  16.     if e != nil {
  17.         return e, nil
  18.     }
  19.     return nil, j
  20. }
  22. func (me *LogEntry) Unmarshal(data []byte) error {
  23.     return json.Unmarshal(data, me)
  24. }

ICmd.go

操作指令接口

  1. package store
  3. import "github.com/boltdb/bolt"
  5. type ICmd interface {
  6.     Marshal() []byte
  7.     Unmarshal(data []byte)
  8.     Apply(tx *bolt.Tx) error
  9. }

ICmdFactory.go

操作指令工厂

  1. package store
  3. import "fmt"
  5. type ICmdFactory interface {
  6.     OfTag(tag int) ICmd
  7.     Tag(cmd ICmd) int
  8. }
  10. type tDefaultCmdFactory int
  12. const gPutCmdTag = 1
  13. const gDelCmdTag = 2
  15. func (me *tDefaultCmdFactory) OfTag(tag int) ICmd {
  16.     switch tag {
  17.     case gPutCmdTag:
  18.         return new(PutCmd)
  20.     case gDelCmdTag:
  21.         return new(DelCmd)
  22.     }
  24.     panic(fmt.Sprintf("unknown tag: %d", tag))
  25. }
  27. func (me *tDefaultCmdFactory) Tag(cmd ICmd) int {
  28.     if _, ok := cmd.(*PutCmd); ok {
  29.         return gPutCmdTag
  30.     }
  32.     if _, ok := cmd.(*DelCmd); ok {
  33.         return gDelCmdTag
  34.     }
  36.     panic(fmt.Sprintf("unknown cmd: %v", cmd))
  37. }
  39. var gCmdFactory = new(tDefaultCmdFactory)

ILogStore.go

日志存储接口

  1. package store
  3. import "learning/gooop/etcd/raft/model"
  5. type ILogStore interface {
  6.     Term() int64
  7.     Index() int64
  8.     Append(entry *model.LogEntry) error
  9.     Commit(index int64) error
  10. }

tCmdBase.go

指令基类

  1. package store
  3. import "encoding/json"
  5. type tCmdBase struct {
  6. }
  8. func (me *tCmdBase) Marshal() []byte {
  9.     j, e := json.Marshal(me)
  10.     if e != nil {
  11.         return nil
  12.     }
  13.     return j
  14. }
  16. func (me *tCmdBase) Unmarshal(data []byte) {
  17.     _ = json.Unmarshal(data, me)
  18. }

PutCmd.go

put指令

  1. package store
  3. import "github.com/boltdb/bolt"
  5. type PutCmd struct {
  6.     tCmdBase
  8.     Key   string
  9.     Value []byte
  10. }
  12. func (me *PutCmd) Apply(tx *bolt.Tx) error {
  13.     b := tx.Bucket(gDataBucket)
  14.     return b.Put([]byte(me.Key), me.Value)
  15. }

DelCmd.go

del指令

  1. package store
  3. import "github.com/boltdb/bolt"
  5. type DelCmd struct {
  6.     tCmdBase
  8.     Key string
  9. }
  11. func (me *DelCmd) Apply(tx *bolt.Tx) error {
  12.     b := tx.Bucket(gDataBucket)
  13.     return b.Delete([]byte(me.Key))
  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.     term  int64
  14.     index int64
  16.     db bolt.DB
  17. }
  19. func NewBoltStore(file string) (error, ILogStore) {
  20.     db, err := bolt.Open(file, 0600, nil)
  21.     if err != nil {
  22.         return err, nil
  23.     }
  25.     store := new(tBoltDBStore)
  26.     err = db.Update(func(tx *bolt.Tx) error {
  27.         b, e := tx.CreateBucketIfNotExists(gMetaBucket)
  28.         if e != nil {
  29.             return e
  30.         }
  32.         v := b.Get(gKeyCommittedTerm)
  33.         if v == nil {
  34.             e = b.Put(gKeyCommittedTerm, int64ToBytes(gDefaultTerm))
  35.             if e != nil {
  36.                 return e
  37.             }
  38.             store.term = gDefaultTerm
  40.         } else {
  41.             store.term = bytesToInt64(v)
  42.         }
  44.         v = b.Get(gKeyCommittedIndex)
  45.         if v == nil {
  46.             e = b.Put(gKeyCommittedIndex, int64ToBytes(gDefaultIndex))
  47.             if e != nil {
  48.                 return e
  49.             }
  50.             store.index = gDefaultIndex
  52.         } else {
  53.             store.index = bytesToInt64(v)
  54.         }
  56.         b, e = tx.CreateBucketIfNotExists(gDataBucket)
  57.         if e != nil {
  58.             return e
  59.         }
  61.         e = tx.DeleteBucket(gUnstableBucket)
  62.         if e != nil {
  63.             return e
  64.         }
  65.         _, e = tx.CreateBucket(gUnstableBucket)
  66.         if e != nil {
  67.             return e
  68.         }
  70.         _, e = tx.CreateBucketIfNotExists(gCommittedBucket)
  71.         if e != nil {
  72.             return e
  73.         }
  75.         return nil
  76.     })
  78.     if err != nil {
  79.         return err, nil
  80.     }
  82.     return nil, store
  83. }
  85. func int64ToBytes(i int64) []byte {
  86.     buf := bytes.NewBuffer(make([]byte, 8))
  87.     _ = binary.Write(buf, binary.BigEndian, i)
  88.     return buf.Bytes()
  89. }
  91. func bytesToInt64(data []byte) int64 {
  92.     var i int64
  93.     buf := bytes.NewBuffer(data)
  94.     _ = binary.Read(buf, binary.BigEndian, &i)
  95.     return i
  96. }
  98. func (me *tBoltDBStore) Term() int64 {
  99.     return me.term
  100. }
  102. func (me *tBoltDBStore) Index() int64 {
  103.     return me.index
  104. }
  106. func (me *tBoltDBStore) Append(entry *model.LogEntry) error {
  107.     cmd := gCmdFactory.OfTag(entry.Tag)
  108.     cmd.Unmarshal(entry.Command)
  110.     e, entryData := entry.Marshal()
  111.     if e != nil {
  112.         return e
  113.     }
  115.     return me.db.Update(func(tx *bolt.Tx) error {
  116.         // save log to unstable
  117.         b := tx.Bucket(gUnstableBucket)
  118.         e = b.Put(int64ToBytes(entry.Index), entryData)
  119.         if e != nil {
  120.             return e
  121.         }
  123.         me.index = entry.Index
  124.         me.term = entry.Term
  126.         return nil
  127.     })
  128. }
  130. func (me *tBoltDBStore) Commit(index int64) error {
  131.     return me.db.Update(func(tx *bolt.Tx) error {
  132.         // read unstable log
  133.         ub := tx.Bucket(gUnstableBucket)
  134.         k := int64ToBytes(index)
  135.         data := ub.Get(k)
  136.         if data == nil {
  137.             return gErrorCommitLogNotFound
  138.         }
  140.         entry := new(model.LogEntry)
  141.         e := entry.Unmarshal(data)
  142.         if e != nil {
  143.             return e
  144.         }
  146.         // apply cmd
  147.         cmd := gCmdFactory.OfTag(entry.Tag)
  148.         cmd.Unmarshal(entry.Command)
  149.         e = cmd.Apply(tx)
  150.         if e != nil {
  151.             return e
  152.         }
  154.         // save to committed log
  155.         cb := tx.Bucket(gCommittedBucket)
  156.         e = cb.Put(k, data)
  157.         if e != nil {
  158.             return e
  159.         }
  161.         // update committed.index, committed.term
  162.         mb := tx.Bucket(gMetaBucket)
  163.         e = mb.Put(gKeyCommittedIndex, int64ToBytes(index))
  164.         if e != nil {
  165.             return e
  166.         }
  168.         e = mb.Put(gKeyCommittedTerm, int64ToBytes(entry.Term))
  169.         if e != nil {
  170.             return e
  171.         }
  173.         // del unstable.index
  174.         e = ub.Delete(k)
  175.         if e != nil {
  176.             return e
  177.         }
  179.         return nil
  180.     })
  181. }
  183. var gMetaBucket = []byte("meta")
  184. var gUnstableBucket = []byte("unstable")
  185. var gCommittedBucket = []byte("committed")
  186. var gDataBucket = []byte("data")
  188. var gKeyCommittedIndex = []byte("committed.index")
  189. var gKeyCommittedTerm = []byte("committed.term")
  191. var gDefaultTerm int64 = 0
  192. var gDefaultIndex int64 = 0
  194. var gErrorCommitLogNotFound = errors.New("committing log not found")

(未完待续)