Implementing multi level index cache

Signed-off-by: Alan Protasio <alanprot@gmail.com>

Author: alanprot

pkg/storage/tsdb/index_cache.go

@@ -60,17 +60,22 @@ func (cfg *IndexCacheConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix str
 
 // Validate the config.
 func (cfg *IndexCacheConfig) Validate() error {
-	if !util.StringsContain(supportedIndexCacheBackends, cfg.Backend) {
-		return errUnsupportedIndexCacheBackend
-	}
 
-	if cfg.Backend == IndexCacheBackendMemcached {
-		if err := cfg.Memcached.Validate(); err != nil {
-			return err
+	splitedBackends := strings.Split(cfg.Backend, ",")
+
+	for _, backend := range splitedBackends {
+		if !util.StringsContain(supportedIndexCacheBackends, backend) {
+			return errUnsupportedIndexCacheBackend
 		}
-	} else if cfg.Backend == IndexCacheBackendRedis {
-		if err := cfg.Redis.Validate(); err != nil {
-			return err
+
+		if backend == IndexCacheBackendMemcached {
+			if err := cfg.Memcached.Validate(); err != nil {
+				return err
+			}
+		} else if backend == IndexCacheBackendRedis {
+			if err := cfg.Redis.Validate(); err != nil {
+				return err
+			}
 		}
 	}
 
@@ -87,16 +92,42 @@ func (cfg *InMemoryIndexCacheConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, pr
 
 // NewIndexCache creates a new index cache based on the input configuration.
 func NewIndexCache(cfg IndexCacheConfig, logger log.Logger, registerer prometheus.Registerer) (storecache.IndexCache, error) {
-	switch cfg.Backend {
-	case IndexCacheBackendInMemory:
-		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
+	splitBackends := strings.Split(cfg.Backend, ",")
+	var caches []storecache.IndexCache
+
+	for i, backend := range splitBackends {
+		iReg := registerer
+
+		// Create the level label if we have more than one cache
+		if len(splitBackends) > 1 {
+			iReg = prometheus.WrapRegistererWith(prometheus.Labels{"level": fmt.Sprintf("L%v", i)}, registerer)
+		}
+
+		switch backend {
+		case IndexCacheBackendInMemory:
+			c, err := newInMemoryIndexCache(cfg.InMemory, logger, iReg)
+			if err != nil {
+				return c, err
+			}
+			caches = append(caches, c)
+		case IndexCacheBackendMemcached:
+			c, err := newMemcachedIndexCache(cfg.Memcached, logger, iReg)
+			if err != nil {
+				return c, err
+			}
+			caches = append(caches, c)
+		case IndexCacheBackendRedis:
+			c, err := newRedisIndexCache(cfg.Redis, logger, iReg)
+			if err != nil {
+				return c, err
+			}
+			caches = append(caches, c)
+		default:
+			return nil, errUnsupportedIndexCacheBackend
+		}
 	}
+
+	return newMultiLevelCache(caches...), nil
 }
 
 func newInMemoryIndexCache(cfg InMemoryIndexCacheConfig, logger log.Logger, registerer prometheus.Registerer) (storecache.IndexCache, error) {

pkg/storage/tsdb/multilevel_cache.go

@@ -0,0 +1,109 @@
+package tsdb
+
+import (
+	"context"
+	"sync"
+
+	"github.com/oklog/ulid"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	storecache "github.com/thanos-io/thanos/pkg/store/cache"
+)
+
+type multiLevelCache struct {
+	caches []storecache.IndexCache
+}
+
+func (m *multiLevelCache) StorePostings(blockID ulid.ULID, l labels.Label, v []byte) {
+	wg := sync.WaitGroup{}
+	wg.Add(len(m.caches))
+	for _, c := range m.caches {
+		cache := c
+		go func() {
+			defer wg.Done()
+			cache.StorePostings(blockID, l, v)
+		}()
+	}
+	wg.Wait()
+}
+
+func (m *multiLevelCache) FetchMultiPostings(ctx context.Context, blockID ulid.ULID, keys []labels.Label) (hits map[labels.Label][]byte, misses []labels.Label) {
+	misses = keys
+	hits = map[labels.Label][]byte{}
+	for _, c := range m.caches {
+		h, m := c.FetchMultiPostings(ctx, blockID, misses)
+		misses = m
+
+		for label, bytes := range h {
+			hits[label] = bytes
+		}
+		if len(misses) == 0 {
+			break
+		}
+	}
+
+	return hits, misses
+}
+
+func (m *multiLevelCache) StoreExpandedPostings(blockID ulid.ULID, matchers []*labels.Matcher, v []byte) {
+	wg := sync.WaitGroup{}
+	wg.Add(len(m.caches))
+	for _, c := range m.caches {
+		cache := c
+		go func() {
+			defer wg.Done()
+			cache.StoreExpandedPostings(blockID, matchers, v)
+		}()
+	}
+	wg.Wait()
+}
+
+func (m *multiLevelCache) FetchExpandedPostings(ctx context.Context, blockID ulid.ULID, matchers []*labels.Matcher) ([]byte, bool) {
+	for _, c := range m.caches {
+		if d, h := c.FetchExpandedPostings(ctx, blockID, matchers); h {
+			return d, h
+		}
+	}
+
+	return []byte{}, false
+}
+
+func (m *multiLevelCache) StoreSeries(blockID ulid.ULID, id storage.SeriesRef, v []byte) {
+	wg := sync.WaitGroup{}
+	wg.Add(len(m.caches))
+	for _, c := range m.caches {
+		cache := c
+		go func() {
+			defer wg.Done()
+			cache.StoreSeries(blockID, id, v)
+		}()
+	}
+	wg.Wait()
+}
+
+func (m *multiLevelCache) FetchMultiSeries(ctx context.Context, blockID ulid.ULID, ids []storage.SeriesRef) (hits map[storage.SeriesRef][]byte, misses []storage.SeriesRef) {
+	misses = ids
+	hits = map[storage.SeriesRef][]byte{}
+	for _, c := range m.caches {
+		h, m := c.FetchMultiSeries(ctx, blockID, misses)
+		misses = m
+
+		for label, bytes := range h {
+			hits[label] = bytes
+		}
+		if len(misses) == 0 {
+			break
+		}
+	}
+
+	return hits, misses
+}
+
+func newMultiLevelCache(c ...storecache.IndexCache) storecache.IndexCache {
+	if len(c) == 0 {
+		return c[0]
+	}
+	return &multiLevelCache{
+		caches: c,
+	}
+}