Chunks caching at bucket level

cmd/thanos/store.go

@@ -55,6 +55,10 @@ func registerStore(m map[string]setupFunc, app *kingpin.Application) {
 		"YAML file that contains index cache configuration. See format details: https://thanos.io/components/store.md/#index-cache",
 		false)
 
+	cachingBucketConfig := extflag.RegisterPathOrContent(cmd, "experimental.caching-bucket.config",
+		"YAML file that contains configuration for caching bucket. Experimental feature, with high risk of changes. See format details: https://thanos.io/components/store.md/#TBD",
+		false)
+
 	chunkPoolSize := cmd.Flag("chunk-pool-size", "Maximum size of concurrently allocatable bytes reserved strictly to reuse for chunks in memory.").
 		Default("2GB").Bytes()
 
@@ -149,6 +153,7 @@ func registerStore(m map[string]setupFunc, app *kingpin.Application) {
 			*webExternalPrefix,
 			*webPrefixHeaderName,
 			*postingOffsetsInMemSampling,
+			cachingBucketConfig,
 		)
 	}
 }
@@ -179,6 +184,7 @@ func runStore(
 	ignoreDeletionMarksDelay time.Duration,
 	externalPrefix, prefixHeader string,
 	postingOffsetsInMemSampling int,
+	cachingBucketConfig *extflag.PathOrContent,
 ) error {
 	grpcProbe := prober.NewGRPC()
 	httpProbe := prober.NewHTTP()
@@ -214,6 +220,18 @@ func runStore(
 		return errors.Wrap(err, "create bucket client")
 	}
 
+	cachingBucketConfigYaml, err := cachingBucketConfig.Content()
+	if err != nil {
+		return errors.Wrap(err, "get caching bucket configuration")
+	}
+	if len(cachingBucketConfigYaml) > 0 {
+		newbkt, err := storecache.NewCachingBucketFromYaml(cachingBucketConfigYaml, bkt, logger, reg)
+		if err != nil {
+			return errors.Wrap(err, "create caching bucket")
+		}
+		bkt = newbkt
+	}
+
 	relabelContentYaml, err := selectorRelabelConf.Content()
 	if err != nil {
 		return errors.Wrap(err, "get content of relabel configuration")

pkg/cache/cache.go

@@ -0,0 +1,20 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package cache
+
+import (
+	"context"
+	"time"
+)
+
+// Generic best-effort cache.
+type Cache interface {
+	// Store data into the cache.
+	//
+	// Note that individual byte buffers may be retained by the cache!
+	Store(ctx context.Context, data map[string][]byte, ttl time.Duration)
+
+	// Fetch multiple keys from cache.
+	Fetch(ctx context.Context, keys []string) (found map[string][]byte, missing []string)
+}

pkg/cache/memcached.go

@@ -0,0 +1,84 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package cache
+
+import (
+	"context"
+	"time"
+
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/promauto"
+
+	"github.com/thanos-io/thanos/pkg/cacheutil"
+)
+
+// MemcachedCache is a memcached-based cache.
+type MemcachedCache struct {
+	logger    log.Logger
+	memcached cacheutil.MemcachedClient
+
+	// Metrics.
+	requests prometheus.Counter
+	hits     prometheus.Counter
+}
+
+// NewMemcachedCache makes a new MemcachedCache.
+func NewMemcachedCache(logger log.Logger, memcached cacheutil.MemcachedClient, reg prometheus.Registerer) *MemcachedCache {
+	c := &MemcachedCache{
+		logger:    logger,
+		memcached: memcached,
+	}
+
+	c.requests = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name: "thanos_cache_memcached_requests_total",
+		Help: "Total number of items requests to the memcached.",
+	})
+
+	c.hits = promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name: "thanos_cache_memcached_hits_total",
+		Help: "Total number of items requests to the cache that were a hit.",
+	})
+
+	level.Info(logger).Log("msg", "created memcached cache")
+
+	return c
+}
+
+// Store data identified by keys.
+// The function enqueues the request and returns immediately: the entry will be
+// asynchronously stored in the cache.
+func (c *MemcachedCache) Store(ctx context.Context, data map[string][]byte, ttl time.Duration) {
+	for key, val := range data {
+		if err := c.memcached.SetAsync(ctx, key, val, ttl); err != nil {
+			level.Error(c.logger).Log("msg", "failed to store data into memcached", "err", err)
+		}
+	}
+}
+
+// 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 *MemcachedCache) Fetch(ctx context.Context, keys []string) (map[string][]byte, []string) {
+	// Fetch the keys from memcached in a single request.
+	c.requests.Add(float64(len(keys)))
+	results := c.memcached.GetMulti(ctx, keys)
+	if len(results) == 0 {
+		return nil, keys
+	}
+
+	// Construct the resulting hits map and list of missing keys. We iterate on the input
+	// list of labels to be able to easily create the list of ones in a single iteration.
+	var misses []string
+	for _, key := range keys {
+		_, ok := results[key]
+		if !ok {
+			misses = append(misses, key)
+			continue
+		}
+	}
+
+	c.hits.Add(float64(len(results)))
+	return results, misses
+}

pkg/cache/memcached_test.go

@@ -0,0 +1,132 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package cache
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/go-kit/kit/log"
+	"github.com/pkg/errors"
+	prom_testutil "github.com/prometheus/client_golang/prometheus/testutil"
+
+	"github.com/thanos-io/thanos/pkg/testutil"
+)
+
+func TestMemcachedIndexCache(t *testing.T) {
+	t.Parallel()
+
+	// 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}
+
+	tests := map[string]struct {
+		setup          map[string][]byte
+		mockedErr      error
+		fetchKeys      []string
+		expectedHits   map[string][]byte
+		expectedMisses []string
+	}{
+		"should return no hits on empty cache": {
+			setup:          nil,
+			fetchKeys:      []string{key1, key2},
+			expectedHits:   nil,
+			expectedMisses: []string{key1, key2},
+		},
+		"should return no misses on 100% hit ratio": {
+			setup: map[string][]byte{
+				key1: value1,
+				key2: value2,
+				key3: value3,
+			},
+			fetchKeys: []string{key1},
+			expectedHits: map[string][]byte{
+				key1: value1,
+			},
+			expectedMisses: nil,
+		},
+		"should return hits and misses on partial hits": {
+			setup: map[string][]byte{
+				key1: value1,
+				key2: value2,
+			},
+			fetchKeys:      []string{key1, key3},
+			expectedHits:   map[string][]byte{key1: value1},
+			expectedMisses: []string{key3},
+		},
+		"should return no hits on memcached error": {
+			setup: map[string][]byte{
+				key1: value1,
+				key2: value2,
+				key3: value3,
+			},
+			mockedErr:      errors.New("mocked error"),
+			fetchKeys:      []string{key1},
+			expectedHits:   nil,
+			expectedMisses: []string{key1},
+		},
+	}
+
+	for testName, testData := range tests {
+		t.Run(testName, func(t *testing.T) {
+			memcached := newMockedMemcachedClient(testData.mockedErr)
+			c := NewMemcachedCache(log.NewNopLogger(), memcached, nil)
+
+			// Store the postings expected before running the test.
+			ctx := context.Background()
+			c.Store(ctx, testData.setup, time.Hour)
+
+			// Fetch postings from cached and assert on it.
+			hits, misses := c.Fetch(ctx, testData.fetchKeys)
+			testutil.Equals(t, testData.expectedHits, hits)
+			testutil.Equals(t, testData.expectedMisses, misses)
+
+			// Assert on metrics.
+			testutil.Equals(t, float64(len(testData.fetchKeys)), prom_testutil.ToFloat64(c.requests))
+			testutil.Equals(t, float64(len(testData.expectedHits)), prom_testutil.ToFloat64(c.hits))
+		})
+	}
+}
+
+type mockedMemcachedClient struct {
+	cache             map[string][]byte
+	mockedGetMultiErr error
+}
+
+func newMockedMemcachedClient(mockedGetMultiErr error) *mockedMemcachedClient {
+	return &mockedMemcachedClient{
+		cache:             map[string][]byte{},
+		mockedGetMultiErr: mockedGetMultiErr,
+	}
+}
+
+func (c *mockedMemcachedClient) GetMulti(_ context.Context, keys []string) map[string][]byte {
+	if c.mockedGetMultiErr != nil {
+		return nil
+	}
+
+	hits := map[string][]byte{}
+
+	for _, key := range keys {
+		if value, ok := c.cache[key]; ok {
+			hits[key] = value
+		}
+	}
+
+	return hits
+}
+
+func (c *mockedMemcachedClient) SetAsync(_ context.Context, key string, value []byte, _ time.Duration) error {
+	c.cache[key] = value
+	return nil
+}
+
+func (c *mockedMemcachedClient) Stop() {
+	// Nothing to do.
+}