raft

Documentation

Index

• Variables
• type Config
• type DynamicLeaderElection
  • func NewDynamicLeaderElection(logger zerolog.Logger, leaderFactory func() (Runnable, error), ...) *DynamicLeaderElection
  • func (d *DynamicLeaderElection) IsRunning() bool
  • func (d *DynamicLeaderElection) Run(ctx context.Context) error
• type FSM
  • func (f *FSM) Apply(log *raft.Log) interface{}
  • func (f *FSM) Restore(rc io.ReadCloser) error
  • func (f *FSM) Snapshot() (raft.FSMSnapshot, error)
• type Node
  • func NewNode(cfg *Config, logger zerolog.Logger) (*Node, error)
  • func (n *Node) AddPeer(nodeID, addr string) error
  • func (n *Node) Broadcast(ctx context.Context, state *RaftBlockState) error
  • func (n *Node) Config() Config
  • func (n *Node) GetState() *RaftBlockState
  • func (n *Node) HasQuorum() bool
  • func (n *Node) IsLeader() bool
  • func (n *Node) NodeID() string
  • func (n *Node) RemovePeer(nodeID string) error
  • func (n *Node) SetApplyCallback(ch chan<- RaftApplyMsg)
  • func (n *Node) Shutdown() error
  • func (n *Node) Start(_ context.Context) error
  • func (n *Node) Stop() error
• type RaftApplyMsg
• type RaftBlockState
• type Runnable

Variables

var ErrLeadershipLost = fmt.Errorf("leader lock lost")

Types

type Config

type Config struct {
	NodeID             string
	RaftAddr           string
	RaftDir            string
	Bootstrap          bool
	Peers              []string
	SnapCount          uint64
	SendTimeout        time.Duration
	HeartbeatTimeout   time.Duration
	LeaderLeaseTimeout time.Duration
}

Config holds raft node configuration

type DynamicLeaderElection

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

func NewDynamicLeaderElection

func NewDynamicLeaderElection(
	logger zerolog.Logger,
	leaderFactory func() (Runnable, error),
	followerFactory func() (Runnable, error),
	node *Node,
) *DynamicLeaderElection

NewDynamicLeaderElection constructor

func (*DynamicLeaderElection) IsRunning

func (d *DynamicLeaderElection) IsRunning() bool

func (*DynamicLeaderElection) Run

func (d *DynamicLeaderElection) Run(ctx context.Context) error

Run starts the leader election process and manages the lifecycle of leader or follower roles based on Raft events. This is a blocking call.

type FSM

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

FSM implements raft.FSM for block state

func (*FSM) Apply

func (f *FSM) Apply(log *raft.Log) interface{}

Apply implements raft.FSM

func (*FSM) Restore

func (f *FSM) Restore(rc io.ReadCloser) error

Restore implements raft.FSM

func (*FSM) Snapshot

func (f *FSM) Snapshot() (raft.FSMSnapshot, error)

Snapshot implements raft.FSM

type Node

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

Node represents a raft consensus node

func NewNode

func NewNode(cfg *Config, logger zerolog.Logger) (*Node, error)

NewNode creates a new raft node

func (*Node) AddPeer

func (n *Node) AddPeer(nodeID, addr string) error

AddPeer adds a peer to the raft cluster

func (*Node) Broadcast

func (n *Node) Broadcast(ctx context.Context, state *RaftBlockState) error

Broadcast proposes a block state to be replicated via raft

func (*Node) Config

func (n *Node) Config() Config

func (*Node) GetState

func (n *Node) GetState() *RaftBlockState

GetState returns the current replicated state

func (*Node) HasQuorum

func (n *Node) HasQuorum() bool

HasQuorum checks if the leader can still contact a quorum of peers. This should be called before producing a block to ensure consensus is possible. Returns false if the leader cannot verify its leadership with the cluster.

func (*Node) IsLeader

func (n *Node) IsLeader() bool

IsLeader returns true if this node is the raft leader

func (*Node) NodeID

func (n *Node) NodeID() string

func (*Node) RemovePeer

func (n *Node) RemovePeer(nodeID string) error

RemovePeer removes a peer from the raft cluster

func (*Node) SetApplyCallback

func (n *Node) SetApplyCallback(ch chan<- RaftApplyMsg)

SetApplyCallback sets a callback channel to receive notifications when a new block state is replicated. The channel must have sufficient buffer space since updates are published only once without blocking. If the channel is full, state updates will be skipped to prevent blocking the raft cluster.

func (*Node) Shutdown

func (n *Node) Shutdown() error

Shutdown stops the raft node

func (*Node) Start

func (n *Node) Start(_ context.Context) error

func (*Node) Stop

func (n *Node) Stop() error

type RaftApplyMsg

type RaftApplyMsg struct {
	Index uint64
	State *RaftBlockState
}

RaftApplyMsg is sent when raft applies a log entry

type RaftBlockState

type RaftBlockState = pb.RaftBlockState

RaftBlockState represents a replicated block state

type Runnable

type Runnable interface {
	// Run runs the main logic of the component using the provided context and returns an error if it fails.
	Run(ctx context.Context) error
	// IsSynced checks whether the component is synced with the given RaftBlockState.
	// -1 means raft is ahead by 1, 0 equal and a positive number the blocks that the local state is ahead of the raft state.
	IsSynced(*RaftBlockState) (int, error)
	Recover(ctx context.Context, state *RaftBlockState) error
}

Runnable represents a component that can be started and performs specific operations while running.