LocalCache: Add cleanup method (#86)

* LocalCache: Add cleanup method

* linter

* Enhanced local cache gc

* Update readme

* Set error

* tests

* tests and minor fix

* tests

* fix tests

Author: lucaslopezf

pkg/metrics/register.go

@@ -23,6 +23,10 @@ var (
 )
 
 func (t *taskMetrics) RegisterMetric(name string, help string, labels []string, handler MetricHandler) error {
+	if handler == nil {
+		panic("handler is mandatory")
+	}
+
 	labels = append(labels, APPNameLabel)
 	var metric prometheus.Collector
 

pkg/zcache/combined_cache.go

@@ -19,7 +19,7 @@ type combinedCache struct {
 	remoteCache        RemoteCache
 	logger             *zap.Logger
 	isRemoteBestEffort bool
-	metricsServer      *metrics.TaskMetrics
+	metricsServer      metrics.TaskMetrics
 	appName            string
 }
 
@@ -111,7 +111,7 @@ func (c *combinedCache) IsNotFoundError(err error) bool {
 }
 
 func (c *combinedCache) SetupAndMonitorMetrics(appName string, metricsServer metrics.TaskMetrics, updateInterval time.Duration) []error {
-	c.metricsServer = &metricsServer
+	c.metricsServer = metricsServer
 	c.appName = appName
 
 	errs := setupAndMonitorCacheMetrics(appName, metricsServer, c, updateInterval)

pkg/zcache/combined_cache_test.go

@@ -3,6 +3,7 @@ package zcache
 import (
 	"context"
 	"github.com/stretchr/testify/suite"
+	"github.com/zondax/golem/pkg/metrics"
 	"go.uber.org/zap"
 	"os"
 	"testing"
@@ -22,20 +23,23 @@ type CombinedCacheTestSuite struct {
 	cacheRemoteBrokenNotBestEffort CombinedCache
 	cacheOkNotBestEffort           CombinedCache
 	cacheRemote                    RemoteCache
+	ms                             metrics.TaskMetrics
 }
 
 func (suite *CombinedCacheTestSuite) SetupSuite() {
 	mr, err := miniredis.Run()
 	suite.Require().NoError(err)
 	suite.mr = mr
-
+	suite.ms = metrics.NewTaskMetrics("", "", "appname")
 	logger, err := zap.NewDevelopment()
 	suite.Require().NoError(err)
 
 	prefix := os.Getenv("PREFIX")
 	suite.cacheRemoteBrokenBestEffort, err = NewCombinedCache(
 		&CombinedConfig{
-			Local: &LocalConfig{},
+			Local: &LocalConfig{
+				MetricServer: suite.ms,
+			},
 			Remote: &RemoteConfig{
 				Addr: "0.0.0.0",
 			},
@@ -46,7 +50,9 @@ func (suite *CombinedCacheTestSuite) SetupSuite() {
 	suite.Nil(err)
 
 	suite.cacheOkNotBestEffort, err = NewCombinedCache(&CombinedConfig{
-		Local: &LocalConfig{},
+		Local: &LocalConfig{
+			MetricServer: suite.ms,
+		},
 		Remote: &RemoteConfig{
 			Addr: mr.Addr(),
 		},
@@ -58,7 +64,9 @@ func (suite *CombinedCacheTestSuite) SetupSuite() {
 
 	suite.cacheRemoteBrokenNotBestEffort, err = NewCombinedCache(
 		&CombinedConfig{
-			Local: &LocalConfig{},
+			Local: &LocalConfig{
+				MetricServer: suite.ms,
+			},
 			Remote: &RemoteConfig{
 				Addr: "0.0.0.0",
 			},

pkg/zcache/config.go

@@ -3,6 +3,7 @@ package zcache
 import (
 	"github.com/allegro/bigcache/v3"
 	"github.com/go-redis/redis/v8"
+	"github.com/zondax/golem/pkg/metrics"
 	"go.uber.org/zap"
 	"time"
 )
@@ -26,8 +27,12 @@ type RemoteConfig struct {
 }
 
 type LocalConfig struct {
-	Prefix string
-	Logger *zap.Logger
+	Prefix          string
+	Logger          *zap.Logger
+	MetricServer    metrics.TaskMetrics
+	CleanupInterval time.Duration
+	BatchSize       int
+	ThrottleTime    time.Duration
 }
 
 func (c *RemoteConfig) ToRedisConfig() *redis.Options {

pkg/zcache/local_cache.go

@@ -7,12 +7,18 @@ import (
 	"fmt"
 	"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
+	neverExpires          = -1
+	errorTypeLabel        = "error_type"
+	cleanupErrorMetricKey = "localCacheCleanupErrors"
+	iterationErrorLabel   = "iteration_error"
+	unmarshalErrorLabel   = "unmarshal_error"
+	deletionErrorLabel    = "deletion_error"
 )
 
 type CacheItem struct {
@@ -44,11 +50,14 @@ type LocalCache interface {
 }
 
 type localCache struct {
-	client        *bigcache.BigCache
-	prefix        string
-	logger        *zap.Logger
-	metricsServer *metrics.TaskMetrics
-	appName       string
+	client          *bigcache.BigCache
+	prefix          string
+	logger          *zap.Logger
+	metricsServer   metrics.TaskMetrics
+	appName         string
+	cleanupInterval time.Duration
+	batchSize       int
+	throttleTime    time.Duration
 }
 
 func (c *localCache) Set(_ context.Context, key string, value interface{}, ttl time.Duration) error {
@@ -126,7 +135,7 @@ func (c *localCache) IsNotFoundError(err error) bool {
 }
 
 func (c *localCache) SetupAndMonitorMetrics(appName string, metricsServer metrics.TaskMetrics, updateInterval time.Duration) []error {
-	c.metricsServer = &metricsServer
+	c.metricsServer = metricsServer
 	c.appName = appName
 
 	errs := setupAndMonitorCacheMetrics(appName, metricsServer, c, updateInterval)
@@ -142,3 +151,61 @@ func (c *localCache) registerInternalCacheMetrics() []error {
 
 	return []error{}
 }
+
+func (c *localCache) startCleanupProcess(interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	go func() {
+		for range ticker.C {
+			c.cleanupExpiredKeys()
+		}
+	}()
+}
+
+func (c *localCache) cleanupExpiredKeys() {
+	iterator := c.client.Iterator()
+	var keysToDelete []string
+
+	for iterator.SetNext() {
+		entry, err := iterator.Value()
+		if err != nil {
+			_ = c.metricsServer.UpdateMetric(cleanupErrorMetricKey, 1, iterationErrorLabel)
+			continue
+		}
+
+		var cachedItem CacheItem
+		if err = json.Unmarshal(entry.Value(), &cachedItem); err != nil {
+			_ = c.metricsServer.UpdateMetric(cleanupErrorMetricKey, 1, unmarshalErrorLabel)
+			continue
+		}
+
+		if cachedItem.IsExpired() {
+			keysToDelete = append(keysToDelete, entry.Key())
+		}
+
+		if len(keysToDelete) >= c.batchSize {
+			c.deleteKeysInBatch(keysToDelete)
+			keysToDelete = keysToDelete[:0]
+			time.Sleep(c.throttleTime)
+		}
+	}
+
+	if len(keysToDelete) > 0 {
+		c.deleteKeysInBatch(keysToDelete)
+	}
+}
+
+func (c *localCache) deleteKeysInBatch(keys []string) {
+	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))
+			}
+		}
+	}
+}
+
+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))
+	}
+}

pkg/zcache/local_cache_test.go

@@ -2,12 +2,21 @@ package zcache
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"github.com/allegro/bigcache/v3"
 	"github.com/stretchr/testify/suite"
+	"github.com/zondax/golem/pkg/metrics"
 	"os"
 	"testing"
 	"time"
 )
 
+const (
+	testValue = "testValue"
+	expireKey = "expireKey"
+)
+
 func TestLocalCacheTestSuite(t *testing.T) {
 	suite.Run(t, new(LocalCacheTestSuite))
 }
@@ -21,7 +30,8 @@ func (suite *LocalCacheTestSuite) SetupSuite() {
 	prefix := os.Getenv("PREFIX")
 	var err error
 	config := LocalConfig{
-		Prefix: prefix,
+		Prefix:       prefix,
+		MetricServer: metrics.NewTaskMetrics("", "", "appname"),
 	}
 	suite.cache, err = NewLocalCache(&config)
 	suite.Nil(err)
@@ -30,7 +40,7 @@ func (suite *LocalCacheTestSuite) SetupSuite() {
 func (suite *LocalCacheTestSuite) TestSetAndGet() {
 	ctx := context.Background()
 	key := "testKey"
-	value := "testValue"
+	value := testValue
 
 	err := suite.cache.Set(ctx, key, value, 0)
 	suite.NoError(err)
@@ -44,7 +54,7 @@ func (suite *LocalCacheTestSuite) TestSetAndGet() {
 func (suite *LocalCacheTestSuite) TestDelete() {
 	ctx := context.Background()
 	key := "testKey"
-	value := "testValue"
+	value := testValue
 
 	suite.NoError(suite.cache.Set(ctx, key, value, 0))
 
@@ -56,17 +66,100 @@ func (suite *LocalCacheTestSuite) TestDelete() {
 }
 
 func (suite *LocalCacheTestSuite) TestCacheItemExpiration() {
-	item := NewCacheItem([]byte("testValue"), 1*time.Second)
+	item := NewCacheItem([]byte(testValue), 1*time.Second)
 	suite.False(item.IsExpired(), "CacheItem should not be expired right after creation")
 	time.Sleep(2 * time.Second)
 
 	suite.True(item.IsExpired(), "CacheItem should be expired after its TTL")
 }
 
 func (suite *LocalCacheTestSuite) TestCacheItemNeverExpires() {
-	item := NewCacheItem([]byte("testValue"), -1)
+	item := NewCacheItem([]byte(testValue), -1)
 	suite.False(item.IsExpired(), "CacheItem with negative TTL should never expire")
 	time.Sleep(2 * time.Second)
 
 	suite.False(item.IsExpired(), "CacheItem with negative TTL should never expire, even after some time")
 }
+
+func (suite *LocalCacheTestSuite) TestCleanupProcess() {
+	cleanupInterval := 1 * time.Second
+	ttl := 10 * time.Millisecond
+
+	cache, err := NewLocalCache(&LocalConfig{
+		Prefix:          "test",
+		CleanupInterval: cleanupInterval,
+		MetricServer:    metrics.NewTaskMetrics("", "", "appname")})
+	suite.NoError(err)
+
+	ctx := context.Background()
+	key := expireKey
+	value := testValue
+
+	err = cache.Set(ctx, key, value, ttl)
+	suite.NoError(err)
+
+	time.Sleep(2 * cleanupInterval)
+
+	var result string
+	err = cache.Get(ctx, key, &result)
+
+	suite.True(errors.Is(err, bigcache.ErrEntryNotFound), "Expected 'key not found' error, but got a different error")
+}
+
+func (suite *LocalCacheTestSuite) TestCleanupProcessBatchLogic() {
+	cleanupInterval := 100 * time.Millisecond
+	testBatchSize := 5
+
+	cache, err := NewLocalCache(&LocalConfig{
+		Prefix:          "testBatch",
+		CleanupInterval: cleanupInterval,
+		MetricServer:    metrics.NewTaskMetrics("", "", "appname"),
+		BatchSize:       testBatchSize,
+	})
+	suite.NoError(err)
+
+	ctx := context.Background()
+
+	for i := 0; i < testBatchSize*2; i++ {
+		key := fmt.Sprintf("key%d", i)
+		value := fmt.Sprintf("value%d", i)
+		err = cache.Set(ctx, key, value, 1*time.Millisecond)
+		suite.NoError(err)
+	}
+
+	time.Sleep(2 * time.Second)
+
+	for i := 0; i < testBatchSize*2; i++ {
+		key := fmt.Sprintf("key%d", i)
+		var result string
+		err = cache.Get(ctx, key, &result)
+
+		suite.True(errors.Is(err, bigcache.ErrEntryNotFound), "Expected 'ErrEntryNotFound' for key: %s, but got a different error or no error: %s", key, err.Error())
+	}
+}
+
+func (suite *LocalCacheTestSuite) TestCleanupProcessItemDoesNotExpire() {
+	cleanupInterval := 1 * time.Second
+
+	cache, err := NewLocalCache(&LocalConfig{
+		Prefix:          "test",
+		CleanupInterval: cleanupInterval,
+		MetricServer:    metrics.NewTaskMetrics("", "", "appname"),
+	})
+	suite.NoError(err)
+
+	ctx := context.Background()
+	key := "permanentKey"
+	value := "thisValueShouldPersist"
+
+	err = cache.Set(ctx, key, value, neverExpires)
+	suite.NoError(err)
+
+	time.Sleep(2 * cleanupInterval)
+
+	var result string
+	err = cache.Get(ctx, key, &result)
+
+	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")
+}