forked from TrueCloudLab/frostfs-node
[#634] shard: Add writecache inhume tests
Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>
This commit is contained in:
parent
e5d18e7a85
commit
0882840bf5
1 changed files with 126 additions and 0 deletions
|
@ -7,10 +7,13 @@ import (
|
|||
|
||||
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"
|
||||
|
@ -167,3 +170,126 @@ func Test_GCDropsLockedExpiredComplexObject(t *testing.T) {
|
|||
_, 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))
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue