Replace custom Redis config struct with go-redis UniversalOptions (adds sentinel & cluster support) (#4306)

cmd/registry/config-cache.yml

@@ -22,11 +22,10 @@ http:
     headers:
         X-Content-Type-Options: [nosniff]
 redis:
-  addr: localhost:6379
+  addrs: [localhost:6379]
-  pool:
+  maxidleconns: 16
-    maxidle: 16
+  poolsize: 64
-    maxactive: 64
+  connmaxidletime: 300s
-    idletimeout: 300s
   dialtimeout: 10ms
   readtimeout: 10ms
   writetimeout: 10ms

configuration/configuration.go

@@ -8,6 +8,8 @@ import (
 	"reflect"
 	"strings"
 	"time"
+
+	"github.com/redis/go-redis/v9"
 )
 
 // Configuration is a versioned registry configuration, intended to be provided by a yaml file, and
@@ -259,44 +261,6 @@ type FileChecker struct {
 	Threshold int `yaml:"threshold,omitempty"`
 }
 
-// Redis configures the redis pool available to the registry webapp.
-type Redis struct {
-	// Addr specifies the redis instance available to the application.
-	Addr string `yaml:"addr,omitempty"`
-
-	// Usernames can be used as a finer-grained permission control since the introduction of the redis 6.0.
-	Username string `yaml:"username,omitempty"`
-
-	// Password string to use when making a connection.
-	Password string `yaml:"password,omitempty"`
-
-	// DB specifies the database to connect to on the redis instance.
-	DB int `yaml:"db,omitempty"`
-
-	// TLS configures settings for redis in-transit encryption
-	TLS struct {
-		Enabled bool `yaml:"enabled,omitempty"`
-	} `yaml:"tls,omitempty"`
-
-	DialTimeout  time.Duration `yaml:"dialtimeout,omitempty"`  // timeout for connect
-	ReadTimeout  time.Duration `yaml:"readtimeout,omitempty"`  // timeout for reads of data
-	WriteTimeout time.Duration `yaml:"writetimeout,omitempty"` // timeout for writes of data
-
-	// Pool configures the behavior of the redis connection pool.
-	Pool struct {
-		// MaxIdle sets the maximum number of idle connections.
-		MaxIdle int `yaml:"maxidle,omitempty"`
-
-		// MaxActive sets the maximum number of connections that should be
-		// opened before blocking a connection request.
-		MaxActive int `yaml:"maxactive,omitempty"`
-
-		// IdleTimeout sets the amount time to wait before closing
-		// inactive connections.
-		IdleTimeout time.Duration `yaml:"idletimeout,omitempty"`
-	} `yaml:"pool,omitempty"`
-}
-
 // HTTPChecker is a type of entry in the health section for checking HTTP URIs.
 type HTTPChecker struct {
 	// Timeout is the duration to wait before timing out the HTTP request
@@ -750,3 +714,172 @@ func Parse(rd io.Reader) (*Configuration, error) {
 
 	return config, nil
 }
+
+type RedisOptions = redis.UniversalOptions
+
+type RedisTLSOptions struct {
+	Certificate string   `yaml:"certificate,omitempty"`
+	Key         string   `yaml:"key,omitempty"`
+	ClientCAs   []string `yaml:"clientcas,omitempty"`
+}
+
+type Redis struct {
+	Options RedisOptions    `yaml:",inline"`
+	TLS     RedisTLSOptions `yaml:"tls,omitempty"`
+}
+
+func (c Redis) MarshalYAML() (interface{}, error) {
+	fields := make(map[string]interface{})
+
+	val := reflect.ValueOf(c.Options)
+	typ := val.Type()
+
+	for i := 0; i < val.NumField(); i++ {
+		field := typ.Field(i)
+		fieldValue := val.Field(i)
+
+		// ignore funcs fields in redis.UniversalOptions
+		if fieldValue.Kind() == reflect.Func {
+			continue
+		}
+
+		fields[strings.ToLower(field.Name)] = fieldValue.Interface()
+	}
+
+	// Add TLS fields if they're not empty
+	if c.TLS.Certificate != "" || c.TLS.Key != "" || len(c.TLS.ClientCAs) > 0 {
+		fields["tls"] = c.TLS
+	}
+
+	return fields, nil
+}
+
+func (c *Redis) UnmarshalYAML(unmarshal func(interface{}) error) error {
+	var fields map[string]interface{}
+	err := unmarshal(&fields)
+	if err != nil {
+		return err
+	}
+
+	val := reflect.ValueOf(&c.Options).Elem()
+	typ := val.Type()
+
+	for i := 0; i < typ.NumField(); i++ {
+		field := typ.Field(i)
+		fieldName := strings.ToLower(field.Name)
+
+		if value, ok := fields[fieldName]; ok {
+			fieldValue := val.Field(i)
+			if fieldValue.CanSet() {
+				switch field.Type {
+				case reflect.TypeOf(time.Duration(0)):
+					durationStr, ok := value.(string)
+					if !ok {
+						return fmt.Errorf("invalid duration value for field: %s", fieldName)
+					}
+					duration, err := time.ParseDuration(durationStr)
+					if err != nil {
+						return fmt.Errorf("failed to parse duration for field: %s, error: %v", fieldName, err)
+					}
+					fieldValue.Set(reflect.ValueOf(duration))
+				default:
+					if err := setFieldValue(fieldValue, value); err != nil {
+						return fmt.Errorf("failed to set value for field: %s, error: %v", fieldName, err)
+					}
+				}
+			}
+		}
+	}
+
+	// Handle TLS fields
+	if tlsData, ok := fields["tls"]; ok {
+		tlsMap, ok := tlsData.(map[interface{}]interface{})
+		if !ok {
+			return fmt.Errorf("invalid TLS data structure")
+		}
+
+		if cert, ok := tlsMap["certificate"]; ok {
+			var isString bool
+			c.TLS.Certificate, isString = cert.(string)
+			if !isString {
+				return fmt.Errorf("Redis TLS certificate must be a string")
+			}
+		}
+		if key, ok := tlsMap["key"]; ok {
+			var isString bool
+			c.TLS.Key, isString = key.(string)
+			if !isString {
+				return fmt.Errorf("Redis TLS (private) key must be a string")
+			}
+		}
+		if cas, ok := tlsMap["clientcas"]; ok {
+			caList, ok := cas.([]interface{})
+			if !ok {
+				return fmt.Errorf("invalid clientcas data structure")
+			}
+			for _, ca := range caList {
+				if caStr, ok := ca.(string); ok {
+					c.TLS.ClientCAs = append(c.TLS.ClientCAs, caStr)
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+func setFieldValue(field reflect.Value, value interface{}) error {
+	if value == nil {
+		return nil
+	}
+
+	switch field.Kind() {
+	case reflect.String:
+		stringValue, ok := value.(string)
+		if !ok {
+			return fmt.Errorf("failed to convert value to string")
+		}
+		field.SetString(stringValue)
+	case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
+		intValue, ok := value.(int)
+		if !ok {
+			return fmt.Errorf("failed to convert value to integer")
+		}
+		field.SetInt(int64(intValue))
+	case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
+		uintValue, ok := value.(uint)
+		if !ok {
+			return fmt.Errorf("failed to convert value to unsigned integer")
+		}
+		field.SetUint(uint64(uintValue))
+	case reflect.Float32, reflect.Float64:
+		floatValue, ok := value.(float64)
+		if !ok {
+			return fmt.Errorf("failed to convert value to float")
+		}
+		field.SetFloat(floatValue)
+	case reflect.Bool:
+		boolValue, ok := value.(bool)
+		if !ok {
+			return fmt.Errorf("failed to convert value to boolean")
+		}
+		field.SetBool(boolValue)
+	case reflect.Slice:
+		slice := reflect.MakeSlice(field.Type(), 0, 0)
+		valueSlice, ok := value.([]interface{})
+		if !ok {
+			return fmt.Errorf("failed to convert value to slice")
+		}
+		for _, item := range valueSlice {
+			sliceValue := reflect.New(field.Type().Elem()).Elem()
+			if err := setFieldValue(sliceValue, item); err != nil {
+				return err
+			}
+			slice = reflect.Append(slice, sliceValue)
+		}
+		field.Set(slice)
+	default:
+		return fmt.Errorf("unsupported field type: %v", field.Type())
+	}
+	return nil
+}

configuration/configuration_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/suite"
 	"gopkg.in/yaml.v2"
 )
@@ -134,22 +135,23 @@ var configStruct = Configuration{
 		},
 	},
 	Redis: Redis{
-		Addr:     "localhost:6379",
+		Options: redis.UniversalOptions{
-		Username: "alice",
+			Addrs:           []string{"localhost:6379"},
-		Password: "123456",
+			Username:        "alice",
-		DB:       1,
+			Password:        "123456",
-		Pool: struct {
+			DB:              1,
-			MaxIdle     int           `yaml:"maxidle,omitempty"`
+			MaxIdleConns:    16,
-			MaxActive   int           `yaml:"maxactive,omitempty"`
+			PoolSize:        64,
-			IdleTimeout time.Duration `yaml:"idletimeout,omitempty"`
+			ConnMaxIdleTime: time.Second * 300,
-		}{
+			DialTimeout:     time.Millisecond * 10,
-			MaxIdle:     16,
+			ReadTimeout:     time.Millisecond * 10,
-			MaxActive:   64,
+			WriteTimeout:    time.Millisecond * 10,
-			IdleTimeout: time.Second * 300,
+		},
+		TLS: RedisTLSOptions{
+			Certificate: "/foo/cert.crt",
+			Key:         "/foo/key.pem",
+			ClientCAs:   []string{"/path/to/ca.pem"},
 		},
-		DialTimeout:  time.Millisecond * 10,
-		ReadTimeout:  time.Millisecond * 10,
-		WriteTimeout: time.Millisecond * 10,
 	},
 	Validation: Validation{
 		Manifests: ValidationManifests{
@@ -197,19 +199,24 @@ notifications:
         actions:
            - pull
 http:
-  clientcas:
+  tls:
-    - /path/to/ca.pem
+    clientcas:
+      - /path/to/ca.pem
   headers:
     X-Content-Type-Options: [nosniff]
 redis:
-  addr: localhost:6379
+  tls:
+    certificate: /foo/cert.crt
+    key: /foo/key.pem
+    clientcas:
+      - /path/to/ca.pem
+  addrs: [localhost:6379]
   username: alice
-  password: 123456
+  password: "123456"
   db: 1
-  pool:
+  maxidleconns: 16
-    maxidle: 16
+  poolsize: 64
-    maxactive: 64
+  connmaxidletime: 300s
-    idletimeout: 300s
   dialtimeout: 10ms
   readtimeout: 10ms
   writetimeout: 10ms
@@ -289,6 +296,7 @@ func (suite *ConfigSuite) TestParseSimple() {
 func (suite *ConfigSuite) TestParseInmemory() {
 	suite.expectedConfig.Storage = Storage{"inmemory": Parameters{}}
 	suite.expectedConfig.Log.Fields = nil
+	suite.expectedConfig.HTTP.TLS.ClientCAs = nil
 	suite.expectedConfig.Redis = Redis{}
 
 	config, err := Parse(bytes.NewReader([]byte(inmemoryConfigYamlV0_1)))
@@ -309,6 +317,7 @@ func (suite *ConfigSuite) TestParseIncomplete() {
 	suite.expectedConfig.Auth = Auth{"silly": Parameters{"realm": "silly"}}
 	suite.expectedConfig.Notifications = Notifications{}
 	suite.expectedConfig.HTTP.Headers = nil
+	suite.expectedConfig.HTTP.TLS.ClientCAs = nil
 	suite.expectedConfig.Redis = Redis{}
 	suite.expectedConfig.Validation.Manifests.Indexes.Platforms = ""
 
@@ -579,8 +588,14 @@ func copyConfig(config Configuration) *Configuration {
 	for k, v := range config.HTTP.Headers {
 		configCopy.HTTP.Headers[k] = v
 	}
+	configCopy.HTTP.TLS.ClientCAs = make([]string, 0, len(config.HTTP.TLS.ClientCAs))
+	configCopy.HTTP.TLS.ClientCAs = append(configCopy.HTTP.TLS.ClientCAs, config.HTTP.TLS.ClientCAs...)
 
 	configCopy.Redis = config.Redis
+	configCopy.Redis.TLS.Certificate = config.Redis.TLS.Certificate
+	configCopy.Redis.TLS.Key = config.Redis.TLS.Key
+	configCopy.Redis.TLS.ClientCAs = make([]string, 0, len(config.Redis.TLS.ClientCAs))
+	configCopy.Redis.TLS.ClientCAs = append(configCopy.Redis.TLS.ClientCAs, config.Redis.TLS.ClientCAs...)
 
 	configCopy.Validation = Validation{
 		Enabled:   config.Validation.Enabled,

docs/content/about/configuration.md

@@ -246,16 +246,20 @@ notifications:
         actions:
            - pull
 redis:
-  addr: localhost:6379
+  tls:
+    certificate: /path/to/cert.crt
+    key: /path/to/key.pem
+    clientcas:
+      - /path/to/ca.pem
+  addrs: [localhost:6379]
   password: asecret
   db: 0
   dialtimeout: 10ms
   readtimeout: 10ms
   writetimeout: 10ms
-  pool:
+  maxidleconns: 16
-    maxidle: 16
+  poolsize: 64
-    maxactive: 64
+  connmaxidletime: 300s
-    idletimeout: 300s
   tls:
     enabled: false
 health:
@@ -1017,72 +1021,46 @@ The `events` structure configures the information provided in event notification
 
 ## `redis`
 
+Declare parameters for constructing the `redis` connections. Registry instances
+may use the Redis instance for several applications. Currently, it caches
+information about immutable blobs. Most of the `redis` options control
+how the registry connects to the `redis` instance.
+
+You should configure Redis with the **allkeys-lru** eviction policy, because the
+registry does not set an expiration value on keys.
+
+Under the hood distribution uses [`go-redis`](https://github.com/redis/go-redis) Go module for
+Redis connectivity and its [`UniversalOptions`](https://pkg.go.dev/github.com/redis/go-redis/v9#UniversalOptions)
+struct.
+
+You can optionally specify TLS configuration on top of the `UniversalOptions` settings.
+
+Use these settings to configure Redis TLS:
+
+| Parameter | Required | Description                                           |
+|-----------|----------|-------------------------------------------------------|
+| `certificate`  | yes  | Absolute path to the x509 certificate file.           |
+| `key`          | yes  | Absolute path to the x509 private key file.           |
+| `clientcas`    | no   | An array of absolute paths to x509 CA files.          |
+
 ```yaml
 redis:
-  addr: localhost:6379
+  tls:
+    certificate: /path/to/cert.crt
+    key: /path/to/key.pem
+    clientcas:
+      - /path/to/ca.pem
+  addrs: [localhost:6379]
   password: asecret
   db: 0
   dialtimeout: 10ms
   readtimeout: 10ms
   writetimeout: 10ms
-  pool:
+  maxidleconns: 16
-    maxidle: 16
+  poolsize: 64
-    maxactive: 64
+  connmaxidletime: 300s
-    idletimeout: 300s
-  tls:
-    enabled: false
 ```
 
-Declare parameters for constructing the `redis` connections. Registry instances
-may use the Redis instance for several applications. Currently, it caches
-information about immutable blobs. Most of the `redis` options control
-how the registry connects to the `redis` instance. You can control the pool's
-behavior with the [pool](#pool) subsection. Additionally, you can control
-TLS connection settings with the [tls](#tls) subsection (in-transit encryption).
-
-You should configure Redis with the **allkeys-lru** eviction policy, because the
-registry does not set an expiration value on keys.
-
-| Parameter | Required | Description                                           |
-|-----------|----------|-------------------------------------------------------|
-| `addr`    | yes      | The address (host and port) of the Redis instance.    |
-| `password`| no       | A password used to authenticate to the Redis instance.|
-| `db`      | no       | The name of the database to use for each connection.  |
-| `dialtimeout` | no   | The timeout for connecting to the Redis instance.     |
-| `readtimeout` | no   | The timeout for reading from the Redis instance.      |
-| `writetimeout` | no  | The timeout for writing to the Redis instance.        |
-
-### `pool`
-
-```yaml
-pool:
-  maxidle: 16
-  maxactive: 64
-  idletimeout: 300s
-```
-
-Use these settings to configure the behavior of the Redis connection pool.
-
-| Parameter | Required | Description                                           |
-|-----------|----------|-------------------------------------------------------|
-| `maxidle` | no       | The maximum number of idle connections in the pool.   |
-| `maxactive`| no      | The maximum number of connections which can be open before blocking a connection request. |
-| `idletimeout`| no    | How long to wait before closing inactive connections. |
-
-### `tls`
-
-```yaml
-tls:
-  enabled: false
-```
-
-Use these settings to configure Redis TLS.
-
-| Parameter | Required | Description                           |
-|-----------|----------|-------------------------------------- |
-| `enabled` | no       | Whether or not to use TLS in-transit. |
-
-
 ## `health`
 
 ```yaml

registry/handlers/app.go

@@ -3,6 +3,8 @@ package handlers
 import (
 	"context"
 	"crypto/rand"
+	"crypto/tls"
+	"crypto/x509"
 	"expvar"
 	"fmt"
 	"math"
@@ -77,7 +79,7 @@ type App struct {
 		source notifications.SourceRecord
 	}
 
-	redis *redis.Client
+	redis redis.UniversalClient
 
 	// isCache is true if this registry is configured as a pull through cache
 	isCache bool
@@ -529,12 +531,41 @@ func (app *App) configureEvents(configuration *configuration.Configuration) {
 }
 
 func (app *App) configureRedis(cfg *configuration.Configuration) {
-	if cfg.Redis.Addr == "" {
+	if len(cfg.Redis.Options.Addrs) == 0 {
 		dcontext.GetLogger(app).Infof("redis not configured")
 		return
 	}
 
-	app.redis = app.createPool(cfg.Redis)
+	// redis TLS config
+	if cfg.Redis.TLS.Certificate != "" || cfg.Redis.TLS.Key != "" {
+		var err error
+		tlsConf := &tls.Config{}
+		tlsConf.Certificates = make([]tls.Certificate, 1)
+		tlsConf.Certificates[0], err = tls.LoadX509KeyPair(cfg.Redis.TLS.Certificate, cfg.Redis.TLS.Key)
+		if err != nil {
+			panic(err)
+		}
+		if len(cfg.Redis.TLS.ClientCAs) != 0 {
+			pool := x509.NewCertPool()
+			for _, ca := range cfg.Redis.TLS.ClientCAs {
+				caPem, err := os.ReadFile(ca)
+				if err != nil {
+					dcontext.GetLogger(app).Errorf("failed reading redis client CA: %v", err)
+					return
+				}
+
+				if ok := pool.AppendCertsFromPEM(caPem); !ok {
+					dcontext.GetLogger(app).Error("could not add CA to pool")
+					return
+				}
+			}
+			tlsConf.ClientAuth = tls.RequireAndVerifyClientCert
+			tlsConf.ClientCAs = pool
+		}
+		cfg.Redis.Options.TLSConfig = tlsConf
+	}
+
+	app.redis = app.createPool(cfg.Redis.Options)
 
 	// Enable metrics instrumentation.
 	if err := redisotel.InstrumentMetrics(app.redis); err != nil {
@@ -556,25 +587,12 @@ func (app *App) configureRedis(cfg *configuration.Configuration) {
 	}))
 }
 
-func (app *App) createPool(cfg configuration.Redis) *redis.Client {
+func (app *App) createPool(cfg redis.UniversalOptions) redis.UniversalClient {
-	return redis.NewClient(&redis.Options{
+	cfg.OnConnect = func(ctx context.Context, cn *redis.Conn) error {
-		Addr: cfg.Addr,
+		res := cn.Ping(ctx)
-		OnConnect: func(ctx context.Context, cn *redis.Conn) error {
+		return res.Err()
-			res := cn.Ping(ctx)
+	}
-			return res.Err()
+	return redis.NewUniversalClient(&cfg)
-		},
-		Username:        cfg.Username,
-		Password:        cfg.Password,
-		DB:              cfg.DB,
-		MaxRetries:      3,
-		DialTimeout:     cfg.DialTimeout,
-		ReadTimeout:     cfg.ReadTimeout,
-		WriteTimeout:    cfg.WriteTimeout,
-		PoolFIFO:        false,
-		MaxIdleConns:    cfg.Pool.MaxIdle,
-		PoolSize:        cfg.Pool.MaxActive,
-		ConnMaxIdleTime: cfg.Pool.IdleTimeout,
-	})
 }
 
 // configureLogHook prepares logging hook parameters.

registry/storage/cache/redis/redis.go

@@ -25,7 +25,7 @@ import (
 // Note that there is no implied relationship between these two caches. The
 // layer may exist in one, both or none and the code must be written this way.
 type redisBlobDescriptorService struct {
-	pool *redis.Client
+	pool redis.UniversalClient
 
 	// TODO(stevvooe): We use a pool because we don't have great control over
 	// the cache lifecycle to manage connections. A new connection if fetched
@@ -37,7 +37,7 @@ var _ distribution.BlobDescriptorService = &redisBlobDescriptorService{}
 
 // NewRedisBlobDescriptorCacheProvider returns a new redis-based
 // BlobDescriptorCacheProvider using the provided redis connection pool.
-func NewRedisBlobDescriptorCacheProvider(pool *redis.Client) cache.BlobDescriptorCacheProvider {
+func NewRedisBlobDescriptorCacheProvider(pool redis.UniversalClient) cache.BlobDescriptorCacheProvider {
 	return metrics.NewPrometheusCacheProvider(
 		&redisBlobDescriptorService{
 			pool: pool,

tests/conf-e2e-cloud-storage.yml

@@ -17,15 +17,14 @@ log:
     formatter: text
     level: debug
 redis:
-   addr: redis:6379
+   addrs: [redis:6379]
    db: 0
    dialtimeout: 5s
    readtimeout: 10ms
    writetimeout: 10ms
-   pool:
+   maxidleconns: 16
-     idletimeout: 60s
+   poolsize: 64
-     maxactive: 64
+   connmaxidletime: 300s
-     maxidle: 16
 storage:
   redirect:
     disable: true