ctlstore

README

ctlstore

ctlstore is a distributed data store that provides very low latency, always-available, "infinitely" scalable reads. The underlying mechanism for this is a SQLite database that runs on every host called the LDB. A daemon called the Reflector plays logged writes from a central database into the LDB. As this involves replicating the full data store on every host, it is only practical for situations where the write rate (<100/s total) and data volumes (<10GB total) are low.

Recommended reading:

• Rick Branson's blog post on ctlstore
• Calvin talking about ctlstore at Synapse

Security

Note that because ctlstore replicates the central database to an LDB on each host with a reflector, that LDB contains all of the control data. In its current state that means that any application which has access to the LDB can access all of the data within it.

The implications of this are that you should not store data in ctlstore that should only be accessed by a subset of the applications that can read the LDB. Things like secrets, passwords, and so on, are an example of this.

The ctlstore system is meant to store non-sensitive configuration data.

Development

A MySQL database is needed to run the tests, which can be started using Docker Compose:

$ docker-compose up -d

Run the tests using make:

$ make test
# For more verbosity (`Q=` trick applies to all targets)
$ make test Q=

A single ctlstore binary is used for all functionality. Build it with make:

$ make build

Sync non-stdlib dependencies and pull them into ./vendor

$ make deps

Ctlstore uses Go modules. To build a docker image, the dependencies must be vendored first:

$ make vendor

Many of ctlstore's unit tests use mocks. To regenerate the mocks using counterfeiter:

$ make generate

Tying the Pieces Together

This project includes a docker-compose file docker-compose-example.yml. This initializes and runs

• mysql (ctlstore SoR)
• executive service (guards the ctlstore SoR)
• reflector (builds the LDB)
• heartbeat (mutates a ctlstore table periodically)
• sidecar (provides HTTP API access to ctlstore reader API)
• supervisor (periodically snapshots LDB)

To start it, run:

$ make deps
$ make vendor
$ docker-compose -f docker-compose-example.yml up -d

Documentation

Index

• Constants
• Variables
• func Initialize(ctx context.Context, appName string, statsHandler stats.Handler)
• func InitializeWithConfig(ctx context.Context, cfg Config)
• type Config
• type LDBReader
  • func NewLDBReaderFromDB(db *sql.DB) *LDBReader
  • func Reader() (*LDBReader, error)
  • func ReaderForPath(path string) (*LDBReader, error)
  • func (reader *LDBReader) Close() error
  • func (reader *LDBReader) GetLastSequence(ctx context.Context) (schema.DMLSequence, error)
  • func (reader *LDBReader) GetLedgerLatency(ctx context.Context) (time.Duration, error)
  • func (reader *LDBReader) GetRowByKey(ctx context.Context, out interface{}, familyName string, tableName string, ...) (found bool, err error)
  • func (reader *LDBReader) GetRowsByKeyPrefix(ctx context.Context, familyName string, tableName string, key ...interface{}) (*Rows, error)
  • func (reader *LDBReader) Ping(ctx context.Context) bool
• type LDBRotatingReader
  • func (r *LDBRotatingReader) GetRowByKey(ctx context.Context, out interface{}, familyName string, tableName string, ...) (found bool, err error)
  • func (r *LDBRotatingReader) GetRowsByKeyPrefix(ctx context.Context, familyName string, tableName string, key ...interface{}) (*Rows, error)
• type LDBTestTableDef
• type LDBTestUtil
  • func NewLDBTestUtil(t testing.TB) (*LDBTestUtil, func())
  • func NewLDBTestUtilLocal(t testing.TB) (*LDBTestUtil, func())
  • func (tu *LDBTestUtil) CreateTable(def LDBTestTableDef)
  • func (tu *LDBTestUtil) DeleteAll(family string, table string)
  • func (tu *LDBTestUtil) InsertRows(family string, table string, rows [][]interface{})
  • func (tu *LDBTestUtil) Reset()
• type RotationPeriod
• type RowRetriever
  • func CustomerRotatingReader(ctx context.Context, minutesPerRotation RotationPeriod, ldbPaths ...string) (RowRetriever, error)
  • func RotatingReader(ctx context.Context, minutesPerRotation RotationPeriod, ldbsCount int) (RowRetriever, error)
• type Rows
  • func (r *Rows) Close() error
  • func (r *Rows) Err() error
  • func (r *Rows) Next() bool
  • func (r *Rows) Scan(target interface{}) error

Constants

const (
	DefaultCtlstorePath      = "/var/spool/ctlstore/"
	DefaultChangelogFilename = "change.log"
)

Variables

var (
	ErrTableHasNoPrimaryKey = errors.New("Table provided has no primary key")
	ErrNeedFullKey          = errors.New("All primary key fields are required")
	ErrNoLedgerUpdates      = errors.New("no ledger updates have been received yet")
)

var Version string

Version is the current ctlstore client library version.

Functions

func Initialize

func Initialize(ctx context.Context, appName string, statsHandler stats.Handler)

Initialize sets up global state for thing including global metrics globalstats data and possibly more as time goes on.

Deprecated: see InitializeWithConfig

func InitializeWithConfig

func InitializeWithConfig(ctx context.Context, cfg Config)

InitializeWithConfig sets up global state for thing including global metrics globalstats data and possibly more as time goes on.

Types

type Config

type Config struct {
	// Stats specifies the config for reporting stats to the global
	// ctlstore stats namespace.
	//
	// By default, global stats are enabled with a set of sane defaults.
	Stats *globalstats.Config

	// LDBVersioning, if enabled, will instruct ctlstore to look for
	// LDBs inside of timestamp-delimited folders, and ctlstore will
	// hot-reload new LDBs as they appear.
	//
	// By default, this is disabled.
	LDBVersioning bool
}

type LDBReader

type LDBReader struct {
	Db *sql.DB
	// contains filtered or unexported fields
}

LDBReader reads data from the LDB. The external interface is thread-safe and it is safe to create as many of these as needed across multiple processes.

func NewLDBReaderFromDB

func NewLDBReaderFromDB(db *sql.DB) *LDBReader

Constructs an LDBReader from a sql.DB. Really only useful for testing.

func Reader

func Reader() (*LDBReader, error)

Reader returns an LDBReader that can be used globally.

func ReaderForPath

func ReaderForPath(path string) (*LDBReader, error)

ReaderForPath opens an LDB at the provided path and returns an LDBReader instance pointed at that LDB.

func (*LDBReader) Close

func (reader *LDBReader) Close() error

func (*LDBReader) GetLastSequence

func (reader *LDBReader) GetLastSequence(ctx context.Context) (schema.DMLSequence, error)

GetLastSequence returns the highest sequence number applied to the DB

func (*LDBReader) GetLedgerLatency

func (reader *LDBReader) GetLedgerLatency(ctx context.Context) (time.Duration, error)

GetLedgerLatency returns the difference between the current time and the timestamp from the last DML ledger update processed by the reflector. ErrNoLedgerUpdates will be returned if no DML statements have been processed.

func (*LDBReader) GetRowByKey

func (reader *LDBReader) GetRowByKey(
	ctx context.Context,
	out interface{},
	familyName string,
	tableName string,
	key ...interface{},
) (found bool, err error)

GetRowByKey fetches a row from the supplied table by the key parameter, filling the data into the out param.

The out param may be one of the following types:

• pointer to struct
• map[string]interface{}

The key parameter can support composite keys by passing a slice type.

func (*LDBReader) GetRowsByKeyPrefix

func (reader *LDBReader) GetRowsByKeyPrefix(ctx context.Context, familyName string, tableName string, key ...interface{}) (*Rows, error)

GetRowsByKeyPrefix returns a *Rows iterator that will supply all of the rows in the family and table match the supplied primary key prefix.

func (*LDBReader) Ping

func (reader *LDBReader) Ping(ctx context.Context) bool

Ping checks if the LDB is available

type LDBRotatingReader

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

LDBRotatingReader reads data from multiple LDBs on a rotating schedule. The main benefit is relieving read pressure on a particular ldb file when it becomes inactive, allowing sqlite maintenance

func (*LDBRotatingReader) GetRowByKey

func (r *LDBRotatingReader) GetRowByKey(ctx context.Context, out interface{}, familyName string, tableName string, key ...interface{}) (found bool, err error)

GetRowByKey delegates to the active LDBReader

func (*LDBRotatingReader) GetRowsByKeyPrefix

func (r *LDBRotatingReader) GetRowsByKeyPrefix(ctx context.Context, familyName string, tableName string, key ...interface{}) (*Rows, error)

GetRowsByKeyPrefix delegates to the active LDBReader

type LDBTestTableDef

type LDBTestTableDef struct {
	Family    string
	Name      string
	Fields    [][]string
	KeyFields []string
	Rows      [][]interface{}
}

LDBTestTableDef is used to pass a table definition to CreateTable for use in tests that need the LDB. The way the parameters are specified mimics the executive interface. Fields are passed as tuples of [name string, type string] where type can be something like "string" or "integer," just like the standard executive interface.

type LDBTestUtil

type LDBTestUtil struct {
	DB *sql.DB
	T  testing.TB
}

LDBTestUtil provides basic unit testing facilities for injecting data into a "fake" LDB.

func NewLDBTestUtil

func NewLDBTestUtil(t testing.TB) (*LDBTestUtil, func())

NewLDBTestUtil changes the global default LDB path to a temporary path.

This function is NOT concurrency safe.

func NewLDBTestUtilLocal

func NewLDBTestUtilLocal(t testing.TB) (*LDBTestUtil, func())

NewLDBTestUtilLocal is just like NewLDBTestUtil above except it does not rely on global state and is therefore threadsafe, at the cost of requiring users to use ensure that the DB property is used to initialize the ctlstore Reader instead of relying on the global/default init.

func (*LDBTestUtil) CreateTable

func (tu *LDBTestUtil) CreateTable(def LDBTestTableDef)

CreateTable creates a table in the target test LDB.

func (*LDBTestUtil) DeleteAll

func (tu *LDBTestUtil) DeleteAll(family string, table string)

DeleteAll deletes all rows from the given table.

func (*LDBTestUtil) InsertRows

func (tu *LDBTestUtil) InsertRows(family string, table string, rows [][]interface{})

InsertRows well, inserts rows into the LDB. Rows are passed as tuples in the table's column order.

func (*LDBTestUtil) Reset

func (tu *LDBTestUtil) Reset()

Reset completely clears the test LDB

type RotationPeriod

type RotationPeriod int

RotationPeriod how many minutes each reader is active for before rotating to the next

const (
	// Every30 rotate on 30 minute mark in an hour
	Every30 RotationPeriod = 30
	// Every20 rotate on 20 minute marks in an hour
	Every20 RotationPeriod = 20
	// Every15 rotate on 15 minute marks in an hour
	Every15 RotationPeriod = 15
	// Every10 rotate on 10 minute marks in an hour
	Every10 RotationPeriod = 10
	// Every6 rotate on 6 minute marks in an hour
	Every6 RotationPeriod = 6
)

type RowRetriever

type RowRetriever interface {
	GetRowsByKeyPrefix(ctx context.Context, familyName string, tableName string, key ...interface{}) (*Rows, error)
	GetRowByKey(ctx context.Context, out interface{}, familyName string, tableName string, key ...interface{}) (found bool, err error)
}

func CustomerRotatingReader

func CustomerRotatingReader(ctx context.Context, minutesPerRotation RotationPeriod, ldbPaths ...string) (RowRetriever, error)

CustomerRotatingReader creates a new reader that rotates which ldb it reads from on a rotation period

func RotatingReader

func RotatingReader(ctx context.Context, minutesPerRotation RotationPeriod, ldbsCount int) (RowRetriever, error)

RotatingReader creates a new reader that rotates which ldb it reads from on a rotation period with the default location in /var/spool/ctlstore

type Rows

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

Rows composes an *sql.Rows and allows scanning ctlstore table rows into structs or maps, similar to how the GetRowByKey reader method works.

The contract around Next/Err/Close is the same was it is for *sql.Rows.

func (*Rows) Close

func (r *Rows) Close() error

Close closes the underlying *sql.Rows.

func (*Rows) Err

func (r *Rows) Err() error

Err returns any error that could have been caused during the invocation of Next(). If Next() returns false, the caller must always check Err() to see if that's why iteration failed.

func (*Rows) Next

func (r *Rows) Next() bool

Next returns true if there's another row available.

func (*Rows) Scan

func (r *Rows) Scan(target interface{}) error

Scan deserializes the current row into the specified target. The target must be either a pointer to a struct, or a map[string]interface{}.