frostfs-node/pkg/local_object_storage/shard/gc_test.go

296 lines
9.1 KiB
Go
Raw Normal View History

package shard
import (
"context"
"errors"
"testing"
objectV2 "git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/object"
objectCore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/common"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/writecache"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/util"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
cidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id/test"
objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
oidtest "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id/test"
"github.com/stretchr/testify/require"
)
func Test_GCDropsLockedExpiredSimpleObject(t *testing.T) {
t.Parallel()
epoch := &epochState{
Value: 100,
}
sh := newCustomShard(t, false, shardOptions{
metaOptions: []meta.Option{meta.WithEpochState(epoch)},
additionalShardOptions: []Option{WithGCWorkerPoolInitializer(func(int) util.WorkerPool {
return util.NewPseudoWorkerPool() // synchronous event processing
})},
})
defer func() { require.NoError(t, sh.Close()) }()
cnr := cidtest.ID()
var objExpirationAttr objectSDK.Attribute
objExpirationAttr.SetKey(objectV2.SysAttributeExpEpoch)
objExpirationAttr.SetValue("101")
obj := testutil.GenerateObjectWithCID(cnr)
obj.SetAttributes(objExpirationAttr)
objID, _ := obj.ID()
var lockExpirationAttr objectSDK.Attribute
lockExpirationAttr.SetKey(objectV2.SysAttributeExpEpoch)
lockExpirationAttr.SetValue("103")
lock := testutil.GenerateObjectWithCID(cnr)
lock.SetType(objectSDK.TypeLock)
lock.SetAttributes(lockExpirationAttr)
lockID, _ := lock.ID()
var putPrm PutPrm
putPrm.SetObject(obj)
_, err := sh.Put(context.Background(), putPrm)
require.NoError(t, err)
err = sh.Lock(context.Background(), cnr, lockID, []oid.ID{objID})
require.NoError(t, err)
putPrm.SetObject(lock)
_, err = sh.Put(context.Background(), putPrm)
require.NoError(t, err)
epoch.Value = 105
sh.gc.handleEvent(context.Background(), EventNewEpoch(epoch.Value))
var getPrm GetPrm
getPrm.SetAddress(objectCore.AddressOf(obj))
_, err = sh.Get(context.Background(), getPrm)
require.True(t, client.IsErrObjectNotFound(err), "expired object must be deleted")
}
func Test_GCDropsLockedExpiredComplexObject(t *testing.T) {
t.Parallel()
epoch := &epochState{
Value: 100,
}
cnr := cidtest.ID()
parentID := oidtest.ID()
splitID := objectSDK.NewSplitID()
var objExpirationAttr objectSDK.Attribute
objExpirationAttr.SetKey(objectV2.SysAttributeExpEpoch)
objExpirationAttr.SetValue("101")
var lockExpirationAttr objectSDK.Attribute
lockExpirationAttr.SetKey(objectV2.SysAttributeExpEpoch)
lockExpirationAttr.SetValue("103")
parent := testutil.GenerateObjectWithCID(cnr)
parent.SetID(parentID)
parent.SetPayload(nil)
parent.SetAttributes(objExpirationAttr)
const childCount = 10
children := make([]*objectSDK.Object, childCount)
childIDs := make([]oid.ID, childCount)
for i := range children {
children[i] = testutil.GenerateObjectWithCID(cnr)
if i != 0 {
children[i].SetPreviousID(childIDs[i-1])
}
if i == len(children)-1 {
children[i].SetParent(parent)
}
children[i].SetSplitID(splitID)
children[i].SetPayload([]byte{byte(i), byte(i + 1), byte(i + 2)})
childIDs[i], _ = children[i].ID()
}
link := testutil.GenerateObjectWithCID(cnr)
link.SetParent(parent)
link.SetParentID(parentID)
link.SetSplitID(splitID)
link.SetChildren(childIDs...)
linkID, _ := link.ID()
sh := newCustomShard(t, false, shardOptions{
metaOptions: []meta.Option{meta.WithEpochState(epoch)},
additionalShardOptions: []Option{WithGCWorkerPoolInitializer(func(int) util.WorkerPool {
return util.NewPseudoWorkerPool() // synchronous event processing
})},
})
defer func() { require.NoError(t, sh.Close()) }()
lock := testutil.GenerateObjectWithCID(cnr)
lock.SetType(objectSDK.TypeLock)
lock.SetAttributes(lockExpirationAttr)
lockID, _ := lock.ID()
var putPrm PutPrm
for _, child := range children {
putPrm.SetObject(child)
_, err := sh.Put(context.Background(), putPrm)
require.NoError(t, err)
}
putPrm.SetObject(link)
_, err := sh.Put(context.Background(), putPrm)
require.NoError(t, err)
err = sh.Lock(context.Background(), cnr, lockID, append(childIDs, parentID, linkID))
require.NoError(t, err)
putPrm.SetObject(lock)
_, err = sh.Put(context.Background(), putPrm)
require.NoError(t, err)
var getPrm GetPrm
getPrm.SetAddress(objectCore.AddressOf(parent))
_, err = sh.Get(context.Background(), getPrm)
var splitInfoError *objectSDK.SplitInfoError
require.True(t, errors.As(err, &splitInfoError), "split info must be provided")
epoch.Value = 105
sh.gc.handleEvent(context.Background(), EventNewEpoch(epoch.Value))
_, err = sh.Get(context.Background(), getPrm)
require.True(t, client.IsErrObjectNotFound(err), "expired complex object must be deleted on epoch after lock expires")
}
func TestGCDropsObjectInhumedFromWritecache(t *testing.T) {
t.Parallel()
t.Run("flush write-cache before inhume", func(t *testing.T) {
t.Parallel()
testGCDropsObjectInhumedFromWritecache(t, true)
})
t.Run("don't flush write-cache before inhume", func(t *testing.T) {
t.Parallel()
testGCDropsObjectInhumedFromWritecache(t, false)
})
}
func testGCDropsObjectInhumedFromWritecache(t *testing.T, flushbeforeInhume bool) {
sh := newCustomShard(t, true, shardOptions{
additionalShardOptions: []Option{WithDisabledGC()},
wcOpts: []writecache.Option{writecache.WithDisableBackgroundFlush()},
})
defer func() { require.NoError(t, sh.Close()) }()
obj := testutil.GenerateObjectWithSize(1024)
var putPrm PutPrm
putPrm.SetObject(obj)
_, err := sh.Put(context.Background(), putPrm)
require.NoError(t, err)
// writecache stores object
wcObj, err := sh.writeCache.Head(context.Background(), objectCore.AddressOf(obj))
require.NoError(t, err)
require.Equal(t, objectCore.AddressOf(obj), objectCore.AddressOf(wcObj))
// blobstore doesn't store object
bsRes, err := sh.blobStor.Get(context.Background(), common.GetPrm{
Address: objectCore.AddressOf(obj),
})
require.ErrorAs(t, err, new(*apistatus.ObjectNotFound))
require.Nil(t, bsRes.Object)
require.Nil(t, bsRes.RawData)
if flushbeforeInhume {
sh.writeCache.Flush(context.Background(), false, false)
}
var inhumePrm InhumePrm
inhumePrm.MarkAsGarbage(objectCore.AddressOf(obj))
_, err = sh.Inhume(context.Background(), inhumePrm)
require.NoError(t, err)
// writecache doesn't store object
wcObj, err = sh.writeCache.Head(context.Background(), objectCore.AddressOf(obj))
require.Error(t, err)
require.Nil(t, wcObj)
if flushbeforeInhume {
// blobstore store object
bsRes, err = sh.blobStor.Get(context.Background(), common.GetPrm{
Address: objectCore.AddressOf(obj),
})
require.NoError(t, err)
require.Equal(t, objectCore.AddressOf(obj), objectCore.AddressOf(bsRes.Object))
} else {
// blobstore doesn't store object
bsRes, err = sh.blobStor.Get(context.Background(), common.GetPrm{
Address: objectCore.AddressOf(obj),
})
require.ErrorAs(t, err, new(*apistatus.ObjectNotFound))
require.Nil(t, bsRes.Object)
require.Nil(t, bsRes.RawData)
}
gcRes := sh.removeGarbage(context.Background())
require.True(t, gcRes.success)
require.Equal(t, uint64(1), gcRes.deleted)
}
func TestGCDontDeleteObjectFromWritecache(t *testing.T) {
sh := newCustomShard(t, true, shardOptions{
additionalShardOptions: []Option{WithDisabledGC()},
wcOpts: []writecache.Option{writecache.WithDisableBackgroundFlush()},
})
defer func() { require.NoError(t, sh.Close()) }()
obj := testutil.GenerateObjectWithSize(1024)
var putPrm PutPrm
putPrm.SetObject(obj)
_, err := sh.Put(context.Background(), putPrm)
require.NoError(t, err)
// writecache stores object
wcObj, err := sh.writeCache.Head(context.Background(), objectCore.AddressOf(obj))
require.NoError(t, err)
require.Equal(t, objectCore.AddressOf(obj), objectCore.AddressOf(wcObj))
// blobstore doesn't store object
bsRes, err := sh.blobStor.Get(context.Background(), common.GetPrm{
Address: objectCore.AddressOf(obj),
})
require.ErrorAs(t, err, new(*apistatus.ObjectNotFound))
require.Nil(t, bsRes.Object)
require.Nil(t, bsRes.RawData)
var metaInhumePrm meta.InhumePrm
metaInhumePrm.SetAddresses(objectCore.AddressOf(obj))
metaInhumePrm.SetLockObjectHandling()
metaInhumePrm.SetGCMark()
_, err = sh.metaBase.Inhume(context.Background(), metaInhumePrm)
require.NoError(t, err)
// logs: WARN shard/delete.go:98 can't remove object: object must be flushed from writecache
gcRes := sh.removeGarbage(context.Background())
require.True(t, gcRes.success)
require.Equal(t, uint64(0), gcRes.deleted)
// writecache stores object
wcObj, err = sh.writeCache.Head(context.Background(), objectCore.AddressOf(obj))
require.NoError(t, err)
require.Equal(t, objectCore.AddressOf(obj), objectCore.AddressOf(wcObj))
}