Support memcached max_item_size (#3929)

* support memcached max_item_size configuration

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

* add new metric

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

* dont cast to uint

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

Author: yeya24

CHANGELOG.md

@@ -102,6 +102,7 @@
 * [ENHANCEMENT] Alertmanager: load alertmanager configurations from object storage concurrently, and only load necessary configurations, speeding configuration synchronization process and executing fewer "GET object" operations to the storage when sharding is enabled. #3898
 * [ENHANCEMENT] Blocks storage: Ingester can now stream entire chunks instead of individual samples to the querier. At the moment this feature must be explicitly enabled either by using `-ingester.stream-chunks-when-using-blocks` flag or `ingester_stream_chunks_when_using_blocks` (boolean) field in runtime config file, but these configuration options are temporary and will be removed when feature is stable. #3889
 * [ENHANCEMENT] Alertmanager: New endpoint `/multitenant_alertmanager/delete_tenant_config` to delete configuration for tenant identified by `X-Scope-OrgID` header. This is an internal endpoint, available even if Alertmanager API is not enabled by using `-experimental.alertmanager.enable-api`. #3900
+* [ENHANCEMENT] MemCached: Add `max_item_size` support. #3929
 * [BUGFIX] Cortex: Fixed issue where fatal errors and various log messages where not logged. #3778
 * [BUGFIX] HA Tracker: don't track as error in the `cortex_kv_request_duration_seconds` metric a CAS operation intentionally aborted. #3745
 * [BUGFIX] Querier / ruler: do not log "error removing stale clients" if the ring is empty. #3761

docs/configuration/config-file-reference.md

@@ -4123,6 +4123,11 @@ The `memcached_client_config` configures the client used to connect to Memcached
 # CLI flag: -<prefix>.memcached.max-idle-conns
 [max_idle_conns: <int> | default = 16]
 
+# The maximum size of an item stored in memcached. Bigger items are not stored.
+# If set to 0, no maximum size is enforced.
+# CLI flag: -<prefix>.memcached.max-item-size
+[max_item_size: <int> | default = 0]
+
 # Period with which to poll DNS for memcache servers.
 # CLI flag: -<prefix>.memcached.update-interval
 [update_interval: <duration> | default = 1m]

pkg/chunk/cache/memcached_client.go

@@ -52,10 +52,13 @@ type memcachedClient struct {
 	cbTimeout  time.Duration
 	cbInterval time.Duration
 
+	maxItemSize int
+
 	quit chan struct{}
 	wait sync.WaitGroup
 
 	numServers prometheus.Gauge
+	skipped    prometheus.Counter
 
 	logger log.Logger
 }
@@ -67,6 +70,7 @@ type MemcachedClientConfig struct {
 	Addresses      string        `yaml:"addresses"` // EXPERIMENTAL.
 	Timeout        time.Duration `yaml:"timeout"`
 	MaxIdleConns   int           `yaml:"max_idle_conns"`
+	MaxItemSize    int           `yaml:"max_item_size"`
 	UpdateInterval time.Duration `yaml:"update_interval"`
 	ConsistentHash bool          `yaml:"consistent_hash"`
 	CBFailures     uint          `yaml:"circuit_breaker_consecutive_failures"`
@@ -86,6 +90,7 @@ func (cfg *MemcachedClientConfig) RegisterFlagsWithPrefix(prefix, description st
 	f.UintVar(&cfg.CBFailures, prefix+"memcached.circuit-breaker-consecutive-failures", 10, description+"Trip circuit-breaker after this number of consecutive dial failures (if zero then circuit-breaker is disabled).")
 	f.DurationVar(&cfg.CBTimeout, prefix+"memcached.circuit-breaker-timeout", 10*time.Second, description+"Duration circuit-breaker remains open after tripping (if zero then 60 seconds is used).")
 	f.DurationVar(&cfg.CBInterval, prefix+"memcached.circuit-breaker-interval", 10*time.Second, description+"Reset circuit-breaker counts after this long (if zero then never reset).")
+	f.IntVar(&cfg.MaxItemSize, prefix+"memcached.max-item-size", 0, description+"The maximum size of an item stored in memcached. Bigger items are not stored. If set to 0, no maximum size is enforced.")
 }
 
 // NewMemcachedClient creates a new MemcacheClient that gets its server list
@@ -107,25 +112,33 @@ func NewMemcachedClient(cfg MemcachedClientConfig, name string, r prometheus.Reg
 	}, r))
 
 	newClient := &memcachedClient{
-		name:       name,
-		Client:     client,
-		serverList: selector,
-		hostname:   cfg.Host,
-		service:    cfg.Service,
-		logger:     logger,
-		provider:   dns.NewProvider(logger, dnsProviderRegisterer, dns.GolangResolverType),
-		cbs:        make(map[string]*gobreaker.CircuitBreaker),
-		cbFailures: cfg.CBFailures,
-		cbInterval: cfg.CBInterval,
-		cbTimeout:  cfg.CBTimeout,
-		quit:       make(chan struct{}),
+		name:        name,
+		Client:      client,
+		serverList:  selector,
+		hostname:    cfg.Host,
+		service:     cfg.Service,
+		logger:      logger,
+		provider:    dns.NewProvider(logger, dnsProviderRegisterer, dns.GolangResolverType),
+		cbs:         make(map[string]*gobreaker.CircuitBreaker),
+		cbFailures:  cfg.CBFailures,
+		cbInterval:  cfg.CBInterval,
+		cbTimeout:   cfg.CBTimeout,
+		maxItemSize: cfg.MaxItemSize,
+		quit:        make(chan struct{}),
 
 		numServers: promauto.With(r).NewGauge(prometheus.GaugeOpts{
 			Namespace:   "cortex",
 			Name:        "memcache_client_servers",
 			Help:        "The number of memcache servers discovered.",
 			ConstLabels: prometheus.Labels{"name": name},
 		}),
+
+		skipped: promauto.With(r).NewCounter(prometheus.CounterOpts{
+			Namespace:   "cortex",
+			Name:        "memcache_client_set_skip_total",
+			Help:        "Total number of skipped set operations because of the value is larger than the max-item-size.",
+			ConstLabels: prometheus.Labels{"name": name},
+		}),
 	}
 	if cfg.CBFailures > 0 {
 		newClient.Client.DialTimeout = newClient.dialViaCircuitBreaker
@@ -183,6 +196,12 @@ func (c *memcachedClient) Stop() {
 }
 
 func (c *memcachedClient) Set(item *memcache.Item) error {
+	// Skip hitting memcached at all if the item is bigger than the max allowed size.
+	if c.maxItemSize > 0 && len(item.Value) > c.maxItemSize {
+		c.skipped.Inc()
+		return nil
+	}
+
 	err := c.Client.Set(item)
 	if err == nil {
 		return nil