package raftbbolt import ( "errors" "git.giftfish.de/ston1th/raftbbolt/msgpack" "github.com/hashicorp/raft" "go.etcd.io/bbolt" ) const ( // Permissions to use on the db file. This is only used if the // database file does not exist and needs to be created. dbFileMode = 0600 ) var ( // Bucket names we perform transactions in dbLogs = []byte("logs") dbConf = []byte("conf") // An error indicating a given key does not exist ErrKeyNotFound = errors.New("not found") ) // Store provides access to BoltDB for Raft to store and retrieve // log entries. It also provides key/value storage, and can be used as // a LogStore and StableStore. type Store struct { // conn is the underlying handle to the db. db *bbolt.DB // The path to the Bolt database file path string } // Options contains all the configuration used to open the BoltDB type Options struct { // Path is the file path to the BoltDB to use Path string // BoltOptions contains any specific BoltDB options you might // want to specify [e.g. open timeout] BoltOptions *bbolt.Options // NoSync causes the database to skip fsync calls after each // write to the log. This is unsafe, so it should be used // with caution. NoSync bool } // readOnly returns true if the contained bolt options say to open // the DB in readOnly mode [this can be useful to tools that want // to examine the log] func (o *Options) readOnly() bool { return o != nil && o.BoltOptions != nil && o.BoltOptions.ReadOnly } // NewStore takes a file path and returns a connected Raft backend. func NewStore(path string) (*Store, error) { return New(Options{Path: path}) } // New uses the supplied options to open the BoltDB and prepare it for use as a raft backend. func New(options Options) (*Store, error) { db, err := bbolt.Open(options.Path, dbFileMode, options.BoltOptions) if err != nil { return nil, err } db.NoSync = options.NoSync // Create the new store store := &Store{ db: db, path: options.Path, } // If the store was opened read-only, don't try and create buckets if !options.readOnly() { // Set up our buckets if err := store.initialize(); err != nil { store.Close() return nil, err } } return store, nil } // initialize is used to set up all of the buckets. func (s *Store) initialize() error { return s.db.Update(func(tx *bbolt.Tx) error { if _, err := tx.CreateBucketIfNotExists(dbLogs); err != nil { return err } _, err := tx.CreateBucketIfNotExists(dbConf) return err }) } // Close is used to gracefully close the DB connection. func (s *Store) Close() error { return s.db.Close() } // FirstIndex returns the first known index from the Raft log. func (s *Store) FirstIndex() (n uint64, err error) { err = s.db.View(func(tx *bbolt.Tx) error { c := tx.Bucket(dbLogs).Cursor() first, _ := c.First() if first == nil { return nil } n = bytesToUint64(first) return nil }) return } // LastIndex returns the last known index from the Raft log. func (s *Store) LastIndex() (n uint64, err error) { err = s.db.View(func(tx *bbolt.Tx) error { c := tx.Bucket(dbLogs).Cursor() last, _ := c.Last() if last == nil { return nil } n = bytesToUint64(last) return nil }) return } // GetLog is used to retrieve a log from BoltDB at a given index. func (s *Store) GetLog(idx uint64, log *raft.Log) error { return s.db.View(func(tx *bbolt.Tx) error { b := tx.Bucket(dbLogs).Get(uint64ToBytes(idx)) if b == nil { return raft.ErrLogNotFound } return msgpack.Unmarshal(b, log) }) } // StoreLog is used to store a single raft log func (s *Store) StoreLog(log *raft.Log) error { return s.StoreLogs([]*raft.Log{log}) } // StoreLogs is used to store a set of raft logs func (s *Store) StoreLogs(logs []*raft.Log) error { return s.db.Update(func(tx *bbolt.Tx) error { bucket := tx.Bucket(dbLogs) for _, log := range logs { b, err := msgpack.Marshal(log) if err != nil { return err } if err := bucket.Put(uint64ToBytes(log.Index), b); err != nil { return err } } return nil }) } // DeleteRange is used to delete logs within a given range inclusively. func (s *Store) DeleteRange(min, max uint64) error { return s.db.Update(func(tx *bbolt.Tx) error { c := tx.Bucket(dbLogs).Cursor() minKey := uint64ToBytes(min) for k, _ := c.Seek(minKey); k != nil; k, _ = c.Next() { // Handle out-of-range log index if bytesToUint64(k) > max { break } // Delete in-range log index if err := c.Delete(); err != nil { return err } } return nil }) } // Set is used to set a key/value set outside of the raft log func (s *Store) Set(k, v []byte) error { return s.db.Update(func(tx *bbolt.Tx) error { return tx.Bucket(dbConf).Put(k, v) }) } // Get is used to retrieve a value from the k/v store by key func (s *Store) Get(k []byte) (b []byte, err error) { err = s.db.View(func(tx *bbolt.Tx) error { b = tx.Bucket(dbConf).Get(k) if b == nil { return ErrKeyNotFound } return nil }) return } // SetUint64 is like Set, but handles uint64 values func (s *Store) SetUint64(k []byte, v uint64) error { return s.Set(k, uint64ToBytes(v)) } // GetUint64 is like Get, but handles uint64 values func (s *Store) GetUint64(k []byte) (uint64, error) { b, err := s.Get(k) if err != nil { return 0, err } return bytesToUint64(b), nil } // Sync performs an fsync on the database file handle. This is not necessary // under normal operation unless NoSync is enabled, in which this forces the // database file to sync against the disk. func (s *Store) Sync() error { return s.db.Sync() }