raft

Documentation

Index

• Constants
• func NewCommandHandler(controller *Controller) channel.TypedReceiveHandler
• func NewHcLogrusLogger() hclog.Logger
• func NewJoinHandler(controller *Controller) channel.TypedReceiveHandler
• func NewRemoveHandler(controller *Controller) channel.TypedReceiveHandler
• type BoltDbFsm
  • func NewFsm(dataDir string, decoders command.Decoders) *BoltDbFsm
  • func (self *BoltDbFsm) Apply(log *raft.Log) interface{}
  • func (self *BoltDbFsm) GetDb() *db.Db
  • func (self *BoltDbFsm) Init() error
  • func (self *BoltDbFsm) RaftInitialized()
  • func (self *BoltDbFsm) Restore(snapshot io.ReadCloser) error
  • func (self *BoltDbFsm) Snapshot() (raft.FSMSnapshot, error)
• type Config
• type Controller
  • func NewController(id *identity.TokenId, config *Config, metricsRegistry metrics.Registry) (*Controller, error)
  • func (self *Controller) ApplyEncodedCommand(encoded []byte) error
  • func (self *Controller) ApplyWithTimeout(log []byte, timeout time.Duration) (interface{}, error)
  • func (self *Controller) Dispatch(cmd command.Command) error
  • func (self *Controller) GetDb() *db.Db
  • func (self *Controller) GetLeaderAddr() string
  • func (self *Controller) GetMesh() mesh.Mesh
  • func (self *Controller) GetRaft() *raft.Raft
  • func (self *Controller) HandleJoin(req *JoinRequest) error
  • func (self *Controller) HandleJoinAsLeader(req *JoinRequest) error
  • func (self *Controller) HandleRemove(req *RemoveRequest) error
  • func (self *Controller) HandleRemoveAsLeader(req *RemoveRequest) error
  • func (self *Controller) Init() error
  • func (self *Controller) IsLeader() bool
  • func (self *Controller) Join(req *JoinRequest) error
  • func (self *Controller) ListMembers() ([]*Member, error)
  • func (self *Controller) RemoveServer(id string) error
• type JoinRequest
  • func (self *JoinRequest) Decode(msg *channel.Message) error
  • func (self *JoinRequest) Encode() (*channel.Message, error)
• type Member
• type MemberModel
• type RemoveRequest
  • func (self *RemoveRequest) Decode(msg *channel.Message) error
  • func (self *RemoveRequest) Encode() (*channel.Message, error)

Constants

const (
	NewLogEntryType     = 2050
	ErrorResponseType   = 2051
	SuccessResponseType = 2052
	JoinRequestType     = 2053
	RemoveRequestType   = 2054

	HeaderErrorCode = 1000

	ErrorCodeBadMessage = 1
	ErrorCodeNotLeader  = 2
	ErrorCodeApiError   = 3
	ErrorCodeGeneric    = 4
)

Functions

func NewCommandHandler

func NewCommandHandler(controller *Controller) channel.TypedReceiveHandler

func NewHcLogrusLogger

func NewHcLogrusLogger() hclog.Logger

func NewJoinHandler

func NewJoinHandler(controller *Controller) channel.TypedReceiveHandler

func NewRemoveHandler

func NewRemoveHandler(controller *Controller) channel.TypedReceiveHandler

Types

type BoltDbFsm

type BoltDbFsm struct {
	// contains filtered or unexported fields
}

func NewFsm

func NewFsm(dataDir string, decoders command.Decoders) *BoltDbFsm

func (*BoltDbFsm) Apply

func (self *BoltDbFsm) Apply(log *raft.Log) interface{}

func (*BoltDbFsm) GetDb

func (self *BoltDbFsm) GetDb() *db.Db

func (*BoltDbFsm) Init

func (self *BoltDbFsm) Init() error

func (*BoltDbFsm) RaftInitialized

func (self *BoltDbFsm) RaftInitialized()

func (*BoltDbFsm) Restore

func (self *BoltDbFsm) Restore(snapshot io.ReadCloser) error

func (*BoltDbFsm) Snapshot

func (self *BoltDbFsm) Snapshot() (raft.FSMSnapshot, error)

type Config

type Config struct {
	Recover               bool
	DataDir               string
	MinClusterSize        uint32
	AdvertiseAddress      string
	BindAddress           string
	BootstrapMembers      []string
	CommandHandlerOptions struct {
		MaxQueueSize uint16
		MaxWorkers   uint16
	}
}

type Controller

type Controller struct {
	Id *identity.TokenId

	Config *Config
	Mesh   mesh.Mesh
	Raft   *raft.Raft
	Fsm    *BoltDbFsm
	// contains filtered or unexported fields
}

Controller manages RAFT related state and operations

func NewController

func NewController(id *identity.TokenId, config *Config, metricsRegistry metrics.Registry) (*Controller, error)

func (*Controller) ApplyEncodedCommand

func (self *Controller) ApplyEncodedCommand(encoded []byte) error

ApplyEncodedCommand applies the command to the RAFT distributed log

func (*Controller) ApplyWithTimeout

func (self *Controller) ApplyWithTimeout(log []byte, timeout time.Duration) (interface{}, error)

ApplyWithTimeout applies the given command to the RAFT distributed log with the given timeout

func (*Controller) Dispatch

func (self *Controller) Dispatch(cmd command.Command) error

Dispatch dispatches the given command to the current leader. If the current node is the leader, the command will be applied and the result returned

func (*Controller) GetDb

func (self *Controller) GetDb() *db.Db

GetDb returns the DB instance

func (*Controller) GetLeaderAddr

func (self *Controller) GetLeaderAddr() string

GetLeaderAddr returns the current leader address, which may be blank if there is no leader currently

func (*Controller) GetMesh

func (self *Controller) GetMesh() mesh.Mesh

GetMesh returns the related Mesh instance

func (*Controller) GetRaft

func (self *Controller) GetRaft() *raft.Raft

GetRaft returns the managed raft instance

func (*Controller) HandleJoin

func (self *Controller) HandleJoin(req *JoinRequest) error

func (*Controller) HandleJoinAsLeader

func (self *Controller) HandleJoinAsLeader(req *JoinRequest) error

func (*Controller) HandleRemove

func (self *Controller) HandleRemove(req *RemoveRequest) error

func (*Controller) HandleRemoveAsLeader

func (self *Controller) HandleRemoveAsLeader(req *RemoveRequest) error

func (*Controller) Init

func (self *Controller) Init() error

Init sets up the Mesh and Raft instances

func (*Controller) IsLeader

func (self *Controller) IsLeader() bool

IsLeader returns true if the current node is the RAFT leader

func (*Controller) Join

func (self *Controller) Join(req *JoinRequest) error

Join adds the given node to the raft cluster

func (*Controller) ListMembers

func (self *Controller) ListMembers() ([]*Member, error)

func (*Controller) RemoveServer

func (self *Controller) RemoveServer(id string) error

RemoveServer removes the node specified by the given id from the raft cluster

type JoinRequest

type JoinRequest struct {
	Addr    string
	Id      string
	IsVoter bool
}

func (*JoinRequest) Decode

func (self *JoinRequest) Decode(msg *channel.Message) error

func (*JoinRequest) Encode

func (self *JoinRequest) Encode() (*channel.Message, error)

type Member

type Member struct {
	Id     string
	Addr   string
	Voter  bool
	Leader bool
}

type MemberModel

type MemberModel interface {
	// ListMembers returns the current set of raft members
	ListMembers() ([]*Member, error)
	// HandleJoin adds a node to the raft cluster
	HandleJoin(req *JoinRequest) error
	// HandleRemove removes a node from the raft cluster
	HandleRemove(req *RemoveRequest) error
}

MemberModel presents information about and operations on RAFT membership

type RemoveRequest

type RemoveRequest struct {
	Id string
}

func (*RemoveRequest) Decode

func (self *RemoveRequest) Decode(msg *channel.Message) error

func (*RemoveRequest) Encode

func (self *RemoveRequest) Encode() (*channel.Message, error)