Cache: support redis cache backend (#4888)

* Cache: support redis cache backend

Signed-off-by: Jimmie Han <[email protected]>

* Cache: add get gate

make linter happy

Signed-off-by: Jimmie Han <[email protected]>

* add concurrent config  control

Signed-off-by: Jimmie Han <[email protected]>

* add redis cache doc

Signed-off-by: Jimmie Han <[email protected]>

* use doWithBatch to optmize concurrent batch code.

Signed-off-by: Jimmie Han <[email protected]>

* support db select

Signed-off-by: Jimmie Han <[email protected]>

* fix query-frontend cache config

Signed-off-by: Jimmie Han <[email protected]>

Author: hanjm

docs/components/query-frontend.md

@@ -95,6 +95,35 @@ config:
   expiration: 24h
 ```
 
+#### Redis
+
+The default redis config is:
+
+```yaml mdox-exec="go run scripts/cfggen/main.go --name=queryfrontend.RedisResponseCacheConfig"
+type: REDIS
+config:
+  addr: ""
+  username: ""
+  password: ""
+  db: 0
+  dial_timeout: 5s
+  read_timeout: 3s
+  write_timeout: 3s
+  pool_size: 100
+  min_idle_conns: 10
+  idle_timeout: 5m0s
+  max_conn_age: 0s
+  max_get_multi_concurrency: 100
+  get_multi_batch_size: 100
+  max_set_multi_concurrency: 100
+  set_multi_batch_size: 100
+  expiration: 24h0m0s
+```
+
+`expiration` specifies redis cache valid time. If set to 0s, so using a default of 24 hours expiration time.
+
+Other cache configuration parameters, you can refer to [redis-index-cache](store.md#redis-index-cache).
+
 ### Slow Query Log
 
 Query Frontend supports `--query-frontend.log-queries-longer-than` flag to log queries running longer than some duration.

docs/components/store.md

@@ -304,11 +304,56 @@ While the remaining settings are **optional**:
 - `dns_provider_update_interval`: the DNS discovery update interval.
 - `auto_discovery`: whether to use the auto-discovery mechanism for memcached.
 
+### Redis index cache
+
+The `redis` index cache allows to use [Redis](https://redis.io) as cache backend. This cache type is configured using `--index-cache.config-file` to reference the configuration file or `--index-cache.config` to put yaml config directly:
+
+```yaml mdox-exec="go run scripts/cfggen/main.go --name=cacheutil.RedisClientConfig"
+type: REDIS
+config:
+  addr: ""
+  username: ""
+  password: ""
+  db: 0
+  dial_timeout: 5s
+  read_timeout: 3s
+  write_timeout: 3s
+  pool_size: 100
+  min_idle_conns: 10
+  idle_timeout: 5m0s
+  max_conn_age: 0s
+  max_get_multi_concurrency: 100
+  get_multi_batch_size: 100
+  max_set_multi_concurrency: 100
+  set_multi_batch_size: 100
+```
+
+The **required** settings are:
+
+- `addr`: redis server address.
+
+While the remaining settings are **optional**:
+
+- `username`: the username to connect redis, only redis 6.0 and grater need this field.
+- `password`: the password to connect redis.
+- `db`: the database to be selected after connecting to the server.
+- `dial_timeout`: the redis dial timeout.
+- `read_timeout`: the redis read timeout.
+- `write_timeout`: the redis write timeout.
+- `pool_size`: maximum number of socket connections.
+- `min_idle_conns`: specifies the minimum number of idle connections which is useful when establishing new connection is slow.
+- `idle_timeout`: amount of time after which client closes idle connections. Should be less than server's timeout.
+- `max_conn_age`: connection age at which client retires (closes) the connection.
+- `max_get_multi_concurrency`: specifies the maximum number of concurrent GetMulti() operations.
+- `get_multi_batch_size`: specifies the maximum size per batch for mget.
+- `max_set_multi_concurrency`: specifies the maximum number of concurrent SetMulti() operations.
+- `set_multi_batch_size`: specifies the maximum size per batch for pipeline set.
+
 ## Caching Bucket
 
 Thanos Store Gateway supports a "caching bucket" with [chunks](../design.md#chunk) and metadata caching to speed up loading of [chunks](../design.md#chunk) from TSDB blocks. To configure caching, one needs to use `--store.caching-bucket.config=<yaml content>` or `--store.caching-bucket.config-file=<file.yaml>`.
 
-Both memcached and in-memory cache "backend"s are supported:
+memcached/in-memory/redis cache "backend"s are supported:
 
 ```yaml
 type: MEMCACHED # Case-insensitive
@@ -333,7 +378,8 @@ metafile_content_ttl: 24h
 metafile_max_size: 1MiB
 ```
 
-`config` field for memcached supports all the same configuration as memcached for [index cache](#memcached-index-cache). `addresses` in the config field is a **required** setting
+- `config` field for memcached supports all the same configuration as memcached for [index cache](#memcached-index-cache). `addresses` in the config field is a **required** setting
+- `config` field for redis supports all the same configuration as redis for [index cache](#redis-index-cache).
 
 Additional options to configure various aspects of [chunks](../design.md#chunk) cache are available:
 

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/Azure/go-autorest/autorest/azure/auth v0.5.8
 	github.com/NYTimes/gziphandler v1.1.1
 	github.com/alecthomas/units v0.0.0-20210927113745-59d0afb8317a
+	github.com/alicebob/miniredis/v2 v2.14.3
 	github.com/aliyun/aliyun-oss-go-sdk v2.0.4+incompatible
 	github.com/baidubce/bce-sdk-go v0.9.81
 	github.com/blang/semver/v4 v4.0.0
@@ -29,11 +30,14 @@ require (
 	github.com/fsnotify/fsnotify v1.5.1
 	github.com/go-kit/log v0.2.0
 	github.com/go-openapi/strfmt v0.21.0
+	github.com/go-redis/redis/v8 v8.11.4
 	github.com/gogo/protobuf v1.3.2
 	github.com/gogo/status v1.1.0
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da
 	github.com/golang/snappy v0.0.4
 	github.com/googleapis/gax-go v2.0.2+incompatible
+	github.com/grafana/dskit v0.0.0-20211021180445-3bd016e9d7f1
+	github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/kit/v2 v2.0.0-20201002093600-73cf2ae9d891
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.0-rc.2.0.20201207153454-9f6bf00c00a7
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0

go.sum

@@ -946,8 +946,9 @@ github.com/gregjones/httpcache v0.0.0-20180305231024-9cad4c3443a7/go.mod h1:Fecb
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.0/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-middleware v1.0.1-0.20190118093823-f849b5445de4/go.mod h1:FiyG127CGDf3tlThmgyCl78X/SZQqEOJBCDaAfeWzPs=
 github.com/grpc-ecosystem/go-grpc-middleware v1.1.0/go.mod h1:f5nM7jw/oeRSadq3xCzHAvxcr8HZnzsqU6ILg/0NiiE=
-github.com/grpc-ecosystem/go-grpc-middleware v1.2.2 h1:FlFbCRLd5Jr4iYXZufAvgWN6Ao0JrI5chLINnUXDDr0=
 github.com/grpc-ecosystem/go-grpc-middleware v1.2.2/go.mod h1:EaizFBKfUKtMIF5iaDEhniwNedqGo9FuLFzppDr3uwI=
+github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 h1:+9834+KizmvFV7pXQGSXQTsaWhq2GjuNUt0aUU0YBYw=
+github.com/grpc-ecosystem/go-grpc-middleware v1.3.0/go.mod h1:z0ButlSOZa5vEBq9m2m2hlwIgKw+rp3sdCBRoJY+30Y=
 github.com/grpc-ecosystem/go-grpc-middleware/providers/kit/v2 v2.0.0-20201002093600-73cf2ae9d891 h1:RhOqTAECcPnehv3hKlYy1fAnpQ7rnZu58l3mpq6gT1k=
 github.com/grpc-ecosystem/go-grpc-middleware/providers/kit/v2 v2.0.0-20201002093600-73cf2ae9d891/go.mod h1:516cTXxZzi4NBUBbKcwmO4Eqbb6GHAEd3o4N+GYyCBY=
 github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.0-20200501113911-9a95f0fdbfea/go.mod h1:GugMBs30ZSAkckqXEAIEGyYdDH6EgqowG8ppA3Zt+AY=

pkg/cache/memcached.go

@@ -18,7 +18,7 @@ import (
 // MemcachedCache is a memcached-based cache.
 type MemcachedCache struct {
 	logger    log.Logger
-	memcached cacheutil.MemcachedClient
+	memcached cacheutil.RemoteCacheClient
 	name      string
 
 	// Metrics.
@@ -27,7 +27,7 @@ type MemcachedCache struct {
 }
 
 // NewMemcachedCache makes a new MemcachedCache.
-func NewMemcachedCache(name string, logger log.Logger, memcached cacheutil.MemcachedClient, reg prometheus.Registerer) *MemcachedCache {
+func NewMemcachedCache(name string, logger log.Logger, memcached cacheutil.RemoteCacheClient, reg prometheus.Registerer) *MemcachedCache {
 	c := &MemcachedCache{
 		logger:    logger,
 		memcached: memcached,

pkg/cache/redis.go

@@ -0,0 +1,69 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package cache
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	"github.com/thanos-io/thanos/pkg/cacheutil"
+)
+
+// RedisCache is a redis cache.
+type RedisCache struct {
+	logger      log.Logger
+	redisClient *cacheutil.RedisClient
+	name        string
+
+	// Metrics.
+	requests prometheus.Counter
+	hits     prometheus.Counter
+}
+
+// NewRedisCache makes a new RedisCache.
+func NewRedisCache(name string, logger log.Logger, redisClient *cacheutil.RedisClient, reg prometheus.Registerer) *RedisCache {
+	c := &RedisCache{
+		logger:      logger,
+		redisClient: redisClient,
+		name:        name,
+	}
+
+	c.requests = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name:        "thanos_cache_redis_requests_total",
+		Help:        "Total number of items requests to redis.",
+		ConstLabels: prometheus.Labels{"name": name},
+	})
+
+	c.hits = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name:        "thanos_cache_redis_hits_total",
+		Help:        "Total number of items requests to the cache that were a hit.",
+		ConstLabels: prometheus.Labels{"name": name},
+	})
+
+	level.Info(logger).Log("msg", "created redis cache")
+
+	return c
+}
+
+// Store data identified by keys.
+func (c *RedisCache) Store(ctx context.Context, data map[string][]byte, ttl time.Duration) {
+	c.redisClient.SetMulti(ctx, data, ttl)
+}
+
+// Fetch fetches multiple keys and returns a map containing cache hits, along with a list of missing keys.
+// In case of error, it logs and return an empty cache hits map.
+func (c *RedisCache) Fetch(ctx context.Context, keys []string) map[string][]byte {
+	c.requests.Add(float64(len(keys)))
+	results := c.redisClient.GetMulti(ctx, keys)
+	c.hits.Add(float64(len(results)))
+	return results
+}
+
+func (c *RedisCache) Name() string {
+	return c.name
+}

pkg/cache/redis_test.go

@@ -0,0 +1,116 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package cache
+
+import (
+	"context"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/go-kit/log"
+	"github.com/prometheus/client_golang/prometheus"
+	prom_testutil "github.com/prometheus/client_golang/prometheus/testutil"
+	"github.com/thanos-io/thanos/pkg/cacheutil"
+	"github.com/thanos-io/thanos/pkg/testutil"
+)
+
+func TestRedisCache(t *testing.T) {
+	// Init some data to conveniently define test cases later one.
+	key1 := "key1"
+	key2 := "key2"
+	key3 := "key3"
+	value1 := []byte{1}
+	value2 := []byte{2}
+	value3 := []byte{3}
+
+	type args struct {
+		data      map[string][]byte
+		fetchKeys []string
+	}
+	type want struct {
+		hits map[string][]byte
+	}
+	tests := []struct {
+		name string
+		args args
+		want want
+	}{
+		{
+			name: "all hit",
+			args: args{
+				data: map[string][]byte{
+					key1: value1,
+					key2: value2,
+					key3: value3,
+				},
+				fetchKeys: []string{key1, key2, key3},
+			},
+			want: want{
+				hits: map[string][]byte{
+					key1: value1,
+					key2: value2,
+					key3: value3,
+				},
+			},
+		},
+		{
+			name: "partial hit",
+			args: args{
+				data: map[string][]byte{
+					key1: value1,
+					key2: value2,
+				},
+				fetchKeys: []string{key1, key2, key3},
+			},
+			want: want{
+				hits: map[string][]byte{
+					key1: value1,
+					key2: value2,
+				},
+			},
+		},
+		{
+			name: "not hit",
+			args: args{
+				data:      map[string][]byte{},
+				fetchKeys: []string{key1, key2, key3},
+			},
+			want: want{
+				hits: map[string][]byte{},
+			},
+		},
+	}
+	s, err := miniredis.Run()
+	if err != nil {
+		testutil.Ok(t, err)
+	}
+	defer s.Close()
+	logger := log.NewLogfmtLogger(os.Stderr)
+	reg := prometheus.NewRegistry()
+	cfg := cacheutil.DefaultRedisClientConfig
+	cfg.Addr = s.Addr()
+	c, err := cacheutil.NewRedisClientWithConfig(logger, t.Name(), cfg, reg)
+	if err != nil {
+		testutil.Ok(t, err)
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			defer s.FlushAll()
+			c := NewRedisCache(tt.name, logger, c, reg)
+			// Store the cache expected before running the test.
+			ctx := context.Background()
+			c.Store(ctx, tt.args.data, time.Hour)
+
+			// Fetch postings from cached and assert on it.
+			hits := c.Fetch(ctx, tt.args.fetchKeys)
+			testutil.Equals(t, tt.want.hits, hits)
+
+			// Assert on metrics.
+			testutil.Equals(t, float64(len(tt.args.fetchKeys)), prom_testutil.ToFloat64(c.requests))
+			testutil.Equals(t, float64(len(tt.want.hits)), prom_testutil.ToFloat64(c.hits))
+		})
+	}
+}