forked from TrueCloudLab/frostfs-node
[#668] shard/test: Properly check event processing
See https://git.frostfs.info/TrueCloudLab/frostfs-node/actions/runs/1594/jobs/2 Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>
This commit is contained in:
parent
39879fa868
commit
36759f8434
3 changed files with 47 additions and 34 deletions
|
@ -160,33 +160,37 @@ func (gc *gc) listenEvents(ctx context.Context) {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
v, ok := gc.mEventHandler[event.typ()]
|
gc.handleEvent(ctx, event)
|
||||||
if !ok {
|
}
|
||||||
continue
|
}
|
||||||
}
|
|
||||||
|
|
||||||
v.cancelFunc()
|
func (gc *gc) handleEvent(ctx context.Context, event Event) {
|
||||||
v.prevGroup.Wait()
|
v, ok := gc.mEventHandler[event.typ()]
|
||||||
|
if !ok {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
var runCtx context.Context
|
v.cancelFunc()
|
||||||
runCtx, v.cancelFunc = context.WithCancel(ctx)
|
v.prevGroup.Wait()
|
||||||
|
|
||||||
v.prevGroup.Add(len(v.handlers))
|
var runCtx context.Context
|
||||||
|
runCtx, v.cancelFunc = context.WithCancel(ctx)
|
||||||
|
|
||||||
for i := range v.handlers {
|
v.prevGroup.Add(len(v.handlers))
|
||||||
h := v.handlers[i]
|
|
||||||
|
|
||||||
err := gc.workerPool.Submit(func() {
|
for i := range v.handlers {
|
||||||
defer v.prevGroup.Done()
|
h := v.handlers[i]
|
||||||
h(runCtx, event)
|
|
||||||
})
|
|
||||||
if err != nil {
|
|
||||||
gc.log.Warn(logs.ShardCouldNotSubmitGCJobToWorkerPool,
|
|
||||||
zap.String("error", err.Error()),
|
|
||||||
)
|
|
||||||
|
|
||||||
v.prevGroup.Done()
|
err := gc.workerPool.Submit(func() {
|
||||||
}
|
defer v.prevGroup.Done()
|
||||||
|
h(runCtx, event)
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
gc.log.Warn(logs.ShardCouldNotSubmitGCJobToWorkerPool,
|
||||||
|
zap.String("error", err.Error()),
|
||||||
|
)
|
||||||
|
|
||||||
|
v.prevGroup.Done()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -4,12 +4,12 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"testing"
|
"testing"
|
||||||
"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/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/util"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
|
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
|
||||||
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"
|
||||||
|
@ -25,7 +25,12 @@ func Test_GCDropsLockedExpiredSimpleObject(t *testing.T) {
|
||||||
Value: 100,
|
Value: 100,
|
||||||
}
|
}
|
||||||
|
|
||||||
sh := newCustomShard(t, false, shardOptions{metaOptions: []meta.Option{meta.WithEpochState(epoch)}})
|
sh := newCustomShard(t, false, shardOptions{
|
||||||
|
metaOptions: []meta.Option{meta.WithEpochState(epoch)},
|
||||||
|
additionalShardOptions: []Option{WithGCWorkerPoolInitializer(func(int) util.WorkerPool {
|
||||||
|
return util.NewPseudoWorkerPool() // synchronous event processing
|
||||||
|
})},
|
||||||
|
})
|
||||||
|
|
||||||
cnr := cidtest.ID()
|
cnr := cidtest.ID()
|
||||||
|
|
||||||
|
@ -60,14 +65,12 @@ func Test_GCDropsLockedExpiredSimpleObject(t *testing.T) {
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
|
||||||
epoch.Value = 105
|
epoch.Value = 105
|
||||||
sh.NotificationChannel() <- EventNewEpoch(epoch.Value)
|
sh.gc.handleEvent(context.Background(), EventNewEpoch(epoch.Value))
|
||||||
|
|
||||||
var getPrm GetPrm
|
var getPrm GetPrm
|
||||||
getPrm.SetAddress(objectCore.AddressOf(obj))
|
getPrm.SetAddress(objectCore.AddressOf(obj))
|
||||||
require.Eventually(t, func() bool {
|
_, err = sh.Get(context.Background(), getPrm)
|
||||||
_, err = sh.Get(context.Background(), getPrm)
|
require.True(t, client.IsErrObjectNotFound(err), "expired object must be deleted")
|
||||||
return client.IsErrObjectNotFound(err)
|
|
||||||
}, 3*time.Second, 1*time.Second, "expired object must be deleted")
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func Test_GCDropsLockedExpiredComplexObject(t *testing.T) {
|
func Test_GCDropsLockedExpiredComplexObject(t *testing.T) {
|
||||||
|
@ -118,7 +121,12 @@ func Test_GCDropsLockedExpiredComplexObject(t *testing.T) {
|
||||||
|
|
||||||
linkID, _ := link.ID()
|
linkID, _ := link.ID()
|
||||||
|
|
||||||
sh := newCustomShard(t, false, shardOptions{metaOptions: []meta.Option{meta.WithEpochState(epoch)}})
|
sh := newCustomShard(t, false, shardOptions{
|
||||||
|
metaOptions: []meta.Option{meta.WithEpochState(epoch)},
|
||||||
|
additionalShardOptions: []Option{WithGCWorkerPoolInitializer(func(int) util.WorkerPool {
|
||||||
|
return util.NewPseudoWorkerPool() // synchronous event processing
|
||||||
|
})},
|
||||||
|
})
|
||||||
|
|
||||||
lock := testutil.GenerateObjectWithCID(cnr)
|
lock := testutil.GenerateObjectWithCID(cnr)
|
||||||
lock.SetType(objectSDK.TypeLock)
|
lock.SetType(objectSDK.TypeLock)
|
||||||
|
@ -152,10 +160,8 @@ func Test_GCDropsLockedExpiredComplexObject(t *testing.T) {
|
||||||
require.True(t, errors.As(err, &splitInfoError), "split info must be provided")
|
require.True(t, errors.As(err, &splitInfoError), "split info must be provided")
|
||||||
|
|
||||||
epoch.Value = 105
|
epoch.Value = 105
|
||||||
sh.NotificationChannel() <- EventNewEpoch(epoch.Value)
|
sh.gc.handleEvent(context.Background(), EventNewEpoch(epoch.Value))
|
||||||
|
|
||||||
require.Eventually(t, func() bool {
|
_, err = sh.Get(context.Background(), getPrm)
|
||||||
_, err = sh.Get(context.Background(), getPrm)
|
require.True(t, client.IsErrObjectNotFound(err), "expired complex object must be deleted on epoch after lock expires")
|
||||||
return client.IsErrObjectNotFound(err)
|
|
||||||
}, 3*time.Second, 1*time.Second, "expired complex object must be deleted on epoch after lock expires")
|
|
||||||
}
|
}
|
||||||
|
|
|
@ -36,6 +36,8 @@ type shardOptions struct {
|
||||||
wcOpts writecacheconfig.Options
|
wcOpts writecacheconfig.Options
|
||||||
bsOpts []blobstor.Option
|
bsOpts []blobstor.Option
|
||||||
metaOptions []meta.Option
|
metaOptions []meta.Option
|
||||||
|
|
||||||
|
additionalShardOptions []Option
|
||||||
}
|
}
|
||||||
|
|
||||||
func newShard(t testing.TB, enableWriteCache bool) *Shard {
|
func newShard(t testing.TB, enableWriteCache bool) *Shard {
|
||||||
|
@ -114,6 +116,7 @@ func newCustomShard(t testing.TB, enableWriteCache bool, o shardOptions) *Shard
|
||||||
}),
|
}),
|
||||||
WithGCRemoverSleepInterval(100 * time.Millisecond),
|
WithGCRemoverSleepInterval(100 * time.Millisecond),
|
||||||
}
|
}
|
||||||
|
opts = append(opts, o.additionalShardOptions...)
|
||||||
|
|
||||||
sh = New(opts...)
|
sh = New(opts...)
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue