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 EncryptionKeyRotator
• type EncryptionKeys
• type LeadershipState
• type Node
  • func NewNode(opts NodeOptions) *Node
  • func (n *Node) CanRemoveMember(id uint64) bool
  • func (n *Node) Cancel()
  • func (n *Node) ChangesBetween(from, to api.Version) ([]state.Change, error)
  • func (n *Node) ClearData()
  • func (n *Node) Done() <-chan struct{}
  • func (n *Node) GetMemberByNodeID(nodeID string) *membership.Member
  • func (n *Node) GetMemberlist() map[uint64]*api.RaftMember
  • func (n *Node) GetNodeIDByRaftID(raftID uint64) (string, error)
  • func (n *Node) GetVersion() *api.Version
  • func (n *Node) IsIDRemoved(id uint64) bool
  • func (n *Node) IsLeader() bool
  • func (n *Node) IsMember() bool
  • func (n *Node) Join(ctx context.Context, req *api.JoinRequest) (*api.JoinResponse, error)
  • func (n *Node) JoinAndStart(ctx context.Context) (err error)
  • func (n *Node) Leader() (uint64, error)
  • func (n *Node) LeaderConn(ctx context.Context) (*grpc.ClientConn, error)
  • func (n *Node) Leave(ctx context.Context, req *api.LeaveRequest) (*api.LeaveResponse, error)
  • func (n *Node) MemoryStore() *store.MemoryStore
  • func (n *Node) NodeRemoved()
  • 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) ReadyForProposals() bool
  • func (n *Node) RemoveMember(ctx context.Context, id uint64) error
  • func (n *Node) ReportSnapshot(id uint64, status raft.SnapshotStatus)
  • func (n *Node) ReportUnreachable(id uint64)
  • func (n *Node) ResolveAddress(ctx context.Context, msg *api.ResolveAddressRequest) (*api.ResolveAddressResponse, error)
  • func (n *Node) Run(ctx context.Context) error
  • func (n *Node) SetAddr(ctx context.Context, addr string) error
  • func (n *Node) Status() raft.Status
  • func (n *Node) StreamRaftMessage(stream api.Raft_StreamRaftMessageServer) error
  • func (n *Node) SubscribeLeadership() (q chan events.Event, cancel func())
  • func (n *Node) SubscribePeers() (q chan events.Event, cancel func())
  • func (n *Node) TransferLeadership(ctx context.Context) error
  • func (n *Node) UpdateNode(id uint64, addr string)
  • func (n *Node) WithContext(ctx context.Context) (context.Context, context.CancelFunc)
• type NodeOptions

Variables

var (
	// 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")
	// 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")
	// 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")
	// ErrNoClusterLeader is thrown when the cluster has no elected leader
	ErrNoClusterLeader = errors.New("raft: no elected cluster leader")
	// ErrMemberUnknown is sent in response to a message from an
	// unrecognized peer.
	ErrMemberUnknown = errors.New("raft: member unknown")
)

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 EncryptionKeyRotator

type EncryptionKeyRotator interface {
	GetKeys() EncryptionKeys
	UpdateKeys(EncryptionKeys) error
	NeedsRotation() bool
	RotationNotify() chan struct{}
}

EncryptionKeyRotator is an interface to find out if any keys need rotating.

type EncryptionKeys

type EncryptionKeys struct {
	CurrentDEK []byte
	PendingDEK []byte
}

EncryptionKeys are the current and, if necessary, pending DEKs with which to encrypt raft data

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 Node

type Node struct {
	Config *raft.Config

	// RemovedFromRaft notifies about node deletion from raft cluster
	RemovedFromRaft chan struct{}
	// contains filtered or unexported fields
}

Node represents the Raft Node useful configuration.

func NewNode

func NewNode(opts NodeOptions) *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) Cancel

func (n *Node) Cancel()

Cancel interrupts all ongoing proposals, and prevents new ones from starting. This is useful for the shutdown sequence because it allows the manager to shut down raft-dependent services that might otherwise block on shutdown if quorum isn't met. Then the raft node can be completely shut down once no more code is using it.

func (*Node) ChangesBetween

func (n *Node) ChangesBetween(from, to api.Version) ([]state.Change, error)

ChangesBetween returns the changes starting after "from", up to and including "to". If these changes are not available because the log has been compacted, an error will be returned.

func (*Node) ClearData

func (n *Node) ClearData()

ClearData tells the raft node to delete its WALs, snapshots, and keys on shutdown.

func (*Node) Done

func (n *Node) Done() <-chan struct{}

Done returns channel which is closed when raft node is fully stopped.

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

func (n *Node) GetNodeIDByRaftID(raftID uint64) (string, error)

GetNodeIDByRaftID returns the generic Node ID of a member given its raft ID. It returns ErrMemberUnknown if the raft ID is unknown.

func (*Node) GetVersion

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

GetVersion returns the sequence information for the current raft round.

func (*Node) IsIDRemoved

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

IsIDRemoved reports if member with id was removed from cluster. Part of transport.Raft interface.

func (*Node) IsLeader

func (n *Node) IsLeader() bool

IsLeader checks if we are the leader or not, with the protection of lock

func (*Node) IsMember

func (n *Node) IsMember() bool

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

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(ctx context.Context) (err error)

JoinAndStart joins and starts the raft server

func (*Node) Leader

func (n *Node) Leader() (uint64, error)

Leader returns the id of the leader, with the protection of lock

func (*Node) LeaderConn

func (n *Node) LeaderConn(ctx context.Context) (*grpc.ClientConn, error)

LeaderConn returns current connection to cluster leader or raftselector.ErrIsLeader if current machine is leader.

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

func (n *Node) NodeRemoved()

NodeRemoved signals that node was removed from cluster and should stop. Part of transport.Raft interface.

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 underlying raft library(etcd/raft) and waits on the commit log action before returning a result

func (*Node) ReadyForProposals

func (n *Node) ReadyForProposals() bool

ReadyForProposals returns true if the node has broadcasted a message saying that it has become the leader. This means it is ready to accept proposals.

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

func (n *Node) ReportSnapshot(id uint64, status raft.SnapshotStatus)

ReportSnapshot reports snapshot status to underlying raft node. Part of transport.Raft interface.

func (*Node) ReportUnreachable

func (n *Node) ReportUnreachable(id uint64)

ReportUnreachable reports to underlying raft node that member with id is unreachable. Part of transport.Raft interface.

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

func (n *Node) SetAddr(ctx context.Context, addr string) error

SetAddr provides the raft node's address. This can be used in cases where opts.Addr was not provided to NewNode, for example when a port was not bound until after the raft node was created.

func (*Node) Status

func (n *Node) Status() raft.Status

Status returns status of underlying etcd.Node.

func (*Node) StreamRaftMessage

func (n *Node) StreamRaftMessage(stream api.Raft_StreamRaftMessageServer) error

StreamRaftMessage is the server endpoint for streaming Raft messages. It accepts a stream of raft messages to be processed on this raft member, returning a StreamRaftMessageResponse when processing of the streamed messages is complete. It is called from the Raft leader, which uses it to stream messages to this raft member. A single stream corresponds to a single raft message, which may be disassembled and streamed by the sender as individual messages. Therefore, each of the messages received by the stream will have the same raft message type and index. Currently, only messages of type raftpb.MsgSnap can be disassembled, sent and received on the stream.

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 no longer interested in events.

func (*Node) SubscribePeers

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

SubscribePeers subscribes to peer updates in cluster. It sends always full list of peers.

func (*Node) TransferLeadership

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

TransferLeadership attempts to transfer leadership to a different node, and wait for the transfer to happen.

func (*Node) UpdateNode

func (n *Node) UpdateNode(id uint64, addr string)

UpdateNode submits a configuration change to change a member's address.

func (*Node) WithContext

func (n *Node) WithContext(ctx context.Context) (context.Context, context.CancelFunc)

WithContext returns context which is cancelled when parent context cancelled or node is stopped.

type NodeOptions

type NodeOptions 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
	// ForceJoin tells us to join even if already part of a cluster.
	ForceJoin bool
	// 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.TransportCredentials
	KeyRotator     EncryptionKeyRotator
	// DisableStackDump prevents Run from dumping goroutine stacks when the
	// store becomes stuck.
	DisableStackDump bool

	// FIPS specifies whether the raft encryption should be FIPS compliant
	FIPS bool
}

NodeOptions provides node-level options.