store

Documentation

Overview

Package store provides a distributed SQLite instance.

Distributed consensus is provided via the Raft algorithm.

Index

• Variables
• func JoinAllowed(raftDir string) (bool, error)
• func NumPeers(raftDir string) (int, error)
• type BackupFormat
• type ClusterState
• type ConsistencyLevel
• type DBConfig
  • func NewDBConfig(dsn string, memory bool) *DBConfig
• type ExecuteRequest
• type QueryRequest
• type Store
  • func New(c *StoreConfig) *Store
  • func (s *Store) APIPeers() (map[string]string, error)
  • func (s *Store) Addr() net.Addr
  • func (s *Store) Apply(l *raft.Log) interface{}
  • func (s *Store) Backup(leader bool, fmt BackupFormat) ([]byte, 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 *ExecuteRequest) ([]*sql.Result, error)
  • func (s *Store) ExecuteOrAbort(ex *ExecuteRequest) (results []*sql.Result, retErr error)
  • func (s *Store) IsLeader() bool
  • func (s *Store) Join(addr string) error
  • func (s *Store) JoinRequired() bool
  • func (s *Store) Leader() string
  • func (s *Store) Nodes() ([]string, error)
  • func (s *Store) Open(enableSingle bool) error
  • func (s *Store) Path() string
  • func (s *Store) Peer(addr string) string
  • func (s *Store) Query(qr *QueryRequest) ([]*sql.Rows, error)
  • func (s *Store) RegisterObserver(o *raft.Observer)
  • func (s *Store) Remove(addr string) error
  • func (s *Store) Restore(rc io.ReadCloser) error
  • func (s *Store) Snapshot() (raft.FSMSnapshot, error)
  • func (s *Store) State() ClusterState
  • func (s *Store) Stats() (map[string]interface{}, error)
  • func (s *Store) UpdateAPIPeers(peers map[string]string) error
  • func (s *Store) WaitForAppliedIndex(idx uint64, timeout time.Duration) error
  • func (s *Store) WaitForLeader(timeout time.Duration) (string, error)
• type StoreConfig
• type Transport

Variables

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

	// 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 JoinAllowed

func JoinAllowed(raftDir string) (bool, error)

JoinAllowed returns whether the config files within raftDir indicate that the node can join a cluster.

func NumPeers

func NumPeers(raftDir string) (int, error)

NumPeers returns the number of peers indicated by the config files within raftDir.

This code makes assumptions about how the Raft module works.

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 ConsistencyLevel

type ConsistencyLevel int

ConsistencyLevel represents the available read consistency levels.

const (
	None ConsistencyLevel = iota
	Weak
	Strong
)

Represents the available consistency levels.

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 ExecuteRequest

type ExecuteRequest struct {
	Queries []string
	Timings bool
	Tx      bool
}

ExecuteRequest represents a query that returns now rows, but does modify the database.

type QueryRequest

type QueryRequest struct {
	Queries []string
	Timings bool
	Tx      bool
	Lvl     ConsistencyLevel
}

QueryRequest represents a query that returns rows, and does not modify the database.

type Store

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

Store is a SQLite database, where all changes are made via Raft consensus.

func New

func New(c *StoreConfig) *Store

New returns a new Store.

func (*Store) APIPeers

func (s *Store) APIPeers() (map[string]string, error)

APIPeers return the map of Raft addresses to API addresses.

func (*Store) Addr

func (s *Store) Addr() net.Addr

Addr returns the address of the store.

func (*Store) Apply

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

Apply applies a Raft log entry to the database.

func (*Store) Backup

func (s *Store) Backup(leader bool, fmt BackupFormat) ([]byte, error)

Backup return a snapshot of the underlying database.

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 should 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 *ExecuteRequest) ([]*sql.Result, error)

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

func (*Store) ExecuteOrAbort

func (s *Store) ExecuteOrAbort(ex *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) 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(addr string) error

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

func (*Store) JoinRequired

func (s *Store) JoinRequired() bool

JoinRequired returns whether the node needs to join a cluster after being opened.

func (*Store) Leader

func (s *Store) Leader() string

Leader returns the current leader. Returns a blank string if there is no leader.

func (*Store) Nodes

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

Nodes returns the list of current peers.

func (*Store) Open

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

Open opens the store. If enableSingle is set, and there are no existing peers, then this node becomes the first node, and therefore leader, of the cluster.

func (*Store) Path

func (s *Store) Path() string

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

func (*Store) Peer

func (s *Store) Peer(addr string) string

Peer returns the API address for the given addr. If there is no peer for the address, it returns the empty string.

func (*Store) Query

func (s *Store) Query(qr *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(addr string) error

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

func (*Store) Restore

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

Restore restores the node to a previous state.

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.

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

func (s *Store) UpdateAPIPeers(peers map[string]string) error

UpdateAPIPeers updates the cluster-wide peer information.

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.
	Logger    *log.Logger    // The logger to use to log stuff.
	PeerStore raft.PeerStore // The PeerStore to use for raft.
}

StoreConfig represents the configuration of the underlying Store.

type Transport

type Transport interface {
	net.Listener

	// Dial is used to create a new outgoing connection
	Dial(address string, timeout time.Duration) (net.Conn, error)
}

Transport is the interface the network service must provide.