Skip to content

Commit

Permalink
Merge pull request #22 from hashicorp/boltmig
Browse files Browse the repository at this point in the history
Upgrade to etcd/bbolt
  • Loading branch information
s-christoff committed Apr 9, 2021
2 parents 4207f1b + fba1f5c commit 03c10cc
Show file tree
Hide file tree
Showing 9 changed files with 864 additions and 2 deletions.
1 change: 0 additions & 1 deletion go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -6,5 +6,4 @@ require (
github.com/boltdb/bolt v1.3.1
github.com/hashicorp/go-msgpack v0.5.5
github.com/hashicorp/raft v1.1.0
golang.org/x/sys v0.0.0-20190602015325-4c4f7f33c9ed // indirect
)
2 changes: 1 addition & 1 deletion go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -42,4 +42,4 @@ github.com/tv42/httpunix v0.0.0-20150427012821-b75d8614f926/go.mod h1:9ESjWnEqri
golang.org/x/net v0.0.0-20181201002055-351d144fa1fc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sys v0.0.0-20190602015325-4c4f7f33c9ed h1:uPxWBzB3+mlnjy9W58qY1j/cjyFjutgw/Vhan2zLy/A=
golang.org/x/sys v0.0.0-20190602015325-4c4f7f33c9ed/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20190602015325-4c4f7f33c9ed/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
6 changes: 6 additions & 0 deletions v2/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
raft-boltdb/v2
===========

This implementation uses the maintained version of BoltDB, [BBolt](https://github.com/etcd-io/bbolt). This is the primary version of `raft-boltdb` and should be used whenever possible.

There is no breaking API change to the library. However, there is the potential for disk format incompatibilities so it was decided to be conservative and making it a separate import path. This separate import path will allow both versions (original and v2) to be imported to perform a safe in-place upgrade of old files read with the old version and written back out with the new one.
88 changes: 88 additions & 0 deletions v2/bench_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
package raftboltdb

import (
"os"
"testing"

"github.com/hashicorp/raft/bench"
)

func BenchmarkBoltStore_FirstIndex(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.FirstIndex(b, store)
}

func BenchmarkBoltStore_LastIndex(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.LastIndex(b, store)
}

func BenchmarkBoltStore_GetLog(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.GetLog(b, store)
}

func BenchmarkBoltStore_StoreLog(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.StoreLog(b, store)
}

func BenchmarkBoltStore_StoreLogs(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.StoreLogs(b, store)
}

func BenchmarkBoltStore_DeleteRange(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.DeleteRange(b, store)
}

func BenchmarkBoltStore_Set(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.Set(b, store)
}

func BenchmarkBoltStore_Get(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.Get(b, store)
}

func BenchmarkBoltStore_SetUint64(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.SetUint64(b, store)
}

func BenchmarkBoltStore_GetUint64(b *testing.B) {
store := testBoltStore(b)
defer store.Close()
defer os.Remove(store.path)

raftbench.GetUint64(b, store)
}
268 changes: 268 additions & 0 deletions v2/bolt_store.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,268 @@
package raftboltdb

import (
"errors"

"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")
)

// BoltStore provides access to Bbolt for Raft to store and retrieve
// log entries. It also provides key/value storage, and can be used as
// a LogStore and StableStore.
type BoltStore struct {
// conn is the underlying handle to the db.
conn *bbolt.DB

// The path to the Bolt database file
path string
}

// Options contains all the configuration used to open the Bbolt
type Options struct {
// Path is the file path to the Bbolt to use
Path string

// BoltOptions contains any specific Bbolt 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
}

// NewBoltStore takes a file path and returns a connected Raft backend.
func NewBoltStore(path string) (*BoltStore, error) {
return New(Options{Path: path})
}

// New uses the supplied options to open the Bbolt and prepare it for use as a raft backend.
func New(options Options) (*BoltStore, error) {
// Try to connect
handle, err := bbolt.Open(options.Path, dbFileMode, options.BoltOptions)
if err != nil {
return nil, err
}
handle.NoSync = options.NoSync

// Create the new store
store := &BoltStore{
conn: handle,
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 (b *BoltStore) initialize() error {
tx, err := b.conn.Begin(true)
if err != nil {
return err
}
defer tx.Rollback()

// Create all the buckets
if _, err := tx.CreateBucketIfNotExists(dbLogs); err != nil {
return err
}
if _, err := tx.CreateBucketIfNotExists(dbConf); err != nil {
return err
}

return tx.Commit()
}

// Close is used to gracefully close the DB connection.
func (b *BoltStore) Close() error {
return b.conn.Close()
}

// FirstIndex returns the first known index from the Raft log.
func (b *BoltStore) FirstIndex() (uint64, error) {
tx, err := b.conn.Begin(false)
if err != nil {
return 0, err
}
defer tx.Rollback()

curs := tx.Bucket(dbLogs).Cursor()
if first, _ := curs.First(); first == nil {
return 0, nil
} else {
return bytesToUint64(first), nil
}
}

// LastIndex returns the last known index from the Raft log.
func (b *BoltStore) LastIndex() (uint64, error) {
tx, err := b.conn.Begin(false)
if err != nil {
return 0, err
}
defer tx.Rollback()

curs := tx.Bucket(dbLogs).Cursor()
if last, _ := curs.Last(); last == nil {
return 0, nil
} else {
return bytesToUint64(last), nil
}
}

// GetLog is used to retrieve a log from Bbolt at a given index.
func (b *BoltStore) GetLog(idx uint64, log *raft.Log) error {
tx, err := b.conn.Begin(false)
if err != nil {
return err
}
defer tx.Rollback()

bucket := tx.Bucket(dbLogs)
val := bucket.Get(uint64ToBytes(idx))

if val == nil {
return raft.ErrLogNotFound
}
return decodeMsgPack(val, log)
}

// StoreLog is used to store a single raft log
func (b *BoltStore) StoreLog(log *raft.Log) error {
return b.StoreLogs([]*raft.Log{log})
}

// StoreLogs is used to store a set of raft logs
func (b *BoltStore) StoreLogs(logs []*raft.Log) error {
tx, err := b.conn.Begin(true)
if err != nil {
return err
}
defer tx.Rollback()

for _, log := range logs {
key := uint64ToBytes(log.Index)
val, err := encodeMsgPack(log)
if err != nil {
return err
}
bucket := tx.Bucket(dbLogs)
if err := bucket.Put(key, val.Bytes()); err != nil {
return err
}
}

return tx.Commit()
}

// DeleteRange is used to delete logs within a given range inclusively.
func (b *BoltStore) DeleteRange(min, max uint64) error {
minKey := uint64ToBytes(min)

tx, err := b.conn.Begin(true)
if err != nil {
return err
}
defer tx.Rollback()

curs := tx.Bucket(dbLogs).Cursor()
for k, _ := curs.Seek(minKey); k != nil; k, _ = curs.Next() {
// Handle out-of-range log index
if bytesToUint64(k) > max {
break
}

// Delete in-range log index
if err := curs.Delete(); err != nil {
return err
}
}

return tx.Commit()
}

// Set is used to set a key/value set outside of the raft log
func (b *BoltStore) Set(k, v []byte) error {
tx, err := b.conn.Begin(true)
if err != nil {
return err
}
defer tx.Rollback()

bucket := tx.Bucket(dbConf)
if err := bucket.Put(k, v); err != nil {
return err
}

return tx.Commit()
}

// Get is used to retrieve a value from the k/v store by key
func (b *BoltStore) Get(k []byte) ([]byte, error) {
tx, err := b.conn.Begin(false)
if err != nil {
return nil, err
}
defer tx.Rollback()

bucket := tx.Bucket(dbConf)
val := bucket.Get(k)

if val == nil {
return nil, ErrKeyNotFound
}
return append([]byte(nil), val...), nil
}

// SetUint64 is like Set, but handles uint64 values
func (b *BoltStore) SetUint64(key []byte, val uint64) error {
return b.Set(key, uint64ToBytes(val))
}

// GetUint64 is like Get, but handles uint64 values
func (b *BoltStore) GetUint64(key []byte) (uint64, error) {
val, err := b.Get(key)
if err != nil {
return 0, err
}
return bytesToUint64(val), 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 (b *BoltStore) Sync() error {
return b.conn.Sync()
}
Loading

0 comments on commit 03c10cc

Please sign in to comment.