Merge 20a00ef into f38069e

CHANGELOG.md

@@ -31,6 +31,7 @@ Changelog for NeoFS Node
 - CLI `--timeout` flag configures whole execution timeout from now (#2124)
 - CLI default timeout for commands with `--await` flag increased to 1m (#2124)
 - BlobStor tries to store object in any sub-storage with free space (#2450)
+- SN does not store container estimations in-mem forever (#2472)
 
 ### Updated
 - `neofs-sdk-go` to `v1.0.0-rc.10`

cmd/neofs-node/container.go

@@ -11,13 +11,15 @@ import (
 	"github.com/nspcc-dev/hrw"
 	containerV2 "github.com/nspcc-dev/neofs-api-go/v2/container"
 	containerGRPC "github.com/nspcc-dev/neofs-api-go/v2/container/grpc"
+	containercontract "github.com/nspcc-dev/neofs-contract/container"
 	"github.com/nspcc-dev/neofs-node/pkg/core/client"
 	containerCore "github.com/nspcc-dev/neofs-node/pkg/core/container"
 	netmapCore "github.com/nspcc-dev/neofs-node/pkg/core/netmap"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine"
 	cntClient "github.com/nspcc-dev/neofs-node/pkg/morph/client/container"
 	"github.com/nspcc-dev/neofs-node/pkg/morph/event"
 	containerEvent "github.com/nspcc-dev/neofs-node/pkg/morph/event/container"
+	netmapEv "github.com/nspcc-dev/neofs-node/pkg/morph/event/netmap"
 	containerTransportGRPC "github.com/nspcc-dev/neofs-node/pkg/network/transport/container/grpc"
 	containerService "github.com/nspcc-dev/neofs-node/pkg/services/container"
 	loadcontroller "github.com/nspcc-dev/neofs-node/pkg/services/container/announcement/load/controller"
@@ -133,23 +135,30 @@ func initContainerService(c *cfg) {
 		cnrWrt.eacls = cachedEACLStorage
 	}
 
+	estimationsLogger := c.log.With(zap.String("component", "container_estimations"))
+
 	localMetrics := &localStorageLoad{
-		log:    c.log,
+		log:    estimationsLogger,
 		engine: c.cfgObject.cfgLocalStorage.localStorage,
 	}
 
 	pubKey := c.key.PublicKey().Bytes()
 
 	resultWriter := &morphLoadWriter{
-		log:            c.log,
+		log:            estimationsLogger,
 		cnrMorphClient: wrapperNoNotary,
 		key:            pubKey,
 	}
 
-	loadAccumulator := loadstorage.New(loadstorage.Prm{})
+	loadAccumulator := loadstorage.New(containercontract.CleanupDelta)
+
+	addNewEpochAsyncNotificationHandler(c, func(e event.Event) {
+		ev := e.(netmapEv.NewEpoch)
+		loadAccumulator.EpochEvent(ev.EpochNumber())
+	})
 
 	loadPlacementBuilder := &loadPlacementBuilder{
-		log:    c.log,
+		log:    estimationsLogger,
 		nmSrc:  c.netMapSource,
 		cnrSrc: cnrSrc,
 	}
@@ -169,7 +178,7 @@ func initContainerService(c *cfg) {
 			},
 			Builder: routeBuilder,
 		},
-		loadroute.WithLogger(c.log),
+		loadroute.WithLogger(estimationsLogger),
 	)
 
 	ctrl := loadcontroller.New(
@@ -179,7 +188,7 @@ func initContainerService(c *cfg) {
 			LocalAnnouncementTarget: loadRouter,
 			ResultReceiver:          loadcontroller.SimpleWriterProvider(resultWriter),
 		},
-		loadcontroller.WithLogger(c.log),
+		loadcontroller.WithLogger(estimationsLogger),
 	)
 
 	setContainerNotificationParser(c, startEstimationNotifyEvent, containerEvent.ParseStartEstimation)

pkg/innerring/blocktimer.go

@@ -95,8 +95,11 @@ func newEpochTimer(args *epochTimerArgs) *timer.BlockTimer {
 				return
 			}
 
+			estimationEpoch := epochN - 1
+			args.l.Info("stop estimation collections", zap.Uint64("epoch", estimationEpoch))
+
 			prm := container.StopEstimationPrm{}
-			prm.SetEpoch(epochN - 1)
+			prm.SetEpoch(estimationEpoch)
 
 			err := args.cnrWrapper.StopEstimation(prm)
 			if err != nil {

pkg/innerring/processors/netmap/process_epoch.go

@@ -45,12 +45,16 @@ func (np *Processor) processNewEpoch(ev netmapEvent.NewEpoch) {
 		return
 	}
 
+	estimationEpoch := epoch - 1
+
 	prm := cntClient.StartEstimationPrm{}
 
-	prm.SetEpoch(epoch - 1)
+	prm.SetEpoch(estimationEpoch)
 	prm.SetHash(ev.TxHash())
 
 	if epoch > 0 { // estimates are invalid in genesis epoch
+		np.log.Info("start estimation collection", zap.Uint64("epoch", estimationEpoch))
+
 		err = np.containerWrp.StartEstimation(prm)
 
 		if err != nil {

pkg/services/container/announcement/load/controller/calls.go

@@ -52,7 +52,7 @@ func (c *Controller) Start(prm StartPrm) {
 }
 
 func (c *announceContext) announce() {
-	c.log.Debug("starting to announce the values of the metrics")
+	c.log.Debug("starting to announce local metrics")
 
 	var (
 		metricsIterator Iterator
@@ -85,6 +85,8 @@ func (c *announceContext) announce() {
 			return true // local metrics don't know about epochs
 		},
 		func(a container.SizeEstimation) error {
+			c.log.Debug("sending local metrics", zap.String("cid", a.Container().EncodeToString()))
+
 			a.SetEpoch(c.epoch) // set epoch explicitly
 			return targetWriter.Put(a)
 		},
@@ -107,7 +109,7 @@ func (c *announceContext) announce() {
 		return
 	}
 
-	c.log.Debug("load announcement successfully finished")
+	c.log.Debug("local load announcement successfully finished")
 }
 
 func (c *Controller) acquireAnnouncement(prm StartPrm) *announceContext {
@@ -126,10 +128,11 @@ func (c *Controller) acquireAnnouncement(prm StartPrm) *announceContext {
 
 	log := c.opts.log.With(
 		zap.Uint64("epoch", prm.Epoch),
+		zap.String("stage", "p2p"),
 	)
 
 	if ctx == nil {
-		log.Debug("announcement is already started")
+		log.Debug("local announcement is already started")
 		return nil
 	}
 
@@ -218,6 +221,7 @@ func (c *Controller) acquireReport(prm StopPrm) *stopContext {
 
 	log := c.opts.log.With(
 		zap.Uint64("epoch", prm.Epoch),
+		zap.String("stage", "report"),
 	)
 
 	if ctx == nil {

pkg/services/container/announcement/load/storage/storage.go

@@ -40,21 +40,22 @@ type storageKey struct {
 type Storage struct {
 	mtx sync.RWMutex
 
-	mItems map[storageKey]*usedSpaceEstimations
+	estLifeCycle uint64
+	mItems       map[storageKey]*usedSpaceEstimations
 }
 
-// Prm groups the required parameters of the Storage's constructor.
-//
-// The component is not parameterizable at the moment.
-type Prm struct{}
-
 // New creates a new instance of the Storage.
 //
 // The created Storage does not require additional
 // initialization and is completely ready for work.
-func New(_ Prm) *Storage {
+//
+// estimationsLifeCycle is a longevity (in epochs) of estimations
+// that are kept in the [Storage] instance. Note, current epoch
+// is controlled with [Storage.EpochEvent].
+func New(estimationsLifeCycle uint64) *Storage {
 	return &Storage{
-		mItems: make(map[storageKey]*usedSpaceEstimations),
+		mItems:       make(map[storageKey]*usedSpaceEstimations),
+		estLifeCycle: estimationsLifeCycle,
 	}
 }
 
@@ -118,6 +119,20 @@ func (s *Storage) Iterate(f loadcontroller.UsedSpaceFilter, h loadcontroller.Use
 	return
 }
 
+// EpochEvent notifies [Storage] about epoch counter updating.
+// Used to remove unused estimations. See [Prm.EstimationsLifeCycle].
+// Blocking operation.
+func (s *Storage) EpochEvent(e uint64) {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+
+	for k := range s.mItems {
+		if k.epoch+s.estLifeCycle < e {
+			delete(s.mItems, k)
+		}
+	}
+}
+
 func finalEstimation(vals []uint64) uint64 {
 	sort.Slice(vals, func(i, j int) bool {
 		return vals[i] < vals[j]

pkg/services/container/announcement/load/storage/storage_test.go

@@ -18,7 +18,7 @@ func TestStorage(t *testing.T) {
 
 	const opinionsNum = 100
 
-	s := New(Prm{})
+	s := New(0)
 
 	opinions := make([]uint64, opinionsNum)
 	for i := range opinions {
@@ -48,3 +48,46 @@ func TestStorage(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, 1, iterCounter)
 }
+
+func TestStorage_NewEpoch(t *testing.T) {
+	const epoch uint64 = 13
+	const lifeCycle = 5
+
+	s := New(lifeCycle)
+
+	var a container.SizeEstimation
+	a.SetContainer(cidtest.ID())
+	a.SetEpoch(epoch)
+
+	err := s.Put(a)
+	require.NoError(t, err)
+
+	for i := uint64(1); i <= lifeCycle+1; i++ {
+		ee := getEstimations(t, s)
+		require.NoError(t, err)
+		require.Len(t, ee, 1)
+
+		s.EpochEvent(epoch + i)
+	}
+
+	ee := getEstimations(t, s)
+	require.NoError(t, err)
+	require.Empty(t, ee)
+}
+
+func getEstimations(t *testing.T, s *Storage) []container.SizeEstimation {
+	var ee []container.SizeEstimation
+
+	err := s.Iterate(
+		func(e container.SizeEstimation) bool {
+			return true
+		},
+		func(e container.SizeEstimation) error {
+			ee = append(ee, e)
+			return nil
+		},
+	)
+	require.NoError(t, err)
+
+	return ee
+}