raft

package module

Go to main page

Versions in this module

v1

v1.2.0

Sep 2, 2020

Changes in this version

type NetworkTransport

+ func (n *NetworkTransport) CloseTargetConns(target ServerAddress)

+ type TargetCloser interface

+ CloseTargetConns func(target ServerAddress)

v1.1.9

Aug 31, 2020

v1.1.7

Aug 31, 2020

v1.1.6

Aug 31, 2020

v1.1.5

Aug 29, 2020

v1.1.4

Aug 29, 2020

v1.1.3

Aug 29, 2020

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) CloseTargetStreams(target ServerAddress)

+ 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) CloseTargetStreams(target ServerAddress)

+ 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

+ CloseTargetStreams func(target ServerAddress)

+ 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