Version: v0.5.3 Latest Latest

This package is not in the latest version of its module.

Go to latest
Published: May 28, 2015 License: Apache-2.0 Imports: 10 Imported by: 0



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.

storage := raft.NewMemoryStorage()
n := raft.StartNode(0x01, []raft.Peer{{ID: 0x02}, {ID: 0x03}}, 3, 1, storage)

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.

An example MemoryStorage is provided in the raft package.

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:
	case rd := <-s.Node.Ready():
		saveToStorage(rd.State, rd.Entries)
	case <-s.done:

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

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.



View Source
const None uint64 = 0

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


View Source
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.

View Source
var (

	// ErrStopped is returned by methods on Nodes that have been stopped.
	ErrStopped = errors.New("raft: stopped")
View Source
var ErrUnavailable = errors.New("requested entry at index is unavailable")


func DescribeEntry

func DescribeEntry(e pb.Entry, f EntryFormatter) string

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

func DescribeMessage

func DescribeMessage(m pb.Message, f EntryFormatter) 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


type EntryFormatter

type EntryFormatter func([]byte) string

EntryFormatter can be implemented by the application to provide human-readable formatting of entry data. Nil is a valid EntryFormatter and will use a default format.

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.
	// 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.
	// 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.
	// 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
	// Status returns the current status of the raft state machine.
	Status() Status
	// Stop performs any necessary termination of the Node

Node represents a node in a raft cluster.

func RestartNode

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

RestartNode is similar to StartNode but does not take a list of peers. The current membership of the cluster will be restored from the Storage. If the caller has an existing state machine, pass in the last log index that has been applied to it; otherwise use zero.

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 Progress

type Progress struct {
	Match, Next uint64
	Wait        int

func (*Progress) String

func (pr *Progress) String() string

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.

	// 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.

	// 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

Possible values for StateType.

func (StateType) MarshalJSON

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

func (StateType) String

func (st StateType) String() string

type Status

type Status struct {
	ID uint64


	Applied  uint64
	Progress map[uint64]Progress

func (Status) MarshalJSON

func (s Status) MarshalJSON() ([]byte, error)

TODO: try to simplify this by introducing ID type into raft

func (Status) String

func (s Status) 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.


Path Synopsis
Package raftpb is a generated protocol buffer package.
Package raftpb is a generated protocol buffer package.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
t or T : Toggle theme light dark auto
y or Y : Canonical URL