Support Snappy compression for gRPC

CHANGELOG.md

@@ -40,6 +40,13 @@
     - `-experimental.tsdb.stripe-size` changed to `-experimental.blocks-storage.tsdb.stripe-size`
     - `-experimental.tsdb.wal-compression-enabled` changed to `-experimental.blocks-storage.tsdb.wal-compression-enabled`
     - `-experimental.tsdb.flush-blocks-on-shutdown` changed to `-experimental.blocks-storage.tsdb.flush-blocks-on-shutdown`
+* [ENHANCEMENT/CHANGE] Added new flags `-*.grpc-compression` to configure compression used by gRPC. Valid values are "gzip", "snappy", "" (no compression, default). Previous flags (`-*.use-gzip-compression`, enumerate) are now deprecated. #2940
+  * `-bigtable.grpc-compression`
+  * `-ingester.client.grpc-compression`
+  * `-querier.frontend-client.grpc-compression`
+  * Deprecated: `-bigtable.grpc-use-gzip-compression`
+  * Deprecated: `-ingester.client.grpc-use-gzip-compression`
+  * Deprecated: `-querier.frontend-client.grpc-use-gzip-compression`
 * [FEATURE] Introduced `ruler.for-outage-tolerance`, Max time to tolerate outage for restoring "for" state of alert. #2783
 * [FEATURE] Introduced `ruler.for-grace-period`, Minimum duration between alert and restored "for" state. This is maintained only for alerts with configured "for" time greater than grace period. #2783
 * [FEATURE] Introduced `ruler.resend-delay`, Minimum amount of time to wait before resending an alert to Alertmanager. #2783

docs/configuration/config-file-reference.md

@@ -1774,10 +1774,15 @@ bigtable:
     # CLI flag: -bigtable.grpc-max-send-msg-size
     [max_send_msg_size: <int> | default = 16777216]
 
-    # Use compression when sending messages.
+    # Deprecated: Use gzip compression when sending messages.
     # CLI flag: -bigtable.grpc-use-gzip-compression
     [use_gzip_compression: <boolean> | default = false]
 
+    # Use compression when sending messages. Supported values are: 'gzip',
+    # 'snappy' and '' (disable compression)
+    # CLI flag: -bigtable.grpc-compression
+    [compression: <string> | default = ""]
+
     # Rate limit for gRPC client; 0 means disabled.
     # CLI flag: -bigtable.grpc-client-rate-limit
     [rate_limit: <float> | default = 0]
@@ -2265,10 +2270,15 @@ grpc_client_config:
   # CLI flag: -ingester.client.grpc-max-send-msg-size
   [max_send_msg_size: <int> | default = 16777216]
 
-  # Use compression when sending messages.
+  # Deprecated: Use gzip compression when sending messages.
   # CLI flag: -ingester.client.grpc-use-gzip-compression
   [use_gzip_compression: <boolean> | default = false]
 
+  # Use compression when sending messages. Supported values are: 'gzip',
+  # 'snappy' and '' (disable compression)
+  # CLI flag: -ingester.client.grpc-compression
+  [compression: <string> | default = ""]
+
   # Rate limit for gRPC client; 0 means disabled.
   # CLI flag: -ingester.client.grpc-client-rate-limit
   [rate_limit: <float> | default = 0]
@@ -2338,10 +2348,15 @@ grpc_client_config:
   # CLI flag: -querier.frontend-client.grpc-max-send-msg-size
   [max_send_msg_size: <int> | default = 16777216]
 
-  # Use compression when sending messages.
+  # Deprecated: Use gzip compression when sending messages.
   # CLI flag: -querier.frontend-client.grpc-use-gzip-compression
   [use_gzip_compression: <boolean> | default = false]
 
+  # Use compression when sending messages. Supported values are: 'gzip',
+  # 'snappy' and '' (disable compression)
+  # CLI flag: -querier.frontend-client.grpc-compression
+  [compression: <string> | default = ""]
+
   # Rate limit for gRPC client; 0 means disabled.
   # CLI flag: -querier.frontend-client.grpc-client-rate-limit
   [rate_limit: <float> | default = 0]

pkg/chunk/gcp/bigtable_index_client.go

@@ -11,9 +11,9 @@ import (
 	"time"
 
 	"cloud.google.com/go/bigtable"
+	"github.com/go-kit/kit/log"
 	ot "github.com/opentracing/opentracing-go"
 	otlog "github.com/opentracing/opentracing-go/log"
-
 	"github.com/pkg/errors"
 
 	"github.com/cortexproject/cortex/pkg/chunk"
@@ -54,6 +54,10 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.GRPCClientConfig.RegisterFlagsWithPrefix("bigtable", f)
 }
 
+func (cfg *Config) Validate(log log.Logger) error {
+	return cfg.GRPCClientConfig.Validate(log)
+}
+
 // storageClientColumnKey implements chunk.storageClient for GCP.
 type storageClientColumnKey struct {
 	cfg       Config

pkg/chunk/storage/factory.go

@@ -105,6 +105,9 @@ func (cfg *Config) Validate() error {
 	if err := cfg.CassandraStorageConfig.Validate(); err != nil {
 		return errors.Wrap(err, "invalid Cassandra Storage config")
 	}
+	if err := cfg.GCPStorageConfig.Validate(util.Logger); err != nil {
+		return errors.Wrap(err, "invalid GCP Storage Storage config")
+	}
 	if err := cfg.Swift.Validate(); err != nil {
 		return errors.Wrap(err, "invalid Swift Storage config")
 	}

pkg/cortex/cortex.go

@@ -178,11 +178,17 @@ func (c *Config) Validate(log log.Logger) error {
 	if err := c.Querier.Validate(); err != nil {
 		return errors.Wrap(err, "invalid querier config")
 	}
+	if err := c.IngesterClient.Validate(log); err != nil {
+		return errors.Wrap(err, "invalid ingester_client config")
+	}
+	if err := c.Worker.Validate(log); err != nil {
+		return errors.Wrap(err, "invalid frontend_worker config")
+	}
 	if err := c.QueryRange.Validate(log); err != nil {
-		return errors.Wrap(err, "invalid queryrange config")
+		return errors.Wrap(err, "invalid query_range config")
 	}
 	if err := c.TableManager.Validate(); err != nil {
-		return errors.Wrap(err, "invalid tablemanager config")
+		return errors.Wrap(err, "invalid table_manager config")
 	}
 	return nil
 }

pkg/ingester/client/client.go

@@ -3,12 +3,14 @@ package client
 import (
 	"flag"
 
+	"github.com/go-kit/kit/log"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"google.golang.org/grpc"
 	_ "google.golang.org/grpc/encoding/gzip" // get gzip compressor registered
 	"google.golang.org/grpc/health/grpc_health_v1"
 
+	_ "github.com/cortexproject/cortex/pkg/util/grpc/encoding/snappy" // get snappy compressor registered
 	"github.com/cortexproject/cortex/pkg/util/grpcclient"
 )
 
@@ -62,3 +64,7 @@ type Config struct {
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.GRPCClientConfig.RegisterFlagsWithPrefix("ingester.client", f)
 }
+
+func (cfg *Config) Validate(log log.Logger) error {
+	return cfg.GRPCClientConfig.Validate(log)
+}

pkg/querier/frontend/worker.go

@@ -40,6 +40,10 @@ func (cfg *WorkerConfig) RegisterFlags(f *flag.FlagSet) {
 	cfg.GRPCClientConfig.RegisterFlagsWithPrefix("querier.frontend-client", f)
 }
 
+func (cfg *WorkerConfig) Validate(log log.Logger) error {
+	return cfg.GRPCClientConfig.Validate(log)
+}
+
 // Worker is the counter-part to the frontend, actually processing requests.
 type worker struct {
 	cfg        WorkerConfig

pkg/util/grpc/encoding/snappy/snappy.go

@@ -0,0 +1,81 @@
+package snappy
+
+import (
+	"io"
+	"sync"
+
+	"github.com/golang/snappy"
+	"google.golang.org/grpc/encoding"
+)
+
+// Name is the name registered for the snappy compressor.
+const Name = "snappy"
+
+func init() {
+	encoding.RegisterCompressor(newCompressor())
+}
+
+type compressor struct {
+	writersPool sync.Pool
+	readersPool sync.Pool
+}
+
+func newCompressor() *compressor {
+	c := &compressor{}
+	c.readersPool = sync.Pool{
+		New: func() interface{} {
+			return &reader{
+				compressor: c,
+				Reader:     snappy.NewReader(nil),
+			}
+		},
+	}
+	c.writersPool = sync.Pool{
+		New: func() interface{} {
+			return &writeCloser{
+				compressor: c,
+				Writer:     snappy.NewBufferedWriter(nil),
+			}
+		},
+	}
+	return c
+}
+
+func (c *compressor) Name() string {
+	return Name
+}
+
+func (c *compressor) Compress(w io.Writer) (io.WriteCloser, error) {
+	wr := c.writersPool.Get().(*writeCloser)
+	wr.Reset(w)
+	return wr, nil
+}
+
+func (c *compressor) Decompress(r io.Reader) (io.Reader, error) {
+	dr := c.readersPool.Get().(*reader)
+	dr.Reset(r)
+	return dr, nil
+}
+
+type writeCloser struct {
+	*compressor
+	*snappy.Writer
+}
+
+func (w *writeCloser) Close() error {
+	defer w.writersPool.Put(w)
+	return w.Writer.Close()
+}
+
+type reader struct {
+	*compressor
+	*snappy.Reader
+}
+
+func (r *reader) Read(p []byte) (n int, err error) {
+	n, err = r.Reader.Read(p)
+	if err == io.EOF {
+		r.readersPool.Put(r)
+	}
+	return n, err
+}

pkg/util/grpcclient/grpcclient.go

@@ -2,11 +2,16 @@ package grpcclient
 
 import (
 	"flag"
+	"fmt"
 
+	"github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/log/level"
 	grpc_middleware "github.com/grpc-ecosystem/go-grpc-middleware"
+	"github.com/pkg/errors"
 	"google.golang.org/grpc"
 
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/tls"
 )
 
@@ -15,11 +20,14 @@ type Config struct {
 	MaxRecvMsgSize     int     `yaml:"max_recv_msg_size"`
 	MaxSendMsgSize     int     `yaml:"max_send_msg_size"`
 	UseGzipCompression bool    `yaml:"use_gzip_compression"`
+	Compression        string  `yaml:"compression"`
 	RateLimit          float64 `yaml:"rate_limit"`
 	RateLimitBurst     int     `yaml:"rate_limit_burst"`
 
 	BackoffOnRatelimits bool               `yaml:"backoff_on_ratelimits"`
 	BackoffConfig       util.BackoffConfig `yaml:"backoff_config"`
+
+	prefix string `yaml:"-"`
 }
 
 // RegisterFlags registers flags.
@@ -29,23 +37,43 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 
 // RegisterFlagsWithPrefix registers flags with prefix.
 func (cfg *Config) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet) {
+	cfg.prefix = prefix
 	f.IntVar(&cfg.MaxRecvMsgSize, prefix+".grpc-max-recv-msg-size", 100<<20, "gRPC client max receive message size (bytes).")
 	f.IntVar(&cfg.MaxSendMsgSize, prefix+".grpc-max-send-msg-size", 16<<20, "gRPC client max send message size (bytes).")
-	f.BoolVar(&cfg.UseGzipCompression, prefix+".grpc-use-gzip-compression", false, "Use compression when sending messages.")
+	f.BoolVar(&cfg.UseGzipCompression, prefix+".grpc-use-gzip-compression", false, "Deprecated: Use gzip compression when sending messages.")
+	f.StringVar(&cfg.Compression, prefix+".grpc-compression", "", "Use compression when sending messages. Supported values are: 'gzip', 'snappy' and '' (disable compression)")
 	f.Float64Var(&cfg.RateLimit, prefix+".grpc-client-rate-limit", 0., "Rate limit for gRPC client; 0 means disabled.")
 	f.IntVar(&cfg.RateLimitBurst, prefix+".grpc-client-rate-limit-burst", 0, "Rate limit burst for gRPC client.")
 	f.BoolVar(&cfg.BackoffOnRatelimits, prefix+".backoff-on-ratelimits", false, "Enable backoff and retry when we hit ratelimits.")
 
 	cfg.BackoffConfig.RegisterFlags(prefix, f)
 }
 
+func (cfg *Config) Validate(log log.Logger) error {
+	if cfg.UseGzipCompression {
+		flagext.DeprecatedFlagsUsed.Inc()
+		level.Warn(log).Log("msg", fmt.Sprintf("running with DEPRECATED flag -%s.grpc-use-gzip-compression, use -%s.grpc-use-compression instead.", cfg.prefix, cfg.prefix))
+	}
+	switch cfg.Compression {
+	case "gzip", "snappy", "":
+		// valid
+	default:
+		return errors.Errorf("unsupported compression type: %s", cfg.Compression)
+	}
+	return nil
+}
+
 // CallOptions returns the config in terms of CallOptions.
 func (cfg *Config) CallOptions() []grpc.CallOption {
 	var opts []grpc.CallOption
 	opts = append(opts, grpc.MaxCallRecvMsgSize(cfg.MaxRecvMsgSize))
 	opts = append(opts, grpc.MaxCallSendMsgSize(cfg.MaxSendMsgSize))
+	compression := cfg.Compression
 	if cfg.UseGzipCompression {
-		opts = append(opts, grpc.UseCompressor("gzip"))
+		compression = "gzip"
+	}
+	if compression != "" {
+		opts = append(opts, grpc.UseCompressor(compression))
 	}
 	return opts
 }
@@ -79,6 +107,10 @@ func (cfg *ConfigWithTLS) RegisterFlagsWithPrefix(prefix string, f *flag.FlagSet
 	cfg.TLS.RegisterFlagsWithPrefix(prefix, f)
 }
 
+func (cfg *ConfigWithTLS) Validate(log log.Logger) error {
+	return cfg.GRPC.Validate(log)
+}
+
 // DialOption returns the config as a grpc.DialOptions
 func (cfg *ConfigWithTLS) DialOption(unaryClientInterceptors []grpc.UnaryClientInterceptor, streamClientInterceptors []grpc.StreamClientInterceptor) ([]grpc.DialOption, error) {
 	opts, err := cfg.TLS.GetGRPCDialOptions()