storage zdb: also check on the mode of the 0-db when allocating volumes

fixes #499

Author: zaibon

pkg/storage/zdb.go

@@ -54,6 +54,7 @@ func (s *storageModule) Allocate(nsID string, diskType pkg.DeviceType, size uint
 	log := log.With().
 		Str("type", string(diskType)).
 		Uint64("size", size).
+		Str("mode", string(mode)).
 		Logger()
 
 	log.Info().Msg("try to allocation space for 0-DB")
@@ -71,6 +72,7 @@ func (s *storageModule) Allocate(nsID string, diskType pkg.DeviceType, size uint
 		}
 
 		for _, volume := range volumes {
+
 			// skip all non-zdb volume
 			if !filesystem.IsZDBVolume(volume) {
 				continue
@@ -97,6 +99,11 @@ func (s *storageModule) Allocate(nsID string, diskType pkg.DeviceType, size uint
 		Free uint64
 	}
 
+	targetMode := zdbpool.IndexModeKeyValue
+	if mode == pkg.ZDBModeSeq {
+		targetMode = zdbpool.IndexModeSequential
+	}
+
 	var candidates []Candidate
 	// okay, so this is a new allocation
 	// we search all the pools for a zdb subvolume that can
@@ -133,6 +140,20 @@ func (s *storageModule) Allocate(nsID string, diskType pkg.DeviceType, size uint
 				continue
 			}
 
+			zdb := zdbpool.New(volume.Path())
+
+			// check if the mode is the same
+			indexMode, err := zdb.IndexMode("default")
+			if err != nil {
+				log.Err(err).Str("namespace", "default").Msg("failed to read index mode")
+				continue
+			}
+
+			if indexMode != targetMode {
+				log.Info().Msg("skip because wrong mode")
+				continue
+			}
+
 			candidates = append(
 				candidates,
 				Candidate{

pkg/storage/zdbpool/index.go

@@ -0,0 +1,49 @@
+package zdbpool
+
+import (
+	"encoding/binary"
+	"io"
+)
+
+// IndexMode represens the mode in which the 0-db is running
+// Adapted from https://github.com/threefoldtech/0-db/blob/development/libzdb/index.h#L4
+type IndexMode uint8
+
+// Enum values for IndexMode
+const (
+	IndexModeKeyValue    IndexMode = 0
+	IndexModeSequential  IndexMode = 1
+	IndexModeDirectKey   IndexMode = 2
+	IndexModeDirectBlock IndexMode = 3
+)
+
+func (i IndexMode) String() string {
+	switch i {
+	case IndexModeKeyValue:
+		return "key-value"
+	case IndexModeSequential:
+		return "sequential"
+	case IndexModeDirectKey:
+		return "direct-key"
+	case IndexModeDirectBlock:
+		return "direct-block"
+	}
+	return "unknown"
+}
+
+// IndexHeader is the structure contains information about an index
+// adapted from https://github.com/threefoldtech/0-db/blob/development/libzdb/index.h#L31
+type IndexHeader struct {
+	Magic   [4]byte   // four bytes magic bytes to recognize the file
+	Version uint32    // file version, for possible upgrade compatibility
+	Created uint64    // unix timestamp of creation time
+	Opened  uint64    // unix timestamp of last opened time
+	Fileid  uint16    // current index file id (sync with dataid)
+	Mode    IndexMode // running mode when index was create
+}
+
+// ReadIndex reads index header
+func ReadIndex(r io.Reader) (header IndexHeader, err error) {
+	err = binary.Read(r, binary.LittleEndian, &header)
+	return
+}

pkg/storage/zdbpool/index_test.go

@@ -0,0 +1,20 @@
+package zdbpool
+
+import (
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestReadIndexHeader(t *testing.T) {
+	f, err := os.Open("./test_data/zdb-index-00000")
+	require.NoError(t, err)
+	defer f.Close()
+
+	h, err := ReadIndex(f)
+	require.NoError(t, err)
+
+	assert.Equal(t, IndexModeKeyValue, h.Mode)
+}

pkg/storage/zdbpool/pool.go

@@ -128,3 +128,21 @@ func (p *ZDBPool) Exists(name string) bool {
 	_, err := os.Stat(path)
 	return err == nil
 }
+
+// IndexMode return the mode of the index of the namespace called name
+func (p *ZDBPool) IndexMode(name string) (mode IndexMode, err error) {
+	path := filepath.Join(p.path, name, "zdb-index-00000")
+
+	f, err := os.Open(path)
+	if err != nil {
+		return mode, err
+	}
+
+	defer f.Close()
+	index, err := ReadIndex(f)
+	if err != nil {
+		return mode, errors.Wrapf(err, "failed to read namespace index header at %s", path)
+	}
+
+	return index.Mode, nil
+}

pkg/storage/zdbpool/pool_test.go

@@ -41,3 +41,13 @@ func TestExists(t *testing.T) {
 	assert.True(t, pool.Exists("test"))
 	assert.False(t, pool.Exists("foo"))
 }
+
+func TestIndexMode(t *testing.T) {
+	pool := ZDBPool{
+		path: "./test_data/pool_layout",
+	}
+
+	mode, err := pool.IndexMode("test")
+	require.NoError(t, err)
+	assert.Equal(t, IndexModeKeyValue, mode)
+}