store

Documentation

Overview

Package store provides a distributed SQLite instance.

Distributed consensus is provided via the Raft algorithm.

Index

• Variables
• func IsNewNode(raftDir string) bool
• type BackupFormat
• type ClusterState
• type DBConfig
  • func NewDBConfig(dsn string, memory bool) *DBConfig
• type Listener
• type Server
• type Servers
  • func (s Servers) Len() int
  • func (s Servers) Less(i, j int) bool
  • func (s Servers) Swap(i, j int)
• type Store
  • func New(ln Listener, c *StoreConfig) *Store
  • func (s *Store) Addr() string
  • func (s *Store) Apply(l *raft.Log) (e interface{})
  • func (s *Store) Backup(leader bool, fmt BackupFormat, dst io.Writer) error
  • func (s *Store) Close(wait bool) error
  • func (s *Store) Database(leader bool) ([]byte, error)
  • func (s *Store) DeregisterObserver(o *raft.Observer)
  • func (s *Store) Execute(ex *command.ExecuteRequest) ([]*sql.Result, error)
  • func (s *Store) ExecuteOrAbort(ex *command.ExecuteRequest) (results []*sql.Result, retErr error)
  • func (s *Store) ID() string
  • func (s *Store) IsLeader() bool
  • func (s *Store) Join(id, addr string, voter bool) error
  • func (s *Store) LeaderAddr() (string, error)
  • func (s *Store) LeaderID() (string, error)
  • func (s *Store) Nodes() ([]*Server, error)
  • func (s *Store) Noop(id string) error
  • func (s *Store) Open(enableBootstrap bool) error
  • func (s *Store) Path() string
  • func (s *Store) Query(qr *command.QueryRequest) ([]*sql.Rows, error)
  • func (s *Store) RegisterObserver(o *raft.Observer)
  • func (s *Store) Remove(id string) error
  • func (s *Store) Restore(rc io.ReadCloser) error
  • func (s *Store) SetRequestCompression(batch, size int)
  • func (s *Store) Snapshot() (raft.FSMSnapshot, error)
  • func (s *Store) State() ClusterState
  • func (s *Store) Stats() (map[string]interface{}, error)
  • func (s *Store) WaitForApplied(timeout time.Duration) error
  • func (s *Store) WaitForAppliedIndex(idx uint64, timeout time.Duration) error
  • func (s *Store) WaitForLeader(timeout time.Duration) (string, error)
• type StoreConfig
• type Transport
  • func NewTransport(ln Listener) *Transport
  • func (t *Transport) Accept() (net.Conn, error)
  • func (t *Transport) Addr() net.Addr
  • func (t *Transport) Close() error
  • func (t *Transport) Dial(addr raft.ServerAddress, timeout time.Duration) (net.Conn, error)

Variables

var (
	// ErrNotLeader is returned when a node attempts to execute a leader-only
	// operation.
	ErrNotLeader = errors.New("not leader")

	// ErrStaleRead is returned if the executing the query would violate the
	// requested freshness.
	ErrStaleRead = errors.New("stale read")

	// ErrOpenTimeout is returned when the Store does not apply its initial
	// logs within the specified time.
	ErrOpenTimeout = errors.New("timeout waiting for initial logs application")

	// ErrInvalidBackupFormat is returned when the requested backup format
	// is not valid.
	ErrInvalidBackupFormat = errors.New("invalid backup format")
)

Functions

func IsNewNode

func IsNewNode(raftDir string) bool

IsNewNode returns whether a node using raftDir would be a brand new node. It also means that the window this node joining a different cluster has passed.

Types

type BackupFormat

type BackupFormat int

BackupFormat represents the format of database backup.

const (
	// BackupSQL is the plaintext SQL command format.
	BackupSQL BackupFormat = iota

	// BackupBinary is a SQLite file backup format.
	BackupBinary
)

type ClusterState

type ClusterState int

ClusterState defines the possible Raft states the current node can be in

const (
	Leader ClusterState = iota
	Follower
	Candidate
	Shutdown
	Unknown
)

Represents the Raft cluster states

type DBConfig

type DBConfig struct {
	DSN    string // Any custom DSN
	Memory bool   // Whether the database is in-memory only.
}

DBConfig represents the configuration of the underlying SQLite database.

func NewDBConfig

func NewDBConfig(dsn string, memory bool) *DBConfig

NewDBConfig returns a new DB config instance.

type Listener

type Listener interface {
	net.Listener
	Dial(address string, timeout time.Duration) (net.Conn, error)
}

Listener is the interface expected by the Store for Transports.

type Server

type Server struct {
	ID       string `json:"id,omitempty"`
	Addr     string `json:"addr,omitempty"`
	Suffrage string `json:"suffrage,omitempty"`
}

Server represents another node in the cluster.

type Servers

type Servers []*Server

Servers is a set of Servers.

func (Servers) Len

func (s Servers) Len() int

func (Servers) Less

func (s Servers) Less(i, j int) bool

func (Servers) Swap

func (s Servers) Swap(i, j int)

type Store

type Store struct {
	ShutdownOnRemove   bool
	SnapshotThreshold  uint64
	SnapshotInterval   time.Duration
	LeaderLeaseTimeout time.Duration
	HeartbeatTimeout   time.Duration
	ElectionTimeout    time.Duration
	ApplyTimeout       time.Duration
	RaftLogLevel       string
	// contains filtered or unexported fields
}

Store is a SQLite database, where all changes are made via Raft consensus. It should implement the Raft FSM interface: https://pkg.go.dev/github.com/hashicorp/raft#FSM

func New

func New(ln Listener, c *StoreConfig) *Store

New returns a new Store.

func (*Store) Addr

func (s *Store) Addr() string

Addr returns the address of the store.

func (*Store) Apply

func (s *Store) Apply(l *raft.Log) (e interface{})

Apply applies a Raft log entry to the database.

func (*Store) Backup

func (s *Store) Backup(leader bool, fmt BackupFormat, dst io.Writer) error

Backup writes a snapshot of the underlying database to dst

If leader is true, this operation is performed with a read consistency level equivalent to "weak". Otherwise no guarantees are made about the read consistency level.

func (*Store) Close

func (s *Store) Close(wait bool) error

Close closes the store. If wait is true, waits for a graceful shutdown.

func (*Store) Database

func (s *Store) Database(leader bool) ([]byte, error)

Database returns a copy of the underlying database. The caller MUST ensure that no transaction is taking place during this call, or an error may be returned. If leader is true, this operation is performed with a read consistency level equivalent to "weak". Otherwise no guarantees are made about the read consistency level.

http://sqlite.org/howtocorrupt.html states it is safe to do this as long as no transaction is in progress.

func (*Store) DeregisterObserver

func (s *Store) DeregisterObserver(o *raft.Observer)

DeregisterObserver deregisters an observer of Raft events

func (*Store) Execute

func (s *Store) Execute(ex *command.ExecuteRequest) ([]*sql.Result, error)

Execute executes queries that return no rows, but do modify the database.

func (*Store) ExecuteOrAbort

func (s *Store) ExecuteOrAbort(ex *command.ExecuteRequest) (results []*sql.Result, retErr error)

ExecuteOrAbort executes the requests, but aborts any active transaction on the underlying database in the case of any error.

func (*Store) ID

func (s *Store) ID() string

ID returns the Raft ID of the store.

func (*Store) IsLeader

func (s *Store) IsLeader() bool

IsLeader is used to determine if the current node is cluster leader

func (*Store) Join

func (s *Store) Join(id, addr string, voter bool) error

Join joins a node, identified by id and located at addr, to this store. The node must be ready to respond to Raft communications at that address.

func (*Store) LeaderAddr

func (s *Store) LeaderAddr() (string, error)

LeaderAddr returns the address of the current leader. Returns a blank string if there is no leader.

func (*Store) LeaderID

func (s *Store) LeaderID() (string, error)

LeaderID returns the node ID of the Raft leader. Returns a blank string if there is no leader, or an error.

func (*Store) Nodes

func (s *Store) Nodes() ([]*Server, error)

Nodes returns the slice of nodes in the cluster, sorted by ID ascending.

func (*Store) Noop

func (s *Store) Noop(id string) error

Noop writes a noop command to the Raft log. A noop command simply consumes a slot in the Raft log, but has no other affect on the system.

func (*Store) Open

func (s *Store) Open(enableBootstrap bool) error

Open opens the Store. If enableBootstrap is set, then this node becomes a standalone node. If not set, then the calling layer must know that this node has pre-existing state, or the calling layer will trigger a join operation after opening the Store.

func (*Store) Path

func (s *Store) Path() string

Path returns the path to the store's storage directory.

func (*Store) Query

func (s *Store) Query(qr *command.QueryRequest) ([]*sql.Rows, error)

Query executes queries that return rows, and do not modify the database.

func (*Store) RegisterObserver

func (s *Store) RegisterObserver(o *raft.Observer)

RegisterObserver registers an observer of Raft events

func (*Store) Remove

func (s *Store) Remove(id string) error

Remove removes a node from the store, specified by ID.

func (*Store) Restore

func (s *Store) Restore(rc io.ReadCloser) error

Restore restores the node to a previous state. The Hashicorp docs state this will not be called concurrently with Apply(), so synchronization with Execute() is not necessary.To prevent problems during queries, which may not go through the log, it blocks all query requests.

func (*Store) SetRequestCompression

func (s *Store) SetRequestCompression(batch, size int)

SetRequestCompression allows low-level control over the compression threshold for the request marshaler.

func (*Store) Snapshot

func (s *Store) Snapshot() (raft.FSMSnapshot, error)

Snapshot returns a snapshot of the database. The caller must ensure that no transaction is taking place during this call. Hashicorp Raft guarantees that this function will not be called concurrently with Apply, as it states Apply and Snapshot are always called from the same thread. This means there is no need to synchronize this function with Execute(). However queries that involve a transaction must be blocked.

http://sqlite.org/howtocorrupt.html states it is safe to do this as long as no transaction is in progress.

func (*Store) State

func (s *Store) State() ClusterState

State returns the current node's Raft state

func (*Store) Stats

func (s *Store) Stats() (map[string]interface{}, error)

Stats returns stats for the store.

func (*Store) WaitForApplied

func (s *Store) WaitForApplied(timeout time.Duration) error

WaitForApplied waits for all Raft log entries to to be applied to the underlying database.

func (*Store) WaitForAppliedIndex

func (s *Store) WaitForAppliedIndex(idx uint64, timeout time.Duration) error

WaitForAppliedIndex blocks until a given log index has been applied, or the timeout expires.

func (*Store) WaitForLeader

func (s *Store) WaitForLeader(timeout time.Duration) (string, error)

WaitForLeader blocks until a leader is detected, or the timeout expires.

type StoreConfig

type StoreConfig struct {
	DBConf *DBConfig   // The DBConfig object for this Store.
	Dir    string      // The working directory for raft.
	Tn     Transport   // The underlying Transport for raft.
	ID     string      // Node ID.
	Logger *log.Logger // The logger to use to log stuff.
}

StoreConfig represents the configuration of the underlying Store.

type Transport

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

Transport is the network service provided to Raft, and wraps a Listener.

func NewTransport

func NewTransport(ln Listener) *Transport

NewTransport returns an initialized Transport.

func (*Transport) Accept

func (t *Transport) Accept() (net.Conn, error)

Accept waits for the next connection.

func (*Transport) Addr

func (t *Transport) Addr() net.Addr

Addr returns the binding address of the transport.

func (*Transport) Close

func (t *Transport) Close() error

Close closes the transport

func (*Transport) Dial

func (t *Transport) Dial(addr raft.ServerAddress, timeout time.Duration) (net.Conn, error)

Dial creates a new network connection.