Index Cache: Change cache key for postings

CHANGELOG.md

@@ -53,6 +53,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#6322](https://github.com/thanos-io/thanos/pull/6322) Logging: Avoid expensive log.Valuer evaluation for disallowed levels.
 - [#6358](https://github.com/thanos-io/thanos/pull/6358) Query: Add +Inf bucket to query duration metrics
 - [#6363](https://github.com/thanos-io/thanos/pull/6363) Store: Check context error when expanding postings.
+- [#6405](https://github.com/thanos-io/thanos/pull/6405) Index Cache: Change postings cache key due to the streaming snappy encoding change.
 
 ### Removed
 

pkg/store/cache/cache.go

@@ -79,6 +79,7 @@ func (c cacheKey) string() string {
 		// which would end up in wrong query results.
 		lbl := c.key.(cacheKeyPostings)
 		lblHash := blake2b.Sum256([]byte(lbl.Name + ":" + lbl.Value))
+		// Add : at the end to force using a new cache key for postings.
 		return "P:" + c.block + ":" + base64.RawURLEncoding.EncodeToString(lblHash[0:])
 	case cacheKeySeries:
 		return "S:" + c.block + ":" + strconv.FormatUint(uint64(c.key.(cacheKeySeries)), 10)

pkg/store/cache/memcached.go

@@ -22,11 +22,17 @@ const (
 	memcachedDefaultTTL = 24 * time.Hour
 )
 
+const (
+	compressionSchemeStreamedSnappy = "dvs"
+)
+
 // RemoteIndexCache is a memcached-based index cache.
 type RemoteIndexCache struct {
 	logger    log.Logger
 	memcached cacheutil.RemoteCacheClient
 
+	compressionScheme string
+
 	// Metrics.
 	postingRequests prometheus.Counter
 	seriesRequests  prometheus.Counter
@@ -37,8 +43,9 @@ type RemoteIndexCache struct {
 // NewRemoteIndexCache makes a new RemoteIndexCache.
 func NewRemoteIndexCache(logger log.Logger, cacheClient cacheutil.RemoteCacheClient, reg prometheus.Registerer) (*RemoteIndexCache, error) {
 	c := &RemoteIndexCache{
-		logger:    logger,
-		memcached: cacheClient,
+		logger:            logger,
+		memcached:         cacheClient,
+		compressionScheme: compressionSchemeStreamedSnappy, // Hardcode it for now. Expose it once we supporter different types of compressions.
 	}
 
 	requests := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
@@ -65,6 +72,9 @@ func NewRemoteIndexCache(logger log.Logger, cacheClient cacheutil.RemoteCacheCli
 // asynchronously stored in the cache.
 func (c *RemoteIndexCache) StorePostings(blockID ulid.ULID, l labels.Label, v []byte) {
 	key := cacheKey{blockID.String(), cacheKeyPostings(l)}.string()
+	if len(c.compressionScheme) > 0 {
+		key += ":" + c.compressionScheme
+	}
 
 	if err := c.memcached.SetAsync(key, v, memcachedDefaultTTL); err != nil {
 		level.Error(c.logger).Log("msg", "failed to cache postings in memcached", "err", err)
@@ -78,8 +88,12 @@ func (c *RemoteIndexCache) FetchMultiPostings(ctx context.Context, blockID ulid.
 	keys := make([]string, 0, len(lbls))
 
 	blockIDKey := blockID.String()
+	suffix := ""
+	if len(c.compressionScheme) > 0 {
+		suffix += ":" + c.compressionScheme
+	}
 	for _, lbl := range lbls {
-		key := cacheKey{blockIDKey, cacheKeyPostings(lbl)}.string()
+		key := cacheKey{blockIDKey, cacheKeyPostings(lbl)}.string() + suffix
 		keys = append(keys, key)
 	}