support snappy cache in query frontend (#3217)

* support snappy cache in query frontend

Signed-off-by: Ben Ye <[email protected]>

* add changelog

Signed-off-by: Ben Ye <[email protected]>

* add supported compression type check

Signed-off-by: Ben Ye <[email protected]>

Author: yeya24

CHANGELOG.md

@@ -10,6 +10,7 @@
 * [CHANGE] Zone-awareness replication for time-series now should be explicitly enabled in the distributor via the `-distributor.zone-awareness-enabled` CLI flag (or its respective YAML config option). Before, zone-aware replication was implicitly enabled if a zone was set on ingesters. #3200
 * [CHANGE] Removed the deprecated CLI flag `-config-yaml`. You should use `-schema-config-file` instead. #3225
 * [FEATURE] Added support for shuffle-sharding queriers in the query-frontend. When configured (`-frontend.max-queriers-per-user` globally, or using per-user limit `max_queriers_per_user`), each user's requests will be handled by different set of queriers. #3113
+* [FEATURE] Query-frontend: added `compression` config to support results cache with compression. #3217
 * [ENHANCEMENT] Added `cortex_query_frontend_connected_clients` metric to show the number of workers currently connected to the frontend. #3207
 * [ENHANCEMENT] Shuffle sharding: improved shuffle sharding in the write path. Shuffle sharding now should be explicitly enabled via `-distributor.sharding-strategy` CLI flag (or its respective YAML config option) and guarantees stability, consistency, shuffling and balanced zone-awareness properties. #3090
 * [ENHANCEMENT] Ingester: added new metric `cortex_ingester_active_series` to track active series more accurately. Also added options to control whether active series tracking is enabled (`-ingester.active-series-enabled`, defaults to false), and how often this metric is updated (`-ingester.active-series-update-period`) and max idle time for series to be considered inactive (`-ingester.active-series-idle-timeout`). #3153

docs/configuration/config-file-reference.md

@@ -817,6 +817,11 @@ results_cache:
     # The CLI flags prefix for this block config is: frontend
     [fifocache: <fifo_cache_config>]
 
+  # Use compression in results cache. Supported values are: 'snappy' and ''
+  # (disable compression).
+  # CLI flag: -frontend.compression
+  [compression: <string> | default = ""]
+
 # Cache query results.
 # CLI flag: -querier.cache-results
 [cache_results: <boolean> | default = false]

pkg/querier/queryrange/results_cache.go

@@ -14,6 +14,7 @@ import (
 	"github.com/gogo/protobuf/types"
 	"github.com/opentracing/opentracing-go"
 	otlog "github.com/opentracing/opentracing-go/log"
+	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/uber/jaeger-client-go"
@@ -41,15 +42,28 @@ type CacheGenNumberLoader interface {
 // ResultsCacheConfig is the config for the results cache.
 type ResultsCacheConfig struct {
 	CacheConfig cache.Config `yaml:"cache"`
+	Compression string       `yaml:"compression"`
 }
 
 // RegisterFlags registers flags.
 func (cfg *ResultsCacheConfig) RegisterFlags(f *flag.FlagSet) {
 	cfg.CacheConfig.RegisterFlagsWithPrefix("frontend.", "", f)
 
+	f.StringVar(&cfg.Compression, "frontend.compression", "", "Use compression in results cache. Supported values are: 'snappy' and '' (disable compression).")
 	flagext.DeprecatedFlag(f, "frontend.cache-split-interval", "Deprecated: The maximum interval expected for each request, results will be cached per single interval. This behavior is now determined by querier.split-queries-by-interval.")
 }
 
+func (cfg *ResultsCacheConfig) Validate() error {
+	switch cfg.Compression {
+	case "snappy", "":
+		// valid
+	default:
+		return errors.Errorf("unsupported compression type: %s", cfg.Compression)
+	}
+
+	return cfg.CacheConfig.Validate()
+}
+
 // Extractor is used by the cache to extract a subset of a response from a cache entry.
 type Extractor interface {
 	// Extract extracts a subset of a response from the `start` and `end` timestamps in milliseconds in the `from` response.
@@ -140,6 +154,9 @@ func NewResultsCacheMiddleware(
 	if err != nil {
 		return nil, nil, err
 	}
+	if cfg.Compression == "snappy" {
+		c = cache.NewSnappy(c, logger)
+	}
 
 	if cacheGenNumberLoader != nil {
 		c = cache.NewCacheGenNumMiddleware(c)

pkg/querier/queryrange/results_cache_test.go

@@ -261,7 +261,7 @@ func TestShouldCache(t *testing.T) {
 	}
 }
 
-func TestPartiton(t *testing.T) {
+func TestPartition(t *testing.T) {
 	for i, tc := range []struct {
 		input                  Request
 		prevCachedResponse     []Extent

pkg/querier/queryrange/roundtrip.go

@@ -82,7 +82,7 @@ func (cfg *Config) Validate(log log.Logger) error {
 		if cfg.SplitQueriesByInterval <= 0 {
 			return errors.New("querier.cache-results may only be enabled in conjunction with querier.split-queries-by-interval. Please set the latter")
 		}
-		if err := cfg.ResultsCacheConfig.CacheConfig.Validate(); err != nil {
+		if err := cfg.ResultsCacheConfig.Validate(); err != nil {
 			return errors.Wrap(err, "invalid ResultsCache config")
 		}
 	}