Skip to content

Commit

Permalink
Introduce tool to migrate objects from Blobovnicza tree to Peapod
Browse files Browse the repository at this point in the history
Add `cmd/blobovnicza-to-peapod` application which accepts YAML
configuration file of the storage node and, for each configured shard,
overtakes data from Blobovnicza tree to Peapod created in the parent
directory.

The tool is going to be used for phased and safe rejection of the
Blobovnicza trees and the transition to Peapods.

Refs nspcc-dev#2453.

Signed-off-by: Leonard Lyubich <[email protected]>
  • Loading branch information
cthulhu-rider committed Aug 11, 2023
1 parent eb3bf7a commit 327fb86
Show file tree
Hide file tree
Showing 4 changed files with 228 additions and 1 deletion.
10 changes: 10 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ Changelog for NeoFS Node
- Stored payload metric per shard (#2023)
- Histogram metrics for RPC and engine operations (#2351)
- SN's version is announced via the attributes automatically but can be overwritten explicitly (#2455)
- New `blobovnicza-to-peapod` tool providing blobovnicza-to-peapod data migration (#2453)

### Fixed
- `neo-go` RPC connection loss handling (#1337)
Expand Down Expand Up @@ -71,6 +72,15 @@ Docker images now contain a single executable file and SSL certificates only.

`neofs-cli control healthcheck` exit code is `0` only for "READY" state.

To migrate data from Blobovnicza trees to Peapods:
```shell
$ blobovnicza-to-peapod -config </path/to/storage/node/config>
```
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`.
Notice that existing Blobovnicza trees are untouched.

## [0.37.0] - 2023-06-15 - Sogado

### Added
Expand Down
95 changes: 95 additions & 0 deletions cmd/blobovnicza-to-peapod/main.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
package main

import (
"flag"
"io/fs"
"log"
"path/filepath"

"github.com/nspcc-dev/neofs-node/cmd/neofs-node/config"
engineconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/engine"
shardconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/engine/shard"
blobovniczaconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/engine/shard/blobstor/blobovnicza"
"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/peapod"
)

func main() {
nodeCfgPath := flag.String("config", "", "Path to storage node's YAML configuration file")

flag.Parse()

if *nodeCfgPath == "" {
log.Fatal("missing storage node config flag")
}

appCfg := config.New(config.Prm{}, config.WithConfigFile(*nodeCfgPath))

err := engineconfig.IterateShards(appCfg, false, func(sc *shardconfig.Config) error {
log.Println("processing shard...")

var bbcz common.Storage
var perm fs.FileMode
storagesCfg := sc.BlobStor().Storages()

for i := range storagesCfg {
if storagesCfg[i].Type() == blobovniczatree.Type {
bbczCfg := blobovniczaconfig.From((*config.Config)(storagesCfg[i]))

perm = storagesCfg[i].Perm()
bbcz = blobovniczatree.NewBlobovniczaTree(
blobovniczatree.WithRootPath(storagesCfg[i].Path()),
blobovniczatree.WithPermissions(storagesCfg[i].Perm()),
blobovniczatree.WithBlobovniczaSize(bbczCfg.Size()),
blobovniczatree.WithBlobovniczaShallowDepth(bbczCfg.ShallowDepth()),
blobovniczatree.WithBlobovniczaShallowWidth(bbczCfg.ShallowWidth()),
blobovniczatree.WithOpenedCacheSize(bbczCfg.OpenedCacheSize()))

break
}
}

if bbcz == nil {
log.Println("Blobovnicza is not configured for the current shard, going to next one...")
return nil
}

bbczPath := bbcz.Path()
if !filepath.IsAbs(bbczPath) {
log.Fatalf("Blobobvnicza tree path '%s' is not absolute, make it like this in the config file first\n", bbczPath)
}

ppdPath := filepath.Join(filepath.Dir(bbcz.Path()), "peapod.db")
ppd := peapod.New(ppdPath, perm)

var compressCfg compression.Config
compressCfg.Enabled = sc.Compress()
compressCfg.UncompressableContentTypes = sc.UncompressableContentTypes()

err := compressCfg.Init()
if err != nil {
log.Fatal("init compression config for the current shard: ", err)
}

bbcz.SetCompressor(&compressCfg)
ppd.SetCompressor(&compressCfg)

log.Printf("migrating data from Blobovnicza tree '%s' to Peapod '%s'...\n", bbcz.Path(), ppd.Path())

err = common.Copy(ppd, bbcz)
if err != nil {
log.Fatal("migration failed: ", err)
}

log.Println("data successfully migrated in the current shard, going to the next one...")

return nil
})
if err != nil {
log.Fatal(err)
}

log.Println("data successfully migrated in all shards, you may now re-configure node to work with Peapod")
}
61 changes: 60 additions & 1 deletion pkg/local_object_storage/blobstor/common/storage.go
Original file line number Diff line number Diff line change
@@ -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.
Expand All @@ -23,3 +27,58 @@ 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(dst, src Storage) error {
err := src.Open(true)
if err != nil {
return fmt.Errorf("open source sub-storage: %w", err)
}

defer func() { _ = src.Close() }()

err = src.Init()
if err != nil {
return fmt.Errorf("initialize source sub-storage: %w", err)
}

err = dst.Open(false)
if err != nil {
return fmt.Errorf("open destination sub-storage: %w", err)
}

defer func() { _ = dst.Close() }()

err = dst.Init()
if err != nil {
return fmt.Errorf("initialize destination sub-storage: %w", err)
}

_, err = src.Iterate(IteratePrm{
Handler: func(el IterationElement) error {
exRes, err := dst.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 = dst.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
}
63 changes: 63 additions & 0 deletions pkg/local_object_storage/blobstor/common/storage_test.go
Original file line number Diff line number Diff line change
@@ -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

src := blobovniczatree.NewBlobovniczaTree(
blobovniczatree.WithBlobovniczaShallowWidth(2),
blobovniczatree.WithBlobovniczaShallowDepth(3),
blobovniczatree.WithRootPath(filepath.Join(dir, "blobovnicza")),
)

require.NoError(t, src.Open(false))
require.NoError(t, src.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 := src.Put(common.PutPrm{
Address: addr,
RawData: data,
})
require.NoError(t, err)
}

require.NoError(t, src.Close())

dst := peapod.New(filepath.Join(dir, "peapod.db"), 0600)

err := common.Copy(dst, src)
require.NoError(t, err)

require.NoError(t, dst.Open(true))
t.Cleanup(func() { _ = dst.Close() })

_, err = dst.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)
}

0 comments on commit 327fb86

Please sign in to comment.