[#332] gc: Fix expired complex object deletion
All checks were successful
ci/woodpecker/push/pre-commit Pipeline was successful
All checks were successful
ci/woodpecker/push/pre-commit Pipeline was successful
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
parent
ab07bad33d
commit
869fcbf591
8 changed files with 108 additions and 69 deletions
|
@ -17,6 +17,7 @@ Changelog for FrostFS Node
|
||||||
- Notary requests parsing according to `neo-go`'s updates (#268)
|
- Notary requests parsing according to `neo-go`'s updates (#268)
|
||||||
- Tree service panic in its internal client cache (#322)
|
- Tree service panic in its internal client cache (#322)
|
||||||
- Iterate over endpoints when create ws client in morph's constructor (#304)
|
- Iterate over endpoints when create ws client in morph's constructor (#304)
|
||||||
|
- Delete complex objects with GC (#332)
|
||||||
|
|
||||||
### Removed
|
### Removed
|
||||||
### Updated
|
### Updated
|
||||||
|
|
|
@ -483,4 +483,5 @@ const (
|
||||||
ShardGCCollectingExpiredLocksCompleted = "collecting expired locks completed"
|
ShardGCCollectingExpiredLocksCompleted = "collecting expired locks completed"
|
||||||
ShardGCRemoveGarbageStarted = "garbage remove started"
|
ShardGCRemoveGarbageStarted = "garbage remove started"
|
||||||
ShardGCRemoveGarbageCompleted = "garbage remove completed"
|
ShardGCRemoveGarbageCompleted = "garbage remove completed"
|
||||||
|
ShardGCFailedToGetExpiredWithLinked = "failed to get expired objects with linked"
|
||||||
)
|
)
|
||||||
|
|
57
pkg/local_object_storage/metabase/children.go
Normal file
57
pkg/local_object_storage/metabase/children.go
Normal file
|
@ -0,0 +1,57 @@
|
||||||
|
package meta
|
||||||
|
|
||||||
|
import (
|
||||||
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||||
|
"go.etcd.io/bbolt"
|
||||||
|
)
|
||||||
|
|
||||||
|
// GetChildren returns parent -> children map.
|
||||||
|
// If an object has no children, then map will contain addr -> empty slice value.
|
||||||
|
func (db *DB) GetChildren(addresses []oid.Address) (map[oid.Address][]oid.Address, error) {
|
||||||
|
db.modeMtx.RLock()
|
||||||
|
defer db.modeMtx.RUnlock()
|
||||||
|
|
||||||
|
if db.mode.NoMetabase() {
|
||||||
|
return nil, ErrDegradedMode
|
||||||
|
}
|
||||||
|
|
||||||
|
result := make(map[oid.Address][]oid.Address, len(addresses))
|
||||||
|
|
||||||
|
buffer := make([]byte, bucketKeySize)
|
||||||
|
err := db.boltDB.View(func(tx *bbolt.Tx) error {
|
||||||
|
for _, addr := range addresses {
|
||||||
|
if _, found := result[addr]; found {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
result[addr] = []oid.Address{}
|
||||||
|
bkt := tx.Bucket(parentBucketName(addr.Container(), buffer))
|
||||||
|
if bkt == nil {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
binObjIDs, err := decodeList(bkt.Get(objectKey(addr.Object(), buffer)))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, binObjID := range binObjIDs {
|
||||||
|
var id oid.ID
|
||||||
|
if err = id.Decode(binObjID); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
var resultAddress oid.Address
|
||||||
|
resultAddress.SetContainer(addr.Container())
|
||||||
|
resultAddress.SetObject(id)
|
||||||
|
result[addr] = append(result[addr], resultAddress)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
})
|
||||||
|
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
return result, nil
|
||||||
|
}
|
|
@ -320,6 +320,12 @@ func (s *Shard) handleExpiredObjects(ctx context.Context, expired []oid.Address)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
expired, err := s.getExpiredWithLinked(expired)
|
||||||
|
if err != nil {
|
||||||
|
s.log.Warn(logs.ShardGCFailedToGetExpiredWithLinked, zap.Error(err))
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
var inhumePrm meta.InhumePrm
|
var inhumePrm meta.InhumePrm
|
||||||
|
|
||||||
inhumePrm.SetAddresses(expired...)
|
inhumePrm.SetAddresses(expired...)
|
||||||
|
@ -345,6 +351,20 @@ func (s *Shard) handleExpiredObjects(ctx context.Context, expired []oid.Address)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (s *Shard) getExpiredWithLinked(source []oid.Address) ([]oid.Address, error) {
|
||||||
|
result := make([]oid.Address, 0, len(source))
|
||||||
|
parentToChildren, err := s.metaBase.GetChildren(source)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
for parent, children := range parentToChildren {
|
||||||
|
result = append(result, parent)
|
||||||
|
result = append(result, children...)
|
||||||
|
}
|
||||||
|
|
||||||
|
return result, nil
|
||||||
|
}
|
||||||
|
|
||||||
func (s *Shard) collectExpiredTombstones(ctx context.Context, e Event) {
|
func (s *Shard) collectExpiredTombstones(ctx context.Context, e Event) {
|
||||||
epoch := e.(newEpoch).epoch
|
epoch := e.(newEpoch).epoch
|
||||||
log := s.log.With(zap.Uint64("epoch", epoch))
|
log := s.log.With(zap.Uint64("epoch", epoch))
|
||||||
|
|
|
@ -2,29 +2,20 @@ package shard_test
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"encoding/binary"
|
|
||||||
"errors"
|
"errors"
|
||||||
"path/filepath"
|
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
|
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
|
||||||
objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
|
objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor"
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/blobovniczatree"
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/fstree"
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
|
||||||
meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
|
meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
|
||||||
cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test"
|
cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test"
|
||||||
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||||
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||||
oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
|
oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
|
||||||
"github.com/panjf2000/ants/v2"
|
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"go.uber.org/zap"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func Test_GCDropsLockedExpiredSimpleObject(t *testing.T) {
|
func Test_GCDropsLockedExpiredSimpleObject(t *testing.T) {
|
||||||
|
@ -34,7 +25,7 @@ func Test_GCDropsLockedExpiredSimpleObject(t *testing.T) {
|
||||||
Value: 100,
|
Value: 100,
|
||||||
}
|
}
|
||||||
|
|
||||||
sh := createAndInitGCTestShard(t, epoch)
|
sh := newCustomShard(t, t.TempDir(), false, nil, nil, []meta.Option{meta.WithEpochState(epoch)})
|
||||||
|
|
||||||
t.Cleanup(func() {
|
t.Cleanup(func() {
|
||||||
releaseShard(sh, t)
|
releaseShard(sh, t)
|
||||||
|
@ -131,7 +122,7 @@ func Test_GCDropsLockedExpiredComplexObject(t *testing.T) {
|
||||||
|
|
||||||
linkID, _ := link.ID()
|
linkID, _ := link.ID()
|
||||||
|
|
||||||
sh := createAndInitGCTestShard(t, epoch)
|
sh := newCustomShard(t, t.TempDir(), false, nil, nil, []meta.Option{meta.WithEpochState(epoch)})
|
||||||
|
|
||||||
t.Cleanup(func() {
|
t.Cleanup(func() {
|
||||||
releaseShard(sh, t)
|
releaseShard(sh, t)
|
||||||
|
@ -176,54 +167,3 @@ func Test_GCDropsLockedExpiredComplexObject(t *testing.T) {
|
||||||
return shard.IsErrNotFound(err)
|
return shard.IsErrNotFound(err)
|
||||||
}, 3*time.Second, 1*time.Second, "expired complex object must be deleted on epoch after lock expires")
|
}, 3*time.Second, 1*time.Second, "expired complex object must be deleted on epoch after lock expires")
|
||||||
}
|
}
|
||||||
|
|
||||||
func createAndInitGCTestShard(t *testing.T, epoch *epochState) *shard.Shard {
|
|
||||||
var sh *shard.Shard
|
|
||||||
|
|
||||||
rootPath := t.TempDir()
|
|
||||||
opts := []shard.Option{
|
|
||||||
shard.WithID(shard.NewIDFromBytes(binary.AppendVarint([]byte{}, int64(48)))),
|
|
||||||
shard.WithLogger(&logger.Logger{Logger: zap.NewNop()}),
|
|
||||||
shard.WithBlobStorOptions(
|
|
||||||
blobstor.WithStorages([]blobstor.SubStorage{
|
|
||||||
{
|
|
||||||
Storage: blobovniczatree.NewBlobovniczaTree(
|
|
||||||
blobovniczatree.WithRootPath(filepath.Join(rootPath, "blob", "blobovnicza")),
|
|
||||||
blobovniczatree.WithBlobovniczaShallowDepth(2),
|
|
||||||
blobovniczatree.WithBlobovniczaShallowWidth(2)),
|
|
||||||
Policy: func(_ *objectSDK.Object, data []byte) bool {
|
|
||||||
return len(data) <= 1<<20
|
|
||||||
},
|
|
||||||
},
|
|
||||||
{
|
|
||||||
Storage: fstree.New(
|
|
||||||
fstree.WithPath(filepath.Join(rootPath, "blob"))),
|
|
||||||
},
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
shard.WithMetaBaseOptions(
|
|
||||||
meta.WithPath(filepath.Join(rootPath, "meta")),
|
|
||||||
meta.WithEpochState(epoch),
|
|
||||||
),
|
|
||||||
shard.WithDeletedLockCallback(func(_ context.Context, addresses []oid.Address) {
|
|
||||||
sh.HandleDeletedLocks(addresses)
|
|
||||||
}),
|
|
||||||
shard.WithExpiredLocksCallback(func(ctx context.Context, epoch uint64, a []oid.Address) {
|
|
||||||
sh.HandleExpiredLocks(ctx, epoch, a)
|
|
||||||
}),
|
|
||||||
shard.WithGCWorkerPoolInitializer(func(sz int) util.WorkerPool {
|
|
||||||
pool, err := ants.NewPool(sz)
|
|
||||||
require.NoError(t, err)
|
|
||||||
|
|
||||||
return pool
|
|
||||||
}),
|
|
||||||
shard.WithGCRemoverSleepInterval(1 * time.Millisecond),
|
|
||||||
}
|
|
||||||
|
|
||||||
sh = shard.New(opts...)
|
|
||||||
|
|
||||||
require.NoError(t, sh.Open())
|
|
||||||
require.NoError(t, sh.Init(context.Background()))
|
|
||||||
|
|
||||||
return sh
|
|
||||||
}
|
|
||||||
|
|
|
@ -87,7 +87,8 @@ func testShardGetRange(t *testing.T, hasWriteCache bool) {
|
||||||
Storage: fstree.New(
|
Storage: fstree.New(
|
||||||
fstree.WithPath(filepath.Join(t.TempDir(), "blob"))),
|
fstree.WithPath(filepath.Join(t.TempDir(), "blob"))),
|
||||||
},
|
},
|
||||||
})})
|
})},
|
||||||
|
nil)
|
||||||
defer releaseShard(sh, t)
|
defer releaseShard(sh, t)
|
||||||
|
|
||||||
for _, tc := range testCases {
|
for _, tc := range testCases {
|
||||||
|
|
|
@ -4,6 +4,7 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"path/filepath"
|
"path/filepath"
|
||||||
"testing"
|
"testing"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/blobovniczatree"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/blobovniczatree"
|
||||||
|
@ -12,8 +13,11 @@ import (
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/writecache"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/writecache"
|
||||||
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util/logger"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
|
||||||
|
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
|
||||||
|
"github.com/panjf2000/ants/v2"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"go.uber.org/zap"
|
"go.uber.org/zap"
|
||||||
"go.uber.org/zap/zaptest"
|
"go.uber.org/zap/zaptest"
|
||||||
|
@ -29,11 +33,13 @@ func (s epochState) CurrentEpoch() uint64 {
|
||||||
|
|
||||||
func newShard(t testing.TB, enableWriteCache bool) *shard.Shard {
|
func newShard(t testing.TB, enableWriteCache bool) *shard.Shard {
|
||||||
return newCustomShard(t, t.TempDir(), enableWriteCache,
|
return newCustomShard(t, t.TempDir(), enableWriteCache,
|
||||||
|
nil,
|
||||||
nil,
|
nil,
|
||||||
nil)
|
nil)
|
||||||
}
|
}
|
||||||
|
|
||||||
func newCustomShard(t testing.TB, rootPath string, enableWriteCache bool, wcOpts []writecache.Option, bsOpts []blobstor.Option) *shard.Shard {
|
func newCustomShard(t testing.TB, rootPath string, enableWriteCache bool, wcOpts []writecache.Option, bsOpts []blobstor.Option, metaOptions []meta.Option) *shard.Shard {
|
||||||
|
var sh *shard.Shard
|
||||||
if enableWriteCache {
|
if enableWriteCache {
|
||||||
rootPath = filepath.Join(rootPath, "wc")
|
rootPath = filepath.Join(rootPath, "wc")
|
||||||
} else {
|
} else {
|
||||||
|
@ -67,8 +73,9 @@ func newCustomShard(t testing.TB, rootPath string, enableWriteCache bool, wcOpts
|
||||||
shard.WithLogger(&logger.Logger{Logger: zap.L()}),
|
shard.WithLogger(&logger.Logger{Logger: zap.L()}),
|
||||||
shard.WithBlobStorOptions(bsOpts...),
|
shard.WithBlobStorOptions(bsOpts...),
|
||||||
shard.WithMetaBaseOptions(
|
shard.WithMetaBaseOptions(
|
||||||
meta.WithPath(filepath.Join(rootPath, "meta")),
|
append([]meta.Option{
|
||||||
meta.WithEpochState(epochState{}),
|
meta.WithPath(filepath.Join(rootPath, "meta")), meta.WithEpochState(epochState{})},
|
||||||
|
metaOptions...)...,
|
||||||
),
|
),
|
||||||
shard.WithPiloramaOptions(pilorama.WithPath(filepath.Join(rootPath, "pilorama"))),
|
shard.WithPiloramaOptions(pilorama.WithPath(filepath.Join(rootPath, "pilorama"))),
|
||||||
shard.WithWriteCache(enableWriteCache),
|
shard.WithWriteCache(enableWriteCache),
|
||||||
|
@ -77,9 +84,21 @@ func newCustomShard(t testing.TB, rootPath string, enableWriteCache bool, wcOpts
|
||||||
[]writecache.Option{writecache.WithPath(filepath.Join(rootPath, "wcache"))},
|
[]writecache.Option{writecache.WithPath(filepath.Join(rootPath, "wcache"))},
|
||||||
wcOpts...)...,
|
wcOpts...)...,
|
||||||
),
|
),
|
||||||
|
shard.WithDeletedLockCallback(func(_ context.Context, addresses []oid.Address) {
|
||||||
|
sh.HandleDeletedLocks(addresses)
|
||||||
|
}),
|
||||||
|
shard.WithExpiredLocksCallback(func(ctx context.Context, epoch uint64, a []oid.Address) {
|
||||||
|
sh.HandleExpiredLocks(ctx, epoch, a)
|
||||||
|
}),
|
||||||
|
shard.WithGCWorkerPoolInitializer(func(sz int) util.WorkerPool {
|
||||||
|
pool, err := ants.NewPool(sz)
|
||||||
|
require.NoError(t, err)
|
||||||
|
return pool
|
||||||
|
}),
|
||||||
|
shard.WithGCRemoverSleepInterval(1 * time.Millisecond),
|
||||||
}
|
}
|
||||||
|
|
||||||
sh := shard.New(opts...)
|
sh = shard.New(opts...)
|
||||||
|
|
||||||
require.NoError(t, sh.Open())
|
require.NoError(t, sh.Open())
|
||||||
require.NoError(t, sh.Init(context.Background()))
|
require.NoError(t, sh.Init(context.Background()))
|
||||||
|
|
|
@ -40,7 +40,7 @@ func TestWriteCacheObjectLoss(t *testing.T) {
|
||||||
writecache.WithSmallObjectSize(smallSize),
|
writecache.WithSmallObjectSize(smallSize),
|
||||||
writecache.WithMaxObjectSize(smallSize * 2)}
|
writecache.WithMaxObjectSize(smallSize * 2)}
|
||||||
|
|
||||||
sh := newCustomShard(t, dir, true, wcOpts, nil)
|
sh := newCustomShard(t, dir, true, wcOpts, nil, nil)
|
||||||
|
|
||||||
var errG errgroup.Group
|
var errG errgroup.Group
|
||||||
for i := range objects {
|
for i := range objects {
|
||||||
|
@ -55,7 +55,7 @@ func TestWriteCacheObjectLoss(t *testing.T) {
|
||||||
require.NoError(t, errG.Wait())
|
require.NoError(t, errG.Wait())
|
||||||
require.NoError(t, sh.Close())
|
require.NoError(t, sh.Close())
|
||||||
|
|
||||||
sh = newCustomShard(t, dir, true, wcOpts, nil)
|
sh = newCustomShard(t, dir, true, wcOpts, nil, nil)
|
||||||
defer releaseShard(sh, t)
|
defer releaseShard(sh, t)
|
||||||
|
|
||||||
var getPrm shard.GetPrm
|
var getPrm shard.GetPrm
|
||||||
|
|
Loading…
Reference in a new issue