datastore

Documentation

Index

• Constants
• Variables
• type Datastore
• type Field
• type FieldLookupError
  • func (self FieldLookupError) Error() string
• type LevelDbShard
  • func NewLevelDbShard(db *levigo.DB, pointBatchSize int) (*LevelDbShard, error)
  • func (self *LevelDbShard) DropDatabase(database string) error
  • func (self *LevelDbShard) IsClosed() bool
  • func (self *LevelDbShard) Query(querySpec *parser.QuerySpec, processor cluster.QueryProcessor) error
  • func (self *LevelDbShard) Write(database string, series *protocol.Series) error
• type LevelDbShardDatastore
  • func NewLevelDbShardDatastore(config *configuration.Configuration) (*LevelDbShardDatastore, error)
  • func (self *LevelDbShardDatastore) BufferWrite(request *protocol.Request)
  • func (self *LevelDbShardDatastore) Close()
  • func (self *LevelDbShardDatastore) DeleteShard(shardId uint32) error
  • func (self *LevelDbShardDatastore) GetOrCreateShard(id uint32) (cluster.LocalShardDb, error)
  • func (self *LevelDbShardDatastore) ReturnShard(id uint32)
  • func (self *LevelDbShardDatastore) SetWriteBuffer(writeBuffer *cluster.WriteBuffer)
  • func (self *LevelDbShardDatastore) Write(request *protocol.Request) error

Constants

const (
	ONE_KILOBYTE                    = 1024
	ONE_MEGABYTE                    = 1024 * 1024
	ONE_GIGABYTE                    = ONE_MEGABYTE * 1024
	TWO_FIFTY_SIX_KILOBYTES         = 256 * 1024
	SIXTY_FOUR_KILOBYTES            = 64 * 1024
	MAX_SERIES_SIZE                 = ONE_MEGABYTE
	DATABASE_DIR                    = "db"
	SHARD_BLOOM_FILTER_BITS_PER_KEY = 10
	SHARD_DATABASE_DIR              = "shard_db"
)

Variables

var (

	// This datastore implements the PersistentAtomicInteger interface. All of the persistent
	// integers start with this prefix, followed by their name
	ATOMIC_INCREMENT_PREFIX = []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFD}
	// NEXT_ID_KEY holds the next id. ids are used to "intern" timeseries and column names
	NEXT_ID_KEY = []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
	// SERIES_COLUMN_INDEX_PREFIX is the prefix of the series to column names index
	SERIES_COLUMN_INDEX_PREFIX = []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFE}
	// DATABASE_SERIES_INDEX_PREFIX is the prefix of the database to series names index
	DATABASE_SERIES_INDEX_PREFIX = []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF}
	MAX_SEQUENCE                 = []byte{0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF}

	TRUE = true
)

Types

type Datastore

type Datastore interface {
	ExecuteQuery(user common.User, database string,
		query *parser.SelectQuery, yield func(*protocol.Series) error,
		ringFilter func(database, series *string, time *int64) bool) error
	// Logs the request to a local store and assigns a sequence number that is unique per server id per day
	LogRequestAndAssignSequenceNumber(request *protocol.Request, replicationFactor *uint8, ownerServerId *uint32) error
	CurrentSequenceNumber(clusterVersion *uint32, replicationFactor *uint8, ownerServerId, originatingServerId *uint32) (uint64, error)
	// will replay all requests from a given number. If the number hasn't occured yet today, it replays from yesterday.
	// So this log replay is only meant to work for outages that last less than maybe 12 hours.
	ReplayRequestsFromSequenceNumber(*uint32, *uint32, *uint32, *uint8, *uint64, func(*[]byte) error) error
	// Increment the named integer by the given amount and return the new value
	AtomicIncrement(name string, val int) (uint64, error)
	WriteSeriesData(database string, series *protocol.Series) error
	DeleteSeriesData(database string, query *parser.DeleteQuery) error
	GetSeriesForDatabase(database string, yield func(string) error) error
	DropDatabase(database string) error
	DropSeries(database, series string) error
	Close()
}

type Field

type Field struct {
	Id   []byte
	Name string
}

type FieldLookupError

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

func (FieldLookupError) Error

func (self FieldLookupError) Error() string

type LevelDbShard

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

func NewLevelDbShard

func NewLevelDbShard(db *levigo.DB, pointBatchSize int) (*LevelDbShard, error)

func (*LevelDbShard) DropDatabase

func (self *LevelDbShard) DropDatabase(database string) error

func (*LevelDbShard) IsClosed

func (self *LevelDbShard) IsClosed() bool

func (*LevelDbShard) Query

func (self *LevelDbShard) Query(querySpec *parser.QuerySpec, processor cluster.QueryProcessor) error

func (*LevelDbShard) Write

func (self *LevelDbShard) Write(database string, series *protocol.Series) error

type LevelDbShardDatastore

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

func NewLevelDbShardDatastore

func NewLevelDbShardDatastore(config *configuration.Configuration) (*LevelDbShardDatastore, error)

func (*LevelDbShardDatastore) BufferWrite

func (self *LevelDbShardDatastore) BufferWrite(request *protocol.Request)

func (*LevelDbShardDatastore) Close

func (self *LevelDbShardDatastore) Close()

func (*LevelDbShardDatastore) DeleteShard

func (self *LevelDbShardDatastore) DeleteShard(shardId uint32) error

func (*LevelDbShardDatastore) GetOrCreateShard

func (self *LevelDbShardDatastore) GetOrCreateShard(id uint32) (cluster.LocalShardDb, error)

func (*LevelDbShardDatastore) ReturnShard

func (self *LevelDbShardDatastore) ReturnShard(id uint32)

func (*LevelDbShardDatastore) SetWriteBuffer

func (self *LevelDbShardDatastore) SetWriteBuffer(writeBuffer *cluster.WriteBuffer)

func (*LevelDbShardDatastore) Write

func (self *LevelDbShardDatastore) Write(request *protocol.Request) error