frostfs-node/pkg/local_object_storage/shard/metrics_test.go

package shard

import (
	"context"
	"path/filepath"
	"sync"
	"testing"

	objectcore "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/core/object"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/blobstor/fstree"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/internal/testutil"
	meta "git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/metabase"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/pilorama"
	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
	objectSDK "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object"
	oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
	"github.com/stretchr/testify/require"
)

type metricsStore struct {
	mtx         sync.Mutex
	objCounters map[string]uint64
	cnrSize     map[string]int64
	pldSize     int64
	mode        mode.Mode
	errCounter  int64
}

func (m *metricsStore) SetShardID(_ string) {}

func (m *metricsStore) SetObjectCounter(objectType string, v uint64) {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.objCounters[objectType] = v
}

func (m *metricsStore) getObjectCounter(objectType string) uint64 {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	return m.objCounters[objectType]
}

func (m *metricsStore) containerSizes() map[string]int64 {
	m.mtx.Lock()
	defer m.mtx.Unlock()

	r := make(map[string]int64, len(m.cnrSize))
	for c, s := range m.cnrSize {
		r[c] = s
	}
	return r
}

func (m *metricsStore) payloadSize() int64 {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	return m.pldSize
}

func (m *metricsStore) AddToObjectCounter(objectType string, delta int) {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	switch {
	case delta > 0:
		m.objCounters[objectType] += uint64(delta)
	case delta < 0:
		uDelta := uint64(-delta)

		if m.objCounters[objectType] >= uDelta {
			m.objCounters[objectType] -= uDelta
		} else {
			m.objCounters[objectType] = 0
		}
	case delta == 0:
		return
	}
}

func (m *metricsStore) IncObjectCounter(objectType string) {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.objCounters[objectType] += 1
}

func (m *metricsStore) DecObjectCounter(objectType string) {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.AddToObjectCounter(objectType, -1)
}

func (m *metricsStore) SetMode(mode mode.Mode) {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.mode = mode
}

func (m *metricsStore) AddToContainerSize(cnr string, size int64) {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.cnrSize[cnr] += size
}

func (m *metricsStore) AddToPayloadSize(size int64) {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.pldSize += size
}

func (m *metricsStore) IncErrorCounter() {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.errCounter += 1
}

func (m *metricsStore) ClearErrorCounter() {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.errCounter = 0
}

func (m *metricsStore) DeleteShardMetrics() {
	m.mtx.Lock()
	defer m.mtx.Unlock()
	m.errCounter = 0
}

func TestCounters(t *testing.T) {
	t.Parallel()

	dir := t.TempDir()
	sh, mm := shardWithMetrics(t, dir)

	sh.SetMode(mode.ReadOnly)
	require.Equal(t, mode.ReadOnly, mm.mode)
	sh.SetMode(mode.ReadWrite)
	require.Equal(t, mode.ReadWrite, mm.mode)

	const objNumber = 10
	oo := make([]*objectSDK.Object, objNumber)
	for i := 0; i < objNumber; i++ {
		oo[i] = testutil.GenerateObject()
	}

	t.Run("defaults", func(t *testing.T) {
		require.Zero(t, mm.getObjectCounter(physical))
		require.Zero(t, mm.getObjectCounter(logical))
		require.Empty(t, mm.containerSizes())
		require.Zero(t, mm.payloadSize())
	})

	var totalPayload int64

	expectedSizes := make(map[string]int64)
	for i := range oo {
		cnr, _ := oo[i].ContainerID()
		oSize := int64(oo[i].PayloadSize())
		expectedSizes[cnr.EncodeToString()] += oSize
		totalPayload += oSize
	}

	var prm PutPrm

	for i := 0; i < objNumber; i++ {
		prm.SetObject(oo[i])

		_, err := sh.Put(context.Background(), prm)
		require.NoError(t, err)
	}

	require.Equal(t, uint64(objNumber), mm.getObjectCounter(physical))
	require.Equal(t, uint64(objNumber), mm.getObjectCounter(logical))
	require.Equal(t, expectedSizes, mm.containerSizes())
	require.Equal(t, totalPayload, mm.payloadSize())

	t.Run("inhume_GC", func(t *testing.T) {
		var prm InhumePrm
		inhumedNumber := objNumber / 4

		for i := 0; i < inhumedNumber; i++ {
			prm.MarkAsGarbage(objectcore.AddressOf(oo[i]))

			_, err := sh.Inhume(context.Background(), prm)
			require.NoError(t, err)
		}

		require.Equal(t, uint64(objNumber), mm.getObjectCounter(physical))
		require.Equal(t, uint64(objNumber-inhumedNumber), mm.getObjectCounter(logical))
		require.Equal(t, expectedSizes, mm.containerSizes())
		require.Equal(t, totalPayload, mm.payloadSize())

		oo = oo[inhumedNumber:]
	})

	t.Run("inhume_TS", func(t *testing.T) {
		var prm InhumePrm
		ts := objectcore.AddressOf(testutil.GenerateObject())

		phy := mm.getObjectCounter(physical)
		logic := mm.getObjectCounter(logical)

		inhumedNumber := int(phy / 4)
		prm.SetTarget(ts, addrFromObjs(oo[:inhumedNumber])...)

		_, err := sh.Inhume(context.Background(), prm)
		require.NoError(t, err)

		require.Equal(t, phy, mm.getObjectCounter(physical))
		require.Equal(t, logic-uint64(inhumedNumber), mm.getObjectCounter(logical))
		require.Equal(t, expectedSizes, mm.containerSizes())
		require.Equal(t, totalPayload, mm.payloadSize())

		oo = oo[inhumedNumber:]
	})

	t.Run("Delete", func(t *testing.T) {
		var prm DeletePrm

		phy := mm.getObjectCounter(physical)
		logic := mm.getObjectCounter(logical)

		deletedNumber := int(phy / 4)
		prm.SetAddresses(addrFromObjs(oo[:deletedNumber])...)

		_, err := sh.Delete(context.Background(), prm)
		require.NoError(t, err)

		require.Equal(t, phy-uint64(deletedNumber), mm.getObjectCounter(physical))
		require.Equal(t, logic-uint64(deletedNumber), mm.getObjectCounter(logical))
		var totalRemovedpayload uint64
		for i := range oo[:deletedNumber] {
			removedPayload := oo[i].PayloadSize()
			totalRemovedpayload += removedPayload

			cnr, _ := oo[i].ContainerID()
			expectedSizes[cnr.EncodeToString()] -= int64(removedPayload)
		}
		require.Equal(t, expectedSizes, mm.containerSizes())
		require.Equal(t, totalPayload-int64(totalRemovedpayload), mm.payloadSize())
	})
}

func shardWithMetrics(t *testing.T, path string) (*Shard, *metricsStore) {
	blobOpts := []blobstor.Option{
		blobstor.WithStorages([]blobstor.SubStorage{
			{
				Storage: fstree.New(
					fstree.WithDirNameLen(2),
					fstree.WithPath(filepath.Join(path, "blob")),
					fstree.WithDepth(1)),
			},
		}),
	}

	mm := &metricsStore{
		objCounters: map[string]uint64{
			"phy":   0,
			"logic": 0,
		},
		cnrSize: make(map[string]int64),
	}

	sh := New(
		WithID(NewIDFromBytes([]byte{})),
		WithBlobStorOptions(blobOpts...),
		WithPiloramaOptions(pilorama.WithPath(filepath.Join(path, "pilorama"))),
		WithMetaBaseOptions(
			meta.WithPath(filepath.Join(path, "meta")),
			meta.WithEpochState(epochState{})),
		WithMetricsWriter(mm),
		WithGCRemoverSleepInterval(time.Hour),
	)
	require.NoError(t, sh.Open(context.Background()))
	require.NoError(t, sh.Init(context.Background()))

	t.Cleanup(func() {
		sh.Close()
	})

	return sh, mm
}

func addrFromObjs(oo []*objectSDK.Object) []oid.Address {
	aa := make([]oid.Address, len(oo))

	for i := 0; i < len(oo); i++ {
		aa[i] = objectcore.AddressOf(oo[i])
	}

	return aa
}