手撸golang etcd raft协议之11

缘起

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

raft分布式一致性算法

分布式存储系统通常会通过维护多个副本来进行容错,
以提高系统的可用性。
这就引出了分布式存储系统的核心问题——如何保证多个副本的一致性?
Raft算法把问题分解成了四个子问题:
1. 领袖选举(leader election)、
2. 日志复制(log replication)、
3. 安全性(safety)
4. 成员关系变化(membership changes)
这几个子问题。
源码gitee地址:
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的实现

package config
import (
 "encoding/json"
 "os"
)
type tRaftConfig struct {
 ID string
 Nodes []*tRaftNodeConfig
}
type tRaftNodeConfig struct {
 ID string
 Endpoint string
}
func (me *tRaftConfig) GetID() string {
 return me.ID
}
func (me *tRaftConfig) GetNodes() []IRaftNodeConfig {
 a := make([]IRaftNodeConfig, len(me.Nodes))
 for i,it := range me.Nodes {
 a[i] = it
 }
 return a
}
func (me *tRaftNodeConfig) GetID() string {
 return me.ID
}
func (me *tRaftNodeConfig) GetEndpoint() string {
 return me.Endpoint
}
func LoadJSONFile(file string) IRaftConfig {
 data, err := os.ReadFile(file)
 if err != nil {
 panic(err)
 }
 c := new(tRaftConfig)
 err = json.Unmarshal(data, c)
 if err != nil {
 panic(err)
 }
 return c
}

lsm/tRaftLSMImplement.go

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

package lsm
import (
 "learning/gooop/etcd/raft/common"
 "learning/gooop/etcd/raft/config"
 "learning/gooop/etcd/raft/logger"
 "learning/gooop/etcd/raft/rpc"
 "learning/gooop/etcd/raft/rpc/client"
 "learning/gooop/etcd/raft/store"
 "sync"
)
type tRaftLSMImplement struct {
 tEventDrivenModel
 mInitOnce sync.Once
 mConfig config.IRaftConfig
 mStore store.ILogStore
 mClientService client.IRaftClientService
 mState IRaftState
}
// trigger: init()
// args: empty
const meInit = "lsm.Init"
// trigger: HandleStateChanged()
// args: IRaftState
const meStateChanged = "lsm.StateChnaged"
func (me *tRaftLSMImplement) init() {
 me.mInitOnce.Do(func() {
 me.initEventHandlers()
 me.raise(meInit)
 })
}
func (me *tRaftLSMImplement) initEventHandlers() {
 // write only
 me.hookEventsForConfig()
 me.hookEventsForStore()
 me.hookEventsForPeerService()
 me.hookEventsForState()
}
func (me *tRaftLSMImplement) hookEventsForConfig() {
 me.hook(meInit, func(e string, args ...interface{}) {
 logger.Logf("tRaftLSMImplement.init, ConfigFile = %v", common.ConfigFile)
 me.mConfig = config.LoadJSONFile(common.ConfigFile)
 })
}
func (me *tRaftLSMImplement) hookEventsForStore() {
 me.hook(meInit, func(e string, args ...interface{}) {
 logger.Logf("tRaftLSMImplement.init, DataFile = %v", common.DataFile)
 err, db := store.NewBoltStore(common.DataFile)
 if err != nil {
 panic(err)
 }
 me.mStore = db
 })
}
func (me *tRaftLSMImplement) hookEventsForPeerService() {
 me.hook(meInit, func(e string, args ...interface{}) {
 me.mClientService = client.NewRaftClientService(me.mConfig)
 })
}
func (me *tRaftLSMImplement) hookEventsForState() {
 me.hook(meInit, func(e string, args ...interface{}) {
 me.mState = newFollowerState(me, me.mStore.LastCommittedTerm())
 me.mState.Start()
 })
 me.hook(meStateChanged, func(e string, args ...interface{}) {
 state := args[0].(IRaftState)
 logger.Logf("tRaftLSMImplement.StateChanged, %v", state.Role())
 me.mState = state
 state.Start()
 })
}
func (me *tRaftLSMImplement) Config() config.IRaftConfig {
 return me.mConfig
}
func (me *tRaftLSMImplement) Store() store.ILogStore {
 return me.mStore
}
func (me *tRaftLSMImplement) HandleStateChanged(state IRaftState) {
 me.raise(meStateChanged, state)
}
func (me *tRaftLSMImplement) RaftClientService() client.IRaftClientService {
 return me.mClientService
}
func (me *tRaftLSMImplement) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error {
 state := me.mState
 e := state.Heartbeat(cmd, ret)
 logger.Logf("tRaftLSMImplement.Heartbeat, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
 return e
}
func (me *tRaftLSMImplement) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error {
 state := me.mState
 e := state.AppendLog(cmd, ret)
 logger.Logf("tRaftLSMImplement.AppendLog, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
 return e
}
func (me *tRaftLSMImplement) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error {
 state := me.mState
 e := state.CommitLog(cmd, ret)
 logger.Logf("tRaftLSMImplement.CommitLog, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
 return e
}
func (me *tRaftLSMImplement) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error {
 state := me.mState
 e := state.RequestVote(cmd, ret)
 logger.Logf("tRaftLSMImplement.RequestVote, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
 return e
}
func (me *tRaftLSMImplement) ExecuteKVCmd(cmd *rpc.KVCmd, ret *rpc.KVRet) error {
 state := me.mState
 e := state.ExecuteKVCmd(cmd, ret)
 logger.Logf("tRaftLSMImplement.ExecuteKVCmd, state=%v, cmd=%v, ret=%v, err=%v", cmd, ret, e)
 return e
}
func (me *tRaftLSMImplement) State() IRaftState {
 return me.mState
}
func NewRaftLSM() IRaftLSM {
 it := new(tRaftLSMImplement)
 it.init()
 return it
}

server/IRaftKVServer.go

server启动器接口

package server
type IRaftKVServer interface {
 BeginServeTCP(port int) error
}

server/tRaftKVServer.go

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

package server
import (
 "fmt"
 "learning/gooop/etcd/raft/lsm"
 rrpc "learning/gooop/etcd/raft/rpc"
 "learning/gooop/saga/mqs/logger"
 "net"
 "net/rpc"
 "time"
)
type tRaftKVServer int
func (me *tRaftKVServer) BeginServeTCP(port int) error {
 logger.Logf("tRaftKVServer.BeginServeTCP, starting, port=%v", port)
 // resolve address
 addy, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("0.0.0.0:%d", port))
 if err != nil {
 return err
 }
 // create raft lsm singleton
 raftLSM := lsm.NewRaftLSM()
 // register raft rpc server
 rserver := &RaftRPCServer {
 mRaftLSM : raftLSM,
 }
 err = rpc.Register(rserver)
 if err != nil {
 return err
 }
 // register kv rpc server
 kserver := &KVStoreRPCServer{
 mRaftLSM: raftLSM,
 }
 err = rpc.Register(kserver)
 if err != nil {
 return err
 }
 inbound, err := net.ListenTCP("tcp", addy)
 if err != nil {
 return err
 }
 go rpc.Accept(inbound)
 logger.Logf("tRaftKVServer.BeginServeTCP, service ready at port=%v", port)
 return nil
}
// RaftRPCServer exposes a raft rpc service
type RaftRPCServer struct {
 mRaftLSM lsm.IRaftLSM
}
// Heartbeat leader to follower
func (me *RaftRPCServer) Heartbeat(cmd *rrpc.HeartbeatCmd, ret *rrpc.HeartbeatRet) error {
 e := me.mRaftLSM.Heartbeat(cmd, ret)
 logger.Logf("RaftRPCServer.Heartbeat, cmd=%v, ret=%v, e=%v", cmd, ret, e)
 return e
}
// AppendLog leader to follower
func (me *RaftRPCServer) AppendLog(cmd *rrpc.AppendLogCmd, ret *rrpc.AppendLogRet) error {
 e := me.mRaftLSM.AppendLog(cmd, ret)
 logger.Logf("RaftRPCServer.AppendLog, cmd=%v, ret=%v, e=%v", cmd, ret, e)
 return e
}
// CommitLog leader to follower
func (me *RaftRPCServer) CommitLog(cmd *rrpc.CommitLogCmd, ret *rrpc.CommitLogRet) error {
 e := me.mRaftLSM.CommitLog(cmd, ret)
 logger.Logf("RaftRPCServer.CommitLog, cmd=%v, ret=%v, e=%v", cmd, ret, e)
 return e
}
// RequestVote candidate to others
func (me *RaftRPCServer) RequestVote(cmd *rrpc.RequestVoteCmd, ret *rrpc.RequestVoteRet) error {
 e := me.mRaftLSM.RequestVote(cmd, ret)
 logger.Logf("RaftRPCServer.RequestVote, cmd=%v, ret=%v, e=%v", cmd, ret, e)
 return e
}
// Ping to keep alive
func (me *RaftRPCServer) Ping(cmd *rrpc.PingCmd, ret *rrpc.PingRet) error {
 ret.SenderID = me.mRaftLSM.Config().GetID()
 ret.Timestamp = time.Now().UnixNano()
 logger.Logf("RaftRPCServer.Ping, cmd=%v, ret=%v", cmd, ret)
 return nil
}
// KVStoreRPCServer expose a kv storage service
type KVStoreRPCServer struct {
 mRaftLSM lsm.IRaftLSM
}
// ExecuteKVCmd leader to follower
func (me *KVStoreRPCServer) ExecuteKVCmd(cmd *rrpc.KVCmd, ret *rrpc.KVRet) error {
 e := me.mRaftLSM.ExecuteKVCmd(cmd, ret)
 logger.Logf("KVStoreRPCServer.ExecuteKVCmd, cmd=%v, ret=%v, e=%v", cmd, ret, e)
 return e
}

(未完待续)

本文来自:简书

感谢作者:老罗话编程

查看原文:手撸golang etcd raft协议之11

• 编辑
• 预览