Move store to bbolt database

.circleci/config.yml

@@ -29,7 +29,6 @@ jobs:
             sudo -E env "PATH=$PATH" go test ./ipam/
             sudo -E env "PATH=$PATH" go test ./log/
             sudo -E env "PATH=$PATH" go test ./netlink/
-            sudo -E env "PATH=$PATH" go test ./store/
             sudo -E env "PATH=$PATH" go test ./telemetry/
             sudo -E env "PATH=$PATH" go test ./cni/ipam/
             sudo -E env "PATH=$PATH" go test ./cnm/network/

boltwrapper/boltwrapper.go

@@ -0,0 +1,87 @@
+package boltwrapper
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+	"time"
+
+	bolt "go.etcd.io/bbolt"
+)
+
+// Bucket we use in the database. No need for multiple buckets.
+const bucketName = "ACN"
+
+// Write writes a value to a bbolt database. Note that if the value is a struct,
+// only exported fields are written (ie those which start with a capital letter).
+func Write(database *bolt.DB, key string, value interface{}) error {
+	if database == nil {
+		return fmt.Errorf("no database supplied writing %q to bucket %q", key, bucketName)
+	}
+	if value == nil {
+		return fmt.Errorf("no value supplied writing %q to bucket %q in %q", key, bucketName, database.Path())
+	}
+	marshalled, err := json.Marshal(value)
+	if err != nil {
+		return fmt.Errorf("failed to marshal writing %q to bucket %q in %q: %s", key, bucketName, database.Path(), err)
+	}
+
+	return database.Update(func(tx *bolt.Tx) error {
+		bucket, err := tx.CreateBucketIfNotExists([]byte(bucketName))
+		if err != nil {
+			return fmt.Errorf("failed creating bucket %q in %s: %s", bucketName, database.Path(), err)
+		}
+		if err = bucket.Put([]byte(key), marshalled); err != nil {
+			return fmt.Errorf("failed writing key %q in bucket %q in %q: %s", key, bucketName, database.Path(), err)
+		}
+		return nil
+	})
+}
+
+var (
+	ErrNotFound = fmt.Errorf("entry not found in database")
+)
+
+// Read reads a value from a bbolt database. We use an out parameter to return the
+// value to avoid impossible type-assertions. The caller is expected to know the
+// type of value and pass in an appropriately typed variable.
+func Read(database *bolt.DB, key string, value interface{}) error {
+	if database == nil {
+		return fmt.Errorf("no database supplied reading %q from bucket %q", key, bucketName)
+	}
+	if value == nil {
+		return fmt.Errorf("no value supplied reading %q from bucket %q in %q", key, bucketName, database.Path())
+	}
+	var bytes []byte
+	if err := database.View(func(tx *bolt.Tx) error {
+		bucket := tx.Bucket([]byte(bucketName))
+		if bucket == nil {
+			return ErrNotFound
+		}
+		bytes = bucket.Get([]byte(key))
+		if len(bytes) == 0 {
+			return ErrNotFound
+		}
+		return nil
+	}); err != nil {
+		return err
+	}
+	if len(bytes) == 0 {
+		return ErrNotFound
+	}
+	if err := json.Unmarshal(bytes, &value); err != nil {
+		return fmt.Errorf("failed unmarshalling %q in bucket %q of %q: %s", key, bucketName, database.Path(), err)
+	}
+	return nil
+}
+
+// GetModificationTime returns the UTC time of last modified
+func GetModificationTime(file string) (time.Time, error) {
+	info, err := os.Stat(file)
+	if err != nil {
+		log.Printf("os.stat() for file %v failed: %v", file, err)
+		return time.Time{}.UTC(), err
+	}
+	return info.ModTime().UTC(), nil
+}

cni/ipam/plugin/main.go

@@ -26,14 +26,14 @@ func main() {
 		os.Exit(1)
 	}
 
-	if err := ipamPlugin.Plugin.InitializeKeyValueStore(&config); err != nil {
-		fmt.Printf("Failed to initialize key-value store of ipam plugin, err:%v.\n", err)
+	if err := ipamPlugin.Plugin.OpenDatabase(&config); err != nil {
+		fmt.Printf("Failed to open database of ipam plugin, err:%v.\n", err)
 		os.Exit(1)
 	}
 
 	defer func() {
-		if errUninit := ipamPlugin.Plugin.UninitializeKeyValueStore(); errUninit != nil {
-			fmt.Printf("Failed to uninitialize key-value store of ipam plugin, err:%v.\n", err)
+		if errUninit := ipamPlugin.Plugin.CloseDatabase(); errUninit != nil {
+			fmt.Printf("Failed to close database of ipam plugin, err:%v.\n", err)
 		}
 
 		if recover() != nil {

cni/network/plugin/main.go

@@ -87,15 +87,15 @@ func main() {
 
 	netPlugin.SetReportManager(reportManager)
 
-	if err = netPlugin.Plugin.InitializeKeyValueStore(&config); err != nil {
-		log.Printf("Failed to initialize key-value store of network plugin, err:%v.\n", err)
+	if err = netPlugin.Plugin.OpenDatabase(&config); err != nil {
+		log.Printf("Failed to open database of network plugin, err:%v.\n", err)
 		reportPluginError(reportManager, err)
 		os.Exit(1)
 	}
 
 	defer func() {
-		if errUninit := netPlugin.Plugin.UninitializeKeyValueStore(); errUninit != nil {
-			log.Printf("Failed to uninitialize key-value store of network plugin, err:%v.\n", err)
+		if errUninit := netPlugin.Plugin.CloseDatabase(); errUninit != nil {
+			log.Printf("Failed to close database of network plugin, err:%v.\n", err)
 		}
 
 		if recover() != nil {

cni/plugin.go

@@ -6,18 +6,19 @@ package cni
 import (
 	"fmt"
 	"os"
+	"path/filepath"
 	"runtime"
 
 	"github.com/Azure/azure-container-networking/common"
 	"github.com/Azure/azure-container-networking/log"
 	"github.com/Azure/azure-container-networking/platform"
-	"github.com/Azure/azure-container-networking/store"
 
 	cniInvoke "github.com/containernetworking/cni/pkg/invoke"
 	cniSkel "github.com/containernetworking/cni/pkg/skel"
 	cniTypes "github.com/containernetworking/cni/pkg/types"
 	cniTypesCurr "github.com/containernetworking/cni/pkg/types/current"
 	cniVers "github.com/containernetworking/cni/pkg/version"
+	bolt "go.etcd.io/bbolt"
 )
 
 // Plugin is the parent class for CNI plugins.
@@ -126,8 +127,7 @@ func (plugin *Plugin) DelegateDel(pluginName string, nwCfg *NetworkConfig) error
 
 	os.Setenv(Cmd, CmdDel)
 
-	err = cniInvoke.DelegateDel(pluginName, nwCfg.Serialize(), nil)
-	if err != nil {
+	if err := cniInvoke.DelegateDel(pluginName, nwCfg.Serialize(), nil); err != nil {
 		return fmt.Errorf("Failed to delegate: %v", err)
 	}
 
@@ -154,39 +154,29 @@ func (plugin *Plugin) Errorf(format string, args ...interface{}) *cniTypes.Error
 	return plugin.Error(fmt.Errorf(format, args...))
 }
 
-// Initialize key-value store
-func (plugin *Plugin) InitializeKeyValueStore(config *common.PluginConfig) error {
-	// Create the key value store.
-	if plugin.Store == nil {
+// Opens the database
+func (plugin *Plugin) OpenDatabase(config *common.PluginConfig) error {
+	// Create the database
+	if plugin.Database == nil {
 		var err error
-		plugin.Store, err = store.NewJsonFileStore(platform.CNIRuntimePath + plugin.Name + ".json")
+		plugin.Database, err = bolt.Open(filepath.Join(platform.CNIRuntimePath, plugin.Name)+".db", 0600, nil)
 		if err != nil {
-			log.Printf("[cni] Failed to create store: %v.", err)
+			log.Printf("[cni] Failed to create database: %v.", err)
 			return err
 		}
 	}
 
-	// Acquire store lock.
-	if err := plugin.Store.Lock(true); err != nil {
-		log.Printf("[cni] Failed to lock store: %v.", err)
-		return err
-	}
-
-	config.Store = plugin.Store
+	config.Database = plugin.Database
 
 	return nil
 }
 
-// Uninitialize key-value store
-func (plugin *Plugin) UninitializeKeyValueStore() error {
-	if plugin.Store != nil {
-		err := plugin.Store.Unlock()
-		if err != nil {
-			log.Printf("[cni] Failed to unlock store: %v.", err)
-			return err
-		}
+// Closes the database
+func (plugin *Plugin) CloseDatabase() error {
+	if plugin.Database != nil {
+		plugin.Database.Close()
 	}
-	plugin.Store = nil
+	plugin.Database = nil
 
 	return nil
 }

cnm/plugin/main.go

@@ -7,14 +7,15 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"path/filepath"
 	"syscall"
 
 	"github.com/Azure/azure-container-networking/cnm/ipam"
 	"github.com/Azure/azure-container-networking/cnm/network"
 	"github.com/Azure/azure-container-networking/common"
 	"github.com/Azure/azure-container-networking/log"
 	"github.com/Azure/azure-container-networking/platform"
-	"github.com/Azure/azure-container-networking/store"
+	bolt "go.etcd.io/bbolt"
 )
 
 const (
@@ -145,17 +146,24 @@ func main() {
 
 	err = common.CreateDirectory(platform.CNMRuntimePath)
 	if err != nil {
-		fmt.Printf("Failed to create File Store directory Error:%v", err.Error())
+		fmt.Printf("Failed to create database directory Error:%v", err.Error())
 		return
 	}
 
-	// Create the key value store.
-	config.Store, err = store.NewJsonFileStore(platform.CNMRuntimePath + name + ".json")
+	// Create the database.
+	config.Database, err = bolt.Open(filepath.Join(platform.CNMRuntimePath, name)+".db", 0600, nil)
 	if err != nil {
-		fmt.Printf("Failed to create store: %v\n", err)
+		fmt.Printf("Failed to open database: %v\n", err)
 		return
 	}
 
+	// Close the database on exit
+	defer func() {
+		if config.Database != nil {
+			config.Database.Close()
+		}
+	}()
+
 	// Create logging provider.
 	log.SetName(name)
 	log.SetLevel(logLevel)

cns/common/service.go

@@ -8,16 +8,16 @@ import (
 
 	acn "github.com/Azure/azure-container-networking/common"
 	"github.com/Azure/azure-container-networking/log"
-	"github.com/Azure/azure-container-networking/store"
+	bolt "go.etcd.io/bbolt"
 )
 
 // Service implements behavior common to all services.
 type Service struct {
-	Name    string
-	Version string
-	Options map[string]interface{}
-	ErrChan chan error
-	Store   store.KeyValueStore
+	Name     string
+	Version  string
+	Options  map[string]interface{}
+	ErrChan  chan error
+	Database *bolt.DB
 }
 
 // ServiceAPI defines base interface.
@@ -34,18 +34,18 @@ type ServiceConfig struct {
 	Version  string
 	Listener *acn.Listener
 	ErrChan  chan error
-	Store    store.KeyValueStore
+	Database *bolt.DB
 }
 
 // NewService creates a new Service object.
-func NewService(name, version string, store store.KeyValueStore) (*Service, error) {
+func NewService(name, version string, database *bolt.DB) (*Service, error) {
 	log.Debugf("[Azure CNS] Going to create a service object with name: %v. version: %v.", name, version)
 
 	svc := &Service{
-		Name:    name,
-		Version: version,
-		Options: make(map[string]interface{}),
-		Store:   store,
+		Name:     name,
+		Version:  version,
+		Options:  make(map[string]interface{}),
+		Database: database,
 	}
 
 	log.Debugf("[Azure CNS] Finished creating service object with name: %v. version: %v.", name, version)
@@ -63,7 +63,7 @@ func (service *Service) Initialize(config *ServiceConfig) error {
 	log.Debugf("[Azure CNS] Going to initialize the service: %+v with config: %+v.", service, config)
 
 	service.ErrChan = config.ErrChan
-	service.Store = config.Store
+	service.Database = config.Database
 	service.Version = config.Version
 
 	log.Debugf("[Azure CNS] nitialized service: %+v with config: %+v.", service, config)