raft

Documentation

Index

• Constants
• type Command
• type ConsistentHashPayload
• type FSM
  • func NewFSM() *FSM
  • func (f *FSM) Apply(log *raft.Log) interface{}
  • func (f *FSM) GetProxyBlock(hash string) (string, bool)
  • func (f *FSM) GetRoundRobinNext(groupName string) uint32
  • func (f *FSM) GetWeightedRRState(groupName, proxyName string) (WeightedProxy, bool)
  • func (f *FSM) Restore(readCloser io.ReadCloser) error
  • func (f *FSM) Snapshot() (raft.FSMSnapshot, error)
• type FSMSnapshot
  • func (f *FSMSnapshot) Persist(sink raft.SnapshotSink) error
  • func (f *FSMSnapshot) Release()
• type HealthStatus
• type Node
  • func NewRaftNode(logger zerolog.Logger, raftConfig config.Raft) (*Node, error)
  • func (n *Node) AddPeer(ctx context.Context, peerID, peerAddr, grpcAddr string) error
  • func (n *Node) AddPeerInternal(ctx context.Context, peerID, peerAddress, grpcAddress string) error
  • func (n *Node) Apply(ctx context.Context, data []byte, timeout time.Duration) error
  • func (n *Node) GetHealthStatus() HealthStatus
  • func (n *Node) GetPeers() []raft.Server
  • func (n *Node) GetState() (raft.RaftState, raft.ServerID)
  • func (n *Node) LeaveCluster(ctx context.Context) error
  • func (n *Node) RemovePeer(ctx context.Context, peerID string) error
  • func (n *Node) RemovePeerInternal(ctx context.Context, peerID string) error
  • func (n *Node) Shutdown() error
  • func (n *Node) StartPeerSynchronizer(ctx context.Context)
• type PeerPayload
• type RoundRobinPayload
• type WeightedProxy
• type WeightedRRBatchPayload
• type WeightedRRPayload

Constants

const (
	CommandAddConsistentHashEntry = "ADD_CONSISTENT_HASH_ENTRY"
	CommandAddRoundRobinNext      = "ADD_ROUND_ROBIN_NEXT"
	CommandUpdateWeightedRR       = "UPDATE_WEIGHTED_RR"
	CommandUpdateWeightedRRBatch  = "UPDATE_WEIGHTED_RR_BATCH"
	CommandAddPeer                = "ADD_PEER"
	CommandRemovePeer             = "REMOVE_PEER"
	RaftLeaderState               = raft.Leader
	LeaderElectionTimeout         = 3 * time.Second

	ApplyTimeout = 2 * time.Second // Timeout for applying commands

)

Configuration constants for Raft operations.

Types

type Command

type Command struct {
	Type    string      `json:"type"`
	Payload interface{} `json:"payload"`
}

Command represents a general command structure for all operations.

type ConsistentHashPayload

type ConsistentHashPayload struct {
	Hash      string `json:"hash"`
	BlockName string `json:"blockName"`
}

ConsistentHashPayload represents the payload for consistent hash operations.

type FSM

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

FSM represents the Finite State Machine for the Raft cluster.

func NewFSM

func NewFSM() *FSM

NewFSM creates a new FSM instance.

func (*FSM) Apply

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

Apply implements the raft.FSM interface.

func (*FSM) GetProxyBlock

func (f *FSM) GetProxyBlock(hash string) (string, bool)

GetProxyBlock safely retrieves the block name for a given hash.

func (*FSM) GetRoundRobinNext

func (f *FSM) GetRoundRobinNext(groupName string) uint32

GetRoundRobinNext retrieves the next index for a given group name.

func (*FSM) GetWeightedRRState

func (f *FSM) GetWeightedRRState(groupName, proxyName string) (WeightedProxy, bool)

GetWeightedRRState retrieves the weight for a given group name and proxy name.

func (*FSM) Restore

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

Restore restores the FSM from a snapshot.

func (*FSM) Snapshot

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

Snapshot returns a snapshot of the FSM.

type FSMSnapshot

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

FSMSnapshot represents a snapshot of the FSM.

func (*FSMSnapshot) Persist

func (f *FSMSnapshot) Persist(sink raft.SnapshotSink) error

Persist writes the FSMSnapshot data to the given SnapshotSink.

func (*FSMSnapshot) Release

func (f *FSMSnapshot) Release()

type HealthStatus

type HealthStatus struct {
	IsHealthy   bool
	HasLeader   bool
	IsLeader    bool
	LastContact time.Duration
	Error       error
}

type Node

type Node struct {
	Fsm *FSM

	Logger zerolog.Logger
	Peers  []config.RaftPeer
	// contains filtered or unexported fields
}

Node represents a node in the Raft cluster.

func NewRaftNode

func NewRaftNode(logger zerolog.Logger, raftConfig config.Raft) (*Node, error)

NewRaftNode creates and initializes a new Raft node.

func (*Node) AddPeer

func (n *Node) AddPeer(ctx context.Context, peerID, peerAddr, grpcAddr string) error

AddPeer adds a new peer to the Raft cluster.

func (*Node) AddPeerInternal

func (n *Node) AddPeerInternal(ctx context.Context, peerID, peerAddress, grpcAddress string) error

AddPeerInternal adds a new peer to the Raft cluster.

func (*Node) Apply

func (n *Node) Apply(ctx context.Context, data []byte, timeout time.Duration) error

Apply is the public method that handles forwarding if necessary.

func (*Node) GetHealthStatus

func (n *Node) GetHealthStatus() HealthStatus

GetHealthStatus returns the health status of the Raft node.

func (*Node) GetPeers

func (n *Node) GetPeers() []raft.Server

GetPeers returns the current list of servers in the Raft configuration. If the Raft node is not initialized, it returns an empty slice.

func (*Node) GetState

func (n *Node) GetState() (raft.RaftState, raft.ServerID)

GetState returns the current Raft state and leader ID.

func (*Node) LeaveCluster

func (n *Node) LeaveCluster(ctx context.Context) error

LeaveCluster gracefully removes the node from the Raft cluster and performs cleanup.

func (*Node) RemovePeer

func (n *Node) RemovePeer(ctx context.Context, peerID string) error

RemovePeer removes a peer from the Raft cluster.

func (*Node) RemovePeerInternal

func (n *Node) RemovePeerInternal(ctx context.Context, peerID string) error

RemovePeerInternal removes a peer from the Raft cluster.

func (*Node) Shutdown

func (n *Node) Shutdown() error

Shutdown gracefully stops the Node by stopping the gRPC server, closing RPC client connections, and shutting down the underlying Raft node. It returns an error if the Raft node fails to shutdown properly, ignoring the ErrRaftShutdown error which indicates the node was already shutdown.

func (*Node) StartPeerSynchronizer

func (n *Node) StartPeerSynchronizer(ctx context.Context)

StartPeerSynchronizer starts a goroutine that synchronizes peers between Raft and FSM.

type PeerPayload

type PeerPayload struct {
	ID          string `json:"id"`
	Address     string `json:"address"`
	GRPCAddress string `json:"grpcAddress"`
}

PeerPayload represents the payload for peer operations.

type RoundRobinPayload

type RoundRobinPayload struct {
	NextIndex uint32 `json:"nextIndex"`
	GroupName string `json:"groupName"`
}

RoundRobinPayload represents the payload for round robin operations.

type WeightedProxy

type WeightedProxy struct {
	CurrentWeight   int `json:"currentWeight"`
	EffectiveWeight int `json:"effectiveWeight"`
}

WeightedProxy represents the weight structure for WeightedRoundRobin operations.

type WeightedRRBatchPayload

type WeightedRRBatchPayload struct {
	GroupName string                   `json:"groupName"`
	Updates   map[string]WeightedProxy `json:"updates"`
}

WeightedRRBatchPayload represents the payload for batch updates of WeightedRoundRobin operations.

type WeightedRRPayload

type WeightedRRPayload struct {
	GroupName string        `json:"groupName"`
	ProxyName string        `json:"proxyName"`
	Weight    WeightedProxy `json:"weight"`
}

WeightedRRPayload represents the payload for WeightedRoundRobin operations.