RedisCache: Use CollectedRequest (#3976)

* RedisCache: Use CollectedRequest

Signed-off-by: Arve Knudsen <[email protected]>

* RedisCache: Apply feedback from review

Signed-off-by: Arve Knudsen <[email protected]>

* Rename innerCtx to ctx

Signed-off-by: Arve Knudsen <[email protected]>

* Add changelog entry

Signed-off-by: Arve Knudsen <[email protected]>

Author: aknuds1

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ## master / unreleased
 
+* [CHANGE] Added instrumentation to Redis client, with the following metrics: #3976
+  - `rediscache_request_duration_seconds`
 * [CHANGE] Enable strict JSON unmarshal for `pkg/util/validation.Limits` struct. The custom `UnmarshalJSON()` will now fail if the input has unknown fields. #4298
 * [CHANGE] Cortex chunks storage has been deprecated and it's now in maintenance mode: all Cortex users are encouraged to migrate to the blocks storage. No new features will be added to the chunks storage. The default Cortex configuration still runs the chunks engine; please check out the [blocks storage doc](https://cortexmetrics.io/docs/blocks-storage/) on how to configure Cortex to run with the blocks storage.  #4268
 * [CHANGE] The example Kubernetes manifests (stored at `k8s/`) have been removed due to a lack of proper support and maintenance. #4268

pkg/chunk/cache/cache.go

@@ -99,7 +99,7 @@ func New(cfg Config, reg prometheus.Registerer, logger log.Logger) (Cache, error
 			cfg.Redis.Expiration = cfg.DefaultValidity
 		}
 		cacheName := cfg.Prefix + "redis"
-		cache := NewRedisCache(cacheName, NewRedisClient(&cfg.Redis), logger)
+		cache := NewRedisCache(cacheName, NewRedisClient(&cfg.Redis), reg, logger)
 		caches = append(caches, NewBackground(cacheName, cfg.Background, Instrument(cacheName, cache, reg), reg))
 	}
 

pkg/chunk/cache/redis_cache.go

@@ -5,48 +5,91 @@ import (
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
+	otlog "github.com/opentracing/opentracing-go/log"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+	instr "github.com/weaveworks/common/instrument"
 
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
+	"github.com/cortexproject/cortex/pkg/util/spanlogger"
 )
 
 // RedisCache type caches chunks in redis
 type RedisCache struct {
-	name   string
-	redis  *RedisClient
-	logger log.Logger
+	name            string
+	redis           *RedisClient
+	logger          log.Logger
+	requestDuration observableVecCollector
 }
 
 // NewRedisCache creates a new RedisCache
-func NewRedisCache(name string, redisClient *RedisClient, logger log.Logger) *RedisCache {
+func NewRedisCache(name string, redisClient *RedisClient, reg prometheus.Registerer, logger log.Logger) *RedisCache {
 	util_log.WarnExperimentalUse("Redis cache")
 	cache := &RedisCache{
 		name:   name,
 		redis:  redisClient,
 		logger: logger,
+		requestDuration: observableVecCollector{
+			v: promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
+				Namespace:   "cortex",
+				Name:        "rediscache_request_duration_seconds",
+				Help:        "Total time spent in seconds doing Redis requests.",
+				Buckets:     prometheus.ExponentialBuckets(0.000016, 4, 8),
+				ConstLabels: prometheus.Labels{"name": name},
+			}, []string{"method", "status_code"}),
+		},
 	}
 	if err := cache.redis.Ping(context.Background()); err != nil {
 		level.Error(logger).Log("msg", "error connecting to redis", "name", name, "err", err)
 	}
 	return cache
 }
 
+func redisStatusCode(err error) string {
+	// TODO: Figure out if there are more error types returned by Redis
+	switch err {
+	case nil:
+		return "200"
+	default:
+		return "500"
+	}
+}
+
 // Fetch gets keys from the cache. The keys that are found must be in the order of the keys requested.
 func (c *RedisCache) Fetch(ctx context.Context, keys []string) (found []string, bufs [][]byte, missed []string) {
-	data, err := c.redis.MGet(ctx, keys)
+	const method = "RedisCache.MGet"
+	var items [][]byte
+	// Run a tracked request, using c.requestDuration to monitor requests.
+	err := instr.CollectedRequest(ctx, method, c.requestDuration, redisStatusCode, func(ctx context.Context) error {
+		log, _ := spanlogger.New(ctx, method)
+		defer log.Finish()
+		log.LogFields(otlog.Int("keys requested", len(keys)))
+
+		var err error
+		items, err = c.redis.MGet(ctx, keys)
+		if err != nil {
+			log.Error(err)
+			level.Error(c.logger).Log("msg", "failed to get from redis", "name", c.name, "err", err)
+			return err
+		}
+
+		log.LogFields(otlog.Int("keys found", len(items)))
+
+		return nil
+	})
 	if err != nil {
-		level.Error(c.logger).Log("msg", "failed to get from redis", "name", c.name, "err", err)
-		missed = make([]string, len(keys))
-		copy(missed, keys)
-		return
+		return found, bufs, keys
 	}
+
 	for i, key := range keys {
-		if data[i] != nil {
+		if items[i] != nil {
 			found = append(found, key)
-			bufs = append(bufs, data[i])
+			bufs = append(bufs, items[i])
 		} else {
 			missed = append(missed, key)
 		}
 	}
+
 	return
 }
 

pkg/chunk/cache/redis_cache_test.go

@@ -63,5 +63,5 @@ func mockRedisCache() (*RedisCache, error) {
 			Addrs: []string{redisServer.Addr()},
 		}),
 	}
-	return NewRedisCache("mock", redisClient, log.NewNopLogger()), nil
+	return NewRedisCache("mock", redisClient, nil, log.NewNopLogger()), nil
 }