Fix

Signed-off-by: Wing924 <[email protected]>

Author: Wing924

CHANGELOG.md

@@ -88,11 +88,8 @@
 * [ENHANCEMENT] Flusher: Added `-flusher.exit-after-flush` option (defaults to true) to control whether Cortex should stop completely after Flusher has finished its work. #2877
 * [ENHANCEMENT] Added metrics `cortex_config_hash` and `cortex_runtime_config_hash` to expose hash of the currently active config file. #2874
 * [ENHANCEMENT] Logger: added JSON logging support, configured via the `-log.format=json` CLI flag or its respective YAML config option. #2386
-<<<<<<< HEAD
 * [ENHANCEMENT] Added new flags `-bigtable.grpc-compression`, `-ingester.client.grpc-compression`, `-querier.frontend-client.grpc-compression` to configure compression used by gRPC. Valid values are `gzip`, `snappy`, or empty string (no compression, default). #2940
-=======
 * [ENHANCEMENT] Clarify limitations of the `/api/v1/series`, `/api/v1/labels` and `/api/v1/label/{name}/values` endpoints. #2953
->>>>>>> master
 * [BUGFIX] Fixed a bug in the index intersect code causing storage to return more chunks/series than required. #2796
 * [BUGFIX] Fixed the number of reported keys in the background cache queue. #2764
 * [BUGFIX] Fix race in processing of headers in sharded queries. #2762

docs/configuration/config-file-reference.md

@@ -1774,14 +1774,15 @@ bigtable:
     # CLI flag: -bigtable.grpc-max-send-msg-size
     [max_send_msg_size: <int> | default = 16777216]
 
-    # Deprecated: Use gzip compression when sending messages.
+    # Deprecated: Use gzip compression when sending messages.  If true,
+    # overrides grpc-compression flag.
     # 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 = ""]
+    [grpc_compression: <string> | default = ""]
 
     # Rate limit for gRPC client; 0 means disabled.
     # CLI flag: -bigtable.grpc-client-rate-limit
@@ -2270,14 +2271,15 @@ grpc_client_config:
   # CLI flag: -ingester.client.grpc-max-send-msg-size
   [max_send_msg_size: <int> | default = 16777216]
 
-  # Deprecated: Use gzip compression when sending messages.
+  # Deprecated: Use gzip compression when sending messages.  If true, overrides
+  # grpc-compression flag.
   # 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 = ""]
+  [grpc_compression: <string> | default = ""]
 
   # Rate limit for gRPC client; 0 means disabled.
   # CLI flag: -ingester.client.grpc-client-rate-limit
@@ -2348,14 +2350,15 @@ grpc_client_config:
   # CLI flag: -querier.frontend-client.grpc-max-send-msg-size
   [max_send_msg_size: <int> | default = 16777216]
 
-  # Deprecated: Use gzip compression when sending messages.
+  # Deprecated: Use gzip compression when sending messages.  If true, overrides
+  # grpc-compression flag.
   # 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 = ""]
+  [grpc_compression: <string> | default = ""]
 
   # Rate limit for gRPC client; 0 means disabled.
   # CLI flag: -querier.frontend-client.grpc-client-rate-limit

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

@@ -24,18 +24,12 @@ func newCompressor() *compressor {
 	c := &compressor{}
 	c.readersPool = sync.Pool{
 		New: func() interface{} {
-			return &reader{
-				pool:   &c.readersPool,
-				Reader: snappy.NewReader(nil),
-			}
+			return snappy.NewReader(nil)
 		},
 	}
 	c.writersPool = sync.Pool{
 		New: func() interface{} {
-			return &writeCloser{
-				pool:   &c.writersPool,
-				Writer: snappy.NewBufferedWriter(nil),
-			}
+			return snappy.NewBufferedWriter(nil)
 		},
 	}
 	return c
@@ -46,36 +40,44 @@ func (c *compressor) Name() string {
 }
 
 func (c *compressor) Compress(w io.Writer) (io.WriteCloser, error) {
-	wr := c.writersPool.Get().(*writeCloser)
+	wr := c.writersPool.Get().(*snappy.Writer)
 	wr.Reset(w)
-	return wr, nil
+	return writeCloser{wr, &c.writersPool}, nil
 }
 
 func (c *compressor) Decompress(r io.Reader) (io.Reader, error) {
-	dr := c.readersPool.Get().(*reader)
+	dr := c.readersPool.Get().(*snappy.Reader)
 	dr.Reset(r)
-	return dr, nil
+	return reader{dr, &c.readersPool}, nil
 }
 
 type writeCloser struct {
 	*snappy.Writer
 	pool *sync.Pool
 }
 
-func (w *writeCloser) Close() error {
-	defer w.pool.Put(w)
-	return w.Writer.Close()
+func (w writeCloser) Close() error {
+	defer func() {
+		w.Writer.Reset(nil)
+		w.pool.Put(w.Writer)
+	}()
+
+	if w.Writer != nil {
+		return w.Writer.Close()
+	}
+	return nil
 }
 
 type reader struct {
-	*snappy.Reader
-	pool *sync.Pool
+	reader *snappy.Reader
+	pool   *sync.Pool
 }
 
-func (r *reader) Read(p []byte) (n int, err error) {
-	n, err = r.Reader.Read(p)
+func (r reader) Read(p []byte) (n int, err error) {
+	n, err = r.reader.Read(p)
 	if err == io.EOF {
-		r.pool.Put(r)
+		r.reader.Reset(nil)
+		r.pool.Put(r.reader)
 	}
 	return n, err
 }

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

@@ -47,9 +47,9 @@ func TestSnappy(t *testing.T) {
 func BenchmarkSnappyCompress(b *testing.B) {
 	data := []byte(strings.Repeat("123456789", 1024))
 	c := newCompressor()
-	w, _ := c.Compress(ioutil.Discard)
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
+		w, _ := c.Compress(ioutil.Discard)
 		_, _ = w.Write(data)
 		_ = w.Close()
 	}

pkg/util/grpcclient/grpcclient.go

@@ -20,7 +20,7 @@ import (
 type Config struct {
 	MaxRecvMsgSize     int     `yaml:"max_recv_msg_size"`
 	MaxSendMsgSize     int     `yaml:"max_send_msg_size"`
-	UseGzipCompression bool    `yaml:"use_gzip_compression"`
+	UseGzipCompression bool    `yaml:"use_gzip_compression"` // TODO: Remove this deprecated option in v1.6.0.
 	GRPCCompression    string  `yaml:"grpc_compression"`
 	RateLimit          float64 `yaml:"rate_limit"`
 	RateLimitBurst     int     `yaml:"rate_limit_burst"`
@@ -68,7 +68,7 @@ func (cfg *Config) CallOptions() []grpc.CallOption {
 	opts = append(opts, grpc.MaxCallSendMsgSize(cfg.MaxSendMsgSize))
 	compression := cfg.GRPCCompression
 	if cfg.UseGzipCompression {
-		compression = "gzip"
+		compression = gzip.Name
 	}
 	if compression != "" {
 		opts = append(opts, grpc.UseCompressor(compression))