缘起

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

  • 虽然Leader State还有细节没处理完，但应该能启动并提供基本服务了
  • 添加外围功能，为首次“点火”做准备：
    • config/tRaftConfig：从本地json文件读取集群节点配置，提供IRaftConfig/IRaftNodeConfig的实现
    • lsm/tRaftLSMImplement： 提供对顶层接口IRaftLSM的实现，将“配置/kv存储/节点通讯”三大块粘合起来
    • server/IRaftKVServer：server启动器接口
    • server/tRaftKVServer: server启动器的实现，监听raft rpc和kv rpc

config/tRaftConfig.go

从本地json文件读取集群节点配置，提供IRaftConfig/IRaftNodeConfig的实现

  1. package config
  3. import (
  4.     "encoding/json"
  5.     "os"
  6. )
  8. type tRaftConfig struct {
  9.     ID string
  10.     Nodes []*tRaftNodeConfig
  11. }
  13. type tRaftNodeConfig struct {
  14.     ID string
  15.     Endpoint string
  16. }
  18. func (me *tRaftConfig) GetID() string {
  19.     return me.ID
  20. }
  22. func (me *tRaftConfig) GetNodes() []IRaftNodeConfig {
  23.     a := make([]IRaftNodeConfig, len(me.Nodes))
  24.     for i,it := range me.Nodes {
  25.         a[i] = it
  26.     }
  27.     return a
  28. }
  31. func (me *tRaftNodeConfig) GetID() string {
  32.     return me.ID
  33. }
  35. func (me *tRaftNodeConfig) GetEndpoint() string {
  36.     return me.Endpoint
  37. }
  39. func LoadJSONFile(file string) IRaftConfig {
  40.     data, err := os.ReadFile(file)
  41.     if err != nil {
  42.         panic(err)
  43.     }
  45.     c := new(tRaftConfig)
  46.     err = json.Unmarshal(data, c)
  47.     if err != nil {
  48.         panic(err)
  49.     }
  51.     return c
  52. }

lsm/tRaftLSMImplement.go

提供对顶层接口IRaftLSM的实现，将“配置/kv存储/节点通讯”三大块粘合起来，并添加诊断日志。

  1. package lsm
  3. import (
  4.     "learning/gooop/etcd/raft/common"
  5.     "learning/gooop/etcd/raft/config"
  6.     "learning/gooop/etcd/raft/logger"
  7.     "learning/gooop/etcd/raft/rpc"
  8.     "learning/gooop/etcd/raft/rpc/client"
  9.     "learning/gooop/etcd/raft/store"
  10.     "sync"
  11. )
  13. type tRaftLSMImplement struct {
  14.     tEventDrivenModel
  15.     mInitOnce sync.Once
  17.     mConfig config.IRaftConfig
  18.     mStore store.ILogStore
  19.     mClientService client.IRaftClientService
  20.     mState IRaftState
  21. }
  24. // trigger: init()
  25. // args: empty
  26. const meInit = "lsm.Init"
  28. // trigger: HandleStateChanged()
  29. // args: IRaftState
  30. const meStateChanged = "lsm.StateChnaged"
  32. func (me *tRaftLSMImplement) init() {
  33.     me.mInitOnce.Do(func() {
  34.         me.initEventHandlers()
  35.         me.raise(meInit)
  36.     })
  37. }
  39. func (me *tRaftLSMImplement) initEventHandlers() {
  40.     // write only
  41.     me.hookEventsForConfig()
  42.     me.hookEventsForStore()
  43.     me.hookEventsForPeerService()
  44.     me.hookEventsForState()
  45. }
  47. func (me *tRaftLSMImplement) hookEventsForConfig() {
  48.     me.hook(meInit, func(e string, args ...interface{}) {
  49.         logger.Logf("tRaftLSMImplement.init, ConfigFile = %v", common.ConfigFile)
  50.         me.mConfig = config.LoadJSONFile(common.ConfigFile)
  51.     })
  52. }
  54. func (me *tRaftLSMImplement) hookEventsForStore() {
  55.     me.hook(meInit, func(e string, args ...interface{}) {
  56.         logger.Logf("tRaftLSMImplement.init, DataFile = %v", common.DataFile)
  57.         err, db := store.NewBoltStore(common.DataFile)
  58.         if err != nil {
  59.             panic(err)
  60.         }
  61.         me.mStore = db
  62.     })
  63. }
  66. func (me *tRaftLSMImplement) hookEventsForPeerService() {
  67.     me.hook(meInit, func(e string, args ...interface{}) {
  68.         me.mClientService = client.NewRaftClientService(me.mConfig)
  69.     })
  70. }
  72. func (me *tRaftLSMImplement) hookEventsForState() {
  73.     me.hook(meInit, func(e string, args ...interface{}) {
  74.         me.mState = newFollowerState(me, me.mStore.LastCommittedTerm())
  75.         me.mState.Start()
  76.     })
  78.     me.hook(meStateChanged, func(e string, args ...interface{}) {
  79.         state := args[0].(IRaftState)
  80.         logger.Logf("tRaftLSMImplement.StateChanged, %v", state.Role())
  82.         me.mState = state
  83.         state.Start()
  84.     })
  85. }
  88. func (me *tRaftLSMImplement) Config() config.IRaftConfig {
  89.     return me.mConfig
  90. }
  92. func (me *tRaftLSMImplement) Store() store.ILogStore {
  93.     return me.mStore
  94. }
  96. func (me *tRaftLSMImplement) HandleStateChanged(state IRaftState) {
  97.     me.raise(meStateChanged, state)
  98. }
  100. func (me *tRaftLSMImplement) RaftClientService() client.IRaftClientService {
  101.     return me.mClientService
  102. }
  104. func (me *tRaftLSMImplement) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
  105.     state := me.mState
  106.     e := state.Heartbeat(cmd, ret)
  107.     logger.Logf("tRaftLSMImplement.Heartbeat, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
  108.     return e
  109. }
  111. func (me *tRaftLSMImplement) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
  112.     state := me.mState
  113.     e := state.AppendLog(cmd, ret)
  114.     logger.Logf("tRaftLSMImplement.AppendLog, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
  115.     return e
  116. }
  118. func (me *tRaftLSMImplement) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
  119.     state := me.mState
  120.     e := state.CommitLog(cmd, ret)
  121.     logger.Logf("tRaftLSMImplement.CommitLog, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
  122.     return e
  123. }
  125. func (me *tRaftLSMImplement) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
  126.     state := me.mState
  127.     e := state.RequestVote(cmd, ret)
  128.     logger.Logf("tRaftLSMImplement.RequestVote, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
  129.     return e
  130. }
  132. func (me *tRaftLSMImplement) ExecuteKVCmd(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
  133.     state := me.mState
  134.     e := state.ExecuteKVCmd(cmd, ret)
  135.     logger.Logf("tRaftLSMImplement.ExecuteKVCmd, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
  136.     return e
  137. }
  139. func (me *tRaftLSMImplement) State() IRaftState {
  140.     return me.mState
  141. }
  143. func NewRaftLSM() IRaftLSM {
  144.     it := new(tRaftLSMImplement)
  145.     it.init()
  146.     return it
  147. }

server/IRaftKVServer.go

server启动器接口

  1. package server
  3. type IRaftKVServer interface {
  4.     BeginServeTCP(port int) error
  5. }

server/tRaftKVServer.go

server启动器的实现，监听raft rpc和kv rpc

  1. package server
  3. import (
  4.     "fmt"
  5.     "learning/gooop/etcd/raft/lsm"
  6.     rrpc "learning/gooop/etcd/raft/rpc"
  7.     "learning/gooop/saga/mqs/logger"
  8.     "net"
  9.     "net/rpc"
  10.     "time"
  11. )
  13. type tRaftKVServer int
  15. func (me *tRaftKVServer) BeginServeTCP(port int) error {
  16.     logger.Logf("tRaftKVServer.BeginServeTCP, starting, port=%v", port)
  18.     // resolve address
  19.     addy, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("0.0.0.0:%d", port))
  20.     if err != nil {
  21.         return err
  22.     }
  24.     // create raft lsm singleton
  25.     raftLSM := lsm.NewRaftLSM()
  27.     // register raft rpc server
  28.     rserver := &RaftRPCServer {
  29.         mRaftLSM : raftLSM,
  30.     }
  31.     err = rpc.Register(rserver)
  32.     if err != nil {
  33.         return err
  34.     }
  36.     // register kv rpc server
  37.     kserver := &KVStoreRPCServer{
  38.         mRaftLSM: raftLSM,
  39.     }
  40.     err = rpc.Register(kserver)
  41.     if err != nil {
  42.         return err
  43.     }
  45.     inbound, err := net.ListenTCP("tcp", addy)
  46.     if err != nil {
  47.         return err
  48.     }
  49.     go rpc.Accept(inbound)
  51.     logger.Logf("tRaftKVServer.BeginServeTCP, service ready at port=%v", port)
  52.     return nil
  53. }
  55. // RaftRPCServer exposes a raft rpc service
  56. type RaftRPCServer struct {
  57.     mRaftLSM lsm.IRaftLSM
  58. }
  60. // Heartbeat leader to follower
  61. func (me *RaftRPCServer) Heartbeat(cmd *rrpc.HeartbeatCmd, ret *rrpc.HeartbeatRet) error {
  62.     e := me.mRaftLSM.Heartbeat(cmd, ret)
  63.     logger.Logf("RaftRPCServer.Heartbeat, cmd=%v, ret=%v, e=%v", cmd, ret, e)
  64.     return e
  65. }
  67. // AppendLog leader to follower
  68. func (me *RaftRPCServer) AppendLog(cmd *rrpc.AppendLogCmd, ret *rrpc.AppendLogRet) error {
  69.     e := me.mRaftLSM.AppendLog(cmd, ret)
  70.     logger.Logf("RaftRPCServer.AppendLog, cmd=%v, ret=%v, e=%v", cmd, ret, e)
  71.     return e
  72. }
  74. // CommitLog leader to follower
  75. func (me *RaftRPCServer) CommitLog(cmd *rrpc.CommitLogCmd, ret *rrpc.CommitLogRet) error {
  76.     e := me.mRaftLSM.CommitLog(cmd, ret)
  77.     logger.Logf("RaftRPCServer.CommitLog, cmd=%v, ret=%v, e=%v", cmd, ret, e)
  78.     return e
  79. }
  81. // RequestVote candidate to others
  82. func (me *RaftRPCServer) RequestVote(cmd *rrpc.RequestVoteCmd, ret *rrpc.RequestVoteRet) error {
  83.     e := me.mRaftLSM.RequestVote(cmd, ret)
  84.     logger.Logf("RaftRPCServer.RequestVote, cmd=%v, ret=%v, e=%v", cmd, ret, e)
  85.     return e
  86. }
  88. // Ping to keep alive
  89. func (me *RaftRPCServer) Ping(cmd *rrpc.PingCmd, ret *rrpc.PingRet) error {
  90.     ret.SenderID = me.mRaftLSM.Config().GetID()
  91.     ret.Timestamp = time.Now().UnixNano()
  93.     logger.Logf("RaftRPCServer.Ping, cmd=%v, ret=%v", cmd, ret)
  94.     return nil
  95. }
  97. // KVStoreRPCServer expose a kv storage service
  98. type KVStoreRPCServer struct {
  99.     mRaftLSM lsm.IRaftLSM
  100. }
  102. // ExecuteKVCmd leader to follower
  103. func (me *KVStoreRPCServer) ExecuteKVCmd(cmd *rrpc.KVCmd, ret *rrpc.KVRet) error {
  104.     e := me.mRaftLSM.ExecuteKVCmd(cmd, ret)
  105.     logger.Logf("KVStoreRPCServer.ExecuteKVCmd, cmd=%v, ret=%v, e=%v", cmd, ret, e)
  106.     return e
  107. }

（未完待续）