Add zcache key metrics (#101)

* feat: add metrics for existing and deleted cache items

* test: gen. TaskMetrics mocks

* test: add tests for cleanup metrics

* fix: use consistent metric names

* test: update tests & fix race condition in the test

* fix: add error logs for updating metrics

* chore: add mock generation process to README

* fix: move readme to the correct location

Author: ziscky

README.md

@@ -1,8 +1,9 @@
 # golem
+
 Zondax's opinionated golang lib for projects
 
 ## Test
 
 The project contains unit tests you can execute them by running the following command:
 
-`go test -v [TEST FOLDER PATH]`
+`go test -v [TEST FOLDER PATH]`

pkg/metrics/readme.md

@@ -111,4 +111,25 @@ metricsServer.IncrementMetric("my_gauge")
 
 // Decrement a gauge
 metricsServer.DecrementMetric("my_gauge")
-```
+```
+
+### Mocking Support
+
+Use MockTaskMetrics for testing.
+
+```go
+func TestMetrics(t *testing.T) {
+    tm := &metrics.MockTaskMetrics{}
+	  tm.On("RegisterMetric", "local_cache_cleanup_errors", mock.Anything, []string{"error_type"}, mock.Anything).Once().Return(nil)
+    // Use tm in your tests
+}
+
+```
+
+To generate mocks:
+
+- install: https://github.com/vektra/mockery
+- pull the repo with the correct version of the interface you want to mock
+- ` mockery --name TaskMetrics --dir ./pkg/metrics --output . --filename task_metrics_mock.go --structname MockTaskMetrics --inpackage`
+- For usage in tests
+

pkg/metrics/task_metrics_mock.go

@@ -5,20 +5,28 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"time"
+
 	"github.com/allegro/bigcache/v3"
 	"github.com/zondax/golem/pkg/metrics"
 	"github.com/zondax/golem/pkg/metrics/collectors"
 	"go.uber.org/zap"
-	"time"
 )
 
 const (
-	neverExpires          = -1
-	errorTypeLabel        = "error_type"
-	cleanupErrorMetricKey = "localCacheCleanupErrors"
-	iterationErrorLabel   = "iteration_error"
-	unmarshalErrorLabel   = "unmarshal_error"
-	deletionErrorLabel    = "deletion_error"
+	neverExpires = -1
+
+	errorTypeLabel         = "error_type"
+	itemCountLabel         = "item_count"
+	residentItemCountLabel = "resident_item_count"
+	deletedItemCountLabel  = "deleted_item_count"
+	iterationErrorLabel    = "iteration_error"
+	unmarshalErrorLabel    = "unmarshal_error"
+	deletionErrorLabel     = "deletion_error"
+
+	cleanupItemCountMetricKey        = "local_cache_cleanup_item_count"
+	cleanupDeletedItemCountMetricKey = "local_cache_cleanup_deleted_item_count"
+	cleanupErrorMetricKey            = "local_cache_cleanup_errors"
 )
 
 type CacheItem struct {
@@ -164,6 +172,8 @@ func (c *localCache) startCleanupProcess(interval time.Duration) {
 func (c *localCache) cleanupExpiredKeys() {
 	iterator := c.client.Iterator()
 	var keysToDelete []string
+	var totalDeleted int
+	var totalResident int
 
 	for iterator.SetNext() {
 		entry, err := iterator.Value()
@@ -178,34 +188,55 @@ func (c *localCache) cleanupExpiredKeys() {
 			continue
 		}
 
+		totalResident++
+
 		if cachedItem.IsExpired() {
 			keysToDelete = append(keysToDelete, entry.Key())
 		}
 
 		if len(keysToDelete) >= c.batchSize {
-			c.deleteKeysInBatch(keysToDelete)
+			totalDeleted += c.deleteKeysInBatch(keysToDelete)
 			keysToDelete = keysToDelete[:0]
 			time.Sleep(c.throttleTime)
 		}
 	}
 
 	if len(keysToDelete) > 0 {
-		c.deleteKeysInBatch(keysToDelete)
+		totalDeleted += c.deleteKeysInBatch(keysToDelete)
+	}
+
+	// update metrics
+	if err := c.metricsServer.UpdateMetric(cleanupItemCountMetricKey, float64(totalResident-totalDeleted), residentItemCountLabel); err != nil {
+		c.logger.Error("Failed to update cleanup item count metric", zap.Error(err))
+	}
+	if err := c.metricsServer.UpdateMetric(cleanupDeletedItemCountMetricKey, float64(totalDeleted), deletedItemCountLabel); err != nil {
+		c.logger.Error("Failed to update deletion cleanup deleted item count metric", zap.Error(err))
 	}
 }
 
-func (c *localCache) deleteKeysInBatch(keys []string) {
+func (c *localCache) deleteKeysInBatch(keys []string) (deleted int) {
 	for _, key := range keys {
 		if err := c.client.Delete(key); err != nil {
 			if err = c.metricsServer.UpdateMetric(cleanupErrorMetricKey, 1, deletionErrorLabel); err != nil {
 				c.logger.Error("Failed to update deletion error metric", zap.Error(err))
 			}
+			continue
 		}
+		deleted++
 	}
+	return
 }
 
 func (c *localCache) registerCleanupMetrics() {
 	if err := c.metricsServer.RegisterMetric(cleanupErrorMetricKey, "Counts different types of errors occurred during cache cleanup process", []string{errorTypeLabel}, &collectors.Counter{}); err != nil {
 		c.logger.Error("Failed to register cleanup metrics", zap.Error(err))
 	}
+
+	if err := c.metricsServer.RegisterMetric(cleanupItemCountMetricKey, "Counts the valid items in the cache during cache cleanup process", []string{itemCountLabel}, &collectors.Gauge{}); err != nil {
+		c.logger.Error("Failed to register cleanup metrics", zap.Error(err))
+	}
+
+	if err := c.metricsServer.RegisterMetric(cleanupDeletedItemCountMetricKey, "Counts the expired (deleted) items in the cache during cache cleanup process", []string{itemCountLabel}, &collectors.Gauge{}); err != nil {
+		c.logger.Error("Failed to register cleanup metrics", zap.Error(err))
+	}
 }

pkg/zcache/local_cache.go

@@ -4,12 +4,15 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/allegro/bigcache/v3"
-	"github.com/stretchr/testify/suite"
-	"github.com/zondax/golem/pkg/metrics"
 	"os"
+	"sync"
 	"testing"
 	"time"
+
+	"github.com/allegro/bigcache/v3"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/suite"
+	"github.com/zondax/golem/pkg/metrics"
 )
 
 const (
@@ -165,3 +168,61 @@ func (suite *LocalCacheTestSuite) TestCleanupProcessItemDoesNotExpire() {
 	suite.NoError(err, "Did not expect an error when retrieving a non-expiring item")
 	suite.Equal(value, result, "The retrieved value should match the original value")
 }
+
+// insert 1 persistent key and 1 key with a ttl.
+// after cleanup, there will be 1 key in the cache and 1 deleted expired key.
+func (suite *LocalCacheTestSuite) TestCleanupProcessMetrics() {
+	cleanupInterval := 1 * time.Second
+	ttl := 10 * time.Millisecond
+
+	// label:count
+	expected := map[string]int{
+		"resident_item_count": 1,
+		"deleted_item_count":  1,
+	}
+	got := sync.Map{}
+
+	tm := &metrics.MockTaskMetrics{}
+	tm.On("RegisterMetric", "local_cache_cleanup_errors", mock.Anything, []string{"error_type"}, mock.Anything).Once().
+		Return(nil)
+	tm.On("RegisterMetric", "local_cache_cleanup_item_count", mock.Anything, []string{"item_count"}, mock.Anything).Once().
+		Return(nil)
+	tm.On("RegisterMetric", "local_cache_cleanup_deleted_item_count", mock.Anything, []string{"item_count"}, mock.Anything).Once().
+		Return(nil)
+
+	tm.On("UpdateMetric", "local_cache_cleanup_errors", mock.Anything, mock.Anything).Return(nil)
+	tm.On("UpdateMetric", "local_cache_cleanup_item_count", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+		total := args.Get(1).(float64)
+		label := args.Get(2).(string)
+		got.Store(label, int(total))
+	}).Return(nil)
+	tm.On("UpdateMetric", "local_cache_cleanup_deleted_item_count", mock.Anything, mock.Anything).Run(func(args mock.Arguments) {
+		total := args.Get(1).(float64)
+		label := args.Get(2).(string)
+		got.Store(label, int(total))
+	}).Return(nil)
+
+	cache, err := NewLocalCache(&LocalConfig{
+		Prefix:          "test",
+		CleanupInterval: cleanupInterval,
+		MetricServer:    tm,
+	})
+	suite.NoError(err)
+
+	ctx := context.Background()
+	key := "permanentKey"
+	value := "thisValueShouldPersist"
+
+	err = cache.Set(ctx, key, value, ttl)
+	suite.NoError(err)
+	err = cache.Set(ctx, key+"2", value, neverExpires)
+	suite.NoError(err)
+
+	time.Sleep(2 * cleanupInterval)
+	for k, v := range expected {
+		gotV, ok := got.Load(k)
+		suite.Assert().True(ok)
+
+		suite.Assert().Equal(v, gotV)
+	}
+}