[#656] policer: Add "bad" testcase

Signed-off-by: Anton Nikiforov <an.nikiforov@yadro.com>
This commit is contained in:
Anton Nikiforov 2023-09-04 12:00:49 +03:00 committed by Evgenii Stratonikov
parent 054e3ef3d3
commit 88d50e4c77
4 changed files with 49 additions and 25 deletions

View file

@ -53,6 +53,7 @@ const (
PolicerRoutineStopped = "routine stopped" PolicerRoutineStopped = "routine stopped"
PolicerFailureAtObjectSelectForReplication = "failure at object select for replication" PolicerFailureAtObjectSelectForReplication = "failure at object select for replication"
PolicerPoolSubmission = "pool submission" PolicerPoolSubmission = "pool submission"
PolicerUnableToProcessObj = "unable to process object"
ReplicatorFinishWork = "finish work" ReplicatorFinishWork = "finish work"
ReplicatorCouldNotGetObjectFromLocalStorage = "could not get object from local storage" ReplicatorCouldNotGetObjectFromLocalStorage = "could not get object from local storage"
ReplicatorCouldNotReplicateObject = "could not replicate object" ReplicatorCouldNotReplicateObject = "could not replicate object"

View file

@ -3,6 +3,7 @@ package policer
import ( import (
"context" "context"
"errors" "errors"
"fmt"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs" "git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
containercore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container" containercore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/container"
@ -16,48 +17,33 @@ import (
"go.uber.org/zap" "go.uber.org/zap"
) )
func (p *Policer) processObject(ctx context.Context, addrWithType objectcore.AddressWithType) { func (p *Policer) processObject(ctx context.Context, addrWithType objectcore.AddressWithType) error {
addr := addrWithType.Address addr := addrWithType.Address
idCnr := addr.Container() idCnr := addr.Container()
idObj := addr.Object() idObj := addr.Object()
cnr, err := p.cnrSrc.Get(idCnr) cnr, err := p.cnrSrc.Get(idCnr)
if err != nil { if err != nil {
p.log.Error(logs.PolicerCouldNotGetContainer,
zap.Stringer("cid", idCnr),
zap.String("error", err.Error()),
)
if client.IsErrContainerNotFound(err) { if client.IsErrContainerNotFound(err) {
existed, err := containercore.WasRemoved(p.cnrSrc, idCnr) existed, errWasRemoved := containercore.WasRemoved(p.cnrSrc, idCnr)
if err != nil { if errWasRemoved != nil {
p.log.Error(logs.PolicerCouldNotConfirmContainerRemoval, return fmt.Errorf("%s: %w", logs.PolicerCouldNotConfirmContainerRemoval, errWasRemoved)
zap.Stringer("cid", idCnr),
zap.Stringer("oid", idObj),
zap.String("error", err.Error()))
} else if existed { } else if existed {
err := p.buryFn(ctx, addrWithType.Address) err := p.buryFn(ctx, addrWithType.Address)
if err != nil { if err != nil {
p.log.Error(logs.PolicerCouldNotInhumeObjectWithMissingContainer, return fmt.Errorf("%s: %w", logs.PolicerCouldNotInhumeObjectWithMissingContainer, err)
zap.Stringer("cid", idCnr),
zap.Stringer("oid", idObj),
zap.String("error", err.Error()))
} }
} }
} }
return return fmt.Errorf("%s: %w", logs.PolicerCouldNotGetContainer, err)
} }
policy := cnr.Value.PlacementPolicy() policy := cnr.Value.PlacementPolicy()
nn, err := p.placementBuilder.BuildPlacement(idCnr, &idObj, policy) nn, err := p.placementBuilder.BuildPlacement(idCnr, &idObj, policy)
if err != nil { if err != nil {
p.log.Error(logs.PolicerCouldNotBuildPlacementVectorForObject, return fmt.Errorf("%s: %w", logs.PolicerCouldNotBuildPlacementVectorForObject, err)
zap.Stringer("cid", idCnr),
zap.String("error", err.Error()),
)
return
} }
c := &placementRequirements{} c := &placementRequirements{}
@ -73,7 +59,7 @@ func (p *Policer) processObject(ctx context.Context, addrWithType objectcore.Add
for i := range nn { for i := range nn {
select { select {
case <-ctx.Done(): case <-ctx.Done():
return return ctx.Err()
default: default:
} }
@ -87,6 +73,7 @@ func (p *Policer) processObject(ctx context.Context, addrWithType objectcore.Add
p.cbRedundantCopy(ctx, addr) p.cbRedundantCopy(ctx, addr)
} }
return nil
} }
type placementRequirements struct { type placementRequirements struct {

View file

@ -12,6 +12,7 @@ import (
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/engine" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/replicator" "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/services/replicator"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client"
apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status" apistatus "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/client/status"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id" cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/netmap"
@ -245,7 +246,8 @@ func TestProcessObject(t *testing.T) {
Type: ti.objType, Type: ti.objType,
} }
p.processObject(context.Background(), addrWithType) err := p.processObject(context.Background(), addrWithType)
require.NoError(t, err)
sort.Ints(gotReplicateTo) sort.Ints(gotReplicateTo)
require.Equal(t, ti.wantRemoveRedundant, gotRemoveRedundant) require.Equal(t, ti.wantRemoveRedundant, gotRemoveRedundant)
@ -254,6 +256,35 @@ func TestProcessObject(t *testing.T) {
} }
} }
func TestProcessObjectError(t *testing.T) {
addr := oidtest.Address()
// Container source
cnr := &container.Container{}
cnr.Value.Init()
source := containerSrc{
get: func(id cid.ID) (*container.Container, error) {
return nil, new(apistatus.ContainerNotFound)
},
deletionInfo: func(id cid.ID) (*container.DelInfo, error) {
return nil, new(apistatus.ContainerNotFound)
},
}
buryFn := func(ctx context.Context, a oid.Address) error {
t.Errorf("unexpected object buried: %v", a)
return nil
}
p := New(
WithContainerSource(source),
WithBuryFunc(buryFn),
)
addrWithType := objectcore.AddressWithType{
Address: addr,
}
require.True(t, client.IsErrContainerNotFound(p.processObject(context.Background(), addrWithType)))
}
func TestIteratorContract(t *testing.T) { func TestIteratorContract(t *testing.T) {
addr := oidtest.Address() addr := oidtest.Address()
objs := []objectcore.AddressWithType{{ objs := []objectcore.AddressWithType{{

View file

@ -52,7 +52,12 @@ func (p *Policer) shardPolicyWorker(ctx context.Context) {
} }
if p.objsInWork.add(addr.Address) { if p.objsInWork.add(addr.Address) {
p.processObject(ctx, addr) err := p.processObject(ctx, addr)
if err != nil {
p.log.Error(logs.PolicerUnableToProcessObj,
zap.Stringer("object", addr.Address),
zap.String("error", err.Error()))
}
p.cache.Add(addr.Address, time.Now()) p.cache.Add(addr.Address, time.Now())
p.objsInWork.remove(addr.Address) p.objsInWork.remove(addr.Address)
} }