add redis as index cache and caching bucket backend

Signed-off-by: Ben Ye <benye@amazon.com>

Author: yeya24

docs/blocks-storage/querier.md

@@ -99,7 +99,7 @@ func TestMetricMetadataToMetricTypeToMetricType(t *testing.T) {
 	for _, tt := range tc {
 		t.Run(tt.desc, func(t *testing.T) {
 			m := MetricMetadataMetricTypeToMetricType(tt.input)
-			testutil.Equals(t, tt.expected, m)
+			assert.Equal(t, tt.expected, m)
 		})
 	}
 }

docs/blocks-storage/store-gateway.md

@@ -23,25 +23,26 @@ import (
 
 const (
 	CacheBackendMemcached = "memcached"
+	CacheBackendRedis     = "redis"
 )
 
 type CacheBackend struct {
 	Backend   string                `yaml:"backend"`
 	Memcached MemcachedClientConfig `yaml:"memcached"`
+	Redis     RedisClientConfig     `yaml:"redis"`
 }
 
 // Validate the config.
 func (cfg *CacheBackend) Validate() error {
-	if cfg.Backend != "" && cfg.Backend != CacheBackendMemcached {
+	switch cfg.Backend {
+	case CacheBackendMemcached:
+		return cfg.Memcached.Validate()
+	case CacheBackendRedis:
+		return cfg.Redis.Validate()
+	case "":
+	default:
 		return fmt.Errorf("unsupported cache backend: %s", cfg.Backend)
 	}
-
-	if cfg.Backend == CacheBackendMemcached {
-		if err := cfg.Memcached.Validate(); err != nil {
-			return err
-		}
-	}
-
 	return nil
 }
 
@@ -58,6 +59,7 @@ func (cfg *ChunksCacheConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix st
 	f.StringVar(&cfg.Backend, prefix+"backend", "", fmt.Sprintf("Backend for chunks cache, if not empty. Supported values: %s.", CacheBackendMemcached))
 
 	cfg.Memcached.RegisterFlagsWithPrefix(f, prefix+"memcached.")
+	cfg.Redis.RegisterFlagsWithPrefix(f, prefix+"redis.")
 
 	f.Int64Var(&cfg.SubrangeSize, prefix+"subrange-size", 16000, "Size of each subrange that bucket object is split into for better caching.")
 	f.IntVar(&cfg.MaxGetRangeRequests, prefix+"max-get-range-requests", 3, "Maximum number of sub-GetRange requests that a single GetRange request can be split into when fetching chunks. Zero or negative value = unlimited number of sub-requests.")
@@ -89,6 +91,7 @@ func (cfg *MetadataCacheConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix
 	f.StringVar(&cfg.Backend, prefix+"backend", "", fmt.Sprintf("Backend for metadata cache, if not empty. Supported values: %s.", CacheBackendMemcached))
 
 	cfg.Memcached.RegisterFlagsWithPrefix(f, prefix+"memcached.")
+	cfg.Redis.RegisterFlagsWithPrefix(f, prefix+"redis.")
 
 	f.DurationVar(&cfg.TenantsListTTL, prefix+"tenants-list-ttl", 15*time.Minute, "How long to cache list of tenants in the bucket.")
 	f.DurationVar(&cfg.TenantBlocksListTTL, prefix+"tenant-blocks-list-ttl", 5*time.Minute, "How long to cache list of blocks for each tenant.")
@@ -111,7 +114,7 @@ func CreateCachingBucket(chunksConfig ChunksCacheConfig, metadataConfig Metadata
 	cfg := cache.NewCachingBucketConfig()
 	cachingConfigured := false
 
-	chunksCache, err := createCache("chunks-cache", chunksConfig.Backend, chunksConfig.Memcached, logger, reg)
+	chunksCache, err := createCache("chunks-cache", chunksConfig.Backend, chunksConfig.Memcached, chunksConfig.Redis, logger, reg)
 	if err != nil {
 		return nil, errors.Wrapf(err, "chunks-cache")
 	}
@@ -121,7 +124,7 @@ func CreateCachingBucket(chunksConfig ChunksCacheConfig, metadataConfig Metadata
 		cfg.CacheGetRange("chunks", chunksCache, isTSDBChunkFile, chunksConfig.SubrangeSize, chunksConfig.AttributesTTL, chunksConfig.SubrangeTTL, chunksConfig.MaxGetRangeRequests)
 	}
 
-	metadataCache, err := createCache("metadata-cache", metadataConfig.Backend, metadataConfig.Memcached, logger, reg)
+	metadataCache, err := createCache("metadata-cache", metadataConfig.Backend, metadataConfig.Memcached, metadataConfig.Redis, logger, reg)
 	if err != nil {
 		return nil, errors.Wrapf(err, "metadata-cache")
 	}
@@ -149,7 +152,7 @@ func CreateCachingBucket(chunksConfig ChunksCacheConfig, metadataConfig Metadata
 	return storecache.NewCachingBucket(bkt, cfg, logger, reg)
 }
 
-func createCache(cacheName string, backend string, memcached MemcachedClientConfig, logger log.Logger, reg prometheus.Registerer) (cache.Cache, error) {
+func createCache(cacheName string, backend string, memcached MemcachedClientConfig, redis RedisClientConfig, logger log.Logger, reg prometheus.Registerer) (cache.Cache, error) {
 	switch backend {
 	case "":
 		// No caching.
@@ -163,6 +166,13 @@ func createCache(cacheName string, backend string, memcached MemcachedClientConf
 		}
 		return cache.NewMemcachedCache(cacheName, logger, client, reg), nil
 
+	case CacheBackendRedis:
+		redisCache, err := cacheutil.NewRedisClientWithConfig(logger, cacheName, redis.ToRedisClientConfig(), reg)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed to create redis client")
+		}
+		return cache.NewRedisCache(cacheName, logger, redisCache, reg), nil
+
 	default:
 		return nil, errors.Errorf("unsupported cache type for cache %s: %s", cacheName, backend)
 	}

docs/configuration/config-file-reference.md

@@ -23,14 +23,17 @@ const (
 	// IndexCacheBackendMemcached is the value for the memcached index cache backend.
 	IndexCacheBackendMemcached = "memcached"
 
+	// IndexCacheBackendRedis is the value for the redis index cache backend.
+	IndexCacheBackendRedis = "redis"
+
 	// IndexCacheBackendDefault is the value for the default index cache backend.
 	IndexCacheBackendDefault = IndexCacheBackendInMemory
 
 	defaultMaxItemSize = model.Bytes(128 * units.MiB)
 )
 
 var (
-	supportedIndexCacheBackends = []string{IndexCacheBackendInMemory, IndexCacheBackendMemcached}
+	supportedIndexCacheBackends = []string{IndexCacheBackendInMemory, IndexCacheBackendMemcached, IndexCacheBackendRedis}
 
 	errUnsupportedIndexCacheBackend = errors.New("unsupported index cache backend")
 	errNoIndexCacheAddresses        = errors.New("no index cache backend addresses")
@@ -40,6 +43,7 @@ type IndexCacheConfig struct {
 	Backend   string                   `yaml:"backend"`
 	InMemory  InMemoryIndexCacheConfig `yaml:"inmemory"`
 	Memcached MemcachedClientConfig    `yaml:"memcached"`
+	Redis     RedisClientConfig        `yaml:"redis"`
 }
 
 func (cfg *IndexCacheConfig) RegisterFlags(f *flag.FlagSet) {
@@ -51,6 +55,7 @@ func (cfg *IndexCacheConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix str
 
 	cfg.InMemory.RegisterFlagsWithPrefix(f, prefix+"inmemory.")
 	cfg.Memcached.RegisterFlagsWithPrefix(f, prefix+"memcached.")
+	cfg.Redis.RegisterFlagsWithPrefix(f, prefix+"redis.")
 }
 
 // Validate the config.
@@ -63,6 +68,10 @@ func (cfg *IndexCacheConfig) Validate() error {
 		if err := cfg.Memcached.Validate(); err != nil {
 			return err
 		}
+	} else if cfg.Backend == IndexCacheBackendRedis {
+		if err := cfg.Redis.Validate(); err != nil {
+			return err
+		}
 	}
 
 	return nil
@@ -83,6 +92,8 @@ func NewIndexCache(cfg IndexCacheConfig, logger log.Logger, registerer prometheu
 		return newInMemoryIndexCache(cfg.InMemory, logger, registerer)
 	case IndexCacheBackendMemcached:
 		return newMemcachedIndexCache(cfg.Memcached, logger, registerer)
+	case IndexCacheBackendRedis:
+		return newRedisIndexCache(cfg.Redis, logger, registerer)
 	default:
 		return nil, errUnsupportedIndexCacheBackend
 	}
@@ -111,3 +122,12 @@ func newMemcachedIndexCache(cfg MemcachedClientConfig, logger log.Logger, regist
 
 	return storecache.NewMemcachedIndexCache(logger, client, registerer)
 }
+
+func newRedisIndexCache(cfg RedisClientConfig, logger log.Logger, registerer prometheus.Registerer) (storecache.IndexCache, error) {
+	client, err := cacheutil.NewRedisClientWithConfig(logger, "index-cache", cfg.ToRedisClientConfig(), registerer)
+	if err != nil {
+		return nil, errors.Wrapf(err, "create index cache redis client")
+	}
+
+	return storecache.NewRemoteIndexCache(logger, client, registerer)
+}

pkg/cortexpb/compat_test.go

@@ -0,0 +1,107 @@
+package tsdb
+
+import (
+	"flag"
+	"time"
+
+	"github.com/pkg/errors"
+	"github.com/thanos-io/thanos/pkg/cacheutil"
+
+	"github.com/cortexproject/cortex/pkg/util/tls"
+)
+
+type RedisClientConfig struct {
+	Addresses  string `yaml:"addresses"`
+	Username   string `yaml:"username"`
+	Password   string `yaml:"password"`
+	DB         int    `yaml:"db"`
+	MasterName string `yaml:"master_name"`
+
+	PoolSize               int `yaml:"pool_size"`
+	MinIdleConns           int `yaml:"min_idle_conns"`
+	MaxGetMultiConcurrency int `yaml:"max_get_multi_concurrency"`
+	GetMultiBatchSize      int `yaml:"get_multi_batch_size"`
+	MaxSetMultiConcurrency int `yaml:"max_set_multi_concurrency"`
+	SetMultiBatchSize      int `yaml:"set_multi_batch_size"`
+
+	DialTimeout  time.Duration `yaml:"dial_timeout"`
+	ReadTimeout  time.Duration `yaml:"read_timeout"`
+	WriteTimeout time.Duration `yaml:"write_timeout"`
+	IdleTimeout  time.Duration `yaml:"idle_timeout"`
+	MaxConnAge   time.Duration `yaml:"max_conn_age"`
+
+	TLSEnabled bool             `yaml:"tls_enabled"`
+	TLS        tls.ClientConfig `yaml:",inline"`
+
+	// If not zero then client-side caching is enabled.
+	// Client-side caching is when data is stored in memory
+	// instead of fetching data each time.
+	// See https://redis.io/docs/manual/client-side-caching/ for info.
+	CacheSize int `yaml:"cache_size"`
+}
+
+func (cfg *RedisClientConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix string) {
+	f.StringVar(&cfg.Addresses, prefix+"addresses", "", "Comma separated list of redis addresses. Supported prefixes are: dns+ (looked up as an A/AAAA query), dnssrv+ (looked up as a SRV query, dnssrvnoa+ (looked up as a SRV query, with no A/AAAA lookup made after that).")
+	f.StringVar(&cfg.Username, prefix+"username", "", "Redis username.")
+	f.StringVar(&cfg.Password, prefix+"password", "", "Redis password.")
+	f.IntVar(&cfg.DB, prefix+"db", 0, "Database to be selected after connecting to the server.")
+	f.DurationVar(&cfg.DialTimeout, prefix+"dial-timeout", time.Second*5, "Client dial timeout.")
+	f.DurationVar(&cfg.ReadTimeout, prefix+"read-timeout", time.Second*3, "Client read timeout.")
+	f.DurationVar(&cfg.WriteTimeout, prefix+"write-timeout", time.Second*3, "Client write timeout.")
+	f.DurationVar(&cfg.IdleTimeout, prefix+"idle-timeout", time.Minute*5, "Amount of time after which client closes idle connections. Should be less than server's timeout. -1 disables idle timeout check.")
+	f.DurationVar(&cfg.MaxConnAge, prefix+"max-conn-age", 0, "Connection age at which client retires (closes) the connection. Default 0 is to not close aged connections.")
+	f.IntVar(&cfg.PoolSize, prefix+"pool-size", 100, "Maximum number of socket connections.")
+	f.IntVar(&cfg.MinIdleConns, prefix+"min-idle-conns", 10, "Specifies the minimum number of idle connections, which is useful when it is slow to establish new connections.")
+	f.IntVar(&cfg.MaxGetMultiConcurrency, prefix+"max-get-multi-concurrency", 100, "The maximum number of concurrent GetMulti() operations. If set to 0, concurrency is unlimited.")
+	f.IntVar(&cfg.GetMultiBatchSize, prefix+"get-multi-batch-size", 100, "The maximum size per batch for mget.")
+	f.IntVar(&cfg.MaxSetMultiConcurrency, prefix+"max-set-multi-concurrency", 100, "The maximum number of concurrent SetMulti() operations. If set to 0, concurrency is unlimited.")
+	f.IntVar(&cfg.SetMultiBatchSize, prefix+"set-multi-batch-size", 100, "The maximum size per batch for pipeline set.")
+	f.StringVar(&cfg.MasterName, prefix+"master-name", "", "Specifies the master's name. Must be not empty for Redis Sentinel.")
+	f.IntVar(&cfg.CacheSize, prefix+"cache-size", 0, "If not zero then client-side caching is enabled. Client-side caching is when data is stored in memory instead of fetching data each time. See https://redis.io/docs/manual/client-side-caching/ for more info.")
+	f.BoolVar(&cfg.TLSEnabled, prefix+"tls-enabled", false, "Whether to enable tls for redis connection.")
+	cfg.TLS.RegisterFlagsWithPrefix(prefix, f)
+}
+
+// Validate the config.
+func (cfg *RedisClientConfig) Validate() error {
+	if cfg.Addresses == "" {
+		return errNoIndexCacheAddresses
+	}
+
+	if cfg.TLSEnabled {
+		if (cfg.TLS.CertPath != "") != (cfg.TLS.KeyPath != "") {
+			return errors.New("both client key and certificate must be provided")
+		}
+	}
+
+	return nil
+}
+
+func (cfg *RedisClientConfig) ToRedisClientConfig() cacheutil.RedisClientConfig {
+	return cacheutil.RedisClientConfig{
+		Addr:                   cfg.Addresses,
+		Username:               cfg.Username,
+		Password:               cfg.Password,
+		DB:                     cfg.DB,
+		MasterName:             cfg.MasterName,
+		DialTimeout:            cfg.DialTimeout,
+		ReadTimeout:            cfg.ReadTimeout,
+		WriteTimeout:           cfg.WriteTimeout,
+		PoolSize:               cfg.PoolSize,
+		MinIdleConns:           cfg.MinIdleConns,
+		IdleTimeout:            cfg.IdleTimeout,
+		MaxConnAge:             cfg.MaxConnAge,
+		MaxGetMultiConcurrency: cfg.MaxGetMultiConcurrency,
+		GetMultiBatchSize:      cfg.GetMultiBatchSize,
+		MaxSetMultiConcurrency: cfg.MaxSetMultiConcurrency,
+		SetMultiBatchSize:      cfg.SetMultiBatchSize,
+		TLSEnabled:             cfg.TLSEnabled,
+		TLSConfig: cacheutil.TLSConfig{
+			CAFile:             cfg.TLS.CAPath,
+			KeyFile:            cfg.TLS.KeyPath,
+			CertFile:           cfg.TLS.CertPath,
+			ServerName:         cfg.TLS.ServerName,
+			InsecureSkipVerify: cfg.TLS.InsecureSkipVerify,
+		},
+	}
+}

pkg/storage/tsdb/caching_bucket.go

@@ -0,0 +1,73 @@
+package tsdb
+
+import (
+	"testing"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/require"
+
+	"github.com/cortexproject/cortex/pkg/util/tls"
+)
+
+func TestRedisIndexCacheConfigValidate(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		conf *RedisClientConfig
+		err  error
+	}{
+		{
+			name: "empty address",
+			conf: &RedisClientConfig{},
+			err:  errNoIndexCacheAddresses,
+		},
+		{
+			name: "provide TLS cert path but not key path",
+			conf: &RedisClientConfig{
+				Addresses:  "aaa",
+				TLSEnabled: true,
+				TLS: tls.ClientConfig{
+					CertPath: "foo",
+				},
+			},
+			err: errors.New("both client key and certificate must be provided"),
+		},
+		{
+			name: "provide TLS key path but not cert path",
+			conf: &RedisClientConfig{
+				Addresses:  "aaa",
+				TLSEnabled: true,
+				TLS: tls.ClientConfig{
+					KeyPath: "foo",
+				},
+			},
+			err: errors.New("both client key and certificate must be provided"),
+		},
+		{
+			name: "success when TLS enabled",
+			conf: &RedisClientConfig{
+				Addresses:  "aaa",
+				TLSEnabled: true,
+				TLS: tls.ClientConfig{
+					KeyPath:  "foo",
+					CertPath: "bar",
+				},
+			},
+		},
+		{
+			name: "success when TLS disabled",
+			conf: &RedisClientConfig{
+				Addresses:  "aaa",
+				TLSEnabled: false,
+			},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			err := tc.conf.Validate()
+			if tc.err == nil {
+				require.NoError(t, err)
+			} else {
+				require.Equal(t, tc.err.Error(), err.Error())
+			}
+		})
+	}
+}