node: Add payload metrics per shard

Closes #2023.

Signed-off-by: Pavel Karpy <[email protected]>

CHANGELOG.md

@@ -8,6 +8,7 @@ Changelog for NeoFS Node
 - `dump-names` command for adm
 - `renew-domain` command for adm
 - Stored payload metric per container (#2116)
+- Stored payload metric per shard (#2023)
 
 ### Fixed
 - `neo-go` RPC connection lost handling by IR (#1337)

pkg/local_object_storage/engine/metrics.go

@@ -23,6 +23,7 @@ type MetricRegister interface {
 	SetReadonly(shardID string, readonly bool)
 
 	AddToContainerSize(cnrID string, size int64)
+	AddToPayloadCounter(shardID string, size int64)
 }
 
 func elapsed(addFunc func(d time.Duration)) func() {

pkg/local_object_storage/engine/shards.go

@@ -53,6 +53,10 @@ func (m *metricsWithID) AddToContainerSize(cnr string, size int64) {
 	m.mw.AddToContainerSize(cnr, size)
 }
 
+func (m *metricsWithID) AddToPayloadSize(size int64) {
+	m.mw.AddToPayloadCounter(m.id, size)
+}
+
 // AddShard adds a new shard to the storage engine.
 //
 // Returns any error encountered that did not allow adding a shard.

pkg/local_object_storage/shard/delete.go

@@ -78,11 +78,16 @@ func (s *Shard) delete(prm DeletePrm) (DeleteRes, error) {
 		return DeleteRes{}, err // stop on metabase error ?
 	}
 
+	var totalRemovedPayload uint64
+
 	s.decObjectCounterBy(physical, res.RawObjectsRemoved())
 	s.decObjectCounterBy(logical, res.AvailableObjectsRemoved())
 	for i := range prm.addr {
-		s.addToContainerSize(prm.addr[i].Container().EncodeToString(), -int64(res.RemovedObjectSizes()[i]))
+		removedPayload := res.RemovedObjectSizes()[i]
+		totalRemovedPayload += removedPayload
+		s.addToContainerSize(prm.addr[i].Container().EncodeToString(), -int64(removedPayload))
 	}
+	s.addToPayloadCounter(-int64(totalRemovedPayload))
 
 	for i := range prm.addr {
 		var delPrm common.DeletePrm

pkg/local_object_storage/shard/metrics_test.go

@@ -19,6 +19,7 @@ import (
 type metricsStore struct {
 	objCounters map[string]uint64
 	cnrSize     map[string]int64
+	pldSize     int64
 	readOnly    bool
 }
 
@@ -61,6 +62,10 @@ func (m metricsStore) AddToContainerSize(cnr string, size int64) {
 	m.cnrSize[cnr] += size
 }
 
+func (m *metricsStore) AddToPayloadSize(size int64) {
+	m.pldSize += size
+}
+
 const physical = "phy"
 const logical = "logic"
 const readonly = "readonly"
@@ -84,12 +89,17 @@ func TestCounters(t *testing.T) {
 		require.Zero(t, mm.objCounters[physical])
 		require.Zero(t, mm.objCounters[logical])
 		require.Empty(t, mm.cnrSize)
+		require.Zero(t, mm.pldSize)
 	})
 
+	var totalPayload int64
+
 	expectedSizes := make(map[string]int64)
 	for i := range oo {
 		cnr, _ := oo[i].ContainerID()
-		expectedSizes[cnr.EncodeToString()] += int64(oo[i].PayloadSize())
+		oSize := int64(oo[i].PayloadSize())
+		expectedSizes[cnr.EncodeToString()] += oSize
+		totalPayload += oSize
 	}
 
 	t.Run("put", func(t *testing.T) {
@@ -105,6 +115,7 @@ func TestCounters(t *testing.T) {
 		require.Equal(t, uint64(objNumber), mm.objCounters[physical])
 		require.Equal(t, uint64(objNumber), mm.objCounters[logical])
 		require.Equal(t, expectedSizes, mm.cnrSize)
+		require.Equal(t, totalPayload, mm.pldSize)
 	})
 
 	t.Run("inhume_GC", func(t *testing.T) {
@@ -121,6 +132,7 @@ func TestCounters(t *testing.T) {
 		require.Equal(t, uint64(objNumber), mm.objCounters[physical])
 		require.Equal(t, uint64(objNumber-inhumedNumber), mm.objCounters[logical])
 		require.Equal(t, expectedSizes, mm.cnrSize)
+		require.Equal(t, totalPayload, mm.pldSize)
 
 		oo = oo[inhumedNumber:]
 	})
@@ -141,6 +153,7 @@ func TestCounters(t *testing.T) {
 		require.Equal(t, phy, mm.objCounters[physical])
 		require.Equal(t, logic-uint64(inhumedNumber), mm.objCounters[logical])
 		require.Equal(t, expectedSizes, mm.cnrSize)
+		require.Equal(t, totalPayload, mm.pldSize)
 
 		oo = oo[inhumedNumber:]
 	})
@@ -159,11 +172,16 @@ func TestCounters(t *testing.T) {
 
 		require.Equal(t, phy-uint64(deletedNumber), mm.objCounters[physical])
 		require.Equal(t, logic-uint64(deletedNumber), mm.objCounters[logical])
+		var totalRemovedpayload uint64
 		for i := range oo[:deletedNumber] {
+			removedPayload := oo[i].PayloadSize()
+			totalRemovedpayload += removedPayload
+
 			cnr, _ := oo[i].ContainerID()
-			expectedSizes[cnr.EncodeToString()] -= int64(oo[i].PayloadSize())
+			expectedSizes[cnr.EncodeToString()] -= int64(removedPayload)
 		}
 		require.Equal(t, expectedSizes, mm.cnrSize)
+		require.Equal(t, totalPayload-int64(totalRemovedpayload), mm.pldSize)
 	})
 }
 

pkg/local_object_storage/shard/shard.go

@@ -56,6 +56,9 @@ type MetricsWriter interface {
 	// AddToContainerSize must add a value to the container size.
 	// Value can be negative.
 	AddToContainerSize(cnr string, value int64)
+	// AddToPayloadSize must add a value to the payload size.
+	// Value can be negative.
+	AddToPayloadSize(value int64)
 	// IncObjectCounter must increment shard's object counter taking into account
 	// object type.
 	IncObjectCounter(objectType string)
@@ -346,6 +349,8 @@ func (s *Shard) updateMetrics() {
 			return
 		}
 
+		var totalPayload uint64
+
 		for i := range cnrList {
 			size, err := s.metaBase.ContainerSize(cnrList[i])
 			if err != nil {
@@ -355,7 +360,10 @@ func (s *Shard) updateMetrics() {
 				continue
 			}
 			s.metricsWriter.AddToContainerSize(cnrList[i].EncodeToString(), int64(size))
+			totalPayload += size
 		}
+
+		s.metricsWriter.AddToPayloadSize(int64(totalPayload))
 	}
 }
 
@@ -379,3 +387,9 @@ func (s *Shard) addToContainerSize(cnr string, size int64) {
 		s.cfg.metricsWriter.AddToContainerSize(cnr, size)
 	}
 }
+
+func (s *Shard) addToPayloadCounter(size int64) {
+	if s.cfg.metricsWriter != nil {
+		s.cfg.metricsWriter.AddToPayloadSize(size)
+	}
+}

pkg/metrics/engine.go

@@ -20,6 +20,7 @@ type (
 		searchDuration                prometheus.Counter
 		listObjectsDuration           prometheus.Counter
 		containerSize                 prometheus.GaugeVec
+		payloadSize                   prometheus.GaugeVec
 	}
 )
 
@@ -110,6 +111,13 @@ func newEngineMetrics() engineMetrics {
 			Name:      "container_size",
 			Help:      "Accumulated size of all objects in a container",
 		}, []string{containerIDLabelKey})
+
+		payloadSize = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+			Namespace: storageNodeNameSpace,
+			Subsystem: engineSubsystem,
+			Name:      "payload_size",
+			Help:      "Accumulated size of all objects in a shard",
+		}, []string{shardIDLabelKey})
 	)
 
 	return engineMetrics{
@@ -125,6 +133,7 @@ func newEngineMetrics() engineMetrics {
 		searchDuration:                searchDuration,
 		listObjectsDuration:           listObjectsDuration,
 		containerSize:                 *containerSize,
+		payloadSize:                   *payloadSize,
 	}
 }
 
@@ -141,6 +150,7 @@ func (m engineMetrics) register() {
 	prometheus.MustRegister(m.searchDuration)
 	prometheus.MustRegister(m.listObjectsDuration)
 	prometheus.MustRegister(m.containerSize)
+	prometheus.MustRegister(m.payloadSize)
 }
 
 func (m engineMetrics) AddListContainersDuration(d time.Duration) {
@@ -194,3 +204,7 @@ func (m engineMetrics) AddToContainerSize(cnrID string, size int64) {
 		},
 	).Add(float64(size))
 }
+
+func (m engineMetrics) AddToPayloadCounter(shardID string, size int64) {
+	m.payloadSize.With(prometheus.Labels{shardIDLabelKey: shardID}).Add(float64(size))
+}