raft

Documentation

Index

• type Command
• type Commit
• type CommitApplier
• type ConfChangeApplier
• type Config
• type LeaderFunc
• type Membership
• type NodeConfig
• type Peer
  • func (p Peer) MarshalLogObject(e zapcore.ObjectEncoder) error
• type Peers
  • func (p Peers) MarshalLogArray(e zapcore.ArrayEncoder) error
• type RaftNode
  • func NewNode(config Config, mesh Membership, recorder Recorder, logger *zap.Logger) *RaftNode
  • func (rc *RaftNode) AddLearner(ctx context.Context, id uint64, address string) error
  • func (rc *RaftNode) AppliedIndex() uint64
  • func (rc *RaftNode) Apply(ctx context.Context, buf []byte) error
  • func (rc *RaftNode) CommittedIndex() uint64
  • func (rc *RaftNode) GetClusterMembers() (*api.GetMembersResponse, error)
  • func (rc *RaftNode) GetStatus(ctx context.Context) *api.GetStatusResponse
  • func (rc *RaftNode) GetTopology(ctx context.Context, in *api.GetTopologyRequest) (*api.GetTopologyResponse, error)
  • func (rc *RaftNode) IsLeader() bool
  • func (rc *RaftNode) IsLearner() bool
  • func (rc *RaftNode) IsRemovedFromCluster() bool
  • func (rc *RaftNode) IsVoter() bool
  • func (rc *RaftNode) Leader() uint64
  • func (rc *RaftNode) Leave(ctx context.Context) error
  • func (rc *RaftNode) Process(ctx context.Context, m raftpb.Message) error
  • func (rc *RaftNode) ProcessMessage(ctx context.Context, message *raftpb.Message) error
  • func (rc *RaftNode) PromoteMember(ctx context.Context, id uint64, address string) error
  • func (rc *RaftNode) Ready() <-chan struct{}
  • func (rc *RaftNode) RemoveMember(ctx context.Context, id uint64, force bool) error
  • func (rc *RaftNode) ReportSnapshot(id uint64, status raft.SnapshotStatus)
  • func (rc *RaftNode) ReportUnreachable(id uint64)
  • func (rc *RaftNode) Run(ctx context.Context, peers []Peer, join bool, config NodeConfig)
  • func (rc *RaftNode) Send(ctx context.Context, messages []raftpb.Message)
• type Recorder
• type SnapshotApplier
• type StableStorage
• type StatsProvider
• type StatsProviderGetter

Types

type Command

type Command struct {
	Ctx     context.Context
	Payload []byte
	ErrCh   chan error
}

type Commit

type Commit struct {
	Index   uint64
	Payload []byte
}

type CommitApplier

type CommitApplier func(context.Context, Commit) error

type ConfChangeApplier

type ConfChangeApplier func(context.Context, uint64, raftpb.ConfChangeI) error

type Config

type Config struct {
	NodeID            uint64
	NodeAddress       string
	ClusterID         string
	DataDir           string
	GetSnapshot       func() ([]byte, error)
	CommitApplier     CommitApplier
	ConfChangeApplier ConfChangeApplier
	SnapshotApplier   SnapshotApplier
}

type LeaderFunc

type LeaderFunc func(context.Context) error

LeaderFunc is a function that will be run on the Leader node. LeaderFunc must stop as soon as the given context is cancelled.

type Membership

type Membership interface {
	Call(id uint64, f func(*grpc.ClientConn) error) error
	Members() []*clusterpb.Member
}

type NodeConfig

type NodeConfig struct {
	AppliedIndex              uint64
	DisableProposalForwarding bool
	LeaderFunc                LeaderFunc
}

type Peer

type Peer struct {
	ID      uint64
	Address string
}

Peer represents a remote raft peer

func (Peer) MarshalLogObject

func (p Peer) MarshalLogObject(e zapcore.ObjectEncoder) error

type Peers

type Peers []Peer

func (Peers) MarshalLogArray

func (p Peers) MarshalLogArray(e zapcore.ArrayEncoder) error

type RaftNode

type RaftNode struct {
	// contains filtered or unexported fields
}

func NewNode

func NewNode(config Config, mesh Membership, recorder Recorder, logger *zap.Logger) *RaftNode

NewNode initiates a raft instance and returns a committed log entry channel and error channel. Proposals for log updates are sent over the provided the proposal channel. All log entries are replayed over the commit channel, followed by a nil message (to indicate the channel is current), then new log entries. To shutdown, close proposeC and read errorC.

func (*RaftNode) AddLearner

func (rc *RaftNode) AddLearner(ctx context.Context, id uint64, address string) error

func (*RaftNode) AppliedIndex

func (rc *RaftNode) AppliedIndex() uint64

func (*RaftNode) Apply

func (rc *RaftNode) Apply(ctx context.Context, buf []byte) error

func (*RaftNode) CommittedIndex

func (rc *RaftNode) CommittedIndex() uint64

func (*RaftNode) GetClusterMembers

func (rc *RaftNode) GetClusterMembers() (*api.GetMembersResponse, error)

func (*RaftNode) GetStatus

func (rc *RaftNode) GetStatus(ctx context.Context) *api.GetStatusResponse

func (*RaftNode) GetTopology

func (rc *RaftNode) GetTopology(ctx context.Context, in *api.GetTopologyRequest) (*api.GetTopologyResponse, error)

func (*RaftNode) IsLeader

func (rc *RaftNode) IsLeader() bool

func (*RaftNode) IsLearner

func (rc *RaftNode) IsLearner() bool

func (*RaftNode) IsRemovedFromCluster

func (rc *RaftNode) IsRemovedFromCluster() bool

func (*RaftNode) IsVoter

func (rc *RaftNode) IsVoter() bool

func (*RaftNode) Leader

func (rc *RaftNode) Leader() uint64

func (*RaftNode) Leave

func (rc *RaftNode) Leave(ctx context.Context) error

func (*RaftNode) Process

func (rc *RaftNode) Process(ctx context.Context, m raftpb.Message) error

func (*RaftNode) ProcessMessage

func (rc *RaftNode) ProcessMessage(ctx context.Context, message *raftpb.Message) error

func (*RaftNode) PromoteMember

func (rc *RaftNode) PromoteMember(ctx context.Context, id uint64, address string) error

func (*RaftNode) Ready

func (rc *RaftNode) Ready() <-chan struct{}

func (*RaftNode) RemoveMember

func (rc *RaftNode) RemoveMember(ctx context.Context, id uint64, force bool) error

func (*RaftNode) ReportSnapshot

func (rc *RaftNode) ReportSnapshot(id uint64, status raft.SnapshotStatus)

func (*RaftNode) ReportUnreachable

func (rc *RaftNode) ReportUnreachable(id uint64)

func (*RaftNode) Run

func (rc *RaftNode) Run(ctx context.Context, peers []Peer, join bool, config NodeConfig)

func (*RaftNode) Send

func (rc *RaftNode) Send(ctx context.Context, messages []raftpb.Message)

type Recorder

type Recorder interface {
	NotifyRaftConfChange(cluster string, cc raftpb.ConfChangeI)
}

Recorder records membership changes

type SnapshotApplier

type SnapshotApplier func(context.Context, uint64, *snap.Snapshotter) error

type StableStorage

type StableStorage interface {
	io.Closer
	ReadAll() (metadata []byte, state raftpb.HardState, ents []raftpb.Entry, err error)
	ReleaseLockTo(index uint64) error
	Save(st raftpb.HardState, ents []raftpb.Entry) error
	Sync() error
	SaveSnapshot(e walpb.Snapshot) error
}

type StatsProvider

type StatsProvider interface {
	Histogram(name string) *prometheus.Histogram
}

type StatsProviderGetter

type StatsProviderGetter func() StatsProvider