diff --git a/cmd/pilosa/config.go b/cmd/pilosa/config.go index 4ddc7484e..81c4a8d70 100644 --- a/cmd/pilosa/config.go +++ b/cmd/pilosa/config.go @@ -4,6 +4,8 @@ import ( "time" "github.com/umbel/pilosa/index" + _ "github.com/umbel/pilosa/index/storage" + "github.com/umbel/pilosa/index/storage/cassandra" "github.com/umbel/pilosa/transport" "github.com/umbel/pilosa/util" ) @@ -91,12 +93,12 @@ func NewConfig() Config { c.HTTP.Port = transport.DefaultHTTPPort c.Log.Path = DefaultLogPath c.Storage.Backend = index.DefaultBackend - c.Storage.Hosts = index.DefaultStorageHosts[:] - c.Storage.Keyspace = index.DefaultStorageKeyspace + c.Storage.Hosts = cassandra.DefaultHosts + c.Storage.Keyspace = cassandra.DefaultKeyspace c.Storage.FragmentBase = DefaultFragmentBase c.Storage.SupportedFrames = DefaultSupportedFrames[:] - c.Storage.CassandraTimeWindow = Duration(index.DefaultCassandraTimeWindow) - c.Storage.CassandraMaxSizeBatch = index.DefaultCassandraMaxSizeBatch + c.Storage.CassandraTimeWindow = Duration(cassandra.DefaultFlushInterval) + c.Storage.CassandraMaxSizeBatch = cassandra.DefaultFlushThreshold c.Statsd.Host = util.DefaultStatsdHost c.ETCD.Hosts = DefaultETCDHosts[:] return c diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index 79a452381..1b4f80544 100644 --- a/cmd/pilosa/main.go +++ b/cmd/pilosa/main.go @@ -108,9 +108,6 @@ func (m *Main) Run(args ...string) error { util.StatsdHost = config.Statsd.Host util.SetupStatsd() - // REMOVED(benbjohnson): config.SetupConfig() - index.SetupCassandra() - // Initialize logging. logger, _ := log.LoggerFromConfigAsBytes([]byte(SeelogProductionConfig(config.Log.Path, *id, config.Log.Level))) log.ReplaceLogger(logger) diff --git a/core/etcd.go b/core/etcd.go index a42cb9062..6b01b6f7f 100644 --- a/core/etcd.go +++ b/core/etcd.go @@ -299,7 +299,7 @@ func (self *TopologyMapper) remove_fragment(node *etcd.Node) error { func flatten(node *etcd.Node) []*etcd.Node { nodes := []*etcd.Node{node} for i := 0; i < len(node.Nodes); i++ { - nodes = append(nodes, flatten(&node.Nodes[i])...) + nodes = append(nodes, flatten(node.Nodes[i])...) } return nodes } diff --git a/index/blocks.go b/index/blocks.go index 482d41502..8e389cbc5 100644 --- a/index/blocks.go +++ b/index/blocks.go @@ -6,13 +6,18 @@ import ( type Blocks []uint64 +// NewBlocks returns a 32-length Block. +func NewBlocks() Blocks { + return make(Blocks, 32) +} + func (a Blocks) bitcount() uint64 { fmt.Println("bitcount:", a) return popcntSlice(a) } func (a Blocks) union(other Blocks) Blocks { - ret := make(Blocks, 32) + ret := NewBlocks() for i, _ := range a { ret[i] = a[i] | other[i] } @@ -20,7 +25,7 @@ func (a Blocks) union(other Blocks) Blocks { } func (a Blocks) invert() Blocks { - other := make(Blocks, 32) + other := NewBlocks() for i, _ := range a { other[i] = ^a[i] } @@ -28,7 +33,7 @@ func (a Blocks) invert() Blocks { } func (a Blocks) copy() Blocks { - other := make(Blocks, 32) + other := NewBlocks() for i, _ := range a { other[i] = a[i] } @@ -41,7 +46,7 @@ func (a Blocks) andcount(other Blocks) uint64 { } func (a Blocks) intersection(other Blocks) Blocks { - ret := make(Blocks, 32) + ret := NewBlocks() for i, _ := range a { ret[i] = a[i] & other[i] } @@ -49,7 +54,7 @@ func (a Blocks) intersection(other Blocks) Blocks { } func (a Blocks) difference(other Blocks) Blocks { - ret := make(Blocks, 32) + ret := NewBlocks() for i, _ := range a { ret[i] = a[i] &^ other[i] } diff --git a/index/brand.go b/index/brand.go index 70ba3b29a..18498f355 100644 --- a/index/brand.go +++ b/index/brand.go @@ -360,7 +360,7 @@ func (b *Brand) getFileName() string { func (b *Brand) Persist() error { log.Info("Brand Persist:", b.getFileName()) - b.storage.FlushBatch() + b.storage.Flush() asize := len(b.bitmap_cache) if asize == 0 { diff --git a/index/brand_test.go b/index/brand_test.go index 646d121a4..fdabd05b2 100644 --- a/index/brand_test.go +++ b/index/brand_test.go @@ -1,16 +1,18 @@ -package index +package index_test import ( "math/rand" "testing" + "github.com/umbel/pilosa/index" + "github.com/umbel/pilosa/index/storage/mem" "github.com/umbel/pilosa/util" ) var ( size int - membrand *Brand - cassbrand *Brand + membrand *index.Brand + cassbrand *index.Brand ) func init() { @@ -19,7 +21,7 @@ func init() { util.SetupStatsd() // SetupCassandra() - membrand = NewBrand("db", "frame", 0, NewMemoryStorage(), size, size, 0) + membrand = index.NewBrand("db", "frame", 0, mem.NewStorage(), size, size, 0) for i := uint64(0); i < uint64(size); i++ { membrand.SetBit(i, 0, 1) } @@ -29,7 +31,7 @@ func init() { // } } -func benchmarkBrand(b *testing.B, size int, fill int, brand *Brand) { +func benchmarkBrand(b *testing.B, size int, fill int, brand *index.Brand) { println(b.N) for i := 0; i < b.N; i++ { bid := rand.Int() % size diff --git a/index/fragment_container.go b/index/fragment_container.go index 1d5dfcb64..a1591de0c 100644 --- a/index/fragment_container.go +++ b/index/fragment_container.go @@ -369,25 +369,22 @@ type Fragment struct { queue_size int } -func getStorage(db string, slice int, frame string, fid util.SUUID) Storage { - switch Backend { - default: - return NewMemoryStorage() - case "leveldb": - full_dir := fmt.Sprintf("%s/%s/%d/%s/%s", LevelDBPath, db, slice, frame, util.SUUID_to_Hex(fid)) - return NewLevelDBStorage(full_dir) - case "cassandra": - return NewCassStorage() - } -} - func NewFragment(frag_id util.SUUID, db string, slice int, frame string) *Fragment { var impl Pilosa log.Warn(fmt.Sprintf("XXXXXXXXXXXXXXXXXXXXXXXXXXX(%s)", frame)) + + storage := NewStorage(Backend, StorageOptions{ + DB: db, + Slice: slice, + Frame: frame, + FragmentID: frag_id, + LevelDBPath: LevelDBPath, + }) + if strings.HasSuffix(frame, ".n") { - impl = NewBrand(db, frame, slice, getStorage(db, slice, frame, frag_id), 50000, 45000, 100) + impl = NewBrand(db, frame, slice, storage, 50000, 45000, 100) } else { - impl = NewGeneral(db, frame, slice, getStorage(db, slice, frame, frag_id)) + impl = NewGeneral(db, frame, slice, storage) } f := new(Fragment) diff --git a/index/fragment_container_test.go b/index/fragment_container_test.go index 25fc8d18f..c92daccce 100644 --- a/index/fragment_container_test.go +++ b/index/fragment_container_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/umbel/pilosa/index" + _ "github.com/umbel/pilosa/index/storage" "github.com/umbel/pilosa/util" ) diff --git a/index/storage.go b/index/storage.go index 99b5eeec9..7ee94f481 100644 --- a/index/storage.go +++ b/index/storage.go @@ -1,14 +1,53 @@ package index +import ( + "github.com/umbel/pilosa/util" +) + +// Storage represents type Storage interface { - Fetch(bitmap_id uint64, db string, frame string, slice int) (*Bitmap, uint64) - Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error - StoreBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, block uint64) error - StoreBit(bid uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, block, count uint64) - RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) - RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) - BeginBatch() - EndBatch() - FlushBatch() - Close() + Open() error + Close() error + Flush() + + Fetch(bitmap_id uint64, db, frame string, slice int) (*Bitmap, uint64) + Store(id uint64, db, frame string, slice int, filter uint64, bitmap *Bitmap) error + + StoreBlock(id uint64, db, frame string, slice int, filter uint64, chunk uint64, block_index int32, block uint64) error + RemoveBlock(id uint64, db, frame string, slice int, chunk uint64, block_index int32) + + StoreBit(bid uint64, db, frame string, slice int, filter uint64, chunk uint64, block_index int32, block, count uint64) + RemoveBit(id uint64, db, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) +} + +var storageFns = make(map[string]NewStorageFunc) + +// NewStorageFunc represents a function that instantiates a new storage engine. +type NewStorageFunc func(opt StorageOptions) Storage + +// RegisterStorage registers a storage engine. +func RegisterStorage(name string, fn NewStorageFunc) { + if storageFns[name] != nil { + panic("storage engine already registered: " + name) + } + storageFns[name] = fn +} + +// NewStorage returns a new Storage instance by name. +func NewStorage(name string, opt StorageOptions) Storage { + fn := storageFns[name] + if fn == nil { + panic("storage type not registered: " + name) + } + return fn(opt) +} + +// StorageOptions represents the options passed to the storage engine. +type StorageOptions struct { + DB string + Slice int + Frame string + FragmentID util.SUUID + + LevelDBPath string } diff --git a/index/storage/cassandra/cassandra.go b/index/storage/cassandra/cassandra.go new file mode 100644 index 000000000..28a44ed4f --- /dev/null +++ b/index/storage/cassandra/cassandra.go @@ -0,0 +1,244 @@ +package cassandra + +import ( + "time" + + log "github.com/cihub/seelog" + "github.com/gocql/gocql" + "github.com/umbel/pilosa/index" + "github.com/umbel/pilosa/util" +) + +func init() { + index.RegisterStorage("cassandra", + func(opt index.StorageOptions) index.Storage { + return NewStorage(opt) + }, + ) +} + +// CREATE KEYSPACE IF NOT EXISTS pilosa WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1" +// create keyspace if not exists pilosa with replication = {'class': 'SimpleStrategy', 'replication_factor' : 1} and durable_writes = true; +// CREATE KEYSPACE pilosa WITH replication = {'class': 'NetworkTopologyStrategy', 'pilpang': '2'} AND durable_writes = true; +// CREATE TABLE IF NOT EXISTS bitmap (bitmap_id bigint, db varchar, frame varchar, slice int, filter int, chunkkey bigint, blockindex int, block bigint, PRIMARY KEY ((bitmap_id, db, frame, slice), chunkkey, blockindex) ) + +// DefaultHosts are the default hosts in the cassandra cluster. +var DefaultHosts = []string{"localhost"} + +const ( + // DefaultKeyspace is the default keyspace used in cassandra. + DefaultKeyspace = "pilosa" + + // DefaultFlushInterval is the default maximum time between flushes. + DefaultFlushInterval = 5 * time.Second + + // DefaultFlushThreshold is the default maximum number of items to batch. + DefaultFlushThreshold = 15 +) + +// Storage represents Cassandra-backed storage for bitmaps. +type Storage struct { + session *gocql.Session + + batch *gocql.Batch + batchTime time.Time + batchN int + + FlushInterval time.Duration + FlushThreshold int + + Hosts []string + Keyspace string +} + +// NewStorage returns a new, uninitialized instance of Storage. +func NewStorage(opt index.StorageOptions) *Storage { + return &Storage{ + batchTime: time.Now(), + + FlushInterval: DefaultFlushInterval, + FlushThreshold: DefaultFlushThreshold, + + Hosts: DefaultHosts, + Keyspace: DefaultKeyspace, + } +} + +// Open opens the connection to the cassandra cluster. +func (s *Storage) Open() error { + // Create cluster configuration. + config := gocql.NewCluster(s.Hosts...) + config.Keyspace = s.Keyspace + config.Consistency = gocql.One + config.Timeout = 5 * time.Second + config.RetryPolicy = &gocql.SimpleRetryPolicy{NumRetries: 10} + + // Connect to cassandra. + session, err := config.CreateSession() + if err != nil { + return err + } + s.session = session + + return nil +} + +// Close closes the connection to the cassandra cluster. +func (s *Storage) Close() error { + if s.session != nil { + s.Flush() + s.session.Close() + } + + return nil +} + +// Fetch returns a bitmap by ID. +func (s *Storage) Fetch(bitmapID uint64, db string, frame string, slice int) (*index.Bitmap, uint64) { + bm := index.NewBitmap() + + // Start benchmark. + start := time.Now() + + // Create iterator over bitmap. + itr := s.session.Query("SELECT filter, Chunkkey, BlockIndex, block FROM bitmap WHERE bitmap_id=? AND db=? AND frame=? AND slice=? ", + u64toi64(bitmapID), db, frame, slice).Iter() + + // Iterate over chunks and materialize bitmap object. + var chunk *index.Chunk + var chunkKey, block, count int64 + var blockIndex uint32 + var filter int + lastKey := int64(-1) + + for itr.Scan(&filter, &chunkKey, &blockIndex, &block) { + if chunkKey != int64(-1) { + if chunkKey != lastKey { + chunk = &index.Chunk{uint64(chunkKey), index.NewBlocks()} + bm.AddChunk(chunk) + } + chunk.Value[uint8(blockIndex)] = uint64(block) + } else { + count = block + } + lastKey = chunkKey + } + + util.SendTimer("cassandra_storage_Fetch", time.Since(start).Nanoseconds()) + util.SendInc("cassandra_storage_Read") + + // Set total bits set. + bm.SetCount(uint64(count)) + return bm, uint64(filter) +} + +// beginBatch starts a batch if one is not already in progress. +func (s *Storage) beginBatch() { + if s.batch == nil { + s.batch = gocql.NewBatch(gocql.UnloggedBatch) + } + s.batchN++ +} + +// endBatch flushes a batch if one is in progress. +func (s *Storage) endBatch() { + if s.batch == nil { + return + } + + start := time.Now() + s.Flush() + util.SendTimer("cassandra_storage_EndBatch", time.Since(start).Nanoseconds()) + util.SendInc("cassandra_storage_Write") +} + +// Flush flushes the current batch to storage. +func (s *Storage) Flush() { + start := time.Now() + + // If batch exists then flush it. + if s.batch != nil { + if err := s.session.ExecuteBatch(s.batch); err != nil { + log.Warn("Batch ERROR: ", err) + } + } + + // Clear batch and threshold time and count. + s.batch = nil + s.batchTime = time.Now() + s.batchN = 0 + + util.SendTimer("cassandra_storage_FlushBatch", time.Since(start).Nanoseconds()) +} + +// Store saves a bitmap to storage. +func (s *Storage) Store(id uint64, db string, frame string, slice int, filter uint64, bm *index.Bitmap) error { + s.beginBatch() + + for i := bm.ChunkIterator(); !i.Limit(); i = i.Next() { + var chunk = i.Item() + for idx, block := range chunk.Value { + if block != 0 { + s.StoreBlock(id, db, frame, slice, filter, chunk.Key, int32(idx), block) + } + } + } + + s.StoreBlock(id, db, frame, slice, filter, index.CounterMask, 0, bm.BitCount()) + s.endBatch() + return nil +} + +// StoreBlock saves a block to storage. +func (s *Storage) StoreBlock(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, blockIndex int32, bblock uint64) error { + id := u64toi64(bid) // these functions ignore overflow + block := u64toi64(bblock) + chunk := u64toi64(bchunk) + + if s.batch == nil { + s.beginBatch() + } + + start := time.Now() + s.batch.Query(`INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?) USING timestamp ?;`, + id, db, frame, slice, int(filter), chunk, blockIndex, block, start.UnixNano()) + util.SendTimer("cassandra_storage_StoreBlock", time.Since(start).Nanoseconds()) + + return nil +} + +// RemoveBlock deletes a block from storage. +func (s *Storage) RemoveBlock(bid uint64, db string, frame string, slice int, bchunk uint64, blockIndex int32) { + log.Trace("RemoveBlock", bid, db, frame, slice, bchunk, blockIndex) + + id := u64toi64(bid) //these fucntions ignore overflow + chunk := u64toi64(bchunk) + + if s.batch == nil { + s.beginBatch() + } + + startTime := time.Now() + + s.batch.Query(`DELETE FROM bitmap USING TIMESTAMP ? WHERE bitmap_id=? AND db=? AND frame=? AND slice=? AND chunkkey=? AND blockindex=?`, + startTime.UnixNano(), id, db, frame, slice, chunk, blockIndex) + + util.SendTimer("cassandra_storage_DeleteBlock", time.Since(startTime).Nanoseconds()) +} + +func (s *Storage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, chunk uint64, blockIndex int32, val, count uint64) { + s.beginBatch() + s.StoreBlock(bid, db, frame, slice, filter, chunk, blockIndex, val) + s.StoreBlock(bid, db, frame, slice, filter, index.CounterMask, 0, count) + s.endBatch() +} + +func (s *Storage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, blockIndex int32, count uint64) { + log.Trace("RemoveBit", id, db, frame, slice, chunk, blockIndex) + s.beginBatch() + s.RemoveBlock(id, db, frame, slice, chunk, blockIndex) + s.StoreBlock(id, db, frame, slice, filter, index.CounterMask, 0, count) + s.endBatch() +} + +func u64toi64(v uint64) int64 { return int64(v) } diff --git a/index/storage/cassandra/cassandra_test.go b/index/storage/cassandra/cassandra_test.go new file mode 100644 index 000000000..702552410 --- /dev/null +++ b/index/storage/cassandra/cassandra_test.go @@ -0,0 +1 @@ +package cassandra_test diff --git a/index/storage/leveldb/leveldb.go b/index/storage/leveldb/leveldb.go new file mode 100644 index 000000000..7320d6da0 --- /dev/null +++ b/index/storage/leveldb/leveldb.go @@ -0,0 +1,222 @@ +package leveldb + +import ( + "bytes" + "encoding/binary" + "path/filepath" + "strconv" + "time" + + "github.com/syndtr/goleveldb/leveldb" + . "github.com/syndtr/goleveldb/leveldb/util" + "github.com/umbel/pilosa/index" + "github.com/umbel/pilosa/util" +) + +func init() { + index.RegisterStorage("leveldb", + func(opt index.StorageOptions) index.Storage { + return NewStorage(opt) + }, + ) +} + +//go get github.com/syndtr/goleveldb/leveldb + +// Storage represents a LevelDB-backed storage engine. +type Storage struct { + path string + db *leveldb.DB + + batch *leveldb.Batch + batchTime time.Time + batchN int +} + +// NewStorage returns a new instance of Storage. +func NewStorage(opt index.StorageOptions) *Storage { + path := filepath.Join( + opt.LevelDBPath, + opt.DB, + strconv.Itoa(opt.Slice), + opt.Frame, + util.SUUID_to_Hex(opt.FragmentID), + ) + + return &Storage{path: path} +} + +// Path returns the path the storage was initialized with. +func (s *Storage) Path() string { return s.path } + +// Open opens and initializes the storage. +func (s *Storage) Open() error { + db, err := leveldb.OpenFile(s.path, nil) + if err != nil { + return err + } + s.db = db + s.batchTime = time.Now().Add(-time.Hour) + + return nil +} + +// Close flushes and closes the storage. +func (s *Storage) Close() error { + s.Flush() + return s.db.Close() +} + +func (s *Storage) Fetch(bitmapID uint64, db string, frame string, slice int) (*index.Bitmap, uint64) { + bm := index.NewBitmap() + + // Begin benchmark. + start := time.Now() + + // Create an iterator on the database. + iter := s.db.NewIterator(&Range{ + Start: marshalKey(bitmapID, 0, 0), + Limit: marshalKey(bitmapID+1, 0, 0), + }, nil) + defer iter.Release() + + // Iterate over blocks in database and create bitmap. + var chunk *index.Chunk + var filter, block, count uint64 + lastKey := uint64(index.CounterMask) + + for iter.Next() { + _, key, idx := unmarshalKey(iter.Key()) + block, filter = unmarshalValue(iter.Value()) + if key != index.CounterMask { + if key != lastKey { + chunk = &index.Chunk{key, index.NewBlocks()} + bm.AddChunk(chunk) + } + chunk.Value[idx] = block + } else { + count = block + } + + lastKey = key + } + + util.SendTimer("leveldb_storage_Fetch", time.Since(start).Nanoseconds()) + + // Set bit count on bitmap. + bm.SetCount(uint64(count)) + + return bm, uint64(filter) +} + +// beginBatch starts a new batch if one is not already started. +func (s *Storage) beginBatch() { + if s.batch == nil { + s.batch = &leveldb.Batch{} + } + s.batchN++ +} + +func (s *Storage) endBatch() { + if s.batch == nil { + return + } + + start := time.Now() + if time.Since(s.batchTime) > 15*time.Second || s.batchN > 300 { + s.Flush() + } + util.SendTimer("leveldb_storage_EndBatch", time.Since(start).Nanoseconds()) +} + +func (s *Storage) Flush() { + start := time.Now() + + // Flush the batch if one exists. + if s.batch != nil { + s.db.Write(s.batch, nil) + } + + // Clear batch and reset timer and count. + s.batch = nil + s.batchTime = time.Now() + s.batchN = 0 + + util.SendTimer("leveldb_storage_FlushBatch", time.Since(start).Nanoseconds()) +} + +// Store saves a bitmap to storage. +func (s *Storage) Store(bitmapID uint64, db string, frame string, slice int, filter uint64, bm *index.Bitmap) error { + s.beginBatch() + + for itr := bm.ChunkIterator(); !itr.Limit(); itr = itr.Next() { + for idx, block := range itr.Item().Value { + if block != 0 { + s.StoreBlock(bitmapID, db, frame, slice, filter, itr.Item().Key, int32(idx), block) + } + } + } + + s.StoreBlock(bitmapID, db, frame, slice, filter, index.CounterMask, 0, bm.BitCount()) + + s.endBatch() + return nil +} + +// StoreBlock saves a block to storage. +func (s *Storage) StoreBlock(bitmapID uint64, db string, frame string, slice int, filter uint64, chunk uint64, index int32, block uint64) error { + start := time.Now() + s.batch.Put(marshalKey(bitmapID, chunk, uint8(index)), marshalValue(block, filter)) + util.SendTimer("leveldb_storage_StoreBlock", time.Since(start).Nanoseconds()) + return nil +} + +// RemoveBlock deletes a block from storage. This is a no-op in the LevelDB store. +func (s *Storage) RemoveBlock(bitmapID uint64, db string, frame string, slice int, chunk uint64, blockIndex int32) { +} + +// StoreBit sets a bit in storage. +func (s *Storage) StoreBit(bitmapID uint64, db string, frame string, slice int, filter uint64, bchunk uint64, blockIndex int32, bblock, count uint64) { + s.beginBatch() + s.StoreBlock(bitmapID, db, frame, slice, filter, bchunk, blockIndex, bblock) + s.StoreBlock(bitmapID, db, frame, slice, filter, index.CounterMask, 0, count) + s.endBatch() +} + +// RemoveBlock unsets a bit in storage. This is a no-op in the LevelDB store. +func (s *Storage) RemoveBit(bitmapID uint64, db string, frame string, slice int, filter uint64, bchunk uint64, blockIndex int32, count uint64) { +} + +// marshalKey encodes the id, chunk key, & block index to a byte slice. +func marshalKey(id, key uint64, index uint8) []byte { + buf := new(bytes.Buffer) + binary.Write(buf, binary.LittleEndian, id) + binary.Write(buf, binary.LittleEndian, key) + binary.Write(buf, binary.LittleEndian, index) + return buf.Bytes() +} + +// marshalValue encodes the block number and filter to a byte slice. +func marshalValue(block, filter uint64) []byte { + buf := new(bytes.Buffer) + binary.Write(buf, binary.LittleEndian, block) + binary.Write(buf, binary.LittleEndian, filter) + return buf.Bytes() +} + +// unmarshalKey decodes the id, chunk key, & block index from a byte slice. +func unmarshalKey(v []byte) (id, key uint64, index uint8) { + buf := bytes.NewReader(v) + binary.Read(buf, binary.LittleEndian, &id) + binary.Read(buf, binary.LittleEndian, &key) + binary.Read(buf, binary.LittleEndian, &index) + return +} + +// unmarshalValue decodes the block number and filter from a byte slice. +func unmarshalValue(v []byte) (block, filter uint64) { + buf := bytes.NewReader(v) + binary.Read(buf, binary.LittleEndian, &block) + binary.Read(buf, binary.LittleEndian, &filter) + return +} diff --git a/index/storage/leveldb/leveldb_test.go b/index/storage/leveldb/leveldb_test.go new file mode 100644 index 000000000..c55cc53e5 --- /dev/null +++ b/index/storage/leveldb/leveldb_test.go @@ -0,0 +1,55 @@ +package leveldb_test + +import ( + "io/ioutil" + "os" + "testing" + + "github.com/umbel/pilosa/index" + "github.com/umbel/pilosa/index/storage/leveldb" + "github.com/umbel/pilosa/util" +) + +func init() { + util.SetupStatsd() +} + +func TestStorage_Fetch(t *testing.T) { + s := MustOpenStorage() + defer s.Close() +} + +// Storage represents a test wrapper for leveldb.Storage. +type Storage struct { + *leveldb.Storage +} + +// NewStorage returns a new instance of Storage with a temporary path. +func NewStorage() *Storage { + // Create temporary path. + f, err := ioutil.TempFile("", "pilosa-leveldb-") + if err != nil { + panic(err) + } + f.Close() + os.Remove(f.Name()) + + return &Storage{leveldb.NewStorage(index.StorageOptions{ + LevelDBPath: f.Name(), + })} +} + +// MustOpenStorage returns a new, opened instance of Storage. +func MustOpenStorage() *Storage { + s := NewStorage() + if err := s.Open(); err != nil { + panic(err) + } + return s +} + +// Close closes the storage and removes the underlying data file. +func (s *Storage) Close() error { + defer os.Remove(s.Path()) + return s.Storage.Close() +} diff --git a/index/storage/mem/mem.go b/index/storage/mem/mem.go new file mode 100644 index 000000000..5924cf408 --- /dev/null +++ b/index/storage/mem/mem.go @@ -0,0 +1,79 @@ +package mem + +import ( + "fmt" + + "github.com/umbel/pilosa/index" +) + +func init() { + index.RegisterStorage("memory", + func(opt index.StorageOptions) index.Storage { + return NewStorage() + }, + ) +} + +// Storage represents in-memory bitmap storage. +type Storage struct { + db map[string]*index.Bitmap +} + +// NewStorage returns a new instance of Storage. +func NewStorage() *Storage { + return &Storage{ + db: make(map[string]*index.Bitmap), + } +} + +// Open initializes the storage. +func (c *Storage) Open() error { return nil } + +// Close shuts down the storage. +func (c *Storage) Close() error { return nil } + +// FlushBatch flushes the batch to disk. This is a no-op for in-memory storage. +func (c *Storage) Flush() {} + +// Fetch retrieves a bitmap by id. +func (c *Storage) Fetch(id uint64, db, frame string, slice int) (*index.Bitmap, uint64) { + key := fmt.Sprintf("%d:%s:%s:%d", id, db, frame, slice) + + // Find bitmap by key. + b, ok := c.db[key] + if ok { + return b, 0 + } + + // If the bitmap doesn't exist then create a new one. + b = index.NewBitmap() + c.db[key] = b + return b, 0 +} + +// Store saves a bitmap to storage. +// This is a no-op for in-memory storage because changes are stored in the cache. +func (c *Storage) Store(id uint64, db, frame string, slice int, filter uint64, b *index.Bitmap) error { + return nil +} + +// StoreBlock saves a block to storage. +// This is a no-op for in-memory storage because changes are stored in the cache. +func (c *Storage) StoreBlock(id uint64, db, frame string, slice int, filter uint64, chunk_key uint64, block_index int32, block uint64) error { + return nil +} + +// RemoveBlock deletes a block from bitmap. +// This is a no-op for in-memory storage because changes are stored in the cache. +func (self *Storage) RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) { +} + +// StoreBit sets a bit in a bitmap. +// This is a no-op for in-memory storage because changes are stored in the cache. +func (self *Storage) StoreBit(id uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) { +} + +// RemoveBit unsets a bit in a bitmap. +// This is a no-op for in-memory storage because changes are stored in the cache. +func (self *Storage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) { +} diff --git a/index/storage/mem/mem_test.go b/index/storage/mem/mem_test.go new file mode 100644 index 000000000..c524644dc --- /dev/null +++ b/index/storage/mem/mem_test.go @@ -0,0 +1,30 @@ +package mem_test + +import ( + "testing" + + "github.com/umbel/pilosa/index/storage/mem" +) + +// Ensure a bitmap can be retrieved from storage. +func TestStorage_Fetch(t *testing.T) { + s := mem.NewStorage() + + // Retrieving a non-existent bitmap should create a new one. + b, _ := s.Fetch(1, "d", "f", 0) + if b == nil { + t.Fatal("expected bitmap") + } + + // Retrieve the bitmap again. + other, _ := s.Fetch(1, "d", "f", 0) + if b != other { + t.Fatal("expected same bitmap") + } + + // Retrieving a different bitmap should return a different reference. + b2, _ := s.Fetch(2, "d", "f", 0) + if b == b2 { + t.Fatal("expected new bitmap") + } +} diff --git a/index/storage/storage.go b/index/storage/storage.go new file mode 100644 index 000000000..40ae92486 --- /dev/null +++ b/index/storage/storage.go @@ -0,0 +1,7 @@ +package storage + +import ( + _ "github.com/umbel/pilosa/index/storage/cassandra" + _ "github.com/umbel/pilosa/index/storage/leveldb" + _ "github.com/umbel/pilosa/index/storage/mem" +) diff --git a/index/storage_cass.go b/index/storage_cass.go deleted file mode 100644 index 5d522b9c1..000000000 --- a/index/storage_cass.go +++ /dev/null @@ -1,224 +0,0 @@ -package index - -// #cgo CFLAGS:-mpopcnt - -import ( - "time" - - log "github.com/cihub/seelog" - "github.com/gocql/gocql" - "github.com/umbel/pilosa/util" -) - -// DefaultStorageHosts are the hosts that stores the data. -var DefaultStorageHosts = [...]string{"localhost"} - -// DefaultStorageKeyspace is the default keyspace used for storage. -const DefaultStorageKeyspace = "pilosa" - -const DefaultCassandraTimeWindow = 5 * time.Second -const DefaultCassandraMaxSizeBatch = 15 - -var StorageHosts = DefaultStorageHosts[:] -var StorageKeyspace = DefaultStorageKeyspace -var CassandraTimeWindow = DefaultCassandraTimeWindow -var CassandraMaxSizeBatch = DefaultCassandraMaxSizeBatch - -type CassandraStorage struct { - db *gocql.Session - batch *gocql.Batch - stmt string - dstmt string - batch_time time.Time - batch_counter int - cass_time_window_secs float64 - cass_flush_size int -} - -var cluster *gocql.ClusterConfig -var session *gocql.Session - -func SetupCassandra() { - var err error - hosts := StorageHosts - keyspace := StorageKeyspace - cluster = gocql.NewCluster(hosts...) - cluster.Keyspace = keyspace - cluster.Consistency = gocql.One - cluster.Timeout = 5 * time.Second - cluster.RetryPolicy = &gocql.SimpleRetryPolicy{NumRetries: 10} - session, err = cluster.CreateSession() - if err != nil { - log.Warn(err) - } -} - -func BuildSchema() { - /* - "CREATE KEYSPACE IF NOT EXISTS pilosa WITH strategy_class = SimpleStrategy AND strategy_options:replication_factor = 1" - create keyspace if not exists pilosa with replication = {'class': 'SimpleStrategy', 'replication_factor' : 1} and durable_writes = true; - CREATE KEYSPACE pilosa WITH replication = {'class': 'NetworkTopologyStrategy', 'pilpang': '2'} AND durable_writes = true; - - CREATE TABLE IF NOT EXISTS bitmap (bitmap_id bigint, db varchar, frame varchar, slice int, filter int, chunkkey bigint, blockindex int, block bigint, PRIMARY KEY ((bitmap_id, db, frame, slice), chunkkey, blockindex) ) - " - */ - -} - -func NewCassStorage() Storage { - obj := new(CassandraStorage) - // cluster.CQLVersion = "3.0.0" - /*session, err := cluster.CreateSession() - if err != nil { - log.Warn(err) - } - */ - - obj.db = session - obj.stmt = `INSERT INTO bitmap ( bitmap_id, db, frame, slice , filter, ChunkKey, BlockIndex, block) VALUES (?,?,?,?,?,?,?,?) USING timestamp ?;` - obj.dstmt = `DELETE FROM bitmap USING TIMESTAMP ? WHERE bitmap_id=? AND db=? AND frame=? AND slice=? AND chunkkey=? AND blockindex=?` - obj.batch = nil - obj.batch_time = time.Now() - obj.batch_counter = 0 - obj.cass_time_window_secs = float64(CassandraTimeWindow.Seconds()) - obj.cass_flush_size = CassandraMaxSizeBatch - return obj -} - -func (c *CassandraStorage) Close() { -} - -func (c *CassandraStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (*Bitmap, uint64) { - last_key, marker := int64(-1), int64(-1) - var id = util.Uint64ToInt64(bitmap_id) - start := time.Now() - var ( - chunk *Chunk - chunk_key, block int64 - block_index uint32 - s8 uint8 - filter int - ) - - bitmap := NewBitmap() - iter := c.db.Query("SELECT filter,Chunkkey,BlockIndex,block FROM bitmap WHERE bitmap_id=? AND db=? AND frame=? AND slice=? ", id, db, frame, slice).Iter() - count := int64(0) - - for iter.Scan(&filter, &chunk_key, &block_index, &block) { - s8 = uint8(block_index) - if chunk_key != marker { - if chunk_key != last_key { - chunk = &Chunk{uint64(chunk_key), make(Blocks, 32)} - bitmap.AddChunk(chunk) - } - chunk.Value[s8] = uint64(block) - - } else { - count = block - } - last_key = chunk_key - - } - delta := time.Since(start) - util.SendTimer("cassandra_storage_Fetch", delta.Nanoseconds()) - util.SendInc("cassandra_storage_Read") - bitmap.SetCount(uint64(count)) - return bitmap, uint64(filter) -} -func (self *CassandraStorage) BeginBatch() { - if self.batch == nil { - self.batch = gocql.NewBatch(gocql.UnloggedBatch) - } - self.batch_counter++ -} -func (self *CassandraStorage) runBatch(batch *gocql.Batch) { - if batch != nil { - err := self.db.ExecuteBatch(batch) - if err != nil { - log.Warn("Batch ERROR: ", err) - } - } -} -func (self *CassandraStorage) FlushBatch() { - start := time.Now() - self.runBatch(self.batch) //maybe this is crazy but i'll give it a whirl - self.batch = nil - self.batch_time = time.Now() - self.batch_counter = 0 - delta := time.Since(start) - util.SendTimer("cassandra_storage_FlushBatch", delta.Nanoseconds()) -} -func (self *CassandraStorage) EndBatch() { - start := time.Now() - if self.batch != nil { - self.FlushBatch() - } else { - log.Warn("NIL BATCH") - } - delta := time.Since(start) - util.SendTimer("cassandra_storage_EndBatch", delta.Nanoseconds()) - util.SendInc("cassandra_storage_Write") - -} - -func (self *CassandraStorage) Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error { - self.BeginBatch() - for i := bitmap.ChunkIterator(); !i.Limit(); i = i.Next() { - var chunk = i.Item() - for idx, block := range chunk.Value { - block_index := int32(idx) - if block != 0 { - self.StoreBlock(id, db, frame, slice, filter, chunk.Key, block_index, block) - } - } - } - cnt := bitmap.BitCount() - self.StoreBlock(id, db, frame, slice, filter, CounterMask, 0, cnt) - self.EndBatch() - return nil -} - -func (self *CassandraStorage) StoreBlock(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock uint64) error { - id := util.Uint64ToInt64(bid) //these fucntions ignore overflow - block := util.Uint64ToInt64(bblock) - chunk := util.Uint64ToInt64(bchunk) - - if self.batch == nil { - self.BeginBatch() - } - start := time.Now() - self.batch.Query(self.stmt, id, db, frame, slice, int(filter), chunk, block_index, block, start.UnixNano()) - delta := time.Since(start) - util.SendTimer("cassandra_storage_StoreBlock", delta.Nanoseconds()) - return nil -} - -func (self *CassandraStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, val, count uint64) { - self.BeginBatch() - self.StoreBlock(bid, db, frame, slice, filter, chunk, block_index, val) - self.StoreBlock(bid, db, frame, slice, filter, CounterMask, 0, count) - self.EndBatch() -} - -func (self *CassandraStorage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) { - log.Trace("RemoveBit", id, db, frame, slice, chunk, block_index) - self.BeginBatch() - self.RemoveBlock(id, db, frame, slice, chunk, block_index) - self.StoreBlock(id, db, frame, slice, filter, CounterMask, 0, count) - self.EndBatch() -} - -func (self *CassandraStorage) RemoveBlock(bid uint64, db string, frame string, slice int, bchunk uint64, block_index int32) { - log.Trace("RemoveBBlock", bid, db, frame, slice, bchunk, block_index) - id := util.Uint64ToInt64(bid) //these fucntions ignore overflow - chunk := util.Uint64ToInt64(bchunk) - - if self.batch == nil { - self.BeginBatch() - } - start := time.Now() - - self.batch.Query(self.dstmt, start.UnixNano(), id, db, frame, slice, chunk, block_index) - delta := time.Since(start) - util.SendTimer("cassandra_storage_DeleteBlock", delta.Nanoseconds()) -} diff --git a/index/storage_leveldb.go b/index/storage_leveldb.go deleted file mode 100644 index eb142f9f0..000000000 --- a/index/storage_leveldb.go +++ /dev/null @@ -1,188 +0,0 @@ -package index - -// #cgo CFLAGS:-mpopcnt - -import ( - "bytes" - "encoding/binary" - "log" - "time" - - "github.com/syndtr/goleveldb/leveldb" - . "github.com/syndtr/goleveldb/leveldb/util" - "github.com/umbel/pilosa/util" -) - -type LevelDBStorage struct { - db *leveldb.DB - batch *leveldb.Batch - batch_time time.Time - batch_counter int -} - -//go get github.com/syndtr/goleveldb/leveldb -func NewLevelDBStorage(file_path string) Storage { - obj := new(LevelDBStorage) - db, _ := leveldb.OpenFile(file_path, nil) - obj.db = db - obj.batch = nil - obj.batch_counter = 0 - obj.batch_time = time.Now().Add(-time.Hour) - return obj -} -func encodeKey(id, chunk_key uint64, block_index uint8) []byte { - buf := new(bytes.Buffer) - binary.Write(buf, binary.LittleEndian, id) - binary.Write(buf, binary.LittleEndian, chunk_key) - binary.Write(buf, binary.LittleEndian, block_index) - return buf.Bytes() -} - -func encodeValue(block, filter uint64) []byte { - buf := new(bytes.Buffer) - binary.Write(buf, binary.LittleEndian, block) - binary.Write(buf, binary.LittleEndian, filter) - return buf.Bytes() -} - -func decodeKey(key []byte) (uint64, uint64, uint8) { - var ( - id, chunk_key uint64 - block_index uint8 - ) - buf := bytes.NewReader(key) - binary.Read(buf, binary.LittleEndian, &id) - binary.Read(buf, binary.LittleEndian, &chunk_key) - binary.Read(buf, binary.LittleEndian, &block_index) - return id, chunk_key, block_index -} - -func decodeValue(value []byte) (uint64, uint64) { - var ( - block, filter uint64 - ) - buf := bytes.NewReader(value) - binary.Read(buf, binary.LittleEndian, &block) - binary.Read(buf, binary.LittleEndian, &filter) - return block, filter -} - -func (self *LevelDBStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (*Bitmap, uint64) { - start := time.Now() - var ( - chunk *Chunk - filter, block, last_key uint64 - ) - - bitmap := NewBitmap() - count := uint64(0) - start_key := encodeKey(bitmap_id, 0, 0) - limit_key := encodeKey(bitmap_id+1, 0, 0) - iter := self.db.NewIterator(&Range{Start: start_key, Limit: limit_key}, nil) - last_key = CounterMask - for iter.Next() { - _, chunk_key, block_index := decodeKey(iter.Key()) - block, filter = decodeValue(iter.Value()) - if chunk_key != CounterMask { - if chunk_key != last_key { - chunk = &Chunk{chunk_key, make(Blocks, 32)} - bitmap.AddChunk(chunk) - } - chunk.Value[block_index] = block - - } else { - count = block - } - last_key = chunk_key - } - iter.Release() - - delta := time.Since(start) - util.SendTimer("leveldb_storage_Fetch", delta.Nanoseconds()) - bitmap.SetCount(uint64(count)) - return bitmap, uint64(filter) -} - -func (self *LevelDBStorage) BeginBatch() { - if self.batch == nil { - self.batch = new(leveldb.Batch) - } - self.batch_counter++ -} -func (self *LevelDBStorage) runBatch(batch *leveldb.Batch) { - if batch != nil { - self.db.Write(batch, nil) - } -} -func (self *LevelDBStorage) FlushBatch() { - start := time.Now() - self.runBatch(self.batch) //maybe this is crazy but i'll give it a whirl - self.batch = nil - self.batch_time = time.Now() - self.batch_counter = 0 - delta := time.Since(start) - util.SendTimer("leveldb_storage_FlushBatch", delta.Nanoseconds()) -} -func (self *LevelDBStorage) EndBatch() { - start := time.Now() - if self.batch != nil { - last := time.Since(self.batch_time) - if last*time.Second > 15 { - self.FlushBatch() - } else if self.batch_counter > 300 { - self.FlushBatch() - } - } else { - log.Println("NIL BATCH") - } - delta := time.Since(start) - util.SendTimer("leveldb_storage_EndBatch", delta.Nanoseconds()) - -} - -func (self *LevelDBStorage) Store(id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error { - self.BeginBatch() - for i := bitmap.ChunkIterator(); !i.Limit(); i = i.Next() { - var chunk = i.Item() - for idx, block := range chunk.Value { - block_index := int32(idx) - if block != 0 { - self.StoreBlock(id, db, frame, slice, filter, chunk.Key, block_index, block) - } - } - } - - self.StoreBlock(id, db, frame, slice, filter, CounterMask, 0, bitmap.BitCount()) - self.EndBatch() - return nil -} - -func (self *LevelDBStorage) StoreBlock(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, block uint64) error { - if self.batch == nil { - panic("NIL BATCH") - } - start := time.Now() - self.batch.Put(encodeKey(id, chunk, uint8(block_index)), encodeValue(block, filter)) - delta := time.Since(start) - util.SendTimer("leveldb_storage_StoreBlock", delta.Nanoseconds()) - return nil -} - -func (self *LevelDBStorage) RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) { -} - -func (self *LevelDBStorage) RemoveBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, count uint64) { -} - -func (self *LevelDBStorage) Close() { - self.FlushBatch() - self.db.Close() -} - -func (self *LevelDBStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) { - self.BeginBatch() - self.StoreBlock(bid, db, frame, slice, filter, bchunk, block_index, bblock) - self.StoreBlock(bid, db, frame, slice, filter, CounterMask, 0, count) - self.EndBatch() - -} diff --git a/index/storage_mem.go b/index/storage_mem.go deleted file mode 100644 index a850ba15f..000000000 --- a/index/storage_mem.go +++ /dev/null @@ -1,53 +0,0 @@ -package index - -// #cgo CFLAGS:-mpopcnt - -import "fmt" - -type MemoryStorage struct { - db map[string]*Bitmap -} - -func NewMemoryStorage() Storage { - obj := new(MemoryStorage) - obj.db = make(map[string]*Bitmap) - - return obj -} - -func (c *MemoryStorage) BeginBatch() {} - -func (c *MemoryStorage) Close() {} - -func (c *MemoryStorage) EndBatch() {} - -func (c *MemoryStorage) FlushBatch() {} - -func (c *MemoryStorage) Fetch(bitmap_id uint64, db string, frame string, slice int) (*Bitmap, uint64) { - key := fmt.Sprintf("%d:%s:%s:%d", bitmap_id, db, frame, slice) - bitmap, found := c.db[key] - if !found { - bitmap = NewBitmap() - c.db[key] = bitmap - } - return bitmap, 0 -} - -func (c *MemoryStorage) Store(bitmap_id uint64, db string, frame string, slice int, filter uint64, bitmap *Bitmap) error { - //only use the cache and throw away everything - return nil -} - -func (c *MemoryStorage) StoreBlock(bitmap_id uint64, db string, frame string, slice int, filter uint64, chunk_key uint64, block_index int32, block uint64) error { - //only use the cache and throw away everything - return nil -} - -func (self *MemoryStorage) StoreBit(bid uint64, db string, frame string, slice int, filter uint64, bchunk uint64, block_index int32, bblock, count uint64) { -} - -func (self *MemoryStorage) RemoveBit(id uint64, db string, frame string, slice int, filter uint64, chunk uint64, block_index int32, count uint64) { -} - -func (self *MemoryStorage) RemoveBlock(id uint64, db string, frame string, slice int, chunk uint64, block_index int32) { -} diff --git a/index/storage_test.go b/index/storage_test.go index bc9f0775d..236d9812e 100644 --- a/index/storage_test.go +++ b/index/storage_test.go @@ -56,7 +56,7 @@ func TestStorage(t *testing.T) { // SetBit(bm, 2) // fmt.Println("STORE") // storage.Store(int64(bitmap_id), db, frame, slice, uint64(filter), bm.(*Bitmap)) - // //storage.FlushBatch() + // //storage.Flush() // fmt.Println("FETCH") // bm2, _ := storage.Fetch(bitmap_id, db, frame, slice) // So(BitCount(bm), ShouldEqual, BitCount(bm2)) diff --git a/util/constants.go b/util/constants.go index 1f64bcd05..c7eea49bf 100644 --- a/util/constants.go +++ b/util/constants.go @@ -29,9 +29,3 @@ func ByteToInt64(data []byte) int64 { binary.Read(buf, binary.BigEndian, &value) return value } -func Uint64ToInt64(before uint64) int64 { - return int64(before) -} -func Int64ToUint64(before int64) uint64 { - return uint64(before) -}