raft

package module

Go to main page

Versions in this module

v1

v1.2.4

Jan 28, 2021

v1.2.3

Jan 28, 2021

v1.2.2

Jan 28, 2021

Changes in this version

+ const SuggestedMaxDataSize

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

+ func EncodeConfiguration(configuration Configuration) []byte

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

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

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

+ type BatchingFSM interface

+ ApplyBatch func([]*Log) []interface{}

type Config

+ LogLevel string

+ NoSnapshotRestoreOnStart bool

type Configuration

+ func DecodeConfiguration(buf []byte) Configuration

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

+ type ConfigurationStore interface

+ StoreConfiguration func(index uint64, configuration Configuration)

type FileSnapshotStore

+ func FileSnapTest(t *testing.T) (string, *FileSnapshotStore)

type InmemTransport

+ func (i *InmemTransport) TimeoutNow(id ServerID, target ServerAddress, args *TimeoutNowRequest, ...) error

+ type LeadershipTransferFuture interface

type Log

+ Extensions []byte

+ type MakeClusterOpts struct

+ Bootstrap bool

+ Conf *Config

+ ConfigStoreFSM bool

+ LongstopTimeout time.Duration

+ MakeFSMFunc func() FSM

+ Peers int

+ type MockFSM struct

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

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

+ func (m *MockFSM) Restore(inp io.ReadCloser) error

+ func (m *MockFSM) Snapshot() (FSMSnapshot, error)

+ type MockFSMConfigStore struct

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

+ type MockSnapshot struct

+ func (m *MockSnapshot) Persist(sink SnapshotSink) error

+ func (m *MockSnapshot) Release()

type NetworkTransport

+ func (n *NetworkTransport) TimeoutNow(id ServerID, target ServerAddress, args *TimeoutNowRequest, ...) error

+ type PeerObservation struct

+ Peer Server

+ Removed bool

type Raft

+ func (r *Raft) ApplyLog(log Log, timeout time.Duration) ApplyFuture

+ func (r *Raft) LeadershipTransfer() Future

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

type RequestVoteRequest

+ LeadershipTransfer bool

+ type TimeoutNowRequest struct

+ func (r *TimeoutNowRequest) GetRPCHeader() RPCHeader

+ type TimeoutNowResponse struct

+ func (r *TimeoutNowResponse) GetRPCHeader() RPCHeader

type Transport

+ TimeoutNow func(id ServerID, target ServerAddress, args *TimeoutNowRequest, ...) error

+ type WrappingFSM interface

+ Underlying func() FSM

v1.0.1

Apr 12, 2019

Changes in this version

+ func NewInmemTransportWithTimeout(addr ServerAddress, timeout time.Duration) (ServerAddress, *InmemTransport)

type NetworkTransport

+ func (n *NetworkTransport) CloseStreams()

v1.0.0

Oct 3, 2017

Changes in this version

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

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

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

+ func BootstrapCluster(conf *Config, logs LogStore, stable StableStore, snaps SnapshotStore, ...) error

+ func HasExistingState(logs LogStore, stable StableStore, snaps SnapshotStore) (bool, error)

+ func RecoverCluster(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps SnapshotStore, ...) error

type AppendEntriesRequest

+ func (r *AppendEntriesRequest) GetRPCHeader() RPCHeader

type AppendEntriesResponse

+ func (r *AppendEntriesResponse) GetRPCHeader() RPCHeader

type Config

+ LocalID ServerID

+ ProtocolVersion ProtocolVersion

+ type Configuration struct

+ Servers []Server

+ func ReadConfigJSON(path string) (Configuration, error)

+ func ReadPeersJSON(path string) (Configuration, error)

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

+ type ConfigurationChangeCommand uint8

+ const AddNonvoter

+ const AddStaging

+ const DemoteVoter

+ const Promote

+ const RemoveServer

+ func (c ConfigurationChangeCommand) String() string

+ type ConfigurationFuture interface

+ Configuration func() Configuration

+ type IndexFuture interface

+ Index func() uint64

+ type InmemSnapshotSink struct

+ func (s *InmemSnapshotSink) Cancel() error

+ func (s *InmemSnapshotSink) Close() error

+ func (s *InmemSnapshotSink) ID() string

+ func (s *InmemSnapshotSink) Write(p []byte) (n int, err error)

+ type InmemSnapshotStore struct

+ func NewInmemSnapshotStore() *InmemSnapshotStore

+ func (m *InmemSnapshotStore) Create(version SnapshotVersion, index, term uint64, configuration Configuration, ...) (SnapshotSink, error)

+ func (m *InmemSnapshotStore) List() ([]*SnapshotMeta, error)

+ func (m *InmemSnapshotStore) Open(id string) (*SnapshotMeta, io.ReadCloser, error)

type InstallSnapshotRequest

+ Configuration []byte

+ ConfigurationIndex uint64

+ SnapshotVersion SnapshotVersion

+ func (r *InstallSnapshotRequest) GetRPCHeader() RPCHeader

type InstallSnapshotResponse

+ func (r *InstallSnapshotResponse) GetRPCHeader() RPCHeader

type LogType

+ const LogAddPeerDeprecated

+ const LogConfiguration

+ const LogRemovePeerDeprecated

type NetworkTransport

+ func NewNetworkTransportWithConfig(config *NetworkTransportConfig) *NetworkTransport

+ func NewTCPTransportWithConfig(bindAddr string, advertise net.Addr, config *NetworkTransportConfig) (*NetworkTransport, error)

+ type NetworkTransportConfig struct

+ Logger *log.Logger

+ MaxPool int

+ ServerAddressProvider ServerAddressProvider

+ Stream StreamLayer

+ Timeout time.Duration

+ type ProtocolVersion int

+ const ProtocolVersionMax

+ const ProtocolVersionMin

+ type RPCHeader struct

+ ProtocolVersion ProtocolVersion

type Raft

+ func (r *Raft) AddNonvoter(id ServerID, address ServerAddress, prevIndex uint64, timeout time.Duration) IndexFuture

+ func (r *Raft) AddVoter(id ServerID, address ServerAddress, prevIndex uint64, timeout time.Duration) IndexFuture

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

+ func (r *Raft) DemoteVoter(id ServerID, prevIndex uint64, timeout time.Duration) IndexFuture

+ func (r *Raft) GetConfiguration() ConfigurationFuture

+ func (r *Raft) RemoveServer(id ServerID, prevIndex uint64, timeout time.Duration) IndexFuture

+ func (r *Raft) Restore(meta *SnapshotMeta, reader io.Reader, timeout time.Duration) error

type RequestVoteRequest

+ func (r *RequestVoteRequest) GetRPCHeader() RPCHeader

type RequestVoteResponse

+ func (r *RequestVoteResponse) GetRPCHeader() RPCHeader

+ type Server struct

+ Address ServerAddress

+ ID ServerID

+ Suffrage ServerSuffrage

+ type ServerAddress string

+ type ServerAddressProvider interface

+ ServerAddr func(id ServerID) (ServerAddress, error)

+ type ServerID string

+ type ServerSuffrage int

+ const Nonvoter

+ const Staging

+ const Voter

+ func (s ServerSuffrage) String() string

+ type SnapshotFuture interface

+ Open func() (*SnapshotMeta, io.ReadCloser, error)

type SnapshotMeta

+ Configuration Configuration

+ ConfigurationIndex uint64

+ Version SnapshotVersion

+ type SnapshotVersion int

+ const SnapshotVersionMax

+ const SnapshotVersionMin

+ type WithRPCHeader interface

+ GetRPCHeader func() RPCHeader

v0

v0.1.0

Sep 29, 2017

Changes in this version

+ const DefaultTimeoutScale

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

+ var ErrKnownPeer = errors.New("peer already known")

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

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

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

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

+ var ErrNothingNewToSnapshot = errors.New("Nothing new to snapshot")

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

+ var ErrPipelineShutdown = errors.New("append pipeline closed")

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

+ var ErrTransportShutdown = errors.New("transport shutdown")

+ var ErrUnknownPeer = errors.New("peer is unknown")

+ func AddUniquePeer(peers []string, peer string) []string

+ func ExcludePeer(peers []string, peer string) []string

+ func NewInmemAddr() string

+ func PeerContained(peers []string, peer string) bool

+ func ValidateConfig(config *Config) error

+ type AppendEntriesRequest struct

+ Entries []*Log

+ Leader []byte

+ LeaderCommitIndex uint64

+ PrevLogEntry uint64

+ PrevLogTerm uint64

+ Term uint64

+ type AppendEntriesResponse struct

+ LastLog uint64

+ NoRetryBackoff bool

+ Success bool

+ Term uint64

+ type AppendFuture interface

+ Request func() *AppendEntriesRequest

+ Response func() *AppendEntriesResponse

+ Start func() time.Time

+ type AppendPipeline interface

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

+ Close func() error

+ Consumer func() <-chan AppendFuture

+ type ApplyFuture interface

+ Index func() uint64

+ Response func() interface{}

+ type Config struct

+ CommitTimeout time.Duration

+ DisableBootstrapAfterElect bool

+ ElectionTimeout time.Duration

+ EnableSingleNode bool

+ HeartbeatTimeout time.Duration

+ LeaderLeaseTimeout time.Duration

+ LogOutput io.Writer

+ Logger *log.Logger

+ MaxAppendEntries int

+ NotifyCh chan<- bool

+ ShutdownOnRemove bool

+ SnapshotInterval time.Duration

+ SnapshotThreshold uint64

+ StartAsLeader bool

+ TrailingLogs uint64

+ func DefaultConfig() *Config

+ type DiscardSnapshotSink struct

+ func (d *DiscardSnapshotSink) Cancel() error

+ func (d *DiscardSnapshotSink) Close() error

+ func (d *DiscardSnapshotSink) ID() string

+ func (d *DiscardSnapshotSink) Write(b []byte) (int, error)

+ type DiscardSnapshotStore struct

+ func NewDiscardSnapshotStore() *DiscardSnapshotStore

+ func (d *DiscardSnapshotStore) Create(index, term uint64, peers []byte) (SnapshotSink, error)

+ func (d *DiscardSnapshotStore) List() ([]*SnapshotMeta, error)

+ func (d *DiscardSnapshotStore) Open(id string) (*SnapshotMeta, io.ReadCloser, error)

+ type FSM interface

+ Apply func(*Log) interface{}

+ Restore func(io.ReadCloser) error

+ Snapshot func() (FSMSnapshot, error)

+ type FSMSnapshot interface

+ Persist func(sink SnapshotSink) error

+ Release func()

+ type FileSnapshotSink struct

+ func (s *FileSnapshotSink) Cancel() error

+ func (s *FileSnapshotSink) Close() error

+ func (s *FileSnapshotSink) ID() string

+ func (s *FileSnapshotSink) Write(b []byte) (int, error)

+ type FileSnapshotStore struct

+ func NewFileSnapshotStore(base string, retain int, logOutput io.Writer) (*FileSnapshotStore, error)

+ func NewFileSnapshotStoreWithLogger(base string, retain int, logger *log.Logger) (*FileSnapshotStore, error)

+ func (f *FileSnapshotStore) Create(index, term uint64, peers []byte) (SnapshotSink, error)

+ func (f *FileSnapshotStore) List() ([]*SnapshotMeta, error)

+ func (f *FileSnapshotStore) Open(id string) (*SnapshotMeta, io.ReadCloser, error)

+ func (f *FileSnapshotStore) ReapSnapshots() error

+ type FilterFn func(o *Observation) bool

+ type Future interface

+ Error func() error

+ type InmemStore struct

+ func NewInmemStore() *InmemStore

+ func (i *InmemStore) DeleteRange(min, max uint64) error

+ func (i *InmemStore) FirstIndex() (uint64, error)

+ func (i *InmemStore) Get(key []byte) ([]byte, error)

+ func (i *InmemStore) GetLog(index uint64, log *Log) error

+ func (i *InmemStore) GetUint64(key []byte) (uint64, error)

+ func (i *InmemStore) LastIndex() (uint64, error)

+ func (i *InmemStore) Set(key []byte, val []byte) error

+ func (i *InmemStore) SetUint64(key []byte, val uint64) error

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

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

+ type InmemTransport struct

+ func NewInmemTransport(addr string) (string, *InmemTransport)

+ func (i *InmemTransport) AppendEntries(target string, args *AppendEntriesRequest, resp *AppendEntriesResponse) error

+ func (i *InmemTransport) AppendEntriesPipeline(target string) (AppendPipeline, error)

+ func (i *InmemTransport) Close() error

+ func (i *InmemTransport) Connect(peer string, t Transport)

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

+ func (i *InmemTransport) DecodePeer(buf []byte) string

+ func (i *InmemTransport) Disconnect(peer string)

+ func (i *InmemTransport) DisconnectAll()

+ func (i *InmemTransport) EncodePeer(p string) []byte

+ func (i *InmemTransport) InstallSnapshot(target string, args *InstallSnapshotRequest, resp *InstallSnapshotResponse, ...) error

+ func (i *InmemTransport) LocalAddr() string

+ func (i *InmemTransport) RequestVote(target string, args *RequestVoteRequest, resp *RequestVoteResponse) error

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

+ type InstallSnapshotRequest struct

+ LastLogIndex uint64

+ LastLogTerm uint64

+ Leader []byte

+ Peers []byte

+ Size int64

+ Term uint64

+ type InstallSnapshotResponse struct

+ Success bool

+ Term uint64

+ type JSONPeers struct

+ func NewJSONPeers(base string, trans Transport) *JSONPeers

+ func (j *JSONPeers) Peers() ([]string, error)

+ func (j *JSONPeers) SetPeers(peers []string) error

+ type LeaderObservation struct

+ Leader string

+ type Log struct

+ Data []byte

+ Index uint64

+ Term uint64

+ Type LogType

+ type LogCache struct

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

+ func (c *LogCache) DeleteRange(min, max uint64) error

+ func (c *LogCache) FirstIndex() (uint64, error)

+ func (c *LogCache) GetLog(idx uint64, log *Log) error

+ func (c *LogCache) LastIndex() (uint64, error)

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

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

+ type LogStore interface

+ DeleteRange func(min, max uint64) error

+ FirstIndex func() (uint64, error)

+ GetLog func(index uint64, log *Log) error

+ LastIndex func() (uint64, error)

+ StoreLog func(log *Log) error

+ StoreLogs func(logs []*Log) error

+ type LogType uint8

+ const LogAddPeer

+ const LogBarrier

+ const LogCommand

+ const LogNoop

+ const LogRemovePeer

+ type LoopbackTransport interface

+ type NetworkTransport struct

+ TimeoutScale int

+ func NewNetworkTransport(stream StreamLayer, maxPool int, timeout time.Duration, logOutput io.Writer) *NetworkTransport

+ func NewNetworkTransportWithLogger(stream StreamLayer, maxPool int, timeout time.Duration, logger *log.Logger) *NetworkTransport

+ func NewTCPTransport(bindAddr string, advertise net.Addr, maxPool int, timeout time.Duration, ...) (*NetworkTransport, error)

+ func NewTCPTransportWithLogger(bindAddr string, advertise net.Addr, maxPool int, timeout time.Duration, ...) (*NetworkTransport, error)

+ func (n *NetworkTransport) AppendEntries(target string, args *AppendEntriesRequest, resp *AppendEntriesResponse) error

+ func (n *NetworkTransport) AppendEntriesPipeline(target string) (AppendPipeline, error)

+ func (n *NetworkTransport) Close() error

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

+ func (n *NetworkTransport) DecodePeer(buf []byte) string

+ func (n *NetworkTransport) EncodePeer(p string) []byte

+ func (n *NetworkTransport) InstallSnapshot(target string, args *InstallSnapshotRequest, resp *InstallSnapshotResponse, ...) error

+ func (n *NetworkTransport) IsShutdown() bool

+ func (n *NetworkTransport) LocalAddr() string

+ func (n *NetworkTransport) RequestVote(target string, args *RequestVoteRequest, resp *RequestVoteResponse) error

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

+ type Observation struct

+ Data interface{}

+ Raft *Raft

+ type Observer struct

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

+ func (or *Observer) GetNumDropped() uint64

+ func (or *Observer) GetNumObserved() uint64

+ type PeerStore interface

+ Peers func() ([]string, error)

+ SetPeers func([]string) error

+ type RPC struct

+ Command interface{}

+ Reader io.Reader

+ RespChan chan<- RPCResponse

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

+ type RPCResponse struct

+ Error error

+ Response interface{}

+ type Raft struct

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

+ func (r *Raft) AddPeer(peer string) Future

+ func (r *Raft) AppliedIndex() uint64

+ func (r *Raft) Apply(cmd []byte, timeout time.Duration) ApplyFuture

+ func (r *Raft) Barrier(timeout time.Duration) Future

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

+ func (r *Raft) LastContact() time.Time

+ func (r *Raft) LastIndex() uint64

+ func (r *Raft) Leader() string

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

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

+ func (r *Raft) RemovePeer(peer string) Future

+ func (r *Raft) SetPeers(p []string) Future

+ func (r *Raft) Shutdown() Future

+ func (r *Raft) Snapshot() Future

+ func (r *Raft) State() RaftState

+ func (r *Raft) Stats() map[string]string

+ func (r *Raft) String() string

+ func (r *Raft) VerifyLeader() Future

+ type RaftState uint32

+ const Candidate

+ const Follower

+ const Leader

+ const Shutdown

+ func (s RaftState) String() string

+ type RequestVoteRequest struct

+ Candidate []byte

+ LastLogIndex uint64

+ LastLogTerm uint64

+ Term uint64

+ type RequestVoteResponse struct

+ Granted bool

+ Peers []byte

+ Term uint64

+ type SnapshotMeta struct

+ ID string

+ Index uint64

+ Peers []byte

+ Size int64

+ Term uint64

+ type SnapshotSink interface

+ Cancel func() error

+ ID func() string

+ type SnapshotStore interface

+ Create func(index, term uint64, peers []byte) (SnapshotSink, error)

+ List func() ([]*SnapshotMeta, error)

+ Open func(id string) (*SnapshotMeta, io.ReadCloser, error)

+ type StableStore interface

+ Get func(key []byte) ([]byte, error)

+ GetUint64 func(key []byte) (uint64, error)

+ Set func(key []byte, val []byte) error

+ SetUint64 func(key []byte, val uint64) error

+ type StaticPeers struct

+ StaticPeers []string

+ func (s *StaticPeers) Peers() ([]string, error)

+ func (s *StaticPeers) SetPeers(p []string) error

+ type StreamLayer interface

+ Dial func(address string, timeout time.Duration) (net.Conn, error)

+ type TCPStreamLayer struct

+ func (t *TCPStreamLayer) Accept() (c net.Conn, err error)

+ func (t *TCPStreamLayer) Addr() net.Addr

+ func (t *TCPStreamLayer) Close() (err error)

+ func (t *TCPStreamLayer) Dial(address string, timeout time.Duration) (net.Conn, error)

+ type Transport interface

+ AppendEntries func(target string, args *AppendEntriesRequest, resp *AppendEntriesResponse) error

+ AppendEntriesPipeline func(target string) (AppendPipeline, error)

+ Consumer func() <-chan RPC

+ DecodePeer func([]byte) string

+ EncodePeer func(string) []byte

+ InstallSnapshot func(target string, args *InstallSnapshotRequest, resp *InstallSnapshotResponse, ...) error

+ LocalAddr func() string

+ RequestVote func(target string, args *RequestVoteRequest, resp *RequestVoteResponse) error

+ SetHeartbeatHandler func(cb func(rpc RPC))

+ type WithClose interface

+ Close func() error

+ type WithPeers interface

+ Connect func(peer string, t Transport)

+ Disconnect func(peer string)

+ DisconnectAll func()

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL