raft

package
v2.0.4 Latest Latest
Warning

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

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

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. 

storage := raft.NewMemoryStorage()
n := raft.StartNode(0x01, []int64{0x02, 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:
		n.Tick()
	case rd := <-s.Node.Ready():
		saveToStorage(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

View Source 
const None uint64 = 0

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

Variables

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")

Functions

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

Types

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.
	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
	// Status returns the current status of the raft state machine.
	Status() Status
	// 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, 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.
	*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 Status 

type Status struct {
	ID uint64

	pb.HardState
	SoftState

	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.

Source Files

View all Source files

Directories

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
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.