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 DescribeEntry(e pb.Entry) string
• func DescribeMessage(m pb.Message) string
• func IsEmptyHardState(st pb.HardState) bool
• func IsEmptySnap(sp pb.Snapshot) bool
• func IsLocalMsg(m pb.Message) bool
• func IsResponseMsg(m pb.Message) bool
• type MemoryStorage
  • func NewMemoryStorage() *MemoryStorage
  • func (ms *MemoryStorage) Append(entries []pb.Entry) error
  • func (ms *MemoryStorage) ApplySnapshot(snap pb.Snapshot) error
  • func (ms *MemoryStorage) Compact(i uint64, cs *pb.ConfState, data []byte) error
  • func (ms *MemoryStorage) Entries(lo, hi uint64) ([]pb.Entry, error)
  • func (ms *MemoryStorage) FirstIndex() (uint64, error)
  • func (ms *MemoryStorage) InitialState() (pb.HardState, pb.ConfState, error)
  • func (ms *MemoryStorage) LastIndex() (uint64, error)
  • func (ms *MemoryStorage) SetHardState(st pb.HardState) error
  • func (ms *MemoryStorage) Snapshot() (pb.Snapshot, error)
  • func (ms *MemoryStorage) Term(i uint64) (uint64, error)
• type Node
  • func RestartNode(id uint64, election, heartbeat int, storage Storage) Node
  • func StartNode(id uint64, peers []Peer, election, heartbeat int, storage Storage) Node
• type Peer
• type Ready
• type SoftState
• type StateType
  • func (st StateType) MarshalJSON() ([]byte, error)
  • func (st StateType) String() string
• type Storage

Constants

const None uint64 = 0

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

Variables

var ErrCompacted = errors.New("requested index is unavailable due to compaction")

ErrCompacted is returned by Storage.Entries/Compact when a requested index is unavailable because it predates the last snapshot.

var (

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

var ErrUnavailable = errors.New("requested entry at index is unavailable")

Functions

func DescribeEntry

func DescribeEntry(e pb.Entry) string

DescribeEntry returns a concise human-readable description of an Entry for debugging.

func DescribeMessage

func DescribeMessage(m pb.Message) string

DescribeMessage returns a concise human-readable description of a Message for debugging.

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.

func IsLocalMsg

func IsLocalMsg(m pb.Message) bool

func IsResponseMsg

func IsResponseMsg(m pb.Message) bool

Types

type MemoryStorage

type MemoryStorage struct {
	// Protects access to all fields. Most methods of MemoryStorage are
	// run on the raft goroutine, but Append() is run on an application
	// goroutine.
	sync.Mutex
	// contains filtered or unexported fields
}

MemoryStorage implements the Storage interface backed by an in-memory array.

func NewMemoryStorage

func NewMemoryStorage() *MemoryStorage

NewMemoryStorage creates an empty MemoryStorage.

func (*MemoryStorage) Append

func (ms *MemoryStorage) Append(entries []pb.Entry) error

Append the new entries to storage.

func (*MemoryStorage) ApplySnapshot

func (ms *MemoryStorage) ApplySnapshot(snap pb.Snapshot) error

ApplySnapshot overwrites the contents of this Storage object with those of the given snapshot.

func (*MemoryStorage) Compact

func (ms *MemoryStorage) Compact(i uint64, cs *pb.ConfState, data []byte) error

Compact discards all log entries prior to i. Creates a snapshot which can be retrieved with the Snapshot() method and can be used to reconstruct the state at that point. If any configuration changes have been made since the last compaction, the result of the last ApplyConfChange must be passed in. It is the application's responsibility to not attempt to compact an index greater than raftLog.applied.

func (*MemoryStorage) Entries

func (ms *MemoryStorage) Entries(lo, hi uint64) ([]pb.Entry, error)

Entries implements the Storage interface.

func (*MemoryStorage) FirstIndex

func (ms *MemoryStorage) FirstIndex() (uint64, error)

FirstIndex implements the Storage interface.

func (*MemoryStorage) InitialState

func (ms *MemoryStorage) InitialState() (pb.HardState, pb.ConfState, error)

InitialState implements the Storage interface.

func (*MemoryStorage) LastIndex

func (ms *MemoryStorage) LastIndex() (uint64, error)

LastIndex implements the Storage interface.

func (*MemoryStorage) SetHardState

func (ms *MemoryStorage) SetHardState(st pb.HardState) error

SetHardState saves the current HardState.

func (*MemoryStorage) Snapshot

func (ms *MemoryStorage) Snapshot() (pb.Snapshot, error)

Snapshot implements the Storage interface.

func (*MemoryStorage) Term

func (ms *MemoryStorage) Term(i uint64) (uint64, error)

Term implements the Storage interface.

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.
	// Returns an opaque ConfState protobuf which must be recorded
	// in snapshots. Will never return nil; it returns a pointer only
	// to match MemoryStorage.Compact.
	ApplyConfChange(cc pb.ConfChange) *pb.ConfState
	// Stop performs any necessary termination of the Node
	Stop()
}

Node represents a node in a raft cluster.

func RestartNode

func RestartNode(id uint64, election, heartbeat int, storage Storage) Node

RestartNode is identical to StartNode but does not take a list of peers. The current membership of the cluster will be restored from the Storage.

func StartNode

func StartNode(id uint64, peers []Peer, election, heartbeat int, storage Storage) 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
}

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

type Storage

type Storage interface {
	// InitialState returns the saved HardState and ConfState information.
	InitialState() (pb.HardState, pb.ConfState, error)
	// Entries returns a slice of log entries in the range [lo,hi).
	Entries(lo, hi uint64) ([]pb.Entry, error)
	// Term returns the term of entry i, which must be in the range
	// [FirstIndex()-1, LastIndex()]. The term of the entry before
	// FirstIndex is retained for matching purposes even though the
	// rest of that entry may not be available.
	Term(i uint64) (uint64, error)
	// LastIndex returns the index of the last entry in the log.
	LastIndex() (uint64, error)
	// FirstIndex returns the index of the first log entry that is
	// possibly available via Entries (older entries have been incorporated
	// into the latest Snapshot; if storage only contains the dummy entry the
	// first log entry is not available).
	FirstIndex() (uint64, error)
	// Snapshot returns the most recent snapshot.
	Snapshot() (pb.Snapshot, error)
}

Storage is an interface that may be implemented by the application to retrieve log entries from storage.

If any Storage method returns an error, the raft instance will become inoperable and refuse to participate in elections; the application is responsible for cleanup and recovery in this case.