raft

package
v0.0.0-...-b0bff92 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Dec 7, 2021 License: AGPL-3.0 Imports: 30 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	// InitialApplyTimeout is the initial timeout for applying a time. When
	// starting up a raft backend, on some machines it might take more than the
	// running apply timeout. For that reason, we allow a grace period when
	// initializing.
	InitialApplyTimeout time.Duration = time.Second * 5
	// ApplyTimeout is the timeout for applying a command in an operation. It
	// is expected that raft can commit a log with in this timeout.
	ApplyTimeout time.Duration = time.Second * 2
)
View Source 
const (

	// LoopTimeout is the max time we will wait until the raft object
	// is constructed and the main loop is started. This is to avoid
	// hard-to-debug problems where the transport hung and so this
	// worker wasn't really started even though it seemed like it
	// was. If it crashes instead the logging will give a path to the
	// problem.
	LoopTimeout = 1 * time.Minute
)

Variables

View Source 
var (
	// ErrWorkerStopped is returned by Worker.Raft if the
	// worker has been explicitly stopped.
	ErrWorkerStopped = errors.New("raft worker stopped")

	// ErrStartTimeout is returned by NewWorker if the worker loop
	// didn't start within LoopTimeout.
	ErrStartTimeout = errors.New("timed out waiting for worker loop")

	// ErrNoLeaderTimeout is returned by the worker loop if we've gone
	// too long without contact from the leader. It gives the worker a
	// chance to see any configuration changes the backstop worker
	// might have force-appended to the raft log.
	ErrNoLeaderTimeout = errors.New("timed out waiting for leader contact")
)

Functions

func Bootstrap 

func Bootstrap(config Config) error

Bootstrap bootstraps the raft cluster, using the given configuration.

This is only to be called once, at the beginning of the raft cluster's lifetime, by the bootstrap machine agent.

func IsNotLeaderError 

func IsNotLeaderError(err error) bool

IsNotLeaderError returns true if the error is the NotLeaderError.

func Manifold 

func Manifold(config ManifoldConfig) dependency.Manifold

Manifold returns a dependency.Manifold that will run a raft worker.

func NewLogStore 

func NewLogStore(dir string, syncMode SyncMode) (*raftboltdb.BoltStore, error)

NewLogStore opens a boltDB logstore in the specified directory. If the directory doesn't already exist it'll be created. If the caller passes NonSyncedAfterWrite as the value of the syncMode argument, the underlying store will NOT perform fsync calls between log writes.

func NewNotLeaderError 

func NewNotLeaderError(serverAddress, serverID string) error

NewNotLeaderError creates a new NotLeaderError with the server address and/or server ID of the current raft state leader.

func NewRaftConfig 

func NewRaftConfig(config Config) (*raft.Config, error)

NewRaftConfig makes a raft config struct from the worker config struct passed in.

func NewSnapshotStore 

func NewSnapshotStore(
	dir string,
	retain int,
	logger Logger,
) (raft.SnapshotStore, error)

NewSnapshotStore opens a file-based snapshot store in the specified directory. If the directory doesn't exist it'll be created.

func NewTarget 

func NewTarget(st *state.State, logger raftleasestore.Logger) raftlease.NotifyTarget

NewTarget creates a new lease notify target using the dependencies in a late fashion.

func NewWorker 

func NewWorker(config Config) (worker.Worker, error)

NewWorker returns a new raft worker, with the given configuration.

func NewWorkerShim 

func NewWorkerShim(config Config) (worker.Worker, error)

NewWorkerShim is suitable for use in ManifoldConfig.NewWorker, and simply calls through to NewWorker.

Types

type Applier 

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

Applier applies a new operation against a raft instance.

func (*Applier) ApplyOperation 

func (a *Applier) ApplyOperation(ops []queue.Operation, applyTimeout time.Duration)

ApplyOperation applies an lease opeartion against the raft instance. If the raft instance isn't the leader, then an error is returned with the leader information if available. This Raft spec outlines this "The first option, which we recommend ..., is for the server to reject the request and return to the client the address of the leader, if known." (see 6.2.1). If the leader is the current raft instance, then attempt to apply it to the fsm.

type ApplierMetrics 

type ApplierMetrics interface {
	// Record times how long a apply operation took, along with if it failed or
	// not. This can be used to understand if we're hitting issues with the
	// underlying raft instance.
	Record(start time.Time, result string)
	// RecordLeaderError calls out that there was a leader error, so didn't
	// follow the usual flow.
	RecordLeaderError(start time.Time)
}

ApplierMetrics defines an interface for recording the application of a log.

type BootstrapFSM 

type BootstrapFSM struct{}

BootstrapFSM is a minimal implementation of raft.FSM for use during bootstrap. Its methods should never be invoked.

func (BootstrapFSM) Apply 

func (BootstrapFSM) Apply(_ *raft.Log) interface{}

Apply is part of raft.FSM.

func (BootstrapFSM) Restore 

func (BootstrapFSM) Restore(io.ReadCloser) error

Restore is part of raft.FSM.

func (BootstrapFSM) Snapshot 

func (BootstrapFSM) Snapshot() (raft.FSMSnapshot, error)

Snapshot is part of raft.FSM.

type BootstrapLeaseApplier 

type BootstrapLeaseApplier struct{}

func (BootstrapLeaseApplier) ApplyOperation 

func (BootstrapLeaseApplier) ApplyOperation([]queue.Operation, time.Duration)

type BootstrapNotifyTarget 

type BootstrapNotifyTarget struct{}

func (BootstrapNotifyTarget) Claimed 

func (BootstrapNotifyTarget) Claimed(lease.Key, string) error

Claimed will be called when a new lease has been claimed.

func (BootstrapNotifyTarget) Expired 

func (BootstrapNotifyTarget) Expired(lease.Key) error

Expired will be called when an existing lease has expired.

type Config 

type Config struct {
	// FSM is the raft.FSM to use for this raft worker. This
	// must be non-nil for NewWorker, and nil for Bootstrap.
	FSM raft.FSM

	// Logger is the logger for this worker.
	Logger Logger

	// StorageDir is the directory in which to store raft
	// artifacts: logs, snapshots, etc. It is expected that
	// this directory is under the full control of the raft
	// worker.
	StorageDir string

	// NonSyncedWritesToRaftLog allows the operator to disable fsync calls
	// after each write to the raft log. This option trades performance for
	// data safety and should be used with caution.
	NonSyncedWritesToRaftLog bool

	// LocalID is the raft.ServerID of this worker.
	LocalID raft.ServerID

	// Transport is the raft.Transport to use for communication
	// between raft servers. This must be non-nil for NewWorker,
	// and nil for Bootstrap.
	//
	// The raft worker expects the server address to exactly
	// match the server ID, which is the stringified agent tag.
	// The transport internally maps the server address to one
	// or more network addresses, i.e. by looking up the API
	// connection information in the state database.
	Transport raft.Transport

	// Clock is used for timeouts in the worker (although not inside
	// raft).
	Clock clock.Clock

	// NoLeaderTimeout, if non-zero, will override the default
	// timeout for leader contact before restarting.
	NoLeaderTimeout time.Duration

	// ElectionTimeout, if non-zero, will override the default
	// raft election timeout.
	ElectionTimeout time.Duration

	// HeartbeatTimeout, if non-zero, will override the default
	// raft heartbeat timeout.
	HeartbeatTimeout time.Duration

	// LeaderLeaseTimeout, if non-zero, will override the default
	// raft leader lease timeout.
	LeaderLeaseTimeout time.Duration

	// SnapshotRetention is the non-negative number of snapshots
	// to retain on disk. If zero, defaults to 2.
	SnapshotRetention int

	// PrometheusRegisterer is used to register the raft metrics.
	PrometheusRegisterer prometheus.Registerer

	// Queue is a blocking queue to apply raft operations.
	Queue Queue

	// NotifyTarget is used to notify the changes from the raft operation
	// applications.
	NotifyTarget raftlease.NotifyTarget

	// NewApplier is used to apply the raft operations on to the raft
	// instance, before notifying a target of the changes.
	NewApplier func(Raft, raftlease.NotifyTarget, ApplierMetrics, clock.Clock, Logger) LeaseApplier
}

Config is the configuration required for running a raft worker.

func (Config) Validate 

func (config Config) Validate() error

Validate validates the raft worker configuration.

type LeaseApplier 

type LeaseApplier interface {
	// ApplyOperation applies a lease opeartion against the raft instance. If
	// the raft instance isn't the leader, then an error is returned with the
	// leader information if available.
	// This Raft spec outlines this "The first option, which we recommend ...,
	// is for the server to reject the request and return to the client the
	// address of the leader, if known." (see 6.2.1).
	// If the leader is the current raft instance, then attempt to apply it to
	// the fsm.
	// The time duration is the applying of a command in an operation, not for
	// the whole operation.
	ApplyOperation([]queue.Operation, time.Duration)
}

LeaseApplier applies operations from the queue onto the underlying raft instance.

func NewApplier 

func NewApplier(raft Raft, target raftlease.NotifyTarget, metrics ApplierMetrics, clock clock.Clock, logger Logger) LeaseApplier

NewApplier creates a new Applier.

type Logger 

type Logger interface {
	Criticalf(message string, args ...interface{})
	Warningf(message string, args ...interface{})
	Errorf(message string, args ...interface{})
	Infof(message string, args ...interface{})
	Debugf(message string, args ...interface{})
	Tracef(message string, args ...interface{})
	Logf(level loggo.Level, message string, args ...interface{})
	IsTraceEnabled() bool
}

Logger represents the logging methods called.

type ManifoldConfig 

type ManifoldConfig struct {
	ClockName     string
	AgentName     string
	TransportName string
	StateName     string

	FSM                  raft.FSM
	Logger               Logger
	PrometheusRegisterer prometheus.Registerer
	NewWorker            func(Config) (worker.Worker, error)
	NewTarget            func(*state.State, raftleasestore.Logger) raftlease.NotifyTarget
	NewApplier           func(Raft, raftlease.NotifyTarget, ApplierMetrics, clock.Clock, Logger) LeaseApplier

	Queue Queue
}

ManifoldConfig holds the information necessary to run a raft worker in a dependency.Engine.

func (ManifoldConfig) Validate 

func (config ManifoldConfig) Validate() error

Validate validates the manifold configuration.

type NotLeaderError 

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

NotLeaderError creates a typed error for when a raft operation is applied, but the raft state shows that it's not the leader. The error will help redirect the consumer of the error to workout where they can try and find the leader.

func (*NotLeaderError) AsMap 

func (e *NotLeaderError) AsMap() map[string]interface{}

AsMap returns a map of the error. Useful when crossing the facade boundary and wanting information in the client.

func (*NotLeaderError) Error 

func (e *NotLeaderError) Error() string

func (*NotLeaderError) ServerAddress 

func (e *NotLeaderError) ServerAddress() string

ServerAddress returns the address of the potential current leader. It's not guaranteed to be the leader, as things may of changed when attempting the same request on the new leader.

func (*NotLeaderError) ServerID 

func (e *NotLeaderError) ServerID() string

ServerID returns the server ID from the raft state. This should align with the controller machine ID of Juju.

type Queue 

type Queue interface {
	// Queue returns the queue of operations. Removing an item from the channel
	// will unblock to allow another to take its place.
	Queue() <-chan []queue.Operation
}

Queue is a blocking queue to guard access and to serialize raft applications, allowing for client side backoff.

type Raft 

type Raft interface {
	// State is used to return the current raft state.
	State() raft.RaftState

	// Leader is used to return the current leader of the cluster.
	// It may return empty string if there is no current leader
	// or the leader is unknown.
	Leader() raft.ServerAddress

	// GetConfiguration returns the latest configuration. This may not yet be
	// committed. The main loop can access this directly.
	GetConfiguration() raft.ConfigurationFuture

	// Apply is used to apply a command to the FSM in a highly consistent
	// manner. This returns a future that can be used to wait on the application.
	// An optional timeout can be provided to limit the amount of time we wait
	// for the command to be started. This must be run on the leader or it
	// will fail.
	Apply([]byte, time.Duration) raft.ApplyFuture
}

Raft defines a local use Raft instance.

type SimpleFSM 

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

SimpleFSM is an implementation of raft.FSM, which simply appends the log data to a slice.

func (*SimpleFSM) Apply 

func (fsm *SimpleFSM) Apply(log *raft.Log) interface{}

Apply is part of the raft.FSM interface.

func (*SimpleFSM) Logs 

func (fsm *SimpleFSM) Logs() [][]byte

Logs returns the accumulated log data.

func (*SimpleFSM) Restore 

func (fsm *SimpleFSM) Restore(rc io.ReadCloser) error

Restore is part of the raft.FSM interface.

func (*SimpleFSM) Snapshot 

func (fsm *SimpleFSM) Snapshot() (raft.FSMSnapshot, error)

Snapshot is part of the raft.FSM interface.

type SimpleSnapshot 

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

SimpleSnapshot is an implementation of raft.FSMSnapshot, returned by the SimpleFSM.Snapshot in this package.

func (*SimpleSnapshot) Persist 

func (snap *SimpleSnapshot) Persist(sink raft.SnapshotSink) error

Persist is part of the raft.FSMSnapshot interface.

func (*SimpleSnapshot) Release 

func (*SimpleSnapshot) Release()

Release is part of the raft.FSMSnapshot interface.

type SyncMode 

type SyncMode bool

SyncMode defines the supported sync modes when writing to the raft log store. 

const (
	// SyncAfterWrite ensures that an fsync call is performed after each write.
	SyncAfterWrite SyncMode = false

	// NoSyncAfterWrite ensures that no fsync
	// calls are performed between writes.
	NoSyncAfterWrite SyncMode = true
)

type Worker 

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

Worker is a worker that manages a raft.Raft instance.

func (*Worker) Kill 

func (w *Worker) Kill()

Kill is part of the worker.Worker interface.

func (*Worker) LogStore 

func (w *Worker) LogStore() (raft.LogStore, error)

LogStore returns the raft.LogStore managed by this worker, or an error if the worker has stopped.

func (*Worker) Raft 

func (w *Worker) Raft() (*raft.Raft, error)

Raft returns the raft.Raft managed by this worker, or an error if the worker has stopped.

func (*Worker) Report 

func (w *Worker) Report() map[string]interface{}

Report is part of the dependency.Reporter interface.

func (*Worker) Wait 

func (w *Worker) Wait() error

Wait is part of the worker.Worker interface.

Source Files

View all Source files

Directories

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.