Do not store container size estimations forever (#2472)

CHANGELOG.md

@@ -32,6 +32,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
+}