const (

	DefaultTimeoutScale = 256 * 1024


	DefaultMaxRPCsInFlight = 2
)

var (


	ErrLeader = errors.New("node is the leader")


	ErrNotLeader = errors.New("node is not the leader")


	ErrNotVoter = errors.New("node is not a voter")


	ErrLeadershipLost = errors.New("leadership lost while committing log")


	ErrAbortedByRestore = errors.New("snapshot restored while committing log")


	ErrRaftShutdown = errors.New("raft is already shutdown")

	ErrEnqueueTimeout = errors.New("timed out enqueuing operation")


	ErrNothingNewToSnapshot = errors.New("nothing new to snapshot")


	ErrUnsupportedProtocol = errors.New("operation not supported with current protocol version")


	ErrCantBootstrap = errors.New("bootstrap only works on new clusters")


	ErrLeadershipTransferInProgress = errors.New("leadership transfer in progress")
)

View Source

var (

	ErrLogNotFound = errors.New("log not found")


	ErrPipelineReplicationNotSupported = errors.New("pipeline replication not supported")
)

func BootstrapCluster(conf *Config, logs LogStore, stable StableStore,
	snaps SnapshotStore, trans Transport, configuration Configuration,
) error

BootstrapCluster initializes a server's storage with the given cluster configuration. This should only be called at the beginning of time for the cluster with an identical configuration listing all Voter servers. There is no need to bootstrap Nonvoter and Staging servers.

A cluster can only be bootstrapped once from a single participating Voter server. Any further attempts to bootstrap will return an error that can be safely ignored.

One approach is to bootstrap a single server with a configuration listing just itself as a Voter, then invoke AddVoter() on it to add other servers to the cluster.

func EncodeConfiguration(configuration Configuration) []byte

EncodeConfiguration serializes a Configuration using MsgPack, or panics on errors.

HasExistingState returns true if the server has any existing state (logs, knowledge of a current term, or any snapshots).

NOTE: This is exposed for middleware testing purposes and is not a stable API

func MakeClusterCustom(t *testing.T, opts *MakeClusterOpts) *cluster

NOTE: This is exposed for middleware testing purposes and is not a stable API

func MakeClusterNoBootstrap(n int, t *testing.T, conf *Config) *cluster

NOTE: This is exposed for middleware testing purposes and is not a stable API

func NewInmemTransport(addr ServerAddress) (ServerAddress, *InmemTransport)

NewInmemTransport is used to initialize a new transport and generates a random local address if none is specified

NewInmemTransportWithTimeout is used to initialize a new transport and generates a random local address if none is specified. The given timeout will be used to decide how long to wait for a connected peer to process the RPCs that we're sending it. See also Connect() and Consumer().

func RecoverCluster(conf *Config, fsm FSM, logs LogStore, stable StableStore,
	snaps SnapshotStore, trans Transport, configuration Configuration,
) error

RecoverCluster is used to manually force a new configuration in order to recover from a loss of quorum where the current configuration cannot be restored, such as when several servers die at the same time. This works by reading all the current state for this server, creating a snapshot with the supplied configuration, and then truncating the Raft log. This is the only safe way to force a given configuration without actually altering the log to insert any new entries, which could cause conflicts with other servers with different state.

WARNING! This operation implicitly commits all entries in the Raft log, so in general this is an extremely unsafe operation. If you've lost your other servers and are performing a manual recovery, then you've also lost the commit information, so this is likely the best you can do, but you should be aware that calling this can cause Raft log entries that were in the process of being replicated but not yet be committed to be committed.

Note the FSM passed here is used for the snapshot operations and will be left in a state that should not be used by the application. Be sure to discard this FSM and any associated state and provide a fresh one when calling NewRaft later.

A typical way to recover the cluster is to shut down all servers and then run RecoverCluster on every server using an identical configuration. When the cluster is then restarted, and election should occur and then Raft will resume normal operation. If it's desired to make a particular server the leader, this can be used to inject a new configuration with that server as the sole voter, and then join up other new clean-state peer servers using the usual APIs in order to bring the cluster back into a known state.

func ValidateConfig(config *Config) error

ValidateConfig is used to validate a sane configuration

AppendEntriesRequest is the command used to append entries to the replicated log.

func (r *AppendEntriesRequest) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

AppendEntriesResponse is the response returned from an AppendEntriesRequest.

func (r *AppendEntriesResponse) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

type AppendFuture interface {
	Future


	Start() time.Time


	Request() *AppendEntriesRequest


	Response() *AppendEntriesResponse
}

AppendFuture is used to return information about a pipelined AppendEntries request.

type AppendPipeline interface {


	AppendEntries(args *AppendEntriesRequest, resp *AppendEntriesResponse) (AppendFuture, error)


	Consumer() <-chan AppendFuture

	Close() error
}

AppendPipeline is used for pipelining AppendEntries requests. It is used to increase the replication throughput by masking latency and better utilizing bandwidth.

type ApplyFuture interface {
	IndexFuture


	Response() interface{}
}

ApplyFuture is used for Apply and can return the FSM response.

type BatchingFSM interface {


	ApplyBatch([]*Log) []interface{}
	FSM
}

BatchingFSM extends the FSM interface to add an ApplyBatch function. This can optionally be implemented by clients to enable multiple logs to be applied to the FSM in batches. Up to MaxAppendEntries could be sent in a batch.

Config provides any necessary configuration for the Raft server.

func DefaultConfig() *Config

DefaultConfig returns a Config with usable defaults.

type Configuration struct {
	Servers []Server
}

Configuration tracks which servers are in the cluster, and whether they have votes. This should include the local server, if it's a member of the cluster. The servers are listed no particular order, but each should only appear once. These entries are appended to the log during membership changes.

func DecodeConfiguration(buf []byte) Configuration

DecodeConfiguration deserializes a Configuration using MsgPack, or panics on errors.

func GetConfiguration(conf *Config, fsm FSM, logs LogStore, stable StableStore,
	snaps SnapshotStore, trans Transport,
) (Configuration, error)

GetConfiguration returns the persisted configuration of the Raft cluster without starting a Raft instance or connecting to the cluster. This function has identical behavior to Raft.GetConfiguration.

ReadConfigJSON reads a new-style peers.json and returns a configuration structure. This can be used to perform manual recovery when running protocol versions that use server IDs.

ReadPeersJSON consumes a legacy peers.json file in the format of the old JSON peer store and creates a new-style configuration structure. This can be used to migrate this data or perform manual recovery when running protocol versions that can interoperate with older, unversioned Raft servers. This should not be used once server IDs are in use, because the old peers.json file didn't have support for these, nor non-voter suffrage types.

func (c *Configuration) Clone() (copy Configuration)

Clone makes a deep copy of a Configuration.

type ConfigurationChangeCommand ¶ added in v1.0.0

type ConfigurationChangeCommand uint8

ConfigurationChangeCommand is the different ways to change the cluster configuration.

const (

	AddVoter ConfigurationChangeCommand = iota

	AddNonvoter

	DemoteVoter

	RemoveServer


	Promote


	AddStaging = 0
)

func (ConfigurationChangeCommand) String ¶ added in v1.0.0

type ConfigurationFuture interface {
	IndexFuture


	Configuration() Configuration
}

ConfigurationFuture is used for GetConfiguration and can return the latest configuration in use by Raft.

type ConfigurationStore interface {
	FSM


	StoreConfiguration(index uint64, configuration Configuration)
}

ConfigurationStore provides an interface that can optionally be implemented by FSMs to store configuration updates made in the replicated log. In general this is only necessary for FSMs that mutate durable state directly instead of applying changes in memory and snapshotting periodically. By storing configuration changes, the persistent FSM state can behave as a complete snapshot, and be able to recover without an external snapshot just for persisting the raft configuration.

type DiscardSnapshotSink struct{}

DiscardSnapshotSink is used to fulfill the SnapshotSink interface while always discarding the . This is useful for when the log should be truncated but no snapshot should be retained. This should never be used for production use, and is only suitable for testing.

Cancel returns successfully with a nil error

Close returns a nil error

ID returns "discard" for DiscardSnapshotSink

Write returns successfully with the length of the input byte slice to satisfy the WriteCloser interface

type DiscardSnapshotStore struct{}

DiscardSnapshotStore is used to successfully snapshot while always discarding the snapshot. This is useful for when the log should be truncated but no snapshot should be retained. This should never be used for production use, and is only suitable for testing.

func NewDiscardSnapshotStore() *DiscardSnapshotStore

NewDiscardSnapshotStore is used to create a new DiscardSnapshotStore.

Create returns a valid type implementing the SnapshotSink which always discards the snapshot.

List returns successfully with a nil for []*SnapshotMeta.

Open returns an error since the DiscardSnapshotStore does not support opening snapshots.

FSM is implemented by clients to make use of the replicated log.

type FSMSnapshot interface {


	Persist(sink SnapshotSink) error

	Release()
}

FSMSnapshot is returned by an FSM in response to a Snapshot It must be safe to invoke FSMSnapshot methods with concurrent calls to Apply.

type FailedHeartbeatObservation struct {
	PeerID      ServerID
	LastContact time.Time
}

FailedHeartbeatObservation is sent when a node fails to heartbeat with the leader

type FileSnapshotSink struct {
}

FileSnapshotSink implements SnapshotSink with a file.

Cancel is used to indicate an unsuccessful end.

Close is used to indicate a successful end.

ID returns the ID of the snapshot, can be used with Open() after the snapshot is finalized.

Write is used to append to the state file. We write to the buffered IO object to reduce the amount of context switches.

type FileSnapshotStore struct {
}

FileSnapshotStore implements the SnapshotStore interface and allows snapshots to be made on the local disk.

NOTE: This is exposed for middleware testing purposes and is not a stable API

NewFileSnapshotStore creates a new FileSnapshotStore based on a base directory. The `retain` parameter controls how many snapshots are retained. Must be at least 1.

NewFileSnapshotStoreWithLogger creates a new FileSnapshotStore based on a base directory. The `retain` parameter controls how many snapshots are retained. Must be at least 1.

Create is used to start a new snapshot

List returns available snapshots in the store.

Open takes a snapshot ID and returns a ReadCloser for that snapshot.

ReapSnapshots reaps any snapshots beyond the retain count.

type FilterFn func(o *Observation) bool

FilterFn is a function that can be registered in order to filter observations. The function reports whether the observation should be included - if it returns false, the observation will be filtered out.

type Future interface {


	Error() error
}

Future is used to represent an action that may occur in the future.

type IndexFuture interface {
	Future


	Index() uint64
}

IndexFuture is used for future actions that can result in a raft log entry being created.

type InmemSnapshotSink struct {
}

InmemSnapshotSink implements SnapshotSink in memory

Cancel returns successfully with a nil error

Close updates the Size and is otherwise a no-op

ID returns the ID of the SnapshotMeta

Write appends the given bytes to the snapshot contents

InmemSnapshotStore implements the SnapshotStore interface and retains only the most recent snapshot

func NewInmemSnapshotStore() *InmemSnapshotStore

NewInmemSnapshotStore creates a blank new InmemSnapshotStore

Create replaces the stored snapshot with a new one using the given args

List returns the latest snapshot taken

Open wraps an io.ReadCloser around the snapshot contents

type InmemStore struct {
}

InmemStore implements the LogStore and StableStore interface. It should NOT EVER be used for production. It is used only for unit tests. Use the MDBStore implementation instead.

func NewInmemStore() *InmemStore

NewInmemStore returns a new in-memory backend. Do not ever use for production. Only for testing.

DeleteRange implements the LogStore interface.

FirstIndex implements the LogStore interface.

Get implements the StableStore interface.

GetLog implements the LogStore interface.

GetUint64 implements the StableStore interface.

LastIndex implements the LogStore interface.

Set implements the StableStore interface.

SetUint64 implements the StableStore interface.

func (i *InmemStore) StoreLog(log *Log) error

StoreLog implements the LogStore interface.

func (i *InmemStore) StoreLogs(logs []*Log) error

StoreLogs implements the LogStore interface.

InmemTransport Implements the Transport interface, to allow Raft to be tested in-memory without going over a network.

AppendEntries implements the Transport interface.

func (i *InmemTransport) AppendEntriesPipeline(id ServerID, target ServerAddress) (AppendPipeline, error)

AppendEntriesPipeline returns an interface that can be used to pipeline AppendEntries requests.

Close is used to permanently disable the transport

func (i *InmemTransport) Connect(peer ServerAddress, t Transport)

Connect is used to connect this transport to another transport for a given peer name. This allows for local routing.

func (i *InmemTransport) Consumer() <-chan RPC

Consumer implements the Transport interface.

func (i *InmemTransport) DecodePeer(buf []byte) ServerAddress

DecodePeer implements the Transport interface.

func (i *InmemTransport) Disconnect(peer ServerAddress)

Disconnect is used to remove the ability to route to a given peer.

func (i *InmemTransport) DisconnectAll()

DisconnectAll is used to remove all routes to peers.

EncodePeer implements the Transport interface.

InstallSnapshot implements the Transport interface.

func (i *InmemTransport) LocalAddr() ServerAddress

LocalAddr implements the Transport interface.

RequestVote implements the Transport interface.

func (*InmemTransport) SetHeartbeatHandler ¶

func (i *InmemTransport) SetHeartbeatHandler(cb func(RPC))

SetHeartbeatHandler is used to set optional fast-path for heartbeats, not supported for this transport.

TimeoutNow implements the Transport interface.

InstallSnapshotRequest is the command sent to a Raft peer to bootstrap its log (and state machine) from a snapshot on another peer.

func (r *InstallSnapshotRequest) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

type InstallSnapshotResponse struct {
	Term    uint64
	Success bool
}

InstallSnapshotResponse is the response returned from an InstallSnapshotRequest.

func (r *InstallSnapshotResponse) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

type LeaderObservation struct {
	Leader     ServerAddress
	LeaderAddr ServerAddress
	LeaderID   ServerID
}

LeaderObservation is used for the data when leadership changes.

type LeadershipTransferFuture interface {
	Future
}

LeadershipTransferFuture is used for waiting on a user-triggered leadership transfer to complete.

Log entries are replicated to all members of the Raft cluster and form the heart of the replicated state machine.

type LogCache struct {
}

LogCache wraps any LogStore implementation to provide an in-memory ring buffer. This is used to cache access to the recently written entries. For implementations that do not cache themselves, this can provide a substantial boost by avoiding disk I/O on recent entries.

func NewLogCache(capacity int, store LogStore) (*LogCache, error)

NewLogCache is used to create a new LogCache with the given capacity and backend store.

func (c *LogCache) IsMonotonic() bool

IsMonotonic implements the MonotonicLogStore interface. This is a shim to expose the underlying store as monotonically indexed or not.

func (c *LogCache) StoreLog(log *Log) error

func (c *LogCache) StoreLogs(logs []*Log) error

LogStore is used to provide an interface for storing and retrieving logs in a durable fashion.

LogType describes various types of log entries.

const (

	LogCommand LogType = iota

	LogNoop


	LogAddPeerDeprecated


	LogRemovePeerDeprecated


	LogBarrier


	LogConfiguration
)

String returns LogType as a human readable string.

type LoopbackTransport interface {
	Transport
	WithPeers
	WithClose
	WithPreVote
}

LoopbackTransport is an interface that provides a loopback transport suitable for testing e.g. InmemTransport. It's there so we don't have to rewrite tests.

type MakeClusterOpts struct {
	Peers           int
	Bootstrap       bool
	Conf            *Config
	ConfigStoreFSM  bool
	MakeFSMFunc     func() FSM
	LongstopTimeout time.Duration
	MonotonicLogs   bool
}

NOTE: This is exposed for middleware testing purposes and is not a stable API

MockFSM is an implementation of the FSM interface, and just stores the logs sequentially.

NOTE: This is exposed for middleware testing purposes and is not a stable API

func (m *MockFSM) Apply(log *Log) interface{}

NOTE: This is exposed for middleware testing purposes and is not a stable API

func (m *MockFSM) Logs() [][]byte

NOTE: This is exposed for middleware testing purposes and is not a stable API

type MockFSMConfigStore struct {
	FSM
}

NOTE: This is exposed for middleware testing purposes and is not a stable API

func (m *MockFSMConfigStore) StoreConfiguration(index uint64, config Configuration)

NOTE: This is exposed for middleware testing purposes and is not a stable API

type MockMonotonicLogStore struct {
}

MockMonotonicLogStore is a LogStore wrapper for testing the MonotonicLogStore interface.

DeleteRange implements the LogStore interface.

FirstIndex implements the LogStore interface.

GetLog implements the LogStore interface.

IsMonotonic implements the MonotonicLogStore interface.

LastIndex implements the LogStore interface.

StoreLog implements the LogStore interface.

StoreLogs implements the LogStore interface.

type MockSnapshot struct {
}

NOTE: This is exposed for middleware testing purposes and is not a stable API

func (m *MockSnapshot) Release()

NOTE: This is exposed for middleware testing purposes and is not a stable API

type MonotonicLogStore interface {
	IsMonotonic() bool
}

MonotonicLogStore is an optional interface for LogStore implementations that cannot tolerate gaps in between the Index values of consecutive log entries. For example, this may allow more efficient indexing because the Index values are densely populated. If true is returned, Raft will avoid relying on gaps to trigger re-synching logs on followers after a snapshot is restored. The LogStore must have an efficient implementation of DeleteLogs for the case where all logs are removed, as this must be called after snapshot restore when gaps are not allowed. We avoid deleting all records for LogStores that do not implement MonotonicLogStore because although it's always correct to do so, it has a major negative performance impact on the BoltDB store that is currently the most widely used.

type NetworkTransport struct {
	TimeoutScale int

}

NetworkTransport provides a network based transport that can be used to communicate with Raft on remote machines. It requires an underlying stream layer to provide a stream abstraction, which can be simple TCP, TLS, etc.

This transport is very simple and lightweight. Each RPC request is framed by sending a byte that indicates the message type, followed by the MsgPack encoded request.

The response is an error string followed by the response object, both are encoded using MsgPack.

InstallSnapshot is special, in that after the RPC request we stream the entire state. That socket is not re-used as the connection state is not known if there is an error.

NewNetworkTransport creates a new network transport with the given dialer and listener. The maxPool controls how many connections we will pool. The timeout is used to apply I/O deadlines. For InstallSnapshot, we multiply the timeout by (SnapshotSize / TimeoutScale).

func NewNetworkTransportWithConfig(
	config *NetworkTransportConfig,
) *NetworkTransport

NewNetworkTransportWithConfig creates a new network transport with the given config struct

NewNetworkTransportWithLogger creates a new network transport with the given logger, dialer and listener. The maxPool controls how many connections we will pool. The timeout is used to apply I/O deadlines. For InstallSnapshot, we multiply the timeout by (SnapshotSize / TimeoutScale).

NewTCPTransport returns a NetworkTransport that is built on top of a TCP streaming transport layer.

NewTCPTransportWithConfig returns a NetworkTransport that is built on top of a TCP streaming transport layer, using the given config struct.

NewTCPTransportWithLogger returns a NetworkTransport that is built on top of a TCP streaming transport layer, with log output going to the supplied Logger

AppendEntries implements the Transport interface.

func (n *NetworkTransport) AppendEntriesPipeline(id ServerID, target ServerAddress) (AppendPipeline, error)

AppendEntriesPipeline returns an interface that can be used to pipeline AppendEntries requests.

Close is used to stop the network transport.

func (n *NetworkTransport) CloseStreams()

CloseStreams closes the current streams.

func (n *NetworkTransport) Consumer() <-chan RPC

Consumer implements the Transport interface.

DecodePeer implements the Transport interface.

EncodePeer implements the Transport interface.

InstallSnapshot implements the Transport interface.

func (n *NetworkTransport) IsShutdown() bool

IsShutdown is used to check if the transport is shutdown.

func (n *NetworkTransport) LocalAddr() ServerAddress

LocalAddr implements the Transport interface.

RequestPreVote implements the Transport interface.

RequestVote implements the Transport interface.

func (*NetworkTransport) SetHeartbeatHandler ¶

func (n *NetworkTransport) SetHeartbeatHandler(cb func(rpc RPC))

SetHeartbeatHandler is used to set up a heartbeat handler as a fast-pass. This is to avoid head-of-line blocking from disk IO.

TimeoutNow implements the Transport interface.

NetworkTransportConfig encapsulates configuration for the network transport layer.

type Observation struct {

	Raft *Raft


	Data interface{}
}

Observation is sent along the given channel to observers when an event occurs.

type Observer struct {
}

Observer describes what to do with a given observation.

func NewObserver(channel chan Observation, blocking bool, filter FilterFn) *Observer

NewObserver creates a new observer that can be registered to make observations on a Raft instance. Observations will be sent on the given channel if they satisfy the given filter.

If blocking is true, the observer will block when it can't send on the channel, otherwise it may discard events.

func (or *Observer) GetNumDropped() uint64

GetNumDropped returns the number of dropped observations due to blocking.

func (or *Observer) GetNumObserved() uint64

GetNumObserved returns the number of observations.

type PeerObservation struct {
	Removed bool
	Peer    Server
}

PeerObservation is sent to observers when peers change.

ProtocolVersion is the version of the protocol (which includes RPC messages as well as Raft-specific log entries) that this server can _understand_. Use the ProtocolVersion member of the Config object to control the version of the protocol to use when _speaking_ to other servers. Note that depending on the protocol version being spoken, some otherwise understood RPC messages may be refused. See dispositionRPC for details of this logic.

There are notes about the upgrade path in the description of the versions below. If you are starting a fresh cluster then there's no reason not to jump right to the latest protocol version. If you need to interoperate with older, version 0 Raft servers you'll need to drive the cluster through the different versions in order.

The version details are complicated, but here's a summary of what's required to get from a version 0 cluster to version 3:

In version N of your app that starts using the new Raft library with versioning, set ProtocolVersion to 1.
Make version N+1 of your app require version N as a prerequisite (all servers must be upgraded). For version N+1 of your app set ProtocolVersion to 2.
Similarly, make version N+2 of your app require version N+1 as a prerequisite. For version N+2 of your app, set ProtocolVersion to 3.

During this upgrade, older cluster members will still have Server IDs equal to their network addresses. To upgrade an older member and give it an ID, it needs to leave the cluster and re-enter:

Remove the server from the cluster with RemoveServer, using its network address as its ServerID.
Update the server's config to use a UUID or something else that is not tied to the machine as the ServerID (restarting the server).
Add the server back to the cluster with AddVoter, using its new ID.

You can do this during the rolling upgrade from N+1 to N+2 of your app, or as a rolling change at any time after the upgrade.

Version History ¶

0: Original Raft library before versioning was added. Servers running this

version of the Raft library use AddPeerDeprecated/RemovePeerDeprecated
for all configuration changes, and have no support for LogConfiguration.

1: First versioned protocol, used to interoperate with old servers, and begin

the migration path to newer versions of the protocol. Under this version
all configuration changes are propagated using the now-deprecated
RemovePeerDeprecated Raft log entry. This means that server IDs are always
set to be the same as the server addresses (since the old log entry type
cannot transmit an ID), and only AddPeer/RemovePeer APIs are supported.
Servers running this version of the protocol can understand the new
LogConfiguration Raft log entry but will never generate one so they can
remain compatible with version 0 Raft servers in the cluster.

2: Transitional protocol used when migrating an existing cluster to the new

server ID system. Server IDs are still set to be the same as server
addresses, but all configuration changes are propagated using the new
LogConfiguration Raft log entry type, which can carry full ID information.
This version supports the old AddPeer/RemovePeer APIs as well as the new
ID-based AddVoter/RemoveServer APIs which should be used when adding
version 3 servers to the cluster later. This version sheds all
interoperability with version 0 servers, but can interoperate with newer
Raft servers running with protocol version 1 since they can understand the
new LogConfiguration Raft log entry, and this version can still understand
their RemovePeerDeprecated Raft log entries. We need this protocol version
as an intermediate step between 1 and 3 so that servers will propagate the
ID information that will come from newly-added (or -rolled) servers using
protocol version 3, but since they are still using their address-based IDs
from the previous step they will still be able to track commitments and
their own voting status properly. If we skipped this step, servers would
be started with their new IDs, but they wouldn't see themselves in the old
address-based configuration, so none of the servers would think they had a
vote.

3: Protocol adding full support for server IDs and new ID-based server APIs

(AddVoter, AddNonvoter, etc.), old AddPeer/RemovePeer APIs are no longer
supported. Version 2 servers should be swapped out by removing them from
the cluster one-by-one and re-adding them with updated configuration for
this protocol version, along with their server ID. The remove/add cycle
is required to populate their server ID. Note that removing must be done
by ID, which will be the old server's address.

const (

	ProtocolVersionMin ProtocolVersion = 0

	ProtocolVersionMax = 3
)

type RPC struct {
	Command  interface{}
	Reader   io.Reader
	RespChan chan<- RPCResponse
}

RPC has a command, and provides a response mechanism.

func (r *RPC) Respond(resp interface{}, err error)

Respond is used to respond with a response, error or both

type RPCHeader struct {
	ProtocolVersion ProtocolVersion
	ID []byte
	Addr []byte
}

RPCHeader is a common sub-structure used to pass along protocol version and other information about the cluster. For older Raft implementations before versioning was added this will default to a zero-valued structure when read by newer Raft versions.

type RPCResponse struct {
	Response interface{}
	Error    error
}

RPCResponse captures both a response and a potential error.

type Raft struct {
}

Raft implements a Raft node.

func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps SnapshotStore, trans Transport) (*Raft, error)

NewRaft is used to construct a new Raft node. It takes a configuration, as well as implementations of various interfaces that are required. If we have any old state, such as snapshots, logs, peers, etc, all those will be restored when creating the Raft node.

AddNonvoter will add the given server to the cluster but won't assign it a vote. The server will receive log entries, but it won't participate in elections or log entry commitment. If the server is already in the cluster, this updates the server's address. This must be run on the leader or it will fail. For prevIndex and timeout, see AddVoter.

func (r *Raft) AddPeer(peer ServerAddress) Future

AddPeer to the cluster configuration. Must be run on the leader, or it will fail.

Deprecated: Use AddVoter/AddNonvoter instead.

AddVoter will add the given server to the cluster as a staging server. If the server is already in the cluster as a voter, this updates the server's address. This must be run on the leader or it will fail. The leader will promote the staging server to a voter once that server is ready. If nonzero, prevIndex is the index of the only configuration upon which this change may be applied; if another configuration entry has been added in the meantime, this request will fail. If nonzero, timeout is how long this server should wait before the configuration change log entry is appended.

func (r *Raft) AppliedIndex() uint64

AppliedIndex returns the last index applied to the FSM. This is generally lagging behind the last index, especially for indexes that are persisted but have not yet been considered committed by the leader. NOTE - this reflects the last index that was sent to the application's FSM over the apply channel but DOES NOT mean that the application's FSM has yet consumed it and applied it to its internal state. Thus, the application's state may lag behind this index.

Apply is used to apply a command to the FSM in a highly consistent manner. This returns a future that can be used to wait on the application. An optional timeout can be provided to limit the amount of time we wait for the command to be started. This must be run on the leader or it will fail.

If the node discovers it is no longer the leader while applying the command, it will return ErrLeadershipLost. There is no way to guarantee whether the write succeeded or failed in this case. For example, if the leader is partitioned it can't know if a quorum of followers wrote the log to disk. If at least one did, it may survive into the next leader's term.

If a user snapshot is restored while the command is in-flight, an ErrAbortedByRestore is returned. In this case the write effectively failed since its effects will not be present in the FSM after the restore.

ApplyLog performs Apply but takes in a Log directly. The only values currently taken from the submitted Log are Data and Extensions. See Apply for details on error cases.

Barrier is used to issue a command that blocks until all preceding operations have been applied to the FSM. It can be used to ensure the FSM reflects all queued writes. An optional timeout can be provided to limit the amount of time we wait for the command to be started. This must be run on the leader, or it will fail.

func (r *Raft) BootstrapCluster(configuration Configuration) Future

BootstrapCluster is equivalent to non-member BootstrapCluster but can be called on an un-bootstrapped Raft instance after it has been created. This should only be called at the beginning of time for the cluster with an identical configuration listing all Voter servers. There is no need to bootstrap Nonvoter and Staging servers.

A cluster can only be bootstrapped once from a single participating Voter server. Any further attempts to bootstrap will return an error that can be safely ignored.

One sane approach is to bootstrap a single server with a configuration listing just itself as a Voter, then invoke AddVoter() on it to add other servers to the cluster.

CommitIndex returns the committed index. This API maybe helpful for server to implement the read index optimization as described in the Raft paper.

CurrentTerm returns the current term.

DemoteVoter will take away a server's vote, if it has one. If present, the server will continue to receive log entries, but it won't participate in elections or log entry commitment. If the server is not in the cluster, this does nothing. This must be run on the leader or it will fail. For prevIndex and timeout, see AddVoter.

func (r *Raft) DeregisterObserver(or *Observer)

DeregisterObserver deregisters an observer.

func (r *Raft) GetConfiguration() ConfigurationFuture

GetConfiguration returns the latest configuration. This may not yet be committed. The main loop can access this directly.

LastContact returns the time of last contact by a leader. This only makes sense if we are currently a follower.

LastIndex returns the last index in stable storage, either from the last log or from the last snapshot.

func (r *Raft) Leader() ServerAddress

Leader is used to return the current leader of the cluster. Deprecated: use LeaderWithID instead It may return empty string if there is no current leader or the leader is unknown. Deprecated: use LeaderWithID instead.

func (r *Raft) LeaderCh() <-chan bool

LeaderCh is used to get a channel which delivers signals on acquiring or losing leadership. It sends true if we become the leader, and false if we lose it.

Receivers can expect to receive a notification only if leadership transition has occurred.

If receivers aren't ready for the signal, signals may drop and only the latest leadership transition. For example, if a receiver receives subsequent `true` values, they may deduce that leadership was lost and regained while the receiver was processing first leadership transition.

func (r *Raft) LeaderWithID() (ServerAddress, ServerID)

LeaderWithID is used to return the current leader address and ID of the cluster. It may return empty strings if there is no current leader or the leader is unknown.

func (r *Raft) LeadershipTransfer() Future

LeadershipTransfer will transfer leadership to a server in the cluster. This can only be called from the leader, or it will fail. The leader will stop accepting client requests, make sure the target server is up to date and starts the transfer with a TimeoutNow message. This message has the same effect as if the election timeout on the target server fires. Since it is unlikely that another server is starting an election, it is very likely that the target server is able to win the election. Note that raft protocol version 3 is not sufficient to use LeadershipTransfer. A recent version of that library has to be used that includes this feature. Using transfer leadership is safe however in a cluster where not every node has the latest version. If a follower cannot be promoted, it will fail gracefully.

func (r *Raft) LeadershipTransferToServer(id ServerID, address ServerAddress) Future

LeadershipTransferToServer does the same as LeadershipTransfer but takes a server in the arguments in case a leadership should be transitioned to a specific server in the cluster. Note that raft protocol version 3 is not sufficient to use LeadershipTransfer. A recent version of that library has to be used that includes this feature. Using transfer leadership is safe however in a cluster where not every node has the latest version. If a follower cannot be promoted, it will fail gracefully.

func (r *Raft) RegisterObserver(or *Observer)

RegisterObserver registers a new observer.

ReloadConfig updates the configuration of a running raft node. If the new configuration is invalid an error is returned and no changes made to the instance. All fields will be copied from rc into the new configuration, even if they are zero valued.

func (r *Raft) ReloadableConfig() ReloadableConfig

ReloadableConfig returns the current state of the reloadable fields in Raft's configuration. This is useful for programs to discover the current state for reporting to users or tests. It is safe to call from any goroutine. It is intended for reporting and testing purposes primarily; external synchronization would be required to safely use this in a read-modify-write pattern for reloadable configuration options.

func (r *Raft) RemovePeer(peer ServerAddress) Future

Deprecated: Use RemoveServer instead.

RemoveServer will remove the given server from the cluster. If the current leader is being removed, it will cause a new election to occur. This must be run on the leader or it will fail. For prevIndex and timeout, see AddVoter.

Restore is used to manually force Raft to consume an external snapshot, such as if restoring from a backup. We will use the current Raft configuration, not the one from the snapshot, so that we can restore into a new cluster. We will also use the max of the index of the snapshot, or the current index, and then add 1 to that, so we force a new state with a hole in the Raft log, so that the snapshot will be sent to followers and used for any new joiners. This can only be run on the leader, and blocks until the restore is complete or an error occurs.

WARNING! This operation has the leader take on the state of the snapshot and then sets itself up so that it replicates that to its followers though the install snapshot process. This involves a potentially dangerous period where the leader commits ahead of its followers, so should only be used for disaster recovery into a fresh cluster, and should not be used in normal operations.

func (r *Raft) Shutdown() Future

Shutdown is used to stop the Raft background routines. This is not a graceful operation. Provides a future that can be used to block until all background routines have exited.

func (r *Raft) Snapshot() SnapshotFuture

Snapshot is used to manually force Raft to take a snapshot. Returns a future that can be used to block until complete, and that contains a function that can be used to open the snapshot.

func (r *Raft) State() RaftState

State returns the state of this raft peer.

Stats is used to return a map of various internal stats. This should only be used for informative purposes or debugging.

Keys are: "state", "term", "last_log_index", "last_log_term", "commit_index", "applied_index", "fsm_pending", "last_snapshot_index", "last_snapshot_term", "latest_configuration", "last_contact", and "num_peers".

The value of "state" is a numeric constant representing one of the possible leadership states the node is in at any given time. the possible states are: "Follower", "Candidate", "Leader", "Shutdown".

The value of "latest_configuration" is a string which contains the id of each server, its suffrage status, and its address.

The value of "last_contact" is either "never" if there has been no contact with a leader, "0" if the node is in the leader state, or the time since last contact with a leader formatted as a string.

The value of "num_peers" is the number of other voting servers in the cluster, not including this node. If this node isn't part of the configuration then this will be "0".

All other values are uint64s, formatted as strings.

String returns a string representation of this Raft node.

func (r *Raft) VerifyLeader() Future

VerifyLeader is used to ensure this peer is still the leader. It may be used to prevent returning stale data from the FSM after the peer has lost leadership.

RaftState captures the state of a Raft node: Follower, Candidate, Leader, or Shutdown.

const (

	Follower RaftState = iota

	Candidate

	Leader

	Shutdown
)

ReadCloserWrapper allows access to an underlying ReadCloser from a wrapper.

ReloadableConfig is the subset of Config that may be reconfigured during runtime using raft.ReloadConfig. We choose to duplicate fields over embedding or accepting a Config but only using specific fields to keep the API clear. Reconfiguring some fields is potentially dangerous so we should only selectively enable it for fields where that is allowed.

RequestPreVoteRequest is the command used by a candidate to ask a Raft peer for a vote in an election.

func (r *RequestPreVoteRequest) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

type RequestPreVoteResponse struct {

	Term uint64

	Granted bool
}

RequestPreVoteResponse is the response returned from a RequestPreVoteRequest.

func (r *RequestPreVoteResponse) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

RequestVoteRequest is the command used by a candidate to ask a Raft peer for a vote in an election.

func (r *RequestVoteRequest) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

type RequestVoteResponse struct {

	Term uint64


	Peers []byte

	Granted bool
}

RequestVoteResponse is the response returned from a RequestVoteRequest.

func (r *RequestVoteResponse) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

type ResumedHeartbeatObservation struct {
	PeerID ServerID
}

ResumedHeartbeatObservation is sent when a node resumes to heartbeat with the leader following failures

type Server struct {

	Suffrage ServerSuffrage

	ID ServerID

	Address ServerAddress
}

Server tracks the information about a single server in a configuration.

ServerAddress is a network address for a server that a transport can contact.

func NewInmemAddr() ServerAddress

NewInmemAddr returns a new in-memory addr with a randomly generate UUID as the ID.

type ServerAddressProvider interface {
	ServerAddr(id ServerID) (ServerAddress, error)
}

ServerAddressProvider is a target address to which we invoke an RPC when establishing a connection

ServerID is a unique string identifying a server for all time.

ServerSuffrage determines whether a Server in a Configuration gets a vote.

const (


	Voter ServerSuffrage = iota


	Nonvoter


	Staging
)

Note: Don't renumber these, since the numbers are written into the log.

SnapshotFuture is used for waiting on a user-triggered snapshot to complete.

SnapshotMeta is for metadata of a snapshot.

SnapshotSink is returned by StartSnapshot. The FSM will Write state to the sink and call Close on completion. On error, Cancel will be invoked.

SnapshotStore interface is used to allow for flexible implementations of snapshot storage and retrieval. For example, a client could implement a shared state store such as S3, allowing new nodes to restore snapshots without streaming from the leader.

SnapshotVersion is the version of snapshots that this server can understand. Currently, it is always assumed that the server generates the latest version, though this may be changed in the future to include a configurable version.

Version History ¶

0: Original Raft library before versioning was added. The peers portion of

these snapshots is encoded in the legacy format which requires decodePeers
to parse. This version of snapshots should only be produced by the
unversioned Raft library.

1: New format which adds support for a full configuration structure and its

associated log index, with support for server IDs and non-voting server
modes. To ease upgrades, this also includes the legacy peers structure but
that will never be used by servers that understand version 1 snapshots.
Since the original Raft library didn't enforce any versioning, we must
include the legacy peers structure for this version, but we can deprecate
it in the next snapshot version.

const (

	SnapshotVersionMin SnapshotVersion = 0

	SnapshotVersionMax = 1
)

StableStore is used to provide stable storage of key configurations to ensure safety.

StreamLayer is used with the NetworkTransport to provide the low level stream abstraction.

type TCPStreamLayer struct {
}

TCPStreamLayer implements StreamLayer interface for plain TCP.

Accept implements the net.Listener interface.

Addr implements the net.Listener interface.

Close implements the net.Listener interface.

Dial implements the StreamLayer interface.

type TimeoutNowRequest struct {
}

TimeoutNowRequest is the command used by a leader to signal another server to start an election.

func (r *TimeoutNowRequest) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

type TimeoutNowResponse struct {
}

TimeoutNowResponse is the response to TimeoutNowRequest.

func (r *TimeoutNowResponse) GetRPCHeader() RPCHeader

GetRPCHeader - See WithRPCHeader.

type Transport interface {


	Consumer() <-chan RPC

	LocalAddr() ServerAddress


	AppendEntriesPipeline(id ServerID, target ServerAddress) (AppendPipeline, error)

	AppendEntries(id ServerID, target ServerAddress, args *AppendEntriesRequest, resp *AppendEntriesResponse) error

	RequestVote(id ServerID, target ServerAddress, args *RequestVoteRequest, resp *RequestVoteResponse) error


	InstallSnapshot(id ServerID, target ServerAddress, args *InstallSnapshotRequest, resp *InstallSnapshotResponse, data io.Reader) error

	EncodePeer(id ServerID, addr ServerAddress) []byte

	DecodePeer([]byte) ServerAddress


	SetHeartbeatHandler(cb func(rpc RPC))

	TimeoutNow(id ServerID, target ServerAddress, args *TimeoutNowRequest, resp *TimeoutNowResponse) error
}

Transport provides an interface for network transports to allow Raft to communicate with other nodes.

type WithClose interface {


	Close() error
}

WithClose is an interface that a transport may provide which allows a transport to be shut down cleanly when a Raft instance shuts down.

It is defined separately from Transport as unfortunately it wasn't in the original interface specification.

type WithPeers interface {
	Connect(peer ServerAddress, t Transport)
	Disconnect(peer ServerAddress)
	DisconnectAll()
}

WithPeers is an interface that a transport may provide which allows for connection and disconnection. Unless the transport is a loopback transport, the transport specified to "Connect" is likely to be nil.

type WithPreVote interface {

	RequestPreVote(id ServerID, target ServerAddress, args *RequestPreVoteRequest, resp *RequestPreVoteResponse) error
}

WithPreVote is an interface that a transport may provide which allows a transport to support a PreVote request.

It is defined separately from Transport as unfortunately it wasn't in the original interface specification.

type WithRPCHeader interface {
}

WithRPCHeader is an interface that exposes the RPC header.

type WrappingFSM interface {
	Underlying() FSM
}

NOTE: This is exposed for middleware testing purposes and is not a stable API