Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

raft

package

Go to main page

Versions in this module

v1

v1.3.1

Jan 25, 2023

v1.3.0

Sep 27, 2022

v1.2.1

Oct 13, 2021

v1.2.0

Mar 18, 2021

v1.1.0

Aug 20, 2020

v1.0.5

Jul 24, 2020

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