Add MinIdleConnectionsHeadroom support to memcached client config

CHANGELOG.md

@@ -101,6 +101,7 @@
   * `<prefix>_cache_operations_total{backend="[memcached|redis]",...}`
   * `<prefix>_cache_requests_total{backend="[memcached|redis]",...}`
 * [ENHANCEMENT] Lifecycler: Added `HealthyInstancesInZoneCount` method returning the number of healthy instances in the ring that are registered in lifecycler's zone, updated during the last heartbeat period. #266
+* [ENHANCEMENT] Memcached: add `MinIdleConnectionsHeadroomPercentage` support. It configures the minimum number of idle connections to keep open as a percentage of the number of recently used idle connections. If negative (default), idle connections are kept open indefinitely. #269
 * [BUGFIX] spanlogger: Support multiple tenant IDs. #59
 * [BUGFIX] Memberlist: fixed corrupted packets when sending compound messages with more than 255 messages or messages bigger than 64KB. #85
 * [BUGFIX] Ring: `ring_member_ownership_percent` and `ring_tokens_owned` metrics are not updated on scale down. #109

cache/memcache_config.go

@@ -14,19 +14,21 @@ var (
 )
 
 type MemcachedConfig struct {
-	Addresses              string        `yaml:"addresses"`
-	Timeout                time.Duration `yaml:"timeout"`
-	MaxIdleConnections     int           `yaml:"max_idle_connections" category:"advanced"`
-	MaxAsyncConcurrency    int           `yaml:"max_async_concurrency" category:"advanced"`
-	MaxAsyncBufferSize     int           `yaml:"max_async_buffer_size" category:"advanced"`
-	MaxGetMultiConcurrency int           `yaml:"max_get_multi_concurrency" category:"advanced"`
-	MaxGetMultiBatchSize   int           `yaml:"max_get_multi_batch_size" category:"advanced"`
-	MaxItemSize            int           `yaml:"max_item_size" category:"advanced"`
+	Addresses                            string        `yaml:"addresses"`
+	Timeout                              time.Duration `yaml:"timeout"`
+	MinIdleConnectionsHeadroomPercentage float64       `yaml:"min_idle_connections_headroom_percentage" category:"advanced"`
+	MaxIdleConnections                   int           `yaml:"max_idle_connections" category:"advanced"`
+	MaxAsyncConcurrency                  int           `yaml:"max_async_concurrency" category:"advanced"`
+	MaxAsyncBufferSize                   int           `yaml:"max_async_buffer_size" category:"advanced"`
+	MaxGetMultiConcurrency               int           `yaml:"max_get_multi_concurrency" category:"advanced"`
+	MaxGetMultiBatchSize                 int           `yaml:"max_get_multi_batch_size" category:"advanced"`
+	MaxItemSize                          int           `yaml:"max_item_size" category:"advanced"`
 }
 
 func (cfg *MemcachedConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix string) {
 	f.StringVar(&cfg.Addresses, prefix+"addresses", "", "Comma-separated list of memcached addresses. Each address can be an IP address, hostname, or an entry specified in the DNS Service Discovery format.")
 	f.DurationVar(&cfg.Timeout, prefix+"timeout", 200*time.Millisecond, "The socket read/write timeout.")
+	f.Float64Var(&cfg.MinIdleConnectionsHeadroomPercentage, prefix+"min-idle-connections-headroom-percentage", -1, "The minimum number of idle connections to keep open as a percentage (0-100) of the number of recently used idle connections. If negative, idle connections are kept open indefinitely.")
 	f.IntVar(&cfg.MaxIdleConnections, prefix+"max-idle-connections", 100, "The maximum number of idle connections that will be maintained per address.")
 	f.IntVar(&cfg.MaxAsyncConcurrency, prefix+"max-async-concurrency", 50, "The maximum number of concurrent asynchronous operations can occur.")
 	f.IntVar(&cfg.MaxAsyncBufferSize, prefix+"max-async-buffer-size", 25000, "The maximum number of enqueued asynchronous operations allowed.")
@@ -52,14 +54,15 @@ func (cfg *MemcachedConfig) Validate() error {
 
 func (cfg *MemcachedConfig) ToMemcachedClientConfig() MemcachedClientConfig {
 	return MemcachedClientConfig{
-		Addresses:                 cfg.GetAddresses(),
-		Timeout:                   cfg.Timeout,
-		MaxIdleConnections:        cfg.MaxIdleConnections,
-		MaxAsyncConcurrency:       cfg.MaxAsyncConcurrency,
-		MaxAsyncBufferSize:        cfg.MaxAsyncBufferSize,
-		MaxGetMultiConcurrency:    cfg.MaxGetMultiConcurrency,
-		MaxGetMultiBatchSize:      cfg.MaxGetMultiBatchSize,
-		MaxItemSize:               flagext.Bytes(cfg.MaxItemSize),
-		DNSProviderUpdateInterval: 30 * time.Second,
+		Addresses:                            cfg.GetAddresses(),
+		Timeout:                              cfg.Timeout,
+		MinIdleConnectionsHeadroomPercentage: cfg.MinIdleConnectionsHeadroomPercentage,
+		MaxIdleConnections:                   cfg.MaxIdleConnections,
+		MaxAsyncConcurrency:                  cfg.MaxAsyncConcurrency,
+		MaxAsyncBufferSize:                   cfg.MaxAsyncBufferSize,
+		MaxGetMultiConcurrency:               cfg.MaxGetMultiConcurrency,
+		MaxGetMultiBatchSize:                 cfg.MaxGetMultiBatchSize,
+		MaxItemSize:                          flagext.Bytes(cfg.MaxItemSize),
+		DNSProviderUpdateInterval:            30 * time.Second,
 	}
 }

cache/memcached_client.go

@@ -36,6 +36,7 @@ type memcachedClientBackend interface {
 	GetMulti(keys []string, opts ...memcache.Option) (map[string]*memcache.Item, error)
 	Set(item *memcache.Item) error
 	Delete(key string) error
+	Close()
 }
 
 // updatableServerSelector extends the interface used for picking a memcached server
@@ -62,6 +63,11 @@ type MemcachedClientConfig struct {
 	// Timeout specifies the socket read/write timeout.
 	Timeout time.Duration `yaml:"timeout"`
 
+	// MinIdleConnectionsHeadroomPercentage specifies the minimum number of idle connections
+	// to keep open as a percentage of the number of recently used idle connections.
+	// If negative, idle connections are kept open indefinitely.
+	MinIdleConnectionsHeadroomPercentage float64 `yaml:"min_idle_connections_headroom_percentage"`
+
 	// MaxIdleConnections specifies the maximum number of idle connections that
 	// will be maintained per address. For better performances, this should be
 	// set to a number higher than your peak parallel requests.
@@ -163,6 +169,7 @@ func NewMemcachedClientWithConfig(logger log.Logger, name string, config Memcach
 
 	client := memcache.NewFromSelector(selector)
 	client.Timeout = config.Timeout
+	client.MinIdleConnsHeadroomPercentage = config.MinIdleConnectionsHeadroomPercentage
 	client.MaxIdleConns = config.MaxIdleConnections
 
 	if reg != nil {
@@ -215,14 +222,15 @@ func newMemcachedClient(
 		Name: clientInfoMetricName,
 		Help: "A metric with a constant '1' value labeled by configuration options from which memcached client was configured.",
 		ConstLabels: prometheus.Labels{
-			"timeout":                      config.Timeout.String(),
-			"max_idle_connections":         strconv.Itoa(config.MaxIdleConnections),
-			"max_async_concurrency":        strconv.Itoa(config.MaxAsyncConcurrency),
-			"max_async_buffer_size":        strconv.Itoa(config.MaxAsyncBufferSize),
-			"max_item_size":                strconv.FormatUint(uint64(config.MaxItemSize), 10),
-			"max_get_multi_concurrency":    strconv.Itoa(config.MaxGetMultiConcurrency),
-			"max_get_multi_batch_size":     strconv.Itoa(config.MaxGetMultiBatchSize),
-			"dns_provider_update_interval": config.DNSProviderUpdateInterval.String(),
+			"timeout": config.Timeout.String(),
+			"min_idle_connections_headroom_percentage": fmt.Sprintf("%f.2", config.MinIdleConnectionsHeadroomPercentage),
+			"max_idle_connections":                     strconv.Itoa(config.MaxIdleConnections),
+			"max_async_concurrency":                    strconv.Itoa(config.MaxAsyncConcurrency),
+			"max_async_buffer_size":                    strconv.Itoa(config.MaxAsyncBufferSize),
+			"max_item_size":                            strconv.FormatUint(uint64(config.MaxItemSize), 10),
+			"max_get_multi_concurrency":                strconv.Itoa(config.MaxGetMultiConcurrency),
+			"max_get_multi_batch_size":                 strconv.Itoa(config.MaxGetMultiBatchSize),
+			"dns_provider_update_interval":             config.DNSProviderUpdateInterval.String(),
 		},
 	},
 		func() float64 { return 1 },
@@ -244,6 +252,9 @@ func (c *memcachedClient) Stop() {
 
 	// Stop running async operations.
 	c.asyncQueue.stop()
+
+	// Stop the underlying client.
+	c.client.Close()
 }
 
 func (c *memcachedClient) SetAsync(ctx context.Context, key string, value []byte, ttl time.Duration) error {

cache/memcached_client_test.go

@@ -98,6 +98,8 @@ func (m *mockMemcachedClientBackend) Delete(key string) error {
 	return nil
 }
 
+func (m *mockMemcachedClientBackend) Close() {}
+
 type mockServerSelector struct{}
 
 func (m mockServerSelector) SetServers(_ ...string) error {

go.mod

@@ -14,7 +14,7 @@ require (
 	github.com/gogo/protobuf v1.3.2
 	github.com/gogo/status v1.1.0
 	github.com/golang/snappy v0.0.4
-	github.com/grafana/gomemcache v0.0.0-20230105173749-11f792309e1f
+	github.com/grafana/gomemcache v0.0.0-20230221082510-6cde04bf2270
 	github.com/hashicorp/consul/api v1.15.3
 	github.com/hashicorp/go-cleanhttp v0.5.2
 	github.com/hashicorp/go-sockaddr v1.0.2

go.sum

@@ -262,8 +262,8 @@ github.com/googleapis/go-type-adapters v1.0.0/go.mod h1:zHW75FOG2aur7gAO2B+MLby+
 github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2zaAs=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
-github.com/grafana/gomemcache v0.0.0-20230105173749-11f792309e1f h1:ANwIMe7kOiMNTK88tusoNDb840pWVskI4rCrdoMv5i0=
-github.com/grafana/gomemcache v0.0.0-20230105173749-11f792309e1f/go.mod h1:PGk3RjYHpxMM8HFPhKKo+vve3DdlPUELZLSDEFehPuU=
+github.com/grafana/gomemcache v0.0.0-20230221082510-6cde04bf2270 h1:cj3uiNKskh+/7QQOr3Lzdf40hmtpdlCRlYVdsV0xBWA=
+github.com/grafana/gomemcache v0.0.0-20230221082510-6cde04bf2270/go.mod h1:PGk3RjYHpxMM8HFPhKKo+vve3DdlPUELZLSDEFehPuU=
 github.com/grafana/memberlist v0.3.1-0.20220708130638-bd88e10a3d91 h1:/NipyHnOmvRsVzj81j2qE0VxsvsqhOB0f4vJIhk2qCQ=
 github.com/grafana/memberlist v0.3.1-0.20220708130638-bd88e10a3d91/go.mod h1:MS2lj3INKhZjWNqd3N0m3J+Jxf3DAOnAH9VT3Sh9MUE=
 github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0/go.mod h1:8NvIoxWQoOIhqOTXgfV/d3M/q6VIi02HzZEHgUlZvzk=