手撸golang etcd raft协定之9,10
缘起
最近浏览 [云原生分布式存储基石：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 10）
- 增加put/get/del kv键值对的rpc接口
- 持续欠缺Leader状态的raft协定响应
设计
- rpc/IKVStoreRPC: kv操作的rpc接口
- store/IKVStore: kv操作的长久化接口
- stoer/ILogStore: 从IKVStore继承，以反对kv长久化
- lsm/IRaftState: 继承rpc.IKVStoreRPC接口，以反对kv操作
- lsm/tLeaderState: 初步实现Leader状态的raft协定解决，事件驱动的逻辑编排，读写拆散的字段治理。
rpc/IKVStoreRPC.go
kv操作的rpc接口
package rpctype IKVStoreRPC interface { ExecuteKVCmd(cmd *KVCmd, ret *KVRet) error}type KVCmd struct { OPCode KVOPCode Key []byte Content []byte}type KVOPCode intconst ( KVGet KVOPCode = iota KVPut KVOPCode = iota KVDel KVOPCode = iota)type KVRet struct { Code KVRetCode Key []byte Content []byte}type KVRetCode intconst ( KVOk KVRetCode = iota KVKeyNotFound KVRetCode = iota KVInternalError KVRetCode = iota)store/IKVStore.go
kv操作的长久化接口
package storetype IKVStore interface { Get(key []byte) (error, []byte) Put(key []byte, content []byte) error Del(key []byte) error}stoer/ILogStore.go
从IKVStore继承，以反对kv长久化
package storeimport ( "learning/gooop/etcd/raft/model")type ILogStore interface { IKVStore LastAppendedTerm() int64 LastAppendedIndex() int64 LastCommittedTerm() int64 LastCommittedIndex() int64 Append(entry *model.LogEntry) error Commit(index int64) error GetLog(index int64) (error, *model.LogEntry)}lsm/IRaftState.go
继承rpc.IKVStoreRPC接口，以反对kv操作
package lsmimport ( "learning/gooop/etcd/raft/roles" "learning/gooop/etcd/raft/rpc")type IRaftState interface { rpc.IRaftRPC rpc.IKVStoreRPC Role() roles.RaftRole Start()}lsm/tLeaderState.go
初步实现Leader状态的raft协定解决，事件驱动的逻辑编排，读写拆散的字段治理。
package lsmimport ( "errors" "learning/gooop/etcd/raft/config" "learning/gooop/etcd/raft/model" "learning/gooop/etcd/raft/roles" "learning/gooop/etcd/raft/rpc" "learning/gooop/etcd/raft/store" "learning/gooop/etcd/raft/timeout" "sync" "time")// tLeaderState presents a leader nodetype tLeaderState struct { tEventDrivenModel context iRaftStateContext mInitOnce sync.Once mStartOnce sync.Once // update: leInit / leLeaderHeartbeat mTerm int64 // update: leInit / leDisposing mDisposedFlag bool // update: leVoteToCandidate mVotedTerm int64 mVotedCandidateID string mVotedTimestamp int64}// trigger: init()// args: emptyconst leInit = "leader.init"// trigger: Start()// args: emptyconst leStart = "leader.Start"// trigger: whenNewLeaderAnnouncedThenSwitchToFollower// args: emptyconst leDiposing = "leader.Disposing"// trigger : Heartbeat() / AppendLog()// args: term int64const leNewLeaderAnnounced = "leader.NewLeaderAnnounced"// trigger: RequestVote()// args: *rpc.RequestVoteCmdconst leBeforeRequestVote = "leader.BeforeRequestVote"// trigger:// args: *rpc.RequestVoteCmdconst leVoteToCandidate = "leader.VoteToCandidate"// trigger: handleHeartbeat()// args: term int64const leHeartbeatRejected = "leader.HeartbeatRejected"func newLeaderState(ctx iRaftStateContext, term int64) IRaftState { it := new(tLeaderState) it.init(ctx, term) return it}func (me *tLeaderState) init(ctx iRaftStateContext, term int64) { me.mInitOnce.Do(func() { me.context = ctx me.mTerm = term me.initEventHandlers() me.raise(leInit) })}func (me *tLeaderState) initEventHandlers() { // write only logic me.hookEventsForDisposedFlag() me.hookEventsForVotedTerm() // read only logic me.hook(leStart, me.whenStartThenBeginHeartbeatToOthers) me.hook(leNewLeaderAnnounced, me.whenNewLeaderAnnouncedThenSwitchToFollower) me.hook(leHeartbeatRejected, me.whenHeartbeatRejectedThenSwitchToFollower)}func (me *tLeaderState) hookEventsForDisposedFlag() { me.hook(leInit, func(e string, args ...interface{}) { me.mDisposedFlag = false }) me.hook(leDiposing, func(e string, args ...interface{}) { me.mDisposedFlag = true })}func (me *tLeaderState) hookEventsForVotedTerm() { me.hook(leBeforeRequestVote, func(e string, args ...interface{}) { // check last vote timeout if me.mVotedTerm == 0 { return } if time.Duration(time.Now().UnixNano() - me.mVotedTimestamp)*time.Nanosecond >= timeout.ElectionTimeout { me.mVotedTerm = 0 me.mVotedTimestamp = 0 me.mVotedCandidateID = "" } }) me.hook(leVoteToCandidate, func(e string, args ...interface{}) { // after vote to candidate cmd := args[0].(*rpc.RequestVoteCmd) me.mVotedTerm = cmd.Term me.mVotedCandidateID = cmd.CandidateID me.mVotedTimestamp = time.Now().UnixNano() })}func (me *tLeaderState) Heartbeat(cmd *rpc.HeartbeatCmd, ret *rpc.HeartbeatRet) error { // check term if cmd.Term <= me.mTerm { ret.Code = rpc.HBTermMismatch return nil } // new leader me.raise(leNewLeaderAnnounced, cmd.Term) // return ok ret.Code = rpc.HBOk return nil}func (me *tLeaderState) AppendLog(cmd *rpc.AppendLogCmd, ret *rpc.AppendLogRet) error { // check term if cmd.Term <= me.mTerm { ret.Code = rpc.ALTermMismatch return nil } // new leader me.raise(leNewLeaderAnnounced, cmd.Term) // return ok ret.Code = rpc.ALInternalError return nil}func (me *tLeaderState) CommitLog(cmd *rpc.CommitLogCmd, ret *rpc.CommitLogRet) error { // just ignore ret.Code = rpc.CLInternalError return nil}func (me *tLeaderState) RequestVote(cmd *rpc.RequestVoteCmd, ret *rpc.RequestVoteRet) error { me.raise(leBeforeRequestVote, cmd) // check voted term if cmd.Term < me.mVotedTerm { ret.Code = rpc.RVTermMismatch return nil } if cmd.Term == me.mVotedTerm { if me.mVotedCandidateID != "" && me.mVotedCandidateID != cmd.CandidateID { // already vote another ret.Code = rpc.RVVotedAnother return nil } else { // already voted ret.Code = rpc.RVOk return nil } } if cmd.Term > me.mVotedTerm { // new term, check log if cmd.LastLogIndex >= me.context.Store().LastCommittedIndex() { // good log me.raise(leVoteToCandidate, cmd) ret.Code = rpc.RVOk } else { // bad log ret.Code = rpc.RVLogMismatch } return nil } // should not reach here ret.Code = rpc.RVTermMismatch return nil}func (me *tLeaderState) Role() roles.RaftRole { return roles.Leader}func (me *tLeaderState) Start() { me.mStartOnce.Do(func() { me.raise(leStart) })}func (me *tLeaderState) whenStartThenBeginHeartbeatToOthers(_ string, _ ...interface{}) { go func() { for !me.mDisposedFlag { _ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error { return me.handleHeartbeat(client) }) time.Sleep(timeout.HeartbeatInterval) } }()}func (me *tLeaderState) boardcast(action func(config.IRaftNodeConfig, rpc.IRaftRPC) error) error { for _,it := range me.context.Config().Nodes() { if it.ID() == me.context.Config().ID() { continue } e := me.context.RaftClientService().Using(it.ID(), func(client rpc.IRaftRPC) error { return action(it, client) }) if e != nil { return e } } return nil}func (me *tLeaderState) handleHeartbeat(client rpc.IRaftRPC) error { cmd := new(rpc.HeartbeatCmd) cmd.Term = me.mTerm cmd.LeaderID = me.context.Config().ID() ret := new(rpc.HeartbeatRet) e := client.Heartbeat(cmd, ret) if e != nil { return e } switch ret.Code { case rpc.HBTermMismatch: me.raise(leHeartbeatRejected, ret.Term) break } return nil}func (me *tLeaderState) whenNewLeaderAnnouncedThenSwitchToFollower(_ string, args ...interface{}) { me.raise(leDiposing) term := args[0].(int64) me.context.HandleStateChanged(newFollowerState(me.context, term))}func (me *tLeaderState) whenHeartbeatRejectedThenSwitchToFollower(_ string, args ...interface{}) { me.raise(leDiposing) term := args[0].(int64) me.context.HandleStateChanged(newFollowerState(me.context, term))}func (me *tLeaderState) ExecuteKVCmd(cmd *rpc.KVCmd, ret *rpc.KVRet) error { switch cmd.OPCode { case rpc.KVGet: return me.handleKVGet(cmd, ret) case rpc.KVPut: return me.handleKVPut(cmd, ret) case rpc.KVDel: return me.handleKVDel(cmd, ret) } return nil}func (me *tLeaderState) handleKVGet(cmd *rpc.KVCmd, ret *rpc.KVRet) error { e, v := me.context.Store().Get(cmd.Key) if e != nil { ret.Code = rpc.KVInternalError return e } ret.Code = rpc.KVOk ret.Content = v return nil}func (me *tLeaderState) handleKVPut(cmd *rpc.KVCmd, ret *rpc.KVRet) error { kvcmd := new(store.PutCmd) kvcmd.Key = cmd.Key kvcmd.Value = cmd.Content // create/append/commit log e := me.broadcastKVCmd(kvcmd, ret) if e != nil { return e } // apply cmd return me.context.Store().Put(cmd.Key, cmd.Content)}func (me *tLeaderState) handleKVDel(cmd *rpc.KVCmd, ret *rpc.KVRet) error { kvcmd := new(store.DelCmd) kvcmd.Key = cmd.Key // create/append/commit log e := me.broadcastKVCmd(kvcmd, ret) if e != nil { return e } // apply cmd return me.context.Store().Put(cmd.Key, cmd.Content)}func (me *tLeaderState) broadcastKVCmd(cmd store.IKVCmd, ret *rpc.KVRet) error { // create log st := me.context.Store() log := new(model.LogEntry) log.Term = me.mTerm log.Index = st.LastCommittedIndex() + 1 log.PrevTerm = st.LastCommittedTerm() log.PrevIndex = st.LastCommittedIndex() log.Command = cmd.Marshal() // append log e := st.Append(log) if e != nil { ret.Code = rpc.KVInternalError return e } // ask other nodes to append log alcmd := new(rpc.AppendLogCmd) alcmd.Term = me.mTerm alcmd.LeaderID = me.context.Config().ID() alcmd.Entry = log sumOk := []int{ 0 } _ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error { alret := new(rpc.AppendLogRet) e := client.AppendLog(alcmd, alret) if e != nil { return e } switch alret.Code { case rpc.ALOk: sumOk[0]++ break case rpc.ALTermMismatch: // todo: fixme break case rpc.ALIndexMismatch: // todo: fixme break } return nil }) // wait for most nodes if sumOk[0] >= len(me.context.Config().Nodes()) / 2 { // commit log clcmd := new(rpc.CommitLogCmd) clcmd.LeaderID = me.context.Config().ID() clcmd.Term = me.mTerm clcmd.Index = log.Index _ = me.boardcast(func(_ config.IRaftNodeConfig, client rpc.IRaftRPC) error { ret := new(rpc.CommitLogRet) e := client.CommitLog(clcmd, ret) if e != nil { return e } switch ret.Code { case rpc.CLInternalError: // todo: fixme break case rpc.CLLogNotFound: // todo: fixme break case rpc.CLOk: return nil } return nil }) // ok return nil } else { return gErrorCannotReachAgreement }}var gErrorCannotReachAgreement = errors.New("cannot reach agreement")（未完待续）