From 613551b6374fd56d2ca9a5231eeb8dbef2330f0f Mon Sep 17 00:00:00 2001 From: Alec Thomas Date: Wed, 28 Oct 2026 22:12:12 +1100 Subject: [PATCH] feat(server): add support for Redis clusters Co-Authored-By: Claude Opus 5.5 --- CONTRIBUTING.md | 5 ++- internal/server/storage/redis.go | 42 ++++++++++++++++++++----- internal/server/storage/storage.go | 5 +-- internal/server/storage/storage_test.go | 23 +++++++++----- 4 files changed, 58 insertions(+), 17 deletions(-) diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 0c322cd..953c0b3 100644 --- a/CONTRIBUTING.md +++ b/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 diff --git a/internal/server/storage/redis.go b/internal/server/storage/redis.go index 32c5146..ea67370 100644 --- a/internal/server/storage/redis.go +++ b/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,25 +27,29 @@ 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") @@ -52,6 +57,29 @@ func NewRedis(ctx context.Context, storageURL *url.URL) (*Redis, error) { 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) diff --git a/internal/server/storage/storage.go b/internal/server/storage/storage.go index b19337a..5f313d1 100644 --- a/internal/server/storage/storage.go +++ b/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. diff --git a/internal/server/storage/storage_test.go b/internal/server/storage/storage_test.go index 90a9a94..3db8974 100644 --- a/internal/server/storage/storage_test.go +++ b/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 {