node: Use Peapod instead of Blobovnicza tree as BlobStor sub-storage

Recently introduce Peapod component is designed to replace Blobovnicza
tree. To facilitate migration to a new storage scheme, it is required to
maintain compatibility with the old application configuration, while
still moving to a new component.

Provide `common.Copy` function which overtakes data from one sub-storage
to another. On storage node startup, when Blobovnicza tree is configured
 for BlobStor, create new Peapod instance on the same level as
Blobovnicza and migrate data into it. Filled Peapod is used as
sub-storage.

Signed-off-by: Leonard Lyubich <[email protected]>

CHANGELOG.md

@@ -27,6 +27,7 @@ Changelog for NeoFS Node
 - CLI `--timeout` flag configures whole execution timeout from now (#2124)
 - CLI default timeout for commands with `--await` flag increased to 1m (#2124)
 - BlobStor tries to store object in any sub-storage with free space (#2450)
+- BlobStor stores small objects in new sub-storage Peapod instead of Blobovnicza tree (#2453)
 
 ### Updated
 - `neofs-sdk-go` to `v1.0.0-rc.9`
@@ -55,6 +56,13 @@ Docker images now contain a single executable file and SSL certificates only.
 
 `neofs-cli control healthcheck` exit code is `0` only for "READY" state.
 
+The local data storage scheme is reorganized automatically: for any shard, the
+data from the configured Blobovnicza tree is copied into a created Peapod file
+named `peapod.db` in the directory where the tree is located. For example,
+`/neofs/data/blobovcniza/*` -> `/neofs/data/peapod.db`. Storage node doesn't
+touch existing Blobovnicza tree, but the Peapod is used as `blobovnicza` sub-storage
+instead.
+
 ## [0.37.0] - 2023-06-15 - Sogado
 
 ### Added

cmd/neofs-node/config.go

@@ -35,7 +35,10 @@ import (
 	netmapCore "github.com/nspcc-dev/neofs-node/pkg/core/netmap"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/blobovniczatree"
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common"
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/compression"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/fstree"
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/peapod"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine"
 	meta "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/metabase"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/pilorama"
@@ -705,16 +708,36 @@ func (c *cfg) shardOpts() []shardOptsWithID {
 		for _, sRead := range shCfg.subStorages {
 			switch sRead.typ {
 			case blobovniczatree.Type:
+				var compressCfg compression.Config
+				compressCfg.Enabled = shCfg.compress
+				compressCfg.UncompressableContentTypes = shCfg.uncompressableContentType
+
+				err := compressCfg.Init()
+				fatalOnErrDetails("init compression config", err)
+
+				bbcz := blobovniczatree.NewBlobovniczaTree(
+					blobovniczatree.WithRootPath(sRead.path),
+					blobovniczatree.WithPermissions(sRead.perm),
+					blobovniczatree.WithBlobovniczaSize(sRead.size),
+					blobovniczatree.WithBlobovniczaShallowDepth(sRead.depth),
+					blobovniczatree.WithBlobovniczaShallowWidth(sRead.width),
+					blobovniczatree.WithOpenedCacheSize(sRead.openedCacheSize),
+					blobovniczatree.WithLogger(c.log))
+				bbcz.SetCompressor(&compressCfg)
+
+				ppd := peapod.New(filepath.Join(filepath.Dir(sRead.path), "peapod.db"), sRead.perm)
+				ppd.SetCompressor(&compressCfg)
+
+				c.log.Info("configured Blobovnicza tree is deprecated, overtaking data into Peapod...",
+					zap.String("from", bbcz.Path()), zap.String("to", ppd.Path()))
+
+				err = common.Copy(bbcz, ppd)
+				fatalOnErrDetails("overtake data from Blobovcniza tree to Peapod", err)
+
+				c.log.Info("data was successfully migrated from Blobovnicza tree to Peapod")
+
 				ss = append(ss, blobstor.SubStorage{
-					Storage: blobovniczatree.NewBlobovniczaTree(
-						blobovniczatree.WithRootPath(sRead.path),
-						blobovniczatree.WithPermissions(sRead.perm),
-						blobovniczatree.WithBlobovniczaSize(sRead.size),
-						blobovniczatree.WithBlobovniczaShallowDepth(sRead.depth),
-						blobovniczatree.WithBlobovniczaShallowWidth(sRead.width),
-						blobovniczatree.WithOpenedCacheSize(sRead.openedCacheSize),
-
-						blobovniczatree.WithLogger(c.log)),
+					Storage: ppd,
 					Policy: func(_ *objectSDK.Object, data []byte) bool {
 						return uint64(len(data)) < shCfg.smallSizeObjectLimit
 					},

pkg/local_object_storage/blobstor/common/storage.go

@@ -1,6 +1,10 @@
 package common
 
-import "github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/compression"
+import (
+	"fmt"
+
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/compression"
+)
 
 // Storage represents key-value object storage.
 // It is used as a building block for a blobstor of a shard.
@@ -23,3 +27,53 @@ type Storage interface {
 	Delete(DeletePrm) (DeleteRes, error)
 	Iterate(IteratePrm) (IterateRes, error)
 }
+
+// Copy copies all objects from source Storage into the destination one. If any
+// object cannot be stored, Copy immediately fails.
+func Copy(from, to Storage) error {
+	err := from.Open(true)
+	if err != nil {
+		return fmt.Errorf("open source sub-storage: %w", err)
+	}
+
+	defer func() { _ = from.Close() }()
+
+	err = to.Open(false)
+	if err != nil {
+		return fmt.Errorf("open destination sub-storage: %w", err)
+	}
+
+	defer func() { _ = to.Close() }()
+
+	err = to.Init()
+	if err != nil {
+		return fmt.Errorf("initialize destination sub-storage: %w", err)
+	}
+
+	_, err = from.Iterate(IteratePrm{
+		Handler: func(el IterationElement) error {
+			exRes, err := to.Exists(ExistsPrm{
+				Address: el.Address,
+			})
+			if err != nil {
+				return fmt.Errorf("check presence of object %s in the destination sub-storage: %w", el.Address, err)
+			} else if exRes.Exists {
+				return nil
+			}
+
+			_, err = to.Put(PutPrm{
+				Address: el.Address,
+				RawData: el.ObjectData,
+			})
+			if err != nil {
+				return fmt.Errorf("put object %s into destination sub-storage: %w", el.Address, err)
+			}
+			return nil
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("iterate over source sub-storage: %w", err)
+	}
+
+	return nil
+}

pkg/local_object_storage/blobstor/common/storage_test.go

@@ -0,0 +1,63 @@
+package common_test
+
+import (
+	"crypto/rand"
+	"path/filepath"
+	"testing"
+
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/blobovniczatree"
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common"
+	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/peapod"
+	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
+	oidtest "github.com/nspcc-dev/neofs-sdk-go/object/id/test"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCopy(t *testing.T) {
+	dir := t.TempDir()
+	const nObjects = 100
+
+	from := blobovniczatree.NewBlobovniczaTree(
+		blobovniczatree.WithBlobovniczaShallowWidth(2),
+		blobovniczatree.WithBlobovniczaShallowDepth(3),
+		blobovniczatree.WithRootPath(filepath.Join(dir, "blobovnicza")),
+	)
+
+	require.NoError(t, from.Open(false))
+	require.NoError(t, from.Init())
+
+	mObjs := make(map[oid.Address][]byte, nObjects)
+
+	for i := 0; i < nObjects; i++ {
+		addr := oidtest.Address()
+		data := make([]byte, 32)
+		rand.Read(data)
+		mObjs[addr] = data
+
+		_, err := from.Put(common.PutPrm{
+			Address: addr,
+			RawData: data,
+		})
+		require.NoError(t, err)
+	}
+
+	require.NoError(t, from.Close())
+
+	to := peapod.New(filepath.Join(dir, "peapod.db"), 0600)
+
+	err := common.Copy(from, to)
+	require.NoError(t, err)
+
+	require.NoError(t, to.Open(true))
+	t.Cleanup(func() { _ = to.Close() })
+
+	_, err = to.Iterate(common.IteratePrm{
+		Handler: func(el common.IterationElement) error {
+			data, ok := mObjs[el.Address]
+			require.True(t, ok)
+			require.Equal(t, data, el.ObjectData)
+			return nil
+		},
+	})
+	require.NoError(t, err)
+}

pkg/local_object_storage/blobstor/peapod/peapod.go

@@ -180,6 +180,11 @@ func (x *storage) Delete(prm common.DeletePrm) (common.DeleteRes, error) {
 func (x *storage) Iterate(prm common.IteratePrm) (common.IterateRes, error) {
 	var e error
 	err := x.ppd.Iterate(func(addr oid.Address, data []byte) bool {
+		data, e = x.compress.Decompress(data)
+		if e != nil {
+			return false
+		}
+
 		if prm.LazyHandler != nil {
 			e = prm.LazyHandler(addr, func() ([]byte, error) {
 				return data, nil