Shulou(Shulou.com)06/03 Report--

Today, I would like to talk to you about what the golang etcd raft agreement is, many people may not know much about it. In order to make you understand better, the editor has summarized the following content for you. I hope you can get something according to this article.

Raft distributed consistency algorithm distributed storage systems usually maintain multiple copies for fault tolerance in order to improve the availability of the system. This leads to the core question of distributed storage systems-how to ensure the consistency of multiple replicas? Raft algorithm decomposes the problem into four sub-problems: 1. Leader Election (leader election), 2. Log replication (log replication), 3. Security (safety) 4. The sub-problems of membership change (membership changes). Source code gitee address: https://gitee.com/ioly/learning.gooop destination

Implement highly available distributed and strongly consistent kv storage according to raft protocol

Subgoal (Day 11)

Although Leader State still has details to deal with, it should be able to start and provide basic services.

Add peripheral features to prepare for the first "ignition":

Config/tRaftConfig: reads the cluster node configuration from the local json file and provides the implementation of IRaftConfig/IRaftNodeConfig

Lsm/tRaftLSMImplement: provides the implementation of the top-level interface IRaftLSM, gluing together the three pieces of "configuration / kv storage / node communication"

Server/IRaftKVServer:server initiator interface

Server/tRaftKVServer: implementation of server initiator, monitoring raft rpc and kv rpc

Config/tRaftConfig.go

Read the cluster node configuration from the local json file and provide the implementation of IRaftConfig/IRaftNodeConfig

Package configimport ("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

Provide the implementation of the top-level interface IRaftLSM, glue together the three blocks of "configuration / kv storage / node communication", and add diagnostic logs.

Package lsmimport ("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: emptyconst meInit = "lsm.Init" / / trigger: HandleStateChanged () / / args: IRaftStateconst 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 initiator interface

Package servertype IRaftKVServer interface {BeginServeTCP (port int) error} server/tRaftKVServer.go

Implementation of server initiator to monitor raft rpc and kv rpc

Package serverimport ("fmt"learning/gooop/etcd/raft/lsm" rrpc "learning/gooop/etcd/raft/rpc"learning/gooop/saga/mqs/logger"net"net/rpc"time") type tRaftKVServer intfunc (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 servicetype RaftRPCServer struct {mRaftLSM lsm.IRaftLSM} / / Heartbeat leader to followerfunc (me * RaftRPCServer) Heartbeat (cmd * rrpc.HeartbeatCmd, ret * rrpc.HeartbeatRet) error {e: = me.mRaftLSM.Heartbeat (cmd) Ret) logger.Logf ("RaftRPCServer.Heartbeat, cmd=%v, ret=%v, eBay% v", cmd, ret, e) return e} / / AppendLog leader to followerfunc (me * RaftRPCServer) AppendLog (cmd * rrpc.AppendLogCmd, ret * rrpc.AppendLogRet) error {e: = me.mRaftLSM.AppendLog (cmd, ret) logger.Logf ("RaftRPCServer.AppendLog, cmd=%v, ret=%v, eBay% v", cmd, ret E) return e} / / CommitLog leader to followerfunc (me * RaftRPCServer) CommitLog (cmd * rrpc.CommitLogCmd, ret * rrpc.CommitLogRet) error {e: = me.mRaftLSM.CommitLog (cmd, ret) logger.Logf ("RaftRPCServer.CommitLog, cmd=%v, ret=%v, eBay% v", cmd, ret, e) return e} / / RequestVote candidate to othersfunc (me * RaftRPCServer) RequestVote (cmd * rrpc.RequestVoteCmd Ret * rrpc.RequestVoteRet) error {e: = me.mRaftLSM.RequestVote (cmd, ret) logger.Logf ("RaftRPCServer.RequestVote, cmd=%v, ret=%v, eBay% v", cmd, ret, e) return e} / / Ping to keep alivefunc (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 servicetype KVStoreRPCServer struct {mRaftLSM lsm.IRaftLSM} / / ExecuteKVCmd leader to followerfunc (me * KVStoreRPCServer) ExecuteKVCmd (cmd * rrpc.KVCmd) Ret * rrpc.KVRet) error {e: = me.mRaftLSM.ExecuteKVCmd (cmd, ret) logger.Logf ("KVStoreRPCServer.ExecuteKVCmd, cmd=%v, ret=%v, eBay% v", cmd, ret, e) return e}

(to be continued)

After reading the above, do you have any further understanding of the golang etcd raft agreement? If you want to know more knowledge or related content, please follow the industry information channel, thank you for your support.

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.