max payload size for caching

cache/async_cache.go

@@ -20,6 +20,8 @@ type AsyncCache struct {
 	TransactionRegistry
 
 	graceTime time.Duration
+
+	MaxPayloadSize int64
 }
 
 func (c *AsyncCache) Close() error {
@@ -103,9 +105,12 @@ func NewAsyncCache(cfg config.Cache, maxExecutionTime time.Duration) (*AsyncCach
 		return nil, err
 	}
 
+	maxPayloadSize := int64(cfg.MaxPayloadSize)
+
 	return &AsyncCache{
 		Cache:               cache,
 		TransactionRegistry: transaction,
 		graceTime:           graceTime,
+		MaxPayloadSize:      maxPayloadSize,
 	}, nil
 }

cache/async_cache_test.go

@@ -213,7 +213,8 @@ func TestAsyncCache_FilesystemCache_instantiation(t *testing.T) {
 			Dir:     asyncTestDir,
 			MaxSize: 8192,
 		},
-		Expire: config.Duration(time.Minute),
+		Expire:         config.Duration(time.Minute),
+		MaxPayloadSize: config.ByteSize(100000000),
 	}
 	if err := os.RemoveAll(testDirAsync); err != nil {
 		log.Fatalf("cannot remove %q: %s", testDirAsync, err)
@@ -248,7 +249,8 @@ func TestAsyncCache_RedisCache_instantiation(t *testing.T) {
 		Redis: config.RedisCacheConfig{
 			Addresses: []string{s.Addr()},
 		},
-		Expire: config.Duration(cacheTTL),
+		Expire:         config.Duration(cacheTTL),
+		MaxPayloadSize: config.ByteSize(100000000),
 	}
 
 	_, err := NewAsyncCache(redisCfg, 1*time.Second)

config/config.go

@@ -32,6 +32,8 @@ var (
 	}
 
 	defaultExecutionTime = Duration(120 * time.Second)
+
+	defaultMaxPayloadSize = ByteSize(1 << 50)
 )
 
 // Config describes server configuration, access and proxy rules
@@ -601,6 +603,9 @@ type Cache struct {
 
 	// Catches all undefined fields
 	XXX map[string]interface{} `yaml:",inline"`
+
+	// Maximum total size of request payload for caching
+	MaxPayloadSize ByteSize `yaml:"max_payload_size,omitempty"`
 }
 
 type FileSystemCacheConfig struct {
@@ -812,6 +817,13 @@ func LoadFile(filename string) (*Config, error) {
 		}
 	}
 
+	for i := range cfg.Caches {
+		c := &cfg.Caches[i]
+		if c.MaxPayloadSize <= 0 {
+			c.MaxPayloadSize = defaultMaxPayloadSize
+		}
+	}
+
 	if maxResponseTime < 0 {
 		maxResponseTime = 0
 	}

config/config_test.go

@@ -21,8 +21,9 @@ var fullConfig = Config{
 				Dir:     "/path/to/longterm/cachedir",
 				MaxSize: ByteSize(100 << 30),
 			},
-			Expire:    Duration(time.Hour),
-			GraceTime: Duration(20 * time.Second),
+			Expire:         Duration(time.Hour),
+			GraceTime:      Duration(20 * time.Second),
+			MaxPayloadSize: ByteSize(100 << 30),
 		},
 		{
 			Name: "shortterm",
@@ -31,7 +32,8 @@ var fullConfig = Config{
 				Dir:     "/path/to/shortterm/cachedir",
 				MaxSize: ByteSize(100 << 20),
 			},
-			Expire: Duration(10 * time.Second),
+			Expire:         Duration(10 * time.Second),
+			MaxPayloadSize: ByteSize(100 << 20),
 		},
 	},
 	HackMePlease: true,
@@ -441,6 +443,16 @@ func TestBadConfig(t *testing.T) {
 			"testdata/bad.heartbeat_section.empty.yml",
 			"`cluster.heartbeat` cannot be unset for \"cluster\"",
 		},
+		{
+			"empty heartbeat section with heartbeat_interval",
+			"testdata/bad.heartbeat_section.empty2.yml",
+			"cannot be use `heartbeat_interval` with `heartbeat` section",
+		},
+		{
+			"max payload size to cache",
+			"testdata/bad.max_payload_size.yml",
+			"cannot parse byte size \"-10B\": it must be positive float followed by optional units. For example, 1.5Gb, 3T",
+		},
 	}
 
 	for _, tc := range testCases {
@@ -815,12 +827,14 @@ caches:
   file_system:
     dir: /path/to/longterm/cachedir
     max_size: 107374182400
+  max_payload_size: 107374182400
 - mode: file_system
   name: shortterm
   expire: 10s
   file_system:
     dir: /path/to/shortterm/cachedir
     max_size: 104857600
+  max_payload_size: 104857600
 param_groups:
 - name: cron-job
   params:

config/testdata/bad.max_payload_size.yml

@@ -0,0 +1,21 @@
+caches:
+  - name: "longterm"
+    mode: "file_system"
+    max_payload_size: "-10B"
+    file_system:
+      dir: "cache_dir"
+      max_size: 100Gb
+
+server:
+  http:
+    listen_addr: ":8080"
+
+users:
+  - name: "dummy"
+    allowed_networks: ["1.2.3.4"]
+    to_cluster: "cluster"
+    to_user: "default"
+
+clusters:
+  - name: "cluster"
+    nodes: ["127.0.1.1:8123"]

config/testdata/full.yml

@@ -26,6 +26,8 @@ caches:
       # Path to directory where cached responses will be stored.
       dir: "/path/to/longterm/cachedir"
 
+    max_payload_size: 100Gb
+
     # Expiration time for cached responses.
     expire: 1h
 
@@ -44,6 +46,7 @@ caches:
     file_system:
       max_size: 100Mb
       dir: "/path/to/shortterm/cachedir"
+    max_payload_size: 100Mb
     expire: 10s
 
 # Optional network lists, might be used as values for `allowed_networks`.

docs/content/en/configuration/caching.md

@@ -31,4 +31,7 @@ Distributed cache relies on external database to share cache across multiple rep
 multiple replicas deployments. Currently only [Redis](https://redis.io/) key value store is supported. 
 Configuration template for distributed cache can be found [here](https://github.com/ContentSquare/chproxy/blob/master/config/#distributed_cache_config).
 
-
+#### Response limitations for caching
+Before caching Clickhouse response, chproxy verifies that the response size 
+is not greater than configured max size. This setting can be specified in config section of the cache `max_payload_size`. The default value
+is set to 1 Petabyte. Therefore, by default this security mechanism is disabled.

metrics.go

@@ -115,6 +115,13 @@ var (
 		},
 		[]string{"cache"},
 	)
+	cacheSkipped = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Name: "cache_payloadsize_too_big_total",
+			Help: "The amount of too big payloads to be cached",
+		},
+		[]string{"cache"},
+	)
 	requestDuration = prometheus.NewSummaryVec(
 		prometheus.SummaryOpts{
 			Name:       "request_duration_seconds",
@@ -194,7 +201,7 @@ func registerMetrics() {
 		limitExcess, hostPenalties, hostHealth, concurrentQueries,
 		requestQueueSize, userQueueOverflow, clusterUserQueueOverflow,
 		requestBodyBytes, responseBodyBytes,
-		cacheHit, cacheMiss, cacheSize, cacheItems,
+		cacheHit, cacheMiss, cacheSize, cacheItems, cacheSkipped,
 		requestDuration, proxiedResponseDuration, cachedResponseDuration,
 		canceledRequest, timeoutRequest,
 		configSuccess, configSuccessTime, badRequest)

proxy.go

@@ -358,9 +358,14 @@ func (rp *reverseProxy) serveFromCache(s *scope, srw *statResponseWriter, req *h
 		if err != nil {
 			err = fmt.Errorf("%s: %w; query: %q", s, err, q)
 			respondWith(srw, err, http.StatusInternalServerError)
-			return
 		}
 	} else {
+		// Do not cache responses greater than max payload size.
+		if contentLength > s.user.cache.MaxPayloadSize {
+			cacheSkipped.With(labels).Inc()
+			log.Debugf("%s: Request will not be cached. Content length (%d) is greater than max payload size (%d)", s, contentLength, s.user.cache.MaxPayloadSize)
+			err = RespondWithData(srw, reader, contentMetadata, 0*time.Second, tmpFileRespWriter.StatusCode())
+		}
 		cacheMiss.With(labels).Inc()
 		log.Debugf("%s: cache miss", s)
 		expiration, err := userCache.Put(reader, contentMetadata, key)

utils.go

@@ -287,3 +287,8 @@ func calcMapHash(m map[string]string) (uint32, error) {
 	}
 	return h.Sum32(), nil
 }
+
+func isToCache(length int64, s *scope) bool {
+	maxPayloadSize := s.user.cache.MaxPayloadSize
+	return length <= maxPayloadSize
+}