feat: add cache interface and implementation for redis,inmemory (#2155)

Makefile

@@ -138,5 +138,6 @@ clear-swarm-data:
 
 test:
 	go test ./pkg/query-service/app/metrics/...
+	go test ./pkg/query-service/cache/...
 	go test ./pkg/query-service/app/...
 	go test ./pkg/query-service/converter/...

go.mod

@@ -7,6 +7,8 @@ require (
 	github.com/SigNoz/govaluate v0.0.0-20220522085550-d19c08c206cb
 	github.com/coreos/go-oidc/v3 v3.4.0
 	github.com/go-kit/log v0.2.1
+	github.com/go-redis/redis/v8 v8.11.5
+	github.com/go-redis/redismock/v8 v8.11.5
 	github.com/google/uuid v1.3.0
 	github.com/gorilla/handlers v1.5.1
 	github.com/gorilla/mux v1.8.0
@@ -20,6 +22,7 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/oklog/oklog v0.3.2
 	github.com/open-telemetry/opamp-go v0.5.0
+	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/pkg/errors v0.9.1
 	github.com/posthog/posthog-go v0.0.0-20220817142604-0b0bbf0f9c0f
 	github.com/prometheus/common v0.39.0
@@ -69,6 +72,7 @@ require (
 	github.com/armon/go-metrics v0.4.0 // indirect
 	github.com/beevik/etree v1.1.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/form3tech-oss/jwt-go v3.2.5+incompatible // indirect
 	github.com/gorilla/websocket v1.5.0 // indirect
 	github.com/jonboulle/clockwork v0.2.2 // indirect
@@ -128,7 +132,7 @@ require (
 	go.opentelemetry.io/otel v1.11.2 // indirect
 	go.opentelemetry.io/otel/trace v1.11.2 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
-	go.uber.org/multierr v1.9.0 // indirect
+	go.uber.org/multierr v1.9.0
 	golang.org/x/crypto v0.1.0
 	golang.org/x/net v0.4.0
 	golang.org/x/oauth2 v0.3.0

pkg/query-service/cache/cache.go

@@ -0,0 +1,69 @@
+package cache
+
+import (
+	"os"
+	"time"
+
+	inmemory "go.signoz.io/signoz/pkg/query-service/cache/inmemory"
+	redis "go.signoz.io/signoz/pkg/query-service/cache/redis"
+	"go.signoz.io/signoz/pkg/query-service/cache/status"
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
+	"gopkg.in/yaml.v2"
+)
+
+type Options struct {
+	Name     string            `yaml:"-"`
+	Provider string            `yaml:"provider"`
+	Redis    *redis.Options    `yaml:"redis,omitempty"`
+	InMemory *inmemory.Options `yaml:"inmemory,omitempty"`
+}
+
+// Cache is the interface for the storage backend
+type Cache interface {
+	Connect() error
+	Store(cacheKey string, data []byte, ttl time.Duration) error
+	Retrieve(cacheKey string, allowExpired bool) ([]byte, status.RetrieveStatus, error)
+	SetTTL(cacheKey string, ttl time.Duration)
+	Remove(cacheKey string)
+	BulkRemove(cacheKeys []string)
+	Close() error
+}
+
+// KeyGenerator is the interface for the key generator
+// The key generator is used to generate the cache keys for the cache entries
+type KeyGenerator interface {
+	// GenerateKeys generates the cache keys for the given query range params
+	// The keys are returned as a map where the key is the query name and the value is the cache key
+	GenerateKeys(*v3.QueryRangeParamsV3) map[string]string
+}
+
+// LoadFromYAMLCacheConfig loads the cache options from the given YAML config bytes
+func LoadFromYAMLCacheConfig(yamlConfig []byte) (*Options, error) {
+	var options Options
+	err := yaml.Unmarshal(yamlConfig, &options)
+	if err != nil {
+		return nil, err
+	}
+	return &options, nil
+}
+
+// LoadFromYAMLCacheConfigFile loads the cache options from the given YAML config file
+func LoadFromYAMLCacheConfigFile(configFile string) (*Options, error) {
+	bytes, err := os.ReadFile(configFile)
+	if err != nil {
+		return nil, err
+	}
+	return LoadFromYAMLCacheConfig(bytes)
+}
+
+// NewCache creates a new cache based on the given options
+func NewCache(options *Options) Cache {
+	switch options.Provider {
+	case "redis":
+		return redis.New(options.Redis)
+	case "inmemory":
+		return inmemory.New(options.InMemory)
+	default:
+		return nil
+	}
+}

pkg/query-service/cache/cache_test.go

@@ -0,0 +1,52 @@
+package cache
+
+import "testing"
+
+func TestNewCacheUnKnownProvider(t *testing.T) {
+	c := NewCache(&Options{
+		Name:     "test",
+		Provider: "unknown",
+	})
+
+	if c != nil {
+		t.Fatalf("expected nil, got %v", c)
+	}
+}
+
+func TestNewCacheInMemory(t *testing.T) {
+	c := NewCache(&Options{
+		Name:     "test",
+		Provider: "inmemory",
+	})
+
+	if c == nil {
+		t.Fatalf("expected non-nil, got nil")
+	}
+}
+
+func TestNewCacheRedis(t *testing.T) {
+	c := NewCache(&Options{
+		Name:     "test",
+		Provider: "redis",
+	})
+
+	if c == nil {
+		t.Fatalf("expected non-nil, got nil")
+	}
+}
+
+func TestLoadFromYAMLCacheConfig(t *testing.T) {
+	_, err := LoadFromYAMLCacheConfig([]byte(`
+provider: inmemory
+`))
+	if err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+}
+
+func TestLoadFromYAMLCacheConfigFile(t *testing.T) {
+	_, err := LoadFromYAMLCacheConfigFile("testdata/cache.yaml")
+	if err != nil {
+		t.Fatalf("unexpected error: %s", err)
+	}
+}

pkg/query-service/cache/inmemory/cache.go

@@ -0,0 +1,73 @@
+package inmemory
+
+import (
+	"time"
+
+	go_cache "github.com/patrickmn/go-cache"
+	"go.signoz.io/signoz/pkg/query-service/cache/status"
+)
+
+// cache implements the Cache interface
+type cache struct {
+	cc *go_cache.Cache
+}
+
+// New creates a new in-memory cache
+func New(opts *Options) *cache {
+	if opts == nil {
+		opts = defaultOptions()
+	}
+	return &cache{cc: go_cache.New(opts.TTL, opts.CleanupInterval)}
+}
+
+// Connect does nothing
+func (c *cache) Connect() error {
+	return nil
+}
+
+// Store stores the data in the cache
+func (c *cache) Store(cacheKey string, data []byte, ttl time.Duration) error {
+	c.cc.Set(cacheKey, data, ttl)
+	return nil
+}
+
+// Retrieve retrieves the data from the cache
+func (c *cache) Retrieve(cacheKey string, allowExpired bool) ([]byte, status.RetrieveStatus, error) {
+	data, found := c.cc.Get(cacheKey)
+	if !found {
+		return nil, status.RetrieveStatusKeyMiss, nil
+	}
+
+	return data.([]byte), status.RetrieveStatusHit, nil
+}
+
+// SetTTL sets the TTL for the cache entry
+func (c *cache) SetTTL(cacheKey string, ttl time.Duration) {
+	item, found := c.cc.Get(cacheKey)
+	if !found {
+		return
+	}
+	c.cc.Replace(cacheKey, item, ttl)
+}
+
+// Remove removes the cache entry
+func (c *cache) Remove(cacheKey string) {
+	c.cc.Delete(cacheKey)
+}
+
+// BulkRemove removes the cache entries
+func (c *cache) BulkRemove(cacheKeys []string) {
+	for _, cacheKey := range cacheKeys {
+		c.cc.Delete(cacheKey)
+	}
+}
+
+// Close does nothing
+func (c *cache) Close() error {
+	return nil
+}
+
+// Configuration returns the cache configuration
+func (c *cache) Configuration() *Options {
+	return nil
+}

pkg/query-service/cache/inmemory/cache_test.go

@@ -0,0 +1,102 @@
+package inmemory
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"go.signoz.io/signoz/pkg/query-service/cache/status"
+)
+
+// TestNew tests the New function
+func TestNew(t *testing.T) {
+	opts := &Options{
+		TTL:             10 * time.Second,
+		CleanupInterval: 10 * time.Second,
+	}
+	c := New(opts)
+	assert.NotNil(t, c)
+	assert.NotNil(t, c.cc)
+}
+
+// TestConnect tests the Connect function
+func TestConnect(t *testing.T) {
+	c := New(nil)
+	assert.NoError(t, c.Connect())
+}
+
+// TestStore tests the Store function
+func TestStore(t *testing.T) {
+	c := New(nil)
+	assert.NoError(t, c.Store("key", []byte("value"), 10*time.Second))
+}
+
+// TestRetrieve tests the Retrieve function
+func TestRetrieve(t *testing.T) {
+	c := New(nil)
+	assert.NoError(t, c.Store("key", []byte("value"), 10*time.Second))
+	data, retrieveStatus, err := c.Retrieve("key", false)
+	assert.NoError(t, err)
+	assert.Equal(t, retrieveStatus, status.RetrieveStatusHit)
+	assert.Equal(t, data, []byte("value"))
+}
+
+// TestSetTTL tests the SetTTL function
+func TestSetTTL(t *testing.T) {
+	c := New(&Options{TTL: 10 * time.Second, CleanupInterval: 1 * time.Second})
+	assert.NoError(t, c.Store("key", []byte("value"), 2*time.Second))
+	time.Sleep(3 * time.Second)
+	data, retrieveStatus, err := c.Retrieve("key", false)
+	assert.NoError(t, err)
+	assert.Equal(t, retrieveStatus, status.RetrieveStatusKeyMiss)
+	assert.Nil(t, data)
+
+	assert.NoError(t, c.Store("key", []byte("value"), 2*time.Second))
+	c.SetTTL("key", 4*time.Second)
+	time.Sleep(3 * time.Second)
+	data, retrieveStatus, err = c.Retrieve("key", false)
+	assert.NoError(t, err)
+	assert.Equal(t, retrieveStatus, status.RetrieveStatusHit)
+	assert.Equal(t, data, []byte("value"))
+}
+
+// TestRemove tests the Remove function
+func TestRemove(t *testing.T) {
+	c := New(nil)
+	assert.NoError(t, c.Store("key", []byte("value"), 10*time.Second))
+	c.Remove("key")
+
+	data, retrieveStatus, err := c.Retrieve("key", false)
+	assert.NoError(t, err)
+	assert.Equal(t, retrieveStatus, status.RetrieveStatusKeyMiss)
+	assert.Nil(t, data)
+}
+
+// TestBulkRemove tests the BulkRemove function
+func TestBulkRemove(t *testing.T) {
+	c := New(nil)
+	assert.NoError(t, c.Store("key1", []byte("value"), 10*time.Second))
+	assert.NoError(t, c.Store("key2", []byte("value"), 10*time.Second))
+	c.BulkRemove([]string{"key1", "key2"})
+
+	data, retrieveStatus, err := c.Retrieve("key1", false)
+	assert.NoError(t, err)
+	assert.Equal(t, retrieveStatus, status.RetrieveStatusKeyMiss)
+	assert.Nil(t, data)
+
+	data, retrieveStatus, err = c.Retrieve("key2", false)
+	assert.NoError(t, err)
+	assert.Equal(t, retrieveStatus, status.RetrieveStatusKeyMiss)
+	assert.Nil(t, data)
+}
+
+// TestCache tests the cache
+func TestCache(t *testing.T) {
+	c := New(nil)
+	assert.NoError(t, c.Store("key", []byte("value"), 10*time.Second))
+	data, retrieveStatus, err := c.Retrieve("key", false)
+	assert.NoError(t, err)
+	assert.Equal(t, retrieveStatus, status.RetrieveStatusHit)
+	assert.Equal(t, data, []byte("value"))
+	c.Remove("key")
+}

pkg/query-service/cache/inmemory/options.go

@@ -0,0 +1,23 @@
+package inmemory
+
+import (
+	"time"
+
+	go_cache "github.com/patrickmn/go-cache"
+)
+
+const (
+	defaultTTL             = go_cache.NoExpiration
+	defaultCleanupInterval = 1 * time.Minute
+)
+
+// Options holds the options for the in-memory cache
+type Options struct {
+	// TTL is the time to live for the cache entries
+	TTL             time.Duration `yaml:"ttl,omitempty"`
+	CleanupInterval time.Duration `yaml:"cleanupInterval,omitempty"`
+}
+
+func defaultOptions() *Options {
+	return &Options{TTL: defaultTTL, CleanupInterval: defaultCleanupInterval}
+}

pkg/query-service/cache/redis/options.go

@@ -0,0 +1,24 @@
+package redis
+
+const (
+	defaultHost     = "localhost"
+	defaultPort     = 6379
+	defaultPassword = ""
+	defaultDB       = 0
+)
+
+type Options struct {
+	Host     string `yaml:"host,omitempty"`
+	Port     int    `yaml:"port,omitempty"`
+	Password string `yaml:"password,omitempty"`
+	DB       int    `yaml:"db,omitempty"`
+}
+
+func defaultOptions() *Options {
+	return &Options{
+		Host:     defaultHost,
+		Port:     defaultPort,
+		Password: defaultPassword,
+		DB:       defaultDB,
+	}
+}