raft

Documentation

Index

• Variables
• func DefaultNodeConfig() *raft.Config
• func DefaultRaftConfig() api.RaftConfig
• func Register(server *grpc.Server, node *Node)
• func WaitForCluster(ctx context.Context, n *Node) (cluster *api.Cluster, err error)
• func WaitForLeader(ctx context.Context, n *Node) error
• type LeadershipState
• type NewNodeOptions
• type Node
  • func NewNode(ctx context.Context, opts NewNodeOptions) *Node
  • func (n *Node) CanRemoveMember(id uint64) bool
  • func (n *Node) ConnectToMember(addr string, timeout time.Duration) (*membership.Member, error)
  • func (n *Node) GetMemberByNodeID(nodeID string) *membership.Member
  • func (n *Node) GetMemberlist() map[uint64]*api.RaftMember
  • func (n *Node) GetVersion() *api.Version
  • func (n *Node) IsLeader() bool
  • func (n *Node) IsMember() bool
  • func (n *Node) IsStopped() bool
  • func (n *Node) Join(ctx context.Context, req *api.JoinRequest) (*api.JoinResponse, error)
  • func (n *Node) JoinAndStart() error
  • func (n *Node) Leader() uint64
  • func (n *Node) LeaderAddr() (string, error)
  • func (n *Node) Leave(ctx context.Context, req *api.LeaveRequest) (*api.LeaveResponse, error)
  • func (n *Node) MemoryStore() *store.MemoryStore
  • func (n *Node) ProcessRaftMessage(ctx context.Context, msg *api.ProcessRaftMessageRequest) (*api.ProcessRaftMessageResponse, error)
  • func (n *Node) ProposeValue(ctx context.Context, storeAction []*api.StoreAction, cb func()) error
  • func (n *Node) RemoveMember(ctx context.Context, id uint64) error
  • func (n *Node) ResolveAddress(ctx context.Context, msg *api.ResolveAddressRequest) (*api.ResolveAddressResponse, error)
  • func (n *Node) Run(ctx context.Context) error
  • func (n *Node) Shutdown()
  • func (n *Node) SubscribeLeadership() (q chan events.Event, cancel func())

Variables

var (
	// ErrHealthCheckFailure is returned when there is an issue with the initial handshake which means
	// that the address provided must be invalid or there is ongoing connectivity issues at join time.
	ErrHealthCheckFailure = errors.New("raft: could not connect to prospective new cluster member using its advertised address")
	// ErrNoRaftMember is thrown when the node is not yet part of a raft cluster
	ErrNoRaftMember = errors.New("raft: node is not yet part of a raft cluster")
	// ErrConfChangeRefused is returned when there is an issue with the configuration change
	ErrConfChangeRefused = errors.New("raft: propose configuration change refused")
	// ErrApplyNotSpecified is returned during the creation of a raft node when no apply method was provided
	ErrApplyNotSpecified = errors.New("raft: apply method was not specified")
	// ErrAppendEntry is thrown when the node fail to append an entry to the logs
	ErrAppendEntry = errors.New("raft: failed to append entry to logs")
	// ErrSetHardState is returned when the node fails to set the hard state
	ErrSetHardState = errors.New("raft: failed to set the hard state for log append entry")
	// ErrApplySnapshot is returned when the node fails to apply a snapshot
	ErrApplySnapshot = errors.New("raft: failed to apply snapshot on raft node")
	// ErrStopped is returned when an operation was submitted but the node was stopped in the meantime
	ErrStopped = errors.New("raft: failed to process the request: node is stopped")
	// ErrLostLeadership is returned when an operation was submitted but the node lost leader status before it became committed
	ErrLostLeadership = errors.New("raft: failed to process the request: node lost leader status")
	// ErrRequestTooLarge is returned when a raft internal message is too large to be sent
	ErrRequestTooLarge = errors.New("raft: raft message is too large and can't be sent")
	// ErrCannotRemoveMember is thrown when we try to remove a member from the cluster but this would result in a loss of quorum
	ErrCannotRemoveMember = errors.New("raft: member cannot be removed, because removing it may result in loss of quorum")
	// ErrMemberRemoved is thrown when a node was removed from the cluster
	ErrMemberRemoved = errors.New("raft: member was removed from the cluster")
	// ErrNoClusterLeader is thrown when the cluster has no elected leader
	ErrNoClusterLeader = errors.New("raft: no elected cluster leader")
)

Functions

func DefaultNodeConfig

func DefaultNodeConfig() *raft.Config

DefaultNodeConfig returns the default config for a raft node that can be modified and customized

func DefaultRaftConfig

func DefaultRaftConfig() api.RaftConfig

DefaultRaftConfig returns a default api.RaftConfig.

func Register

func Register(server *grpc.Server, node *Node)

Register registers the node raft server

func WaitForCluster

func WaitForCluster(ctx context.Context, n *Node) (cluster *api.Cluster, err error)

WaitForCluster waits until node observes that the cluster wide config is committed to raft. This ensures that we can see and serve informations related to the cluster.

func WaitForLeader

func WaitForLeader(ctx context.Context, n *Node) error

WaitForLeader waits until node observe some leader in cluster. It returns error if ctx was cancelled before leader appeared.

Types

type LeadershipState

type LeadershipState int

LeadershipState indicates whether the node is a leader or follower.

const (
	// IsLeader indicates that the node is a raft leader.
	IsLeader LeadershipState = iota
	// IsFollower indicates that the node is a raft follower.
	IsFollower
)

type NewNodeOptions

type NewNodeOptions struct {
	// ID is the node's ID, from its certificate's CN field.
	ID string
	// Addr is the address of this node's listener
	Addr string
	// ForceNewCluster defines if we have to force a new cluster
	// because we are recovering from a backup data directory.
	ForceNewCluster bool
	// JoinAddr is the cluster to join. May be an empty string to create
	// a standalone cluster.
	JoinAddr string
	// Config is the raft config.
	Config *raft.Config
	// StateDir is the directory to store durable state.
	StateDir string
	// TickInterval interval is the time interval between raft ticks.
	TickInterval time.Duration
	// ClockSource is a Clock interface to use as a time base.
	// Leave this nil except for tests that are designed not to run in real
	// time.
	ClockSource clock.Clock
	// SendTimeout is the timeout on the sending messages to other raft
	// nodes. Leave this as 0 to get the default value.
	SendTimeout    time.Duration
	TLSCredentials credentials.TransportAuthenticator
}

NewNodeOptions provides arguments for NewNode

type Node

type Node struct {
	raft.Node

	Server *grpc.Server
	Ctx    context.Context

	Address  string
	StateDir string
	Error    error

	Config *raft.Config
	// contains filtered or unexported fields
}

Node represents the Raft Node useful configuration.

func NewNode

func NewNode(ctx context.Context, opts NewNodeOptions) *Node

NewNode generates a new Raft node

func (*Node) CanRemoveMember

func (n *Node) CanRemoveMember(id uint64) bool

CanRemoveMember checks if a member can be removed from the context of the current node.

func (*Node) ConnectToMember

func (n *Node) ConnectToMember(addr string, timeout time.Duration) (*membership.Member, error)

ConnectToMember returns a member object with an initialized connection to communicate with other raft members

func (*Node) GetMemberByNodeID

func (n *Node) GetMemberByNodeID(nodeID string) *membership.Member

GetMemberByNodeID returns member information based on its generic Node ID.

func (*Node) GetMemberlist

func (n *Node) GetMemberlist() map[uint64]*api.RaftMember

GetMemberlist returns the current list of raft members in the cluster.

func (*Node) GetVersion

func (n *Node) GetVersion() *api.Version

GetVersion returns the sequence information for the current raft round.

func (*Node) IsLeader

func (n *Node) IsLeader() bool

IsLeader checks if we are the leader or not

func (*Node) IsMember

func (n *Node) IsMember() bool

IsMember checks if the raft node has effectively joined a cluster of existing members.

func (*Node) IsStopped

func (n *Node) IsStopped() bool

IsStopped checks if the raft node is stopped or not

func (*Node) Join

func (n *Node) Join(ctx context.Context, req *api.JoinRequest) (*api.JoinResponse, error)

Join asks to a member of the raft to propose a configuration change and add us as a member thus beginning the log replication process. This method is called from an aspiring member to an existing member

func (*Node) JoinAndStart

func (n *Node) JoinAndStart() error

JoinAndStart joins and starts the raft server

func (*Node) Leader

func (n *Node) Leader() uint64

Leader returns the id of the leader

func (*Node) LeaderAddr

func (n *Node) LeaderAddr() (string, error)

LeaderAddr returns address of current cluster leader. With this method Node satisfies raftpicker.AddrSelector interface.

func (*Node) Leave

func (n *Node) Leave(ctx context.Context, req *api.LeaveRequest) (*api.LeaveResponse, error)

Leave asks to a member of the raft to remove us from the raft cluster. This method is called from a member who is willing to leave its raft membership to an active member of the raft

func (*Node) MemoryStore

func (n *Node) MemoryStore() *store.MemoryStore

MemoryStore returns the memory store that is kept in sync with the raft log.

func (*Node) ProcessRaftMessage

func (n *Node) ProcessRaftMessage(ctx context.Context, msg *api.ProcessRaftMessageRequest) (*api.ProcessRaftMessageResponse, error)

ProcessRaftMessage calls 'Step' which advances the raft state machine with the provided message on the receiving node

func (*Node) ProposeValue

func (n *Node) ProposeValue(ctx context.Context, storeAction []*api.StoreAction, cb func()) error

ProposeValue calls Propose on the raft and waits on the commit log action before returning a result

func (*Node) RemoveMember

func (n *Node) RemoveMember(ctx context.Context, id uint64) error

RemoveMember submits a configuration change to remove a member from the raft cluster after checking if the operation would not result in a loss of quorum.

func (*Node) ResolveAddress

func (n *Node) ResolveAddress(ctx context.Context, msg *api.ResolveAddressRequest) (*api.ResolveAddressResponse, error)

ResolveAddress returns the address reaching for a given node ID.

func (*Node) Run

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

Run is the main loop for a Raft node, it goes along the state machine, acting on the messages received from other Raft nodes in the cluster.

Before running the main loop, it first starts the raft node based on saved cluster state. If no saved state exists, it starts a single-node cluster.

func (*Node) Shutdown

func (n *Node) Shutdown()

Shutdown stops the raft node processing loop. Calling Shutdown on an already stopped node will result in a panic.

func (*Node) SubscribeLeadership

func (n *Node) SubscribeLeadership() (q chan events.Event, cancel func())

SubscribeLeadership returns channel to which events about leadership change will be sent in form of raft.LeadershipState. Also cancel func is returned - it should be called when listener is not longer interested in events.