feat(server): add support for Redis clusters

CONTRIBUTING.md

@@ -67,7 +67,10 @@ and `docker.io/blockossreleases/spectre`.
 `spectre serve` collects the quarantines that mirrors report. It listens on
 `127.0.0.1:50070` by default and keeps them in memory, so a restart loses them.
 To keep them in Redis instead, pass a Redis URL to `--storage`, such as
-`--storage=redis://localhost:6379/0`.
+`--storage=redis://localhost:6379/0`. For a Redis Cluster, use the
+`redis-cluster` scheme, or `rediss-cluster` for TLS, and list further seed
+nodes with `addr`, such as
+`--storage=redis-cluster://localhost:7000?addr=localhost:7001`.
 
 Every mirror reports to a server, so `spectre supervise` requires its URL,
 `--server-url`, and the mirror's identity, `--service`, `--environment`, and

internal/server/storage/redis.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"net/url"
 	"strconv"
+	"strings"
 
 	"github.com/alecthomas/errors"
 	"github.com/redis/go-redis/v9"
@@ -14,7 +15,7 @@ import (
 )
 
 func init() {
-	for _, scheme := range []string{"redis", "rediss"} {
+	for _, scheme := range []string{"redis", "rediss", "redis" + clusterSuffix, "rediss" + clusterSuffix} {
 		register(scheme, func(ctx context.Context, storageURL *url.URL) (Store, error) {
 			// Returning a nil *Redis directly would make a non-nil Store.
 			store, err := NewRedis(ctx, storageURL)
@@ -26,32 +27,59 @@ func init() {
 	}
 }
 
+// clusterSuffix marks a scheme that connects to a Redis Cluster. go-redis parses
+// cluster URLs with the plain schemes, so the suffix is the only way to tell.
+const clusterSuffix = "-cluster"
+
 // divergencesKey is the Redis hash that maps each sourceField to its divergence
 // as JSON.
 const divergencesKey = "spectre:divergences"
 
-// Redis keeps divergences in a Redis hash, so they outlive the SPECTRE server
-// and are shared by every server using the same Redis database.
+// Redis keeps divergences in one hash, shared by servers using the same database.
+// Commands touch only that key, so in a cluster one node's slot serves them all.
 type Redis struct {
-	client *redis.Client
+	client redis.UniversalClient
 }
 
 var _ Store = (*Redis)(nil)
 
-// NewRedis connects to the Redis server at a redis:// or rediss:// URL.
+// NewRedis connects to the Redis server at a redis:// or rediss:// URL, or to
+// the Redis Cluster at a redis-cluster:// or rediss-cluster:// URL.
 func NewRedis(ctx context.Context, storageURL *url.URL) (*Redis, error) {
-	options, err := redis.ParseURL(storageURL.String())
+	client, err := newRedisClient(storageURL)
 	if err != nil {
 		return nil, errors.Wrap(err, "redis URL")
 	}
-	client := redis.NewClient(options)
 	if err := client.Ping(ctx).Err(); err != nil {
 		_ = client.Close() //nolint:errcheck // The connection failed, so closing has nothing to report.
 		return nil, errors.Wrap(err, "connect to Redis")
 	}
 	return &Redis{client: client}, nil
 }
 
+func newRedisClient(storageURL *url.URL) (redis.UniversalClient, error) {
+	scheme, cluster := strings.CutSuffix(storageURL.Scheme, clusterSuffix)
+	goRedisURL := *storageURL
+	goRedisURL.Scheme = scheme
+	if !cluster {
+		options, err := redis.ParseURL(goRedisURL.String())
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+		return redis.NewClient(options), nil
+	}
+	// go-redis ignores the path of a cluster URL, which would otherwise silently
+	// store divergences in database 0 instead of the one the path names.
+	if strings.Trim(goRedisURL.Path, "/") != "" {
+		return nil, errors.Errorf("a Redis Cluster has no databases to select with path %q", goRedisURL.Path)
+	}
+	options, err := redis.ParseClusterURL(goRedisURL.String())
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return redis.NewClusterClient(options), nil
+}
+
 // Record keeps a divergence unless its source has already reported a later quarantine.
 func (r *Redis) Record(ctx context.Context, divergence model.Divergence) error {
 	field := sourceField(divergence.Service, divergence.Environment, divergence.Direction)

internal/server/storage/storage.go

@@ -1,5 +1,6 @@
 // Package storage keeps the divergences that mirrors report to the SPECTRE
-// server. A URL selects the backend by its scheme, such as memory:// or redis://.
+// server. A URL selects the backend by its scheme, such as memory://, redis://,
+// or redis-cluster://.
 package storage
 
 import (
@@ -25,7 +26,7 @@ func register(scheme string, constructor Constructor) {
 
 // Config selects where the SPECTRE server keeps divergences.
 type Config struct {
-	Storage *url.URL `default:"memory://" placeholder:"URL" help:"Where to keep divergences: memory://, which a restart loses, or a Redis URL such as redis://localhost:6379/0."`
+	Storage *url.URL `default:"memory://" placeholder:"URL" help:"Where to keep divergences: memory://, which a restart loses, a Redis URL such as redis://localhost:6379/0, or a Redis Cluster URL such as redis-cluster://localhost:7000?addr=localhost:7001."`
 }
 
 // Validate checks that a backend serves the storage URL's scheme.

internal/server/storage/storage_test.go

@@ -23,7 +23,8 @@ func TestStores(t *testing.T) {
 		open func(t *testing.T) storage.Store
 	}{
 		{name: "Memory", open: func(*testing.T) storage.Store { return storage.NewMemory() }},
-		{name: "Redis", open: openRedis},
+		{name: "Redis", open: openRedis("redis")},
+		{name: "RedisCluster", open: openRedis("redis-cluster")},
 	}
 	tests := []struct {
 		name string
@@ -55,8 +56,12 @@ func TestOpen(t *testing.T) {
 	}{
 		{name: "Memory", url: "memory://"},
 		{name: "Redis", url: "redis://" + server.Addr()},
+		{name: "RedisCluster", url: "redis-cluster://" + server.Addr()},
 		{name: "InvalidRedisURL", url: "redis://localhost/notadatabase", err: "redis URL"},
+		{name: "RedisClusterDatabase", url: "redis-cluster://" + server.Addr() + "/1", err: "no databases"},
+		{name: "InvalidRedisClusterURL", url: "redis-cluster://" + server.Addr() + "?db=1", err: "unexpected option: db"},
 		{name: "UnreachableRedis", url: "redis://127.0.0.1:1", err: "connect to Redis"},
+		{name: "UnreachableRedisCluster", url: "redis-cluster://127.0.0.1:1", err: "connect to Redis"},
 		{name: "UnknownScheme", url: "disk:///var/lib/spectre", err: `unknown storage scheme "disk"`},
 	}
 	for _, test := range tests {
@@ -88,12 +93,16 @@ func TestRedisRejectsUnknownFields(t *testing.T) {
 	assert.Contains(t, err.Error(), `unknown field "method"`)
 }
 
-func openRedis(t *testing.T) storage.Store {
-	t.Helper()
-	server := miniredis.RunT(t)
-	store, err := storage.NewRedis(t.Context(), parseURL(t, "redis://"+server.Addr()))
-	assert.NoError(t, err)
-	return store
+// openRedis opens a store on a fresh miniredis, which acts as a cluster of one
+// node owning every slot when the scheme selects a cluster client.
+func openRedis(scheme string) func(t *testing.T) storage.Store {
+	return func(t *testing.T) storage.Store {
+		t.Helper()
+		server := miniredis.RunT(t)
+		store, err := storage.NewRedis(t.Context(), parseURL(t, scheme+"://"+server.Addr()))
+		assert.NoError(t, err)
+		return store
+	}
 }
 
 func parseURL(t *testing.T, raw string) *url.URL {