feat: support configure refinery to use redis in cluster mode

config/config.go

@@ -62,35 +62,7 @@ type Config interface {
 
 	GetPeerManagementType() string
 
-	// GetRedisHost returns the address of a Redis instance to use for peer
-	// management.
-	GetRedisHost() string
-
-	// GetRedisUsername returns the username of a Redis instance to use for peer
-	// management.
-	GetRedisUsername() string
-
-	// GetRedisPassword returns the password of a Redis instance to use for peer
-	// management.
-	GetRedisPassword() string
-
-	// GetRedisAuthCode returns the AUTH string to use for connecting to a Redis
-	// instance to use for peer management
-	GetRedisAuthCode() string
-
-	// GetRedisPrefix returns the prefix string used in the keys for peer
-	// management.
-	GetRedisPrefix() string
-
-	// GetRedisDatabase returns the ID of the Redis database to use for peer management.
-	GetRedisDatabase() int
-
-	// GetUseTLS returns true when TLS must be enabled to dial the Redis instance to
-	// use for peer management.
-	GetUseTLS() bool
-
-	// UseTLSInsecure returns true when certificate checks are disabled
-	GetUseTLSInsecure() bool
+	GetRedisPeerManagement() RedisPeerManagementConfig
 
 	// GetHoneycombAPI returns the base URL (protocol, hostname, and port) of
 	// the upstream Honeycomb API server

config/config_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/honeycombio/refinery/internal/configwatcher"
 	"github.com/honeycombio/refinery/pubsub"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 	"gopkg.in/yaml.v3"
 )
 
@@ -90,7 +91,7 @@ func TestRedisHostEnvVar(t *testing.T) {
 	c, err := getConfig([]string{"--no-validate", "--config", "../config.yaml", "--rules_config", "../rules.yaml"})
 	assert.NoError(t, err)
 
-	if d := c.GetRedisHost(); d != host {
+	if d := c.GetRedisPeerManagement().Host; d != host {
 		t.Error("received", d, "expected", host)
 	}
 }
@@ -103,7 +104,7 @@ func TestRedisUsernameEnvVar(t *testing.T) {
 	c, err := getConfig([]string{"--no-validate", "--config", "../config.yaml", "--rules_config", "../rules.yaml"})
 	assert.NoError(t, err)
 
-	if d := c.GetRedisUsername(); d != username {
+	if d := c.GetRedisPeerManagement().Username; d != username {
 		t.Error("received", d, "expected", username)
 	}
 }
@@ -116,7 +117,7 @@ func TestRedisPasswordEnvVar(t *testing.T) {
 	c, err := getConfig([]string{"--no-validate", "--config", "../config.yaml", "--rules_config", "../rules.yaml"})
 	assert.NoError(t, err)
 
-	if d := c.GetRedisPassword(); d != password {
+	if d := c.GetRedisPeerManagement().Password; d != password {
 		t.Error("received", d, "expected", password)
 	}
 }
@@ -129,7 +130,7 @@ func TestRedisAuthCodeEnvVar(t *testing.T) {
 	c, err := getConfig([]string{"--no-validate", "--config", "../config.yaml", "--rules_config", "../rules.yaml"})
 	assert.NoError(t, err)
 
-	if d := c.GetRedisAuthCode(); d != authCode {
+	if d := c.GetRedisPeerManagement().AuthCode; d != authCode {
 		t.Error("received", d, "expected", authCode)
 	}
 }
@@ -403,11 +404,11 @@ func TestPeerManagementType(t *testing.T) {
 		t.Error("received", d, "expected", "redis")
 	}
 
-	if s := c.GetRedisPrefix(); s != "testPrefix" {
+	if s := c.GetRedisPeerManagement().Prefix; s != "testPrefix" {
 		t.Error("received", s, "expected", "testPrefix")
 	}
 
-	if db := c.GetRedisDatabase(); db != 9 {
+	if db := c.GetRedisPeerManagement().Database; db != 9 {
 		t.Error("received", db, "expected", 9)
 	}
 }
@@ -454,6 +455,27 @@ func TestDryRun(t *testing.T) {
 	}
 }
 
+func TestRedisClusterHosts(t *testing.T) {
+	clusterHosts := []string{"localhost:7001", "localhost:7002"}
+	cm := makeYAML(
+		"General.ConfigurationVersion", 2,
+		"PeerManagement.Type", "redis",
+		"RedisPeerManagement.ClusterHosts", clusterHosts,
+		"RedisPeerManagement.Prefix", "test",
+		"RedisPeerManagement.Database", 9,
+	)
+	rm := makeYAML("ConfigVersion", 2)
+	config, rules := createTempConfigs(t, cm, rm)
+	defer os.Remove(rules)
+	defer os.Remove(config)
+	c, err := getConfig([]string{"--no-validate", "--config", config, "--rules_config", rules})
+	assert.NoError(t, err)
+
+	d := c.GetRedisPeerManagement().ClusterHosts
+	require.NotNil(t, d)
+	require.EqualValues(t, clusterHosts, d)
+}
+
 func TestMaxAlloc(t *testing.T) {
 	cm := makeYAML("General.ConfigurationVersion", 2, "Collection.CacheCapacity", 1000, "Collection.MaxAlloc", 17179869184)
 	rm := makeYAML("ConfigVersion", 2)

config/file_config.go

@@ -285,6 +285,7 @@ type PeerManagementConfig struct {
 
 type RedisPeerManagementConfig struct {
 	Host           string   `yaml:"Host" cmdenv:"RedisHost"`
+	ClusterHosts   []string `yaml:"ClusterHosts"`
 	Username       string   `yaml:"Username" cmdenv:"RedisUsername"`
 	Password       string   `yaml:"Password" cmdenv:"RedisPassword"`
 	AuthCode       string   `yaml:"AuthCode" cmdenv:"RedisAuthCode"`
@@ -646,13 +647,27 @@ func (f *fileConfig) GetPeers() []string {
 	return f.mainConfig.PeerManagement.Peers
 }
 
+func (f *fileConfig) GetRedisPeerManagement() RedisPeerManagementConfig {
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+
+	return f.mainConfig.RedisPeerManagement
+}
+
 func (f *fileConfig) GetRedisHost() string {
 	f.mux.RLock()
 	defer f.mux.RUnlock()
 
 	return f.mainConfig.RedisPeerManagement.Host
 }
 
+func (f *fileConfig) GetRedisClusterHosts() []string {
+	f.mux.RLock()
+	defer f.mux.RUnlock()
+
+	return f.mainConfig.RedisPeerManagement.ClusterHosts
+}
+
 func (f *fileConfig) GetRedisUsername() string {
 	f.mux.RLock()
 	defer f.mux.RUnlock()

config/metadata/configMeta.yaml

@@ -987,6 +987,21 @@ groups:
         description: >
           Must be in the form `host:port`.
 
+      - name: ClusterHosts
+        type: stringarray
+        valuetype: stringarray
+        example: "- localhost:6379"
+        firstversion: v2.8
+        validations:
+          - type: elementType
+            arg: hostport
+        reload: false
+        summary: is a list of host and port pairs for the instances in a Redis Cluster, used for managing peer cluster membership.
+        description: >
+          This configuration enables Refinery to connect to a Redis deployment setup in Cluster Mode.
+          Each entry in the list should follow the format `host:port`.
+          If `ClusterHosts` is specified, the `Host` setting will be ignored.
+
       - name: Username
         v1group: PeerManagement
         v1name: RedisUsername
@@ -1253,6 +1268,7 @@ groups:
       - name: ShutdownDelay
         type: duration
         valuetype: nondefault
+        firstversion: v2.8
         default: 15s
         reload: true
         summary: controls the maximum time Refinery can use while draining traces at shutdown.

config/mock.go

@@ -26,14 +26,7 @@ type MockConfig struct {
 	GetStdoutLoggerConfigVal         StdoutLoggerConfig
 	GetLoggerLevelVal                Level
 	GetPeersVal                      []string
-	GetRedisHostVal                  string
-	GetRedisUsernameVal              string
-	GetRedisPasswordVal              string
-	GetRedisAuthCodeVal              string
-	GetRedisDatabaseVal              int
-	GetRedisPrefixVal                string
-	GetUseTLSVal                     bool
-	GetUseTLSInsecureVal             bool
+	GetRedisPeerManagementVal        RedisPeerManagementConfig
 	GetSamplerTypeName               string
 	GetSamplerTypeVal                interface{}
 	GetMetricsTypeVal                string
@@ -209,60 +202,11 @@ func (m *MockConfig) GetPeers() []string {
 	return m.GetPeersVal
 }
 
-func (m *MockConfig) GetRedisHost() string {
+func (m *MockConfig) GetRedisPeerManagement() RedisPeerManagementConfig {
 	m.Mux.RLock()
 	defer m.Mux.RUnlock()
 
-	return m.GetRedisHostVal
-}
-
-func (m *MockConfig) GetRedisUsername() string {
-	m.Mux.RLock()
-	defer m.Mux.RUnlock()
-
-	return m.GetRedisUsernameVal
-}
-
-func (m *MockConfig) GetRedisPassword() string {
-	m.Mux.RLock()
-	defer m.Mux.RUnlock()
-
-	return m.GetRedisPasswordVal
-}
-
-func (m *MockConfig) GetRedisAuthCode() string {
-	m.Mux.RLock()
-	defer m.Mux.RUnlock()
-
-	return m.GetRedisAuthCodeVal
-}
-
-func (m *MockConfig) GetRedisPrefix() string {
-	m.Mux.RLock()
-	defer m.Mux.RUnlock()
-
-	return m.GetRedisPrefixVal
-}
-
-func (m *MockConfig) GetRedisDatabase() int {
-	m.Mux.RLock()
-	defer m.Mux.RUnlock()
-
-	return m.GetRedisDatabaseVal
-}
-
-func (m *MockConfig) GetUseTLS() bool {
-	m.Mux.RLock()
-	defer m.Mux.RUnlock()
-
-	return m.GetUseTLSVal
-}
-
-func (m *MockConfig) GetUseTLSInsecure() bool {
-	m.Mux.RLock()
-	defer m.Mux.RUnlock()
-
-	return m.GetUseTLSInsecureVal
+	return m.GetRedisPeerManagementVal
 }
 
 func (m *MockConfig) GetGeneralConfig() GeneralConfig {

pubsub/pubsub_goredis.go

@@ -3,7 +3,6 @@ package pubsub
 import (
 	"context"
 	"crypto/tls"
-	"strings"
 	"sync"
 
 	"go.opentelemetry.io/otel/trace"
@@ -50,36 +49,42 @@ type GoRedisSubscription struct {
 var _ Subscription = (*GoRedisSubscription)(nil)
 
 func (ps *GoRedisPubSub) Start() error {
-	options := &redis.UniversalOptions{}
-	authcode := ""
-
-	ps.Metrics.Register("redis_pubsub_published", "counter")
-	ps.Metrics.Register("redis_pubsub_received", "counter")
+	options := new(redis.UniversalOptions)
+	var (
+		authcode           string
+		clusterModeEnabled bool
+	)
 
 	if ps.Config != nil {
-		host := ps.Config.GetRedisHost()
-		username := ps.Config.GetRedisUsername()
-		pw := ps.Config.GetRedisPassword()
-		authcode = ps.Config.GetRedisAuthCode()
-
-		// we may have multiple hosts, separated by commas, so split them up and
-		// use them as the addrs for the client (if there are multiples, it will
-		// create a cluster client)
-		hosts := strings.Split(host, ",")
+		redisCfg := ps.Config.GetRedisPeerManagement()
+		hosts := []string{redisCfg.Host}
+		// if we have a cluster host, use that instead of the regular host
+		if len(redisCfg.ClusterHosts) > 0 {
+			hosts = redisCfg.ClusterHosts
+			clusterModeEnabled = true
+		}
+
+		authcode = redisCfg.AuthCode
+
 		options.Addrs = hosts
-		options.Username = username
-		options.Password = pw
-		options.DB = ps.Config.GetRedisDatabase()
+		options.Username = redisCfg.Username
+		options.Password = redisCfg.Password
+		options.DB = redisCfg.Database
 
-		if ps.Config.GetUseTLS() {
+		if redisCfg.UseTLS {
 			options.TLSConfig = &tls.Config{
 				MinVersion:         tls.VersionTLS12,
-				InsecureSkipVerify: ps.Config.GetUseTLSInsecure(),
+				InsecureSkipVerify: redisCfg.UseTLSInsecure,
 			}
 		}
 	}
 
-	client := redis.NewUniversalClient(options)
+	var client redis.UniversalClient
+	if clusterModeEnabled {
+		client = redis.NewClusterClient(options.Cluster())
+	} else {
+		client = redis.NewUniversalClient(options)
+	}
 
 	// if an authcode was provided, use it to authenticate the connection
 	if authcode != "" {
@@ -90,6 +95,9 @@ func (ps *GoRedisPubSub) Start() error {
 		}
 	}
 
+	ps.Metrics.Register("redis_pubsub_published", "counter")
+	ps.Metrics.Register("redis_pubsub_received", "counter")
+
 	ps.client = client
 	ps.subs = make([]*GoRedisSubscription, 0)
 	return nil