raftstorage

Documentation

Overview

Package raftstorage implements a Raft-backed storage provider.

Index

• Constants
• Variables
• type Consensus
  • func (r *Consensus) AddObserver(ctx context.Context, peer *v1.StoragePeer) error
  • func (r *Consensus) AddVoter(ctx context.Context, peer *v1.StoragePeer) error
  • func (r *Consensus) DemoteVoter(ctx context.Context, peer *v1.StoragePeer) error
  • func (r *Consensus) GetLeader(ctx context.Context) (*v1.StoragePeer, error)
  • func (r *Consensus) IsLeader() bool
  • func (r *Consensus) IsMember() bool
  • func (r *Consensus) RemovePeer(ctx context.Context, peer *v1.StoragePeer, wait bool) error
• type LeaderObservation
• type MonotonicLogStore
  • func (m *MonotonicLogStore) IsMonotonic() bool
• type Observation
• type ObservationCallback
• type Options
  • func NewOptions(nodeID string) Options
  • func (o *Options) RaftConfig(ctx context.Context, nodeID string) *raft.Config
• type PeerObservation
• type Provider
  • func NewProvider(opts Options) *Provider
  • func (r *Provider) ApplyRaftLog(ctx context.Context, log *v1.RaftLogEntry) (*v1.RaftApplyResponse, error)
  • func (r *Provider) Bootstrap(ctx context.Context) error
  • func (r *Provider) Close() error
  • func (r *Provider) Consensus() storage.Consensus
  • func (r *Provider) GetRaftConfiguration() (raft.Configuration, error)
  • func (r *Provider) ListenPort() uint16
  • func (r *Provider) MeshStorage() storage.MeshStorage
  • func (r *Provider) OnObservation(cb ObservationCallback)
  • func (r *Provider) Start(ctx context.Context) error
  • func (r *Provider) Status() *v1.StorageStatus
• type RaftStorage
  • func (rs *RaftStorage) Close() error
  • func (rs *RaftStorage) Delete(ctx context.Context, key string) error
  • func (rs *RaftStorage) GetValue(ctx context.Context, key string) (string, error)
  • func (rs *RaftStorage) IterPrefix(ctx context.Context, prefix string, fn storage.PrefixIterator) error
  • func (rs *RaftStorage) List(ctx context.Context, prefix string) ([]string, error)
  • func (rs *RaftStorage) PutValue(ctx context.Context, key, value string, ttl time.Duration) error
  • func (rs *RaftStorage) Subscribe(ctx context.Context, key string, fn storage.SubscribeFunc) (context.CancelFunc, error)
• type SnapshotMeta

Constants

const (
	// DefaultListenPort is the default raft listen port
	DefaultListenPort = 9000
	// DefaultListenAddress is the default raft listen address
	DefaultListenAddress = "[::]:9000"
)

const (
	Follower  = raft.Follower
	Candidate = raft.Candidate
	Leader    = raft.Leader
	Shutdown  = raft.Shutdown
)

Raft states.

const (
	Voter    = raft.Voter
	Nonvoter = raft.Nonvoter
)

Raft suffrage states.

const BarrierThreshold = 10

BarrierThreshold is the threshold for sending a barrier after a write operation. TODO: make this configurable.

Variables

var DefaultDataDir = func() string {
	if runtime.GOOS == "windows" {
		return "C:\\ProgramData\\webmesh\\store"
	}
	return "/var/lib/webmesh/store"
}()

DefaultDataDir is the default data directory.

Types

type Consensus

type Consensus struct {
	*Provider
}

RaftConsensus is the Raft consensus implementation.

func (*Consensus) AddObserver

func (r *Consensus) AddObserver(ctx context.Context, peer *v1.StoragePeer) error

AddObserver adds an observer to the consensus group.

func (*Consensus) AddVoter

func (r *Consensus) AddVoter(ctx context.Context, peer *v1.StoragePeer) error

AddVoter adds a voter to the consensus group.

func (*Consensus) DemoteVoter

func (r *Consensus) DemoteVoter(ctx context.Context, peer *v1.StoragePeer) error

DemoteVoter demotes a voter to an observer.

func (*Consensus) GetLeader

func (r *Consensus) GetLeader(ctx context.Context) (*v1.StoragePeer, error)

GetLeader returns the leader of the cluster.

func (*Consensus) IsLeader

func (r *Consensus) IsLeader() bool

IsLeader returns true if the Raft node is the leader.

func (*Consensus) IsMember

func (r *Consensus) IsMember() bool

IsMember returns true if the Raft node is a member of the cluster.

func (*Consensus) RemovePeer

func (r *Consensus) RemovePeer(ctx context.Context, peer *v1.StoragePeer, wait bool) error

RemovePeer removes a peer from the consensus group.

type LeaderObservation

type LeaderObservation = raft.LeaderObservation

LeaderObservation is an alias for raft.LeaderObservation.

type MonotonicLogStore

type MonotonicLogStore struct {
	raft.LogStore
}

MonotonicLogStore is a LogStore that is monotonic.

func (*MonotonicLogStore) IsMonotonic

func (m *MonotonicLogStore) IsMonotonic() bool

IsMonotonic returns true if the log store is monotonic.

type Observation

type Observation = raft.Observation

Observation is an alias for raft.Observation.

type ObservationCallback

type ObservationCallback func(ctx context.Context, obs Observation)

ObservationCallback is a callback that can be registered for when an observation is received.

type Options

type Options struct {
	// NodeID is the node ID.
	NodeID string
	// Transport is the Raft transport to use for communicating with
	// other Raft nodes.
	Transport transport.RaftTransport
	// DataDir is the directory to store data in.
	DataDir string
	// ClearDataDir is if the data directory should be cleared on startup.
	ClearDataDir bool
	// InMemory is if the store should be in memory. This should only be used for testing and ephemeral nodes.
	InMemory bool
	// ConnectionPoolCount is the number of connections to pool. If 0, no connection pooling is used.
	ConnectionPoolCount int
	// ConnectionTimeout is the timeout for connections.
	ConnectionTimeout time.Duration
	// HeartbeatTimeout is the timeout for heartbeats.
	HeartbeatTimeout time.Duration
	// ElectionTimeout is the timeout for elections.
	ElectionTimeout time.Duration
	// ApplyTimeout is the timeout for applying.
	ApplyTimeout time.Duration
	// CommitTimeout is the timeout for committing.
	CommitTimeout time.Duration
	// MaxAppendEntries is the maximum number of append entries.
	MaxAppendEntries int
	// LeaderLeaseTimeout is the timeout for leader leases.
	LeaderLeaseTimeout time.Duration
	// SnapshotInterval is the interval to take snapshots.
	SnapshotInterval time.Duration
	// SnapshotThreshold is the threshold to take snapshots.
	SnapshotThreshold uint64
	// SnapshotRetention is the number of snapshots to retain.
	SnapshotRetention uint64
	// ObserverChanBuffer is the buffer size for the observer channel.
	ObserverChanBuffer int
	// LogLevel is the log level for the raft backend.
	LogLevel string
}

Options are the raft options.

func NewOptions

func NewOptions(nodeID string) Options

NewOptions returns new raft options with sensible defaults.

func (*Options) RaftConfig

func (o *Options) RaftConfig(ctx context.Context, nodeID string) *raft.Config

RaftConfig builds a raft config.

type PeerObservation

type PeerObservation = raft.PeerObservation

PeerObservation is an alias for raft.PeerObservation.

type Provider

type Provider struct {
	Options
	// contains filtered or unexported fields
}

RaftStorage is a storage provider that uses Raft for consensus. BadgerDB is used for the underlying storage.

func NewProvider

func NewProvider(opts Options) *Provider

NewProvider returns a new RaftStorageProvider.

func (*Provider) ApplyRaftLog

func (r *Provider) ApplyRaftLog(ctx context.Context, log *v1.RaftLogEntry) (*v1.RaftApplyResponse, error)

ApplyRaftLog applies a raft log entry.

func (*Provider) Bootstrap

func (r *Provider) Bootstrap(ctx context.Context) error

Bootstrap bootstraps the raft storage provider.

func (*Provider) Close

func (r *Provider) Close() error

Close closes the mesh storage and shuts down the raft instance.

func (*Provider) Consensus

func (r *Provider) Consensus() storage.Consensus

Consensus returns the underlying Consensus instance.

func (*Provider) GetRaftConfiguration

func (r *Provider) GetRaftConfiguration() (raft.Configuration, error)

GetRaftConfiguration returns the current raft configuration.

func (*Provider) ListenPort

func (r *Provider) ListenPort() uint16

ListenPort returns the TCP port that the storage provider is listening on.

func (*Provider) MeshStorage

func (r *Provider) MeshStorage() storage.MeshStorage

MeshStorage returns the underlying MeshStorage instance.

func (*Provider) OnObservation

func (r *Provider) OnObservation(cb ObservationCallback)

OnObservation registers a callback for when an observation is received.

func (*Provider) Start

func (r *Provider) Start(ctx context.Context) error

Start starts the raft storage provider.

func (*Provider) Status

func (r *Provider) Status() *v1.StorageStatus

Status returns the status of the storage provider.

type RaftStorage

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

RaftStorage wraps the storage.Storage interface to force write operations through the Raft log.

func (*RaftStorage) Close

func (rs *RaftStorage) Close() error

Close closes the storage.

func (*RaftStorage) Delete

func (rs *RaftStorage) Delete(ctx context.Context, key string) error

Delete removes a key.

func (*RaftStorage) GetValue

func (rs *RaftStorage) GetValue(ctx context.Context, key string) (string, error)

GetValue gets the value of a key.

func (*RaftStorage) IterPrefix

func (rs *RaftStorage) IterPrefix(ctx context.Context, prefix string, fn storage.PrefixIterator) error

IterPrefix iterates over all keys with a given prefix.

func (*RaftStorage) List

func (rs *RaftStorage) List(ctx context.Context, prefix string) ([]string, error)

List returns a list of keys.

func (*RaftStorage) PutValue

func (rs *RaftStorage) PutValue(ctx context.Context, key, value string, ttl time.Duration) error

Put sets the value of a key.

func (*RaftStorage) Subscribe

func (rs *RaftStorage) Subscribe(ctx context.Context, key string, fn storage.SubscribeFunc) (context.CancelFunc, error)

Subscribe subscribes to changes to a key.

type SnapshotMeta

type SnapshotMeta = raft.SnapshotMeta

SnapshotMeta is an alias for raft.SnapshotMeta.