- Fix boltdb issues

Signed-off-by: Valery Piashchynski <[email protected]>

5 files changed, 261 insertions, 17 deletions

diff --git a/plugins/kv/drivers/boltdb/config.go b/plugins/kv/drivers/boltdb/config.go
index ebe73c25..0beb209b 100644
--- a/plugins/kv/drivers/boltdb/config.go
+++ b/plugins/kv/drivers/boltdb/config.go
@@ -1,34 +1,27 @@
 package boltdb
 
 type Config struct {
-	// Dir is a directory to store the DB files
-	Dir string
 	// File is boltDB file. No need to create it by your own,
 	// boltdb driver is able to create the file, or read existing
 	File string
 	// Bucket to store data in boltDB
-	Bucket string
+	bucket string
 	// db file permissions
 	Permissions int
 	// timeout
-	Interval uint `mapstructure:"interval"`
+	Interval int `mapstructure:"interval"`
 }
 
 // InitDefaults initializes default values for the boltdb
 func (s *Config) InitDefaults() {
-	if s.Dir == "" {
-		s.Dir = "." // current dir
-	}
-	if s.Bucket == "" {
-		s.Bucket = "rr" // default bucket name
-	}
+	s.bucket = "default"
 
 	if s.File == "" {
 		s.File = "rr.db" // default file name
 	}
 
 	if s.Permissions == 0 {
-		s.Permissions = 777 // free for all
+		s.Permissions = 0777 // free for all
 	}
 
 	if s.Interval == 0 {
diff --git a/plugins/kv/drivers/boltdb/driver.go b/plugins/kv/drivers/boltdb/driver.go
index 253b9d33..5f4d98b1 100644
--- a/plugins/kv/drivers/boltdb/driver.go
+++ b/plugins/kv/drivers/boltdb/driver.go
@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"encoding/gob"
 	"os"
-	"path"
 	"strings"
 	"sync"
 	"time"
@@ -47,14 +46,21 @@ func NewBoltDBDriver(log logger.Logger, key string, cfgPlugin config.Configurer,
 		return nil, errors.E(op, err)
 	}
 
-	d.bucket = []byte(d.cfg.Bucket)
+	// add default values
+	d.cfg.InitDefaults()
+
+	d.bucket = []byte(d.cfg.bucket)
 	d.timeout = time.Duration(d.cfg.Interval) * time.Second
 	d.gc = sync.Map{}
 
-	// add default values
-	d.cfg.InitDefaults()
+	db, err := bolt.Open(d.cfg.File, os.FileMode(d.cfg.Permissions), &bolt.Options{
+		Timeout:        time.Second * 20,
+		NoGrowSync:     false,
+		NoFreelistSync: false,
+		ReadOnly:       false,
+		NoSync:         false,
+	})
 
-	db, err := bolt.Open(path.Join(d.cfg.Dir, d.cfg.File), os.FileMode(d.cfg.Permissions), nil)
 	if err != nil {
 		return nil, errors.E(op, err)
 	}
@@ -65,7 +71,7 @@ func NewBoltDBDriver(log logger.Logger, key string, cfgPlugin config.Configurer,
 	// tx.Commit invokes via the db.Update
 	err = db.Update(func(tx *bolt.Tx) error {
 		const upOp = errors.Op("boltdb_plugin_update")
-		_, err = tx.CreateBucketIfNotExists([]byte(d.cfg.Bucket))
+		_, err = tx.CreateBucketIfNotExists([]byte(d.cfg.bucket))
 		if err != nil {
 			return errors.E(op, upOp)
 		}
diff --git a/tests/plugins/kv/configs/.rr-kv-bolt-no-interval.yaml b/tests/plugins/kv/configs/.rr-kv-bolt-no-interval.yaml
new file mode 100644
index 00000000..f58de3e4
--- /dev/null
+++ b/tests/plugins/kv/configs/.rr-kv-bolt-no-interval.yaml
@@ -0,0 +1,14 @@
+rpc:
+  listen: tcp://127.0.0.1:6001
+
+
+kv:
+  boltdb-south:
+    driver: boltdb
+    file: "rr.db"
+    permissions: 755
+
+  boltdb-africa:
+    driver: boltdb
+    file: "africa.db"
+    permissions: 755
diff --git a/tests/plugins/kv/configs/.rr-kv-bolt-perms.yaml b/tests/plugins/kv/configs/.rr-kv-bolt-perms.yaml
new file mode 100644
index 00000000..08b3bfad
--- /dev/null
+++ b/tests/plugins/kv/configs/.rr-kv-bolt-perms.yaml
@@ -0,0 +1,13 @@
+rpc:
+  listen: tcp://127.0.0.1:6001
+
+
+kv:
+  boltdb-south:
+    driver: boltdb
+    file: "rr.db"
+
+  boltdb-africa:
+    driver: boltdb
+    file: "africa.db"
+    permissions: 0777
diff --git a/tests/plugins/kv/storage_plugin_test.go b/tests/plugins/kv/storage_plugin_test.go
index e7e7735a..0ded9fbd 100644
--- a/tests/plugins/kv/storage_plugin_test.go
+++ b/tests/plugins/kv/storage_plugin_test.go
@@ -102,6 +102,224 @@ func TestKVInit(t *testing.T) {
 	_ = os.RemoveAll("africa.db")
 }
 
+func TestKVNoInterval(t *testing.T) {
+	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel))
+	assert.NoError(t, err)
+
+	cfg := &config.Viper{
+		Path:   "configs/.rr-kv-bolt-no-interval.yaml",
+		Prefix: "rr",
+	}
+
+	err = cont.RegisterAll(
+		cfg,
+		&boltdb.Plugin{},
+		&rpcPlugin.Plugin{},
+		&logger.ZapLogger{},
+		&kv.Plugin{},
+	)
+	assert.NoError(t, err)
+
+	err = cont.Init()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ch, err := cont.Serve()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	stopCh := make(chan struct{}, 1)
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case e := <-ch:
+				assert.Fail(t, "error", e.Error.Error())
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+			case <-sig:
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+				return
+			case <-stopCh:
+				// timeout
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+				return
+			}
+		}
+	}()
+
+	time.Sleep(time.Second * 1)
+	t.Run("KvSetTest", kvSetTest)
+	t.Run("KvHasTest", kvHasTest)
+
+	stopCh <- struct{}{}
+
+	wg.Wait()
+
+	_ = os.RemoveAll("rr.db")
+	_ = os.RemoveAll("africa.db")
+}
+
+func TestKVCreateToReopenWithPerms(t *testing.T) {
+	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel))
+	assert.NoError(t, err)
+
+	cfg := &config.Viper{
+		Path:   "configs/.rr-kv-bolt-perms.yaml",
+		Prefix: "rr",
+	}
+
+	err = cont.RegisterAll(
+		cfg,
+		&boltdb.Plugin{},
+		&rpcPlugin.Plugin{},
+		&logger.ZapLogger{},
+		&kv.Plugin{},
+	)
+	assert.NoError(t, err)
+
+	err = cont.Init()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ch, err := cont.Serve()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	stopCh := make(chan struct{}, 1)
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case e := <-ch:
+				assert.Fail(t, "error", e.Error.Error())
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+			case <-sig:
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+				return
+			case <-stopCh:
+				// timeout
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+				return
+			}
+		}
+	}()
+
+	time.Sleep(time.Second * 1)
+	stopCh <- struct{}{}
+	wg.Wait()
+}
+
+func TestKVCreateToReopenWithPerms2(t *testing.T) {
+	cont, err := endure.NewContainer(nil, endure.SetLogLevel(endure.ErrorLevel))
+	assert.NoError(t, err)
+
+	cfg := &config.Viper{
+		Path:   "configs/.rr-kv-bolt-perms.yaml",
+		Prefix: "rr",
+	}
+
+	err = cont.RegisterAll(
+		cfg,
+		&boltdb.Plugin{},
+		&rpcPlugin.Plugin{},
+		&logger.ZapLogger{},
+		&kv.Plugin{},
+	)
+	assert.NoError(t, err)
+
+	err = cont.Init()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ch, err := cont.Serve()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGINT, syscall.SIGTERM)
+
+	wg := &sync.WaitGroup{}
+	wg.Add(1)
+
+	stopCh := make(chan struct{}, 1)
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case e := <-ch:
+				assert.Fail(t, "error", e.Error.Error())
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+			case <-sig:
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+				return
+			case <-stopCh:
+				// timeout
+				err = cont.Stop()
+				if err != nil {
+					assert.FailNow(t, "error", err.Error())
+				}
+				return
+			}
+		}
+	}()
+
+	time.Sleep(time.Second * 1)
+	t.Run("KvSetTest", kvSetTest)
+	t.Run("KvHasTest", kvHasTest)
+
+	stopCh <- struct{}{}
+
+	wg.Wait()
+
+	_ = os.RemoveAll("rr.db")
+	_ = os.RemoveAll("africa.db")
+}
+
 func kvSetTest(t *testing.T) {
 	conn, err := net.Dial("tcp", "127.0.0.1:6001")
 	assert.NoError(t, err)