frostfs-node/pkg/services/policer/process.go

package policer

import (
	"context"
	"errors"
	"sync"
	"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"
)

func (p *Policer) Run(ctx context.Context) {
	p.shardPolicyWorker(ctx)
	p.log.Info(logs.PolicerRoutineStopped)
}

func (p *Policer) shardPolicyWorker(ctx context.Context) {
	defer p.taskPool.Release()

	for {
		select {
		case <-ctx.Done():
			return
		default:
		}

		addrs, err := p.keySpaceIterator.Next(ctx, p.batchSize)
		if err != nil {
			if errors.Is(err, engine.ErrEndOfListing) {
				p.keySpaceIterator.Rewind()
				time.Sleep(p.sleepDuration) // finished whole cycle, sleep a bit
				continue
			}
			p.log.Warn(logs.PolicerFailureAtObjectSelectForReplication, zap.Error(err))
		}

		skipMap := newSkipMap()
		for _, addr := range addrs {
			select {
			case <-ctx.Done():
				return
			default:
			}
			if p.objsInWork.inWork(addr.Address) {
				// do not process an object that is in work
				continue
			}

			if err := p.submitPolicerTask(ctx, addr, skipMap); err != nil {
				p.log.Warn(logs.PolicerPoolSubmission, zap.Error(err))
			}
		}
	}
}

func (p *Policer) submitPolicerTask(ctx context.Context, addr object.Info, skipMap *errMap) error {
	return 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
		}

		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.removeOrphanLocksFn(ctx, addr.Address); err != nil {
				p.log.Warn(logs.PolicerUnableToRemoveOrphanLocks,
					zap.Stringer("object", addr.Address),
					zap.Error(err),
				)
			}
		}

		p.cache.Add(addr.Address, time.Now())
		p.objsInWork.remove(addr.Address)
		p.metrics.IncProcessedObjects()
	})
}

type errMap struct {
	sync.Mutex
	skipMap map[cid.ID][]error
}

func newSkipMap() *errMap {
	return &errMap{
		skipMap: make(map[cid.ID][]error),
	}
}

// addSeenError marks err as seen error for the container.
// Returns true is the error has already been added.
func (m *errMap) addSeenError(cnr cid.ID, err error) bool {
	m.Lock()
	defer m.Unlock()

	for _, e := range m.skipMap[cnr] {
		if errors.Is(err, e) {
			return true
		}
	}

	// Restrict list length to avoid possible OOM if some random error is added in future.
	const maxErrListLength = 10

	lst := m.skipMap[cnr]
	if len(lst) < maxErrListLength {
		m.skipMap[cnr] = append(lst, err)
	}
	return false
}