forked from TrueCloudLab/frostfs-node
[#1204] shard: Save ID in the metabase
`AddShard` must return shard id, so we temporarily open metabase there. Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
parent
49ae91d720
commit
a193db3a3d
6 changed files with 112 additions and 10 deletions
|
@ -3,8 +3,10 @@ package engine
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
"os"
|
"os"
|
||||||
|
"path/filepath"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
|
"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/shard"
|
||||||
cidtest "github.com/nspcc-dev/neofs-sdk-go/container/id/test"
|
cidtest "github.com/nspcc-dev/neofs-sdk-go/container/id/test"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
)
|
)
|
||||||
|
@ -48,3 +50,38 @@ func TestExecBlocks(t *testing.T) {
|
||||||
// try to resume
|
// try to resume
|
||||||
require.Error(t, e.ResumeExecution())
|
require.Error(t, e.ResumeExecution())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestPersistentShardID(t *testing.T) {
|
||||||
|
dir, err := os.MkdirTemp("", "*")
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
|
e, _, id := newEngineWithErrorThreshold(t, dir, 1)
|
||||||
|
|
||||||
|
checkShardState(t, e, id[0], shard.ModeReadWrite)
|
||||||
|
require.NoError(t, e.Close())
|
||||||
|
|
||||||
|
e, _, newID := newEngineWithErrorThreshold(t, dir, 1)
|
||||||
|
require.Equal(t, id, newID)
|
||||||
|
require.NoError(t, e.Close())
|
||||||
|
|
||||||
|
p1 := e.shards[id[0].String()].DumpInfo().MetaBaseInfo.Path
|
||||||
|
p2 := e.shards[id[1].String()].DumpInfo().MetaBaseInfo.Path
|
||||||
|
tmp := filepath.Join(dir, "tmp")
|
||||||
|
require.NoError(t, os.Rename(p1, tmp))
|
||||||
|
require.NoError(t, os.Rename(p2, p1))
|
||||||
|
require.NoError(t, os.Rename(tmp, p2))
|
||||||
|
|
||||||
|
e, _, newID = newEngineWithErrorThreshold(t, dir, 1)
|
||||||
|
require.Equal(t, id[1], newID[0])
|
||||||
|
require.Equal(t, id[0], newID[1])
|
||||||
|
require.NoError(t, e.Close())
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
func checkShardState(t *testing.T, e *StorageEngine, id *shard.ID, mode shard.Mode) {
|
||||||
|
e.mtx.RLock()
|
||||||
|
sh := e.shards[id.String()]
|
||||||
|
e.mtx.RUnlock()
|
||||||
|
|
||||||
|
require.Equal(t, mode, sh.GetMode())
|
||||||
|
}
|
||||||
|
|
|
@ -25,26 +25,34 @@ func (e *StorageEngine) AddShard(opts ...shard.Option) (*shard.ID, error) {
|
||||||
e.mtx.Lock()
|
e.mtx.Lock()
|
||||||
defer e.mtx.Unlock()
|
defer e.mtx.Unlock()
|
||||||
|
|
||||||
id, err := generateShardID()
|
|
||||||
if err != nil {
|
|
||||||
return nil, fmt.Errorf("could not generate shard ID: %w", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
pool, err := ants.NewPool(int(e.shardPoolSize), ants.WithNonblocking(true))
|
pool, err := ants.NewPool(int(e.shardPoolSize), ants.WithNonblocking(true))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
strID := id.String()
|
id, err := generateShardID()
|
||||||
|
if err != nil {
|
||||||
|
return nil, fmt.Errorf("could not generate shard ID: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
e.shards[strID] = shard.New(append(opts,
|
sh := shard.New(append(opts,
|
||||||
shard.WithID(id),
|
shard.WithID(id),
|
||||||
shard.WithExpiredObjectsCallback(e.processExpiredTombstones),
|
shard.WithExpiredObjectsCallback(e.processExpiredTombstones),
|
||||||
)...)
|
)...)
|
||||||
|
|
||||||
|
if err := sh.UpdateID(); err != nil {
|
||||||
|
return nil, fmt.Errorf("could not open shard: %w", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
strID := sh.ID().String()
|
||||||
|
if _, ok := e.shards[strID]; ok {
|
||||||
|
return nil, fmt.Errorf("shard with id %s was already added", strID)
|
||||||
|
}
|
||||||
|
|
||||||
|
e.shards[strID] = sh
|
||||||
e.shardPools[strID] = pool
|
e.shardPools[strID] = pool
|
||||||
|
|
||||||
return id, nil
|
return sh.ID(), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func generateShardID() (*shard.ID, error) {
|
func generateShardID() (*shard.ID, error) {
|
||||||
|
|
36
pkg/local_object_storage/metabase/shard_id.go
Normal file
36
pkg/local_object_storage/metabase/shard_id.go
Normal file
|
@ -0,0 +1,36 @@
|
||||||
|
package meta
|
||||||
|
|
||||||
|
import (
|
||||||
|
"github.com/nspcc-dev/neo-go/pkg/util/slice"
|
||||||
|
"go.etcd.io/bbolt"
|
||||||
|
)
|
||||||
|
|
||||||
|
var (
|
||||||
|
shardInfoBucket = []byte(invalidBase58String + "i")
|
||||||
|
shardIDKey = []byte("id")
|
||||||
|
)
|
||||||
|
|
||||||
|
// ReadShardID reads shard id from db.
|
||||||
|
// If id is missing, returns nil, nil.
|
||||||
|
func (db *DB) ReadShardID() ([]byte, error) {
|
||||||
|
var id []byte
|
||||||
|
err := db.boltDB.View(func(tx *bbolt.Tx) error {
|
||||||
|
b := tx.Bucket(shardInfoBucket)
|
||||||
|
if b != nil {
|
||||||
|
id = slice.Copy(b.Get(shardIDKey))
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
return id, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// WriteShardID writes shard it to db.
|
||||||
|
func (db *DB) WriteShardID(id []byte) error {
|
||||||
|
return db.boltDB.Update(func(tx *bbolt.Tx) error {
|
||||||
|
b, err := tx.CreateBucketIfNotExists(shardInfoBucket)
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
return b.Put(shardIDKey, id)
|
||||||
|
})
|
||||||
|
}
|
|
@ -26,7 +26,6 @@ func (s *Shard) Open() error {
|
||||||
return fmt.Errorf("could not open %T: %w", component, err)
|
return fmt.Errorf("could not open %T: %w", component, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -23,3 +23,25 @@ func (id ID) String() string {
|
||||||
func (s *Shard) ID() *ID {
|
func (s *Shard) ID() *ID {
|
||||||
return s.info.ID
|
return s.info.ID
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// UpdateID reads shard ID saved in the metabase and updates it if it is missing.
|
||||||
|
func (s *Shard) UpdateID() (err error) {
|
||||||
|
if err = s.metaBase.Open(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
defer func() {
|
||||||
|
cErr := s.metaBase.Close()
|
||||||
|
if err == nil {
|
||||||
|
err = cErr
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
id, err := s.metaBase.ReadShardID()
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
if len(id) != 0 {
|
||||||
|
s.info.ID = NewIDFromBytes(id)
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
return s.metaBase.WriteShardID(*s.info.ID)
|
||||||
|
}
|
||||||
|
|
|
@ -96,7 +96,7 @@ func New(opts ...Option) *Shard {
|
||||||
return s
|
return s
|
||||||
}
|
}
|
||||||
|
|
||||||
// WithID returns option to set shard identifier.
|
// WithID returns option to set the default shard identifier.
|
||||||
func WithID(id *ID) Option {
|
func WithID(id *ID) Option {
|
||||||
return func(c *cfg) {
|
return func(c *cfg) {
|
||||||
c.info.ID = id
|
c.info.ID = id
|
||||||
|
|
Loading…
Reference in a new issue