Adding TLS to memcached client

Author: alanprot

go.mod

@@ -250,6 +250,8 @@ replace git.apache.org/thrift.git => github.com/apache/thrift v0.0.0-20180902110
 // Use fork of gocql that has gokit logs and Prometheus metrics.
 replace github.com/gocql/gocql => github.com/grafana/gocql v0.0.0-20200605141915-ba5dc39ece85
 
+replace github.com/bradfitz/gomemcache => github.com/aws-observability/gomemcache v0.0.0-20240306233126-1bbebfb3ba1c
+
 // Replace memberlist with Grafana's fork which includes some fixes that haven't been merged upstream yet
 replace github.com/hashicorp/memberlist => github.com/grafana/memberlist v0.3.1-0.20220714140823-09ffed8adbbe
 

go.sum

@@ -596,6 +596,8 @@ github.com/armon/go-radix v1.0.0/go.mod h1:ufUuZ+zHj4x4TnLV4JWEpy2hxWSpsRywHrMgI
 github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5/go.mod h1:wHh0iHkYZB8zMSxRWpUBQtwG5a7fFgvEO+odwuTv2gs=
 github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2 h1:DklsrG3dyBCFEj5IhUbnKptjxatkF07cF2ak3yi77so=
 github.com/asaskevich/govalidator v0.0.0-20230301143203-a9d515a09cc2/go.mod h1:WaHUgvxTVq04UNunO+XhnAqY/wQc+bxr74GqbsZ/Jqw=
+github.com/aws-observability/gomemcache v0.0.0-20240306233126-1bbebfb3ba1c h1:mynlyHS6sYNrE1tw3i253/M4YHJOlTKBUdHvozOluGQ=
+github.com/aws-observability/gomemcache v0.0.0-20240306233126-1bbebfb3ba1c/go.mod h1:tudJZmQ+rri4slsqtM2BP7WI7eYE034wnvhkiIpO58Q=
 github.com/aws/aws-sdk-go v1.27.0/go.mod h1:KmX6BPdI08NWTb3/sm4ZGu5ShLoqVDhKgpiN924inxo=
 github.com/aws/aws-sdk-go v1.38.35/go.mod h1:hcU610XS61/+aQV88ixoOzUoG7v3b31pl2zKMmprdro=
 github.com/aws/aws-sdk-go v1.50.32 h1:POt81DvegnpQKM4DMDLlHz1CO6OBnEoQ1gRhYFd7QRY=
@@ -649,8 +651,6 @@ github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM
 github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
 github.com/boombuler/barcode v1.0.0/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/boombuler/barcode v1.0.1/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
-github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874 h1:N7oVaKyGp8bttX0bfZGmcGkjz7DLQXhAn3DNd3T0ous=
-github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874/go.mod h1:r5xuitiExdLAJ09PR7vBVENGvp4ZuTBeWTGtxuX3K+c=
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=

pkg/chunk/cache/memcached_client.go

@@ -18,6 +18,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/sony/gobreaker"
 	"github.com/thanos-io/thanos/pkg/discovery/dns"
+	"github.com/thanos-io/thanos/pkg/tls"
 
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 )
@@ -65,17 +66,23 @@ type memcachedClient struct {
 
 // MemcachedClientConfig defines how a MemcachedClient should be constructed.
 type MemcachedClientConfig struct {
-	Host           string        `yaml:"host"`
-	Service        string        `yaml:"service"`
-	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"`
-	CBTimeout      time.Duration `yaml:"circuit_breaker_timeout"`  // reset error count after this long
-	CBInterval     time.Duration `yaml:"circuit_breaker_interval"` // remain closed for this long after CBFailures errors
+	Host                  string        `yaml:"host"`
+	Service               string        `yaml:"service"`
+	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"`
+	CBTimeout             time.Duration `yaml:"circuit_breaker_timeout"`  // reset error count after this long
+	CBInterval            time.Duration `yaml:"circuit_breaker_interval"` // remain closed for this long after CBFailures errors
+	TLSEnabled            bool          `yaml:"tls_enabled"`
+	TLSCertPath           string        `yaml:"tls_cert_path"`
+	TLSKeyPath            string        `yaml:"tls_key_path"`
+	TLSCAPath             string        `yaml:"tls_ca_path"`
+	TLSServerName         string        `yaml:"tls_ca_name"`
+	TLSInsecureSkipVerify bool          `yaml:"tls_insecure_skip_verify"`
 }
 
 // RegisterFlagsWithPrefix adds the flags required to config this to the given FlagSet
@@ -91,6 +98,12 @@ func (cfg *MemcachedClientConfig) RegisterFlagsWithPrefix(prefix, description st
 	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.")
+	f.BoolVar(&cfg.TLSEnabled, prefix+"tls-enabled", false, "Enable TLS in the memcached client. This flag needs to be enabled when any other TLS flag is set. If set to false, insecure connection to memcached server will be used.")
+	f.StringVar(&cfg.TLSCertPath, prefix+"tls-cert-path", "", "Path to the client certificate file, which will be used for authenticating with the server. Also requires the key path to be configured.")
+	f.StringVar(&cfg.TLSKeyPath, prefix+"tls-key-path", "", "Path to the key file for the client certificate. Also requires the client certificate to be configured.")
+	f.StringVar(&cfg.TLSCAPath, prefix+"tls-ca-path", "", "Path to the CA certificates file to validate server certificate against. If not set, the host's root CA certificates are used.")
+	f.StringVar(&cfg.TLSServerName, prefix+"tls-server-name", "", "Override the expected name on the server certificate.")
+	f.BoolVar(&cfg.TLSInsecureSkipVerify, prefix+"tls-insecure-skip-verify", false, "Skip validating server certificate.")
 }
 
 // NewMemcachedClient creates a new MemcacheClient that gets its server list
@@ -103,7 +116,14 @@ func NewMemcachedClient(cfg MemcachedClientConfig, name string, r prometheus.Reg
 		selector = &memcache.ServerList{}
 	}
 
-	client := memcache.NewFromSelector(selector)
+	var client *memcache.Client
+	if cfg.TLSEnabled {
+		tlsConfig, _ := tls.NewClientConfig(logger, cfg.TLSCertPath, cfg.TLSKeyPath, cfg.TLSCAPath, cfg.TLSServerName, cfg.TLSInsecureSkipVerify)
+		client = memcache.NewWithTLSFromSelector(tlsConfig, selector)
+	} else {
+		client = memcache.NewFromSelector(selector)
+	}
+
 	client.Timeout = cfg.Timeout
 	client.MaxIdleConns = cfg.MaxIdleConns
 
@@ -141,7 +161,7 @@ func NewMemcachedClient(cfg MemcachedClientConfig, name string, r prometheus.Reg
 		}),
 	}
 	if cfg.CBFailures > 0 {
-		newClient.Client.DialContext = newClient.dialViaCircuitBreaker
+		newClient.Client.DialTimeout = newClient.dialViaCircuitBreaker
 	}
 
 	if len(cfg.Addresses) > 0 {
@@ -163,7 +183,7 @@ func (c *memcachedClient) circuitBreakerStateChange(name string, from gobreaker.
 	level.Info(c.logger).Log("msg", "circuit-breaker state change", "name", name, "from-state", from, "to-state", to)
 }
 
-func (c *memcachedClient) dialViaCircuitBreaker(_ context.Context, network, address string) (net.Conn, error) {
+func (c *memcachedClient) dialViaCircuitBreaker(network, address string, timeout time.Duration) (net.Conn, error) {
 	c.Lock()
 	cb := c.cbs[address]
 	if cb == nil {
@@ -181,7 +201,7 @@ func (c *memcachedClient) dialViaCircuitBreaker(_ context.Context, network, addr
 	c.Unlock()
 
 	conn, err := cb.Execute(func() (interface{}, error) {
-		return net.DialTimeout(network, address, c.cbTimeout)
+		return net.DialTimeout(network, address, timeout)
 	})
 	if err != nil {
 		return nil, err

pkg/storage/tsdb/memcache_client_config.go

@@ -17,9 +17,16 @@ type MemcachedClientConfig struct {
 	MaxAsyncBufferSize     int                  `yaml:"max_async_buffer_size"`
 	MaxGetMultiConcurrency int                  `yaml:"max_get_multi_concurrency"`
 	MaxGetMultiBatchSize   int                  `yaml:"max_get_multi_batch_size"`
-	MaxItemSize            int                  `yaml:"max_item_size"`
-	AutoDiscovery          bool                 `yaml:"auto_discovery"`
 	SetAsyncCircuitBreaker CircuitBreakerConfig `yaml:"set_async_circuit_breaker_config"`
+
+	MaxItemSize           int    `yaml:"max_item_size"`
+	AutoDiscovery         bool   `yaml:"auto_discovery"`
+	TLSEnabled            bool   `yaml:"tls_enabled"`
+	TLSCertPath           string `yaml:"tls_cert_path"`
+	TLSKeyPath            string `yaml:"tls_key_path"`
+	TLSCAPath             string `yaml:"tls_ca_path"`
+	TLSServerName         string `yaml:"tls_ca_name"`
+	TLSInsecureSkipVerify bool   `yaml:"tls_insecure_skip_verify"`
 }
 
 func (cfg *MemcachedClientConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefix string) {
@@ -33,6 +40,12 @@ func (cfg *MemcachedClientConfig) RegisterFlagsWithPrefix(f *flag.FlagSet, prefi
 	f.IntVar(&cfg.MaxItemSize, prefix+"max-item-size", 1024*1024, "The maximum size of an item stored in memcached. Bigger items are not stored. If set to 0, no maximum size is enforced.")
 	f.BoolVar(&cfg.AutoDiscovery, prefix+"auto-discovery", false, "Use memcached auto-discovery mechanism provided by some cloud provider like GCP and AWS")
 	cfg.SetAsyncCircuitBreaker.RegisterFlagsWithPrefix(f, prefix+"set-async.")
+	f.BoolVar(&cfg.TLSEnabled, prefix+"tls-enabled", false, "Enable TLS in the memcached client. This flag needs to be enabled when any other TLS flag is set. If set to false, insecure connection to memcached server will be used.")
+	f.StringVar(&cfg.TLSCertPath, prefix+"tls-cert-path", "", "Path to the client certificate file, which will be used for authenticating with the server. Also requires the key path to be configured.")
+	f.StringVar(&cfg.TLSKeyPath, prefix+"tls-key-path", "", "Path to the key file for the client certificate. Also requires the client certificate to be configured.")
+	f.StringVar(&cfg.TLSCAPath, prefix+"tls-ca-path", "", "Path to the CA certificates file to validate server certificate against. If not set, the host's root CA certificates are used.")
+	f.StringVar(&cfg.TLSServerName, prefix+"tls-server-name", "", "Override the expected name on the server certificate.")
+	f.BoolVar(&cfg.TLSInsecureSkipVerify, prefix+"tls-insecure-skip-verify", false, "Skip validating server certificate.")
 }
 
 func (cfg *MemcachedClientConfig) GetAddresses() []string {
@@ -72,5 +85,11 @@ func (cfg MemcachedClientConfig) ToMemcachedClientConfig() cacheutil.MemcachedCl
 			ConsecutiveFailures: uint32(cfg.SetAsyncCircuitBreaker.ConsecutiveFailures),
 			FailurePercent:      cfg.SetAsyncCircuitBreaker.FailurePercent,
 		},
+		TlsEnabled:            cfg.TLSEnabled,
+		TLSCertPath:           cfg.TLSCertPath,
+		TLSKeyPath:            cfg.TLSKeyPath,
+		TLSCAPath:             cfg.TLSCAPath,
+		TLSServerName:         cfg.TLSServerName,
+		TLSInsecureSkipVerify: cfg.TLSInsecureSkipVerify,
 	}
 }