raft

Documentation

Overview

Package raft provides an implementation of the raft consensus algorithm.

The primary object in raft is a Node. You either start a Node from scratch using raft.StartNode or start a Node from some initial state using raft.RestartNode.

n := raft.StartNode(0x01, []int64{0x02, 0x03}, 3, 1)

Now that you are holding onto a Node you have a few responsibilities:

First, you need to push messages that you receive from other machines into the Node with n.Step().

func recvRaftRPC(ctx context.Context, m raftpb.Message) {
	n.Step(ctx, m)
}

Second, you need to save log entries to storage, process committed log entries through your application and then send pending messages to peers by reading the channel returned by n.Ready(). It is important that the user persist any entries that require stable storage before sending messages to other peers to ensure fault-tolerance.

And finally you need to service timeouts with Tick(). Raft has two important timeouts: heartbeat and the election timeout. However, internally to the raft package time is represented by an abstract "tick". The user is responsible for calling Tick() on their raft.Node on a regular interval in order to service these timeouts.

The total state machine handling loop will look something like this:

for {
	select {
	case <-s.Ticker:
		n.Tick()
	case rd := <-s.Node.Ready():
		saveToStable(rd.State, rd.Entries)
		send(rd.Messages)
		process(rd.CommittedEntries)
		s.Node.Advance()
	case <-s.done:
		return
	}
}

To propose changes to the state machine from your node take your application data, serialize it into a byte slice and call:

n.Propose(ctx, data)

If the proposal is committed, data will appear in committed entries with type raftpb.EntryNormal.

To add or remove node in a cluster, build ConfChange struct 'cc' and call:

n.ProposeConfChange(ctx, cc)

After config change is committed, some committed entry with type raftpb.EntryConfChange will be returned. You should apply it to node through:

var cc raftpb.ConfChange
cc.Unmarshal(data)
n.ApplyConfChange(cc)

Note: An ID represents a unique node in a cluster. A given ID MUST be used only once even if the old node has been removed.

Index

• Constants
• Variables
• func IsEmptyHardState(st pb.HardState) bool
• func IsEmptySnap(sp pb.Snapshot) bool
• type Node
  • func RestartNode(id uint64, election, heartbeat int, snapshot *pb.Snapshot, st pb.HardState, ...) Node
  • func StartNode(id uint64, peers []Peer, election, heartbeat int) Node
• type Peer
• type Ready
• type SoftState
• type StateType
  • func (st StateType) MarshalJSON() ([]byte, error)
  • func (st StateType) String() string

Constants

const None uint64 = 0

None is a placeholder node ID used when there is no leader.

Variables

var (

	// ErrStopped is returned by methods on Nodes that have been stopped.
	ErrStopped = errors.New("raft: stopped")
)

Functions

func IsEmptyHardState

func IsEmptyHardState(st pb.HardState) bool

IsEmptyHardState returns true if the given HardState is empty.

func IsEmptySnap

func IsEmptySnap(sp pb.Snapshot) bool

IsEmptySnap returns true if the given Snapshot is empty.

Types

type Node

type Node interface {
	// Tick increments the internal logical clock for the Node by a single tick. Election
	// timeouts and heartbeat timeouts are in units of ticks.
	Tick()
	// Campaign causes the Node to transition to candidate state and start campaigning to become leader.
	Campaign(ctx context.Context) error
	// Propose proposes that data be appended to the log.
	Propose(ctx context.Context, data []byte) error
	// ProposeConfChange proposes config change.
	// At most one ConfChange can be in the process of going through consensus.
	// Application needs to call ApplyConfChange when applying EntryConfChange type entry.
	ProposeConfChange(ctx context.Context, cc pb.ConfChange) error
	// Step advances the state machine using the given message. ctx.Err() will be returned, if any.
	Step(ctx context.Context, msg pb.Message) error
	// Ready returns a channel that returns the current point-in-time state
	// Users of the Node must call Advance after applying the state returned by Ready
	Ready() <-chan Ready
	// Advance notifies the Node that the application has applied and saved progress up to the last Ready.
	// It prepares the node to return the next available Ready.
	Advance()
	// ApplyConfChange applies config change to the local node.
	// TODO: reject existing node when add node
	// TODO: reject non-existant node when remove node
	ApplyConfChange(cc pb.ConfChange)
	// Stop performs any necessary termination of the Node
	Stop()
	// Compact discards the entrire log up to the given index. It also
	// generates a raft snapshot containing the given nodes configuration
	// and the given snapshot data.
	// It is the caller's responsibility to ensure the given configuration
	// and snapshot data match the actual point-in-time configuration and snapshot
	// at the given index.
	Compact(index uint64, nodes []uint64, d []byte)
}

Node represents a node in a raft cluster.

func RestartNode

func RestartNode(id uint64, election, heartbeat int, snapshot *pb.Snapshot, st pb.HardState, ents []pb.Entry) Node

RestartNode is identical to StartNode but takes an initial State and a slice of entries. Generally this is used when restarting from a stable storage log.

func StartNode

func StartNode(id uint64, peers []Peer, election, heartbeat int) Node

StartNode returns a new Node given a unique raft id, a list of raft peers, and the election and heartbeat timeouts in units of ticks. It appends a ConfChangeAddNode entry for each given peer to the initial log.

type Peer

type Peer struct {
	ID      uint64
	Context []byte
}

type Ready

type Ready struct {
	// The current volatile state of a Node.
	// SoftState will be nil if there is no update.
	// It is not required to consume or store SoftState.
	*SoftState

	// The current state of a Node to be saved to stable storage BEFORE
	// Messages are sent.
	// HardState will be equal to empty state if there is no update.
	pb.HardState

	// Entries specifies entries to be saved to stable storage BEFORE
	// Messages are sent.
	Entries []pb.Entry

	// Snapshot specifies the snapshot to be saved to stable storage.
	Snapshot pb.Snapshot

	// CommittedEntries specifies entries to be committed to a
	// store/state-machine. These have previously been committed to stable
	// store.
	CommittedEntries []pb.Entry

	// Messages specifies outbound messages to be sent AFTER Entries are
	// committed to stable storage.
	Messages []pb.Message
}

Ready encapsulates the entries and messages that are ready to read, be saved to stable storage, committed or sent to other peers. All fields in Ready are read-only.

type SoftState

type SoftState struct {
	Lead      uint64
	RaftState StateType
	Nodes     []uint64
}

SoftState provides state that is useful for logging and debugging. The state is volatile and does not need to be persisted to the WAL.

type StateType

type StateType uint64

StateType represents the role of a node in a cluster.

const (
	StateFollower StateType = iota
	StateCandidate
	StateLeader
)

Possible values for StateType.

func (StateType) MarshalJSON

func (st StateType) MarshalJSON() ([]byte, error)

func (StateType) String

func (st StateType) String() string