core

Documentation

Index

• type ClusterConfig
  • func ParseClusterConfig(cfgFilePath string) (*ClusterConfig, error)
• type RaftService
• type Snapshot
• type StateMachine
  • func NewStateMachine(config StateMachineConfig) (*StateMachine, error)
  • func (sm *StateMachine) AppendData(ctx context.Context, req *konsen.AppendDataReq) (*konsen.AppendDataResp, error)
  • func (sm *StateMachine) AppendEntries(ctx context.Context, req *konsen.AppendEntriesReq) (*konsen.AppendEntriesResp, error)
  • func (sm *StateMachine) Close() error
  • func (sm *StateMachine) GetSnapshot(ctx context.Context) (*Snapshot, error)
  • func (sm *StateMachine) GetValue(ctx context.Context, key []byte) ([]byte, error)
  • func (sm *StateMachine) RequestVote(ctx context.Context, req *konsen.RequestVoteReq) (*konsen.RequestVoteResp, error)
  • func (sm *StateMachine) Run(ctx context.Context)
  • func (sm *StateMachine) SetKeyValue(ctx context.Context, kv *konsen.KVList) error
• type StateMachineConfig

Types

type ClusterConfig

type ClusterConfig struct {
	Servers         map[string]string `yaml:"servers"`                   // All servers in the cluster, a map of "serverName": "serverAddress".
	HttpServers     map[string]string `yaml:"httpServers"`               // All HTTP servers in the cluster, a map of "serverName": "httpServerAddress".
	LocalServerName string            `yaml:"localServerName,omitempty"` // Local server name.
}

ClusterConfig is the configuration of a cluster.

func ParseClusterConfig

func ParseClusterConfig(cfgFilePath string) (*ClusterConfig, error)

ParseClusterConfig parses given config YAML file.

type RaftService

type RaftService interface {
	// AppendEntries sends AppendEntries request to the remote server.
	AppendEntries(ctx context.Context, in *konsen.AppendEntriesReq) (*konsen.AppendEntriesResp, error)

	// RequestVote sends RequestVote request to the remote server.
	RequestVote(ctx context.Context, in *konsen.RequestVoteReq) (*konsen.RequestVoteResp, error)

	// AppendData sends AppendData request to the remote server.
	AppendData(ctx context.Context, in *konsen.AppendDataReq) (*konsen.AppendDataResp, error)
}

RaftService defines methods exposed by a Raft service.

type Snapshot

type Snapshot struct {
	CurrentTerm   uint64            // Current term.
	CommitIndex   uint64            // Index of highest log entry known to be committed.
	LastApplied   uint64            // Index of highest log entry applied to state machine.
	Role          konsen.Role       // Current role.
	CurrentLeader string            // Current leader.
	NextIndex     map[string]uint64 // For each server, index of the next log entry to send to that server (initialized to leader last log index + 1).
	MatchIndex    map[string]uint64 // For each server, index of highest log entry known to be replicated on that server (initialized to 0, increases monotonically).
	LogIndices    []uint64          // Logs indices.
	LogTerms      []uint64          // Log terms.
	LogBytes      []int             // Log binary sizes.
}

Snapshot is a snapshot of the internal state of a state machine. TODO: Reduce the fields or remove this, as this is for debug purpose.

type StateMachine

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

StateMachine is the state machine that implements Raft algorithm: https://raft.github.io/raft.pdf. The state machine maintains a message queue (mailbox) internally, all requests/responses to the state machine are processed asynchronously (although the caller still observes a synchronized behavior): they are firstly put onto the message queue, then the message worker (goroutine) in turns takes a message at a time and processes it, and passes the result back to caller. The internal state is never directly accessed by goroutines other than the message worker. Internal errors will always cause a crash on the server since otherwise the state machine may be left in an inconsistent state.

func NewStateMachine

func NewStateMachine(config StateMachineConfig) (*StateMachine, error)

NewStateMachine creates a new instance of the state machine.

func (*StateMachine) AppendData

func (sm *StateMachine) AppendData(ctx context.Context, req *konsen.AppendDataReq) (*konsen.AppendDataResp, error)

AppendData stores the given data into state machine, and it returns after the data is replicated onto quorum.

func (*StateMachine) AppendEntries

func (sm *StateMachine) AppendEntries(ctx context.Context, req *konsen.AppendEntriesReq) (*konsen.AppendEntriesResp, error)

AppendEntries puts the incoming AppendEntries request in main message channel and waits for result.

func (*StateMachine) Close

func (sm *StateMachine) Close() error

func (*StateMachine) GetSnapshot

func (sm *StateMachine) GetSnapshot(ctx context.Context) (*Snapshot, error)

func (*StateMachine) GetValue

func (sm *StateMachine) GetValue(ctx context.Context, key []byte) ([]byte, error)

func (*StateMachine) RequestVote

func (sm *StateMachine) RequestVote(ctx context.Context, req *konsen.RequestVoteReq) (*konsen.RequestVoteResp, error)

RequestVote puts the incoming RequestVote request in main message channel and waits for result.

func (*StateMachine) Run

func (sm *StateMachine) Run(ctx context.Context)

Run starts the state machine and blocks until done.

func (*StateMachine) SetKeyValue

func (sm *StateMachine) SetKeyValue(ctx context.Context, kv *konsen.KVList) error

type StateMachineConfig

type StateMachineConfig struct {
	Storage datastore.Storage      // Local storage instance.
	Cluster *ClusterConfig         // Cluster configuration.
	Clients map[string]RaftService // A map of "server name": "Raft service".
}

StateMachineConfig