cassandra

README

Cassandra DB driver

This implementation is geared towards yugabyte.

Migration from v0.4.0 to v0.5.0

The addition of row_id as a TIMEUUID as a simplistic version of state hash. Since row_id can be null, gungnir will work with both database schemas. In order to do long polling, gungnir db driver will need to be updated. Svalinn is not backwards compatible as the insert statement has changed to include the TIMEUUID.

The following is the migration script from v0.4.0 to v0.5.0

ALTER TABLE devices.events ADD row_id TIMEUUID;
CREATE INDEX search_by_row_id ON devices.events
    (device_id, row_id) 
    WITH CLUSTERING ORDER BY (row_id DESC)
    AND default_time_to_live = 2768400
    AND transactions = {'enabled': 'false', 'consistency_level':'user_enforced'};

Documentation

Overview

package cassandra provides a way to connect to a cassandra database to keep track of device events.

Index

• Constants
• func Metrics() []xmetrics.Metric
• type Config
• type Connection
  • func CreateDbConnection(config Config, provider provider.Provider, health *health.Health) (*Connection, error)
  • func (c *Connection) Close() error
  • func (c *Connection) GetBlacklist() (list []blacklist.BlackListedItem, err error)
  • func (c *Connection) GetDeviceList(startDate time.Time, endDate time.Time, offset int, limit int) ([]string, error)
  • func (c *Connection) GetRecords(deviceID string, limit int, stateHash string) ([]db.Record, error)
  • func (c *Connection) GetRecordsOfType(deviceID string, limit int, eventType db.EventType, stateHash string) ([]db.Record, error)
  • func (c *Connection) GetStateHash(records []db.Record) (string, error)
  • func (c *Connection) InsertRecords(records ...db.Record) error
  • func (c *Connection) Ping() error
• type Measures
  • func NewMeasures(p provider.Provider) Measures

Constants

const (
	RetryCounter                = "retry_count"
	PoolOpenConnectionsGauge    = "pool_open_connections"
	PoolInUseConnectionsGauge   = "pool_in_use_connections"
	PoolIdleConnectionsGauge    = "pool_idle_connections"
	SQLWaitCounter              = "sql_wait_count"
	SQLWaitDurationCounter      = "sql_wait_duration_seconds"
	SQLMaxIdleClosedCounter     = "sql_max_idle_closed"
	SQLMaxLifetimeClosedCounter = "sql_max_lifetime_closed"

	SQLDurationSeconds        = "sql_duration_seconds"
	SQLQuerySuccessCounter    = "sql_query_success_count"
	SQLQueryFailureCounter    = "sql_query_failure_count"
	SQLInsertedRecordsCounter = "sql_inserted_rows_count"
	SQLReadRecordsCounter     = "sql_read_rows_count"
	SQLDeletedRecordsCounter  = "sql_deleted_rows_count"
)

const CountLabel = "count"

Functions

func Metrics

func Metrics() []xmetrics.Metric

Metrics returns the Metrics relevant to this package

Types

type Config

type Config struct {
	// Hosts to  connect to. Must have at least one
	Hosts []string

	// Database aka Keyspace for cassandra
	Database string

	// OpTimeout
	OpTimeout time.Duration

	// SSLRootCert used for enabling tls to the cluster. SSLKey, and SSLCert must also be set.
	SSLRootCert string
	// SSLKey used for enabling tls to the cluster. SSLRootCert, and SSLCert must also be set.
	SSLKey string
	// SSLCert used for enabling tls to the cluster. SSLRootCert, and SSLRootCert must also be set.
	SSLCert string
	// If you want to verify the hostname and server cert (like a wildcard for cass cluster) then you should turn this on
	// This option is basically the inverse of InSecureSkipVerify
	// See InSecureSkipVerify in http://golang.org/pkg/crypto/tls/ for more info
	EnableHostVerification bool

	// Username to authenticate into the cluster. Password must also be provided.
	Username string
	// Password to authenticate into the cluster. Username must also be provided.
	Password string

	// NumRetries for connecting to the db
	NumRetries int

	// WaitTimeMult the amount of time to wait before retrying to connect to the db
	WaitTimeMult time.Duration

	// MaxConnsPerHost max number of connections per host
	MaxConnsPerHost int
}

type Connection

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

func CreateDbConnection

func CreateDbConnection(config Config, provider provider.Provider, health *health.Health) (*Connection, error)

func (*Connection) Close

func (c *Connection) Close() error

Close closes the database connection.

func (*Connection) GetBlacklist

func (c *Connection) GetBlacklist() (list []blacklist.BlackListedItem, err error)

GetBlacklist returns a list of blacklisted devices.

func (*Connection) GetDeviceList

func (c *Connection) GetDeviceList(startDate time.Time, endDate time.Time, offset int, limit int) ([]string, error)

GetDeviceList returns a list of device ids where the device id is greater than the offset device id.

func (*Connection) GetRecords

func (c *Connection) GetRecords(deviceID string, limit int, stateHash string) ([]db.Record, error)

GetRecords returns a list of records for a given device.

func (*Connection) GetRecordsOfType

func (c *Connection) GetRecordsOfType(deviceID string, limit int, eventType db.EventType, stateHash string) ([]db.Record, error)

GetRecords returns a list of records for a given device and event type.

func (*Connection) GetStateHash

func (c *Connection) GetStateHash(records []db.Record) (string, error)

GetStateHash returns a hash for the latest record added to the database.

func (*Connection) InsertRecords

func (c *Connection) InsertRecords(records ...db.Record) error

InsertEvent adds a list of records to the table.

func (*Connection) Ping

func (c *Connection) Ping() error

Ping is for pinging the database to verify that the connection is still good.

type Measures

type Measures struct {
	PoolInUseConnections metrics.Gauge
	SQLDuration          metrics.Histogram
	SQLQuerySuccessCount metrics.Counter
	SQLQueryFailureCount metrics.Counter
	SQLInsertedRecords   metrics.Counter
	SQLReadRecords       metrics.Counter
	SQLDeletedRecords    metrics.Counter
}

func NewMeasures

func NewMeasures(p provider.Provider) Measures