a-savchuk/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Pull requests Activity

[#1445] policer: Make policer remove dangling locks

Browse source 
Signed-off-by: Aleksey Savchuk <a.savchuk@yadro.com>
This commit is contained in:
Aleksey Savchuk 2024-11-08 11:03:41 +03:00
parent 4f22585202
commit bee86f1928
Signed by: a-savchuk
GPG key ID: 70C0A7FF6F9C4639
4 changed files with 59 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
cmd/frostfs-node/object.go
View file

@ -287,8 +287,7 @@ func addPolicer(c *cfg, keyStorage *util.KeyStorage, clientConstructor *cache.Cl
var inhumePrm engine.InhumePrm
inhumePrm.MarkAsGarbage(addr)
_, err := ls.Inhume(ctx, inhumePrm)
if err != nil {
if _, err := ls.Inhume(ctx, inhumePrm); err != nil {
c.log.Warn(logs.FrostFSNodeCouldNotInhumeMarkRedundantCopyAsGarbage,
zap.String("error", err.Error()),
)
@ -297,6 +296,7 @@ func addPolicer(c *cfg, keyStorage *util.KeyStorage, clientConstructor *cache.Cl
policer.WithPool(c.cfgObject.pool.replication),
policer.WithMetrics(c.metricsCollector.PolicerMetrics()),
policer.WithKeyStorage(keyStorage),
policer.WithRemoveDanglingLocksFunc(ls.RemoveDanglingLocks),
)
c.workers = append(c.workers, worker{

1
internal/logs/logs.go
View file

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

10
pkg/services/policer/option.go
View file

@ -51,6 +51,8 @@ type RemoteObjectGetFunc func(context.Context, netmapSDK.NodeInfo, oid.Address)
type LocalObjectGetFunc func(context.Context, oid.Address) (*objectSDK.Object, error)
type RemoveDanglingLocksFunc func(context.Context, oid.Address) error
type cfg struct {
headTimeout time.Duration
@ -87,6 +89,8 @@ type cfg struct {
localObject LocalObjectGetFunc
keyStorage *util.KeyStorage
removeDanglingLocksFn RemoveDanglingLocksFunc
}
func defaultCfg() *cfg {
@ -212,3 +216,9 @@ func WithKeyStorage(ks *util.KeyStorage) Option {
c.keyStorage = ks
}
}
func WithRemoveDanglingLocksFunc(v RemoveDanglingLocksFunc) Option {
return func(c *cfg) {
c.removeDanglingLocksFn = v
}
}

76
pkg/services/policer/process.go
View file

@ -7,6 +7,7 @@ import (
"time"
"git.frostfs.info/TrueCloudLab/frostfs-node/internal/logs"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/engine"
cid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/container/id"
"go.uber.org/zap"
@ -37,41 +38,56 @@ func (p *Policer) shardPolicyWorker(ctx context.Context) {
}
skipMap := newSkipMap()
for i := range addrs {
select {
case <-ctx.Done():
p.taskPool.Release()
for _, addr := range addrs {
p.shardPolicyWorkerForAddr(ctx, addr, skipMap)
}
}
}
func (p *Policer) shardPolicyWorkerForAddr(ctx context.Context, addr object.Info, skipMap *errMap) {
select {
case <-ctx.Done():
p.taskPool.Release()
return
default:
if p.objsInWork.inWork(addr.Address) {
// do not process an object
// that is in work
return
}
err := p.taskPool.Submit(func() {
v, ok := p.cache.Get(addr.Address)
if ok && time.Since(v) < p.evictDuration {
return
default:
addr := addrs[i]
if p.objsInWork.inWork(addr.Address) {
// do not process an object
// that is in work
continue
}
}
err := p.taskPool.Submit(func() {
v, ok := p.cache.Get(addr.Address)
if ok && time.Since(v) < p.evictDuration {
return
}
if !p.objsInWork.add(addr.Address) {
return
}
if p.objsInWork.add(addr.Address) {
err := p.processObject(ctx, addr)
if err != nil && !skipMap.addSeenError(addr.Address.Container(), err) {
p.log.Error(logs.PolicerUnableToProcessObj,
zap.Stringer("object", addr.Address),
zap.String("error", err.Error()))
}
p.cache.Add(addr.Address, time.Now())
p.objsInWork.remove(addr.Address)
p.metrics.IncProcessedObjects()
}
})
if err != nil {
p.log.Warn(logs.PolicerPoolSubmission, zap.Error(err))
err := p.processObject(ctx, addr)
if err != nil && !skipMap.addSeenError(addr.Address.Container(), err) {
p.log.Error(logs.PolicerUnableToProcessObj,
zap.Stringer("object", addr.Address),
zap.String("error", err.Error()))
}
if addr.IsLocked {
if err := p.removeDanglingLocksFn(ctx, addr.Address); err != nil {
p.log.Warn(logs.PolicerUnableToRemoveDanglingLocks,
zap.Stringer("object", addr.Address),
zap.Error(err),
)
}
}
p.cache.Add(addr.Address, time.Now())
p.objsInWork.remove(addr.Address)
p.metrics.IncProcessedObjects()
})
if err != nil {
p.log.Warn(logs.PolicerPoolSubmission, zap.Error(err))
}
}
}