initial commit

store.go

@@ -0,0 +1,223 @@
+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()
+}