手撸golang etcd raft协议之2

缘起

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

raft分布式一致性算法

分布式存储系统通常会通过维护多个副本来进行容错，
以提高系统的可用性。
这就引出了分布式存储系统的核心问题——如何保证多个副本的一致性？

Raft算法把问题分解成了领袖选举（leader election）、
日志复制（log replication）、安全性（safety）
和成员关系变化（membership changes）这几个子问题。

Raft算法的基本操作只需2种RPC即可完成。
RequestVote RPC是在选举过程中通过旧的Leader触发的，
AppendEntries RPC是领导人触发的，目的是向其他节点复制日志条目和发送心跳（heartbeat）。

目标

• 根据raft协议，实现高可用分布式强一致的kv存储

子目标(Day 2)

• 定义raft rpc接口
• 定义raft lsm有限状态自动机接口（状态模式）

设计

• config/IRaftConfig.go: 集群配置接口。简单起见， 使用静态配置模式定义节点数量和地址。
• config/IRaftNodeConfig.go: 节点配置接口
• roles/roles.go：raft三种角色常量
• timeout/timeout.go：超时时间常量
• rpc/IRaftRPC.go: raft协议的基本RPC接口及参数定义。简单起见，拟采用net/rpc实现之。
• rpc/IRaftRPCServer.go: 支持raft协议的服务器接口。简单起见，拟采用net/rpc实现之。
• lsm/IRaftLSM.go: raft有限状态机接口
• lsm/IRaftState.go: 状态接口
• lsm/tRaftStateBase.go: 基本状态数据
• lsm/tFollowerState: follower状态的实现，未完成

config/IRaftConfig.go

集群配置接口。简单起见， 使用静态配置模式定义节点数量和地址。

package config

type IRaftConfig interface {
    ID() string
    Nodes() []IRaftNodeConfig
}

config/IRaftNodeConfig.go

节点配置接口

package config

type IRaftNodeConfig interface {
    ID() string
    Endpoint() string
}

roles/roles.go

raft三种角色常量

package roles

type RaftRole int

const Follower RaftRole = 1
const Candidate RaftRole = 2
const Leader RaftRole = 3

timeout/timeout.go

超时时间常量

package timeout

import "time"

const HeartbeatInterval = 150 * time.Millisecond
const HeartbeatTimeout = 5 * HeartbeatInterval
const ElectionTimeout = HeartbeatTimeout

rpc/IRaftRPC.go

raft协议的基本RPC接口及参数定义。简单起见，拟采用net/rpc实现之。

package rpc

type IRaftRPC interface {
    RequestVote(cmd *RequestVoteCmd, ret *RequestVoteRet) error
    AppendEntries(cmd *AppendEntriesCmd, ret *AppendEntriesRet) error
}

type RequestVoteCmd struct {
    CandidateID  string
    Term         int
    LastLogIndex int
    LastLogTerm  int
}

type RequestVoteRet struct {
    Term        int
    VoteGranted bool
}

type AppendEntriesCmd struct {
    Term         int
    LeaderID     string
    PrevLogTerm  int
    PrevLogIndex int
    LeaderCommit int
    Entries      []*LogEntry
}

type LogEntry struct {
    Tag     int
    Content []byte
}

type AppendEntriesRet struct {
    Term    int
    Success bool
}

rpc/IRaftRPCServer.go

支持raft协议的服务器接口。简单起见，拟采用net/rpc实现之。

package rpc

type IRaftRPCServer interface {
    BeginServeTCP(port int, r IRaftRPC)
}

lsm/IRaftLSM.go

raft有限状态机接口

package lsm

import "learning/gooop/etcd/raft/rpc"

// IRaftLSM raft有限状态自动机
type IRaftLSM interface {
    rpc.IRaftRPC

    State() IRaftState
}

lsm/IRaftState.go

状态接口

package lsm

import (
    "learning/gooop/etcd/raft/roles"
    "learning/gooop/etcd/raft/rpc"
)

type IRaftState interface {
    rpc.IRaftRPC

    Role() roles.RaftRole
    Start()
}

lsm/tRaftStateBase.go

基本状态数据

package lsm

import (
    "learning/gooop/etcd/raft/config"
    "learning/gooop/etcd/raft/roles"
)

//
type tRaftStateBase struct {
    // 当前角色
    role roles.RaftRole

    // 当前任期号
    term int

    // leader.id
    leaderID string

    // 集群配置
    cfg config.IRaftConfig
}

func newRaftStateBase(term int, cfg config.IRaftConfig) *tRaftStateBase {
    it := new(tRaftStateBase)
    it.init(term, cfg)
    return it
}

// init initialize self, with term and config specified
func (me *tRaftStateBase) init(term int, cfg config.IRaftConfig) {
    me.cfg = cfg
    me.role = roles.Follower
    me.term = term
    me.leaderID = ""
}

func (me *tRaftStateBase) Role() roles.RaftRole {
    return me.role
}

lsm/tFollowerState

follower状态的实现，未完成

package lsm

import (
    "learning/gooop/etcd/raft/config"
    "learning/gooop/etcd/raft/timeout"
    "sync"
    "time"
)

type tFollowerState struct {
    tRaftStateBase

    mInitOnce  sync.Once
    mStartOnce sync.Once
    mEventMap  map[tFollowerEvent][]tFollowerEventHandler
}

type tFollowerEvent int

const evStart tFollowerEvent = 1

type tFollowerEventHandler func(e tFollowerEvent, args ...interface{})

func newFollowerState(term int, cfg config.IRaftConfig) *tFollowerState {
    it := new(tFollowerState)
    it.init(term, cfg)

    // todo: to implement IRaftState
    return it
}

func (me *tFollowerState) init(term int, cfg config.IRaftConfig) {
    me.mInitOnce.Do(func() {
        me.tRaftStateBase = *newRaftStateBase(term, cfg)

        // init event map
        me.mEventMap = make(map[tFollowerEvent][]tFollowerEventHandler)
        me.registerEventHandlers()
    })
}

func (me *tFollowerState) registerEventHandlers() {
    me.mEventMap[evStart] = []tFollowerEventHandler{
        me.afterStartThenBeginWatchLeaderTimeout,
    }
}

func (me *tFollowerState) raise(e tFollowerEvent, args ...interface{}) {
    if handlers, ok := me.mEventMap[e]; ok {
        for _, it := range handlers {
            it(e, args...)
        }
    }
}

func (me *tFollowerState) Start() {
    me.mStartOnce.Do(func() {
        me.raise(evStart)
    })
}

func (me *tFollowerState) afterStartThenBeginWatchLeaderTimeout(e tFollowerEvent, args ...interface{}) {
    go func() {
        iCheckingTimeoutInterval := timeout.HeartbeatTimeout / 3
        for range time.Tick(iCheckingTimeoutInterval) {
            // todo: watch leader.AppendEntries rpc timeout
        }
    }()
}

(未完待续)