手撸 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
}
(未完待续)