feat: support redis cluster mode

Author: hoalongnatsu

cmd/main.go

@@ -145,13 +145,11 @@ func main() {
 
 	// Shared Redis client used by both on-call and the agent worker. We open
 	// it once here so both subsystems share connections.
-	var sharedRedis *redis.Client
+	var sharedRedis redis.UniversalClient
 
 	if cfg.OnCall.Enable || cfg.OnCall.InitializedOnly {
-		redisOptions := handlerRedisOptions(cfg.Redis)
-
-		// Initialize Redis client
-		redisClient := redis.NewClient(redisOptions)
+		// Initialize Redis client (cluster-aware when redis.cluster is set)
+		redisClient := newRedisClient(cfg.Redis)
 
 		// Test Redis connection
 		if err := redisClient.Ping(context.Background()).Err(); err != nil {
@@ -179,7 +177,7 @@ func main() {
 		// in-memory cursors when Redis isn't reachable).
 		rdb := sharedRedis
 		if rdb == nil && cfg.Redis.Host != "" {
-			rdb = redis.NewClient(handlerRedisOptions(cfg.Redis))
+			rdb = newRedisClient(cfg.Redis)
 			if err := rdb.Ping(context.Background()).Err(); err != nil {
 				log.Printf("agent: Redis unavailable (%v); cursors will be in-memory only", err)
 				rdb = nil
@@ -225,7 +223,7 @@ func main() {
 // startAgent constructs the worker, starts it in a goroutine, and registers
 // admin routes on the fiber app. It returns the catalog so the caller can
 // hold a reference (and so future hot-reload code has a handle to it).
-func startAgent(ctx context.Context, app *fiber.App, cfg c.AgentConfig, gatewaySecret string, store storage.Provider, rdb *redis.Client) (*agent.Catalog, error) {
+func startAgent(ctx context.Context, app *fiber.App, cfg c.AgentConfig, gatewaySecret string, store storage.Provider, rdb redis.UniversalClient) (*agent.Catalog, error) {
 	// On the Postgres backend, install the typed signal-table
 	// catalog store so the log catalog reads/writes the explicit
 	// vs_patterns/vs_logs/vs_services tables (searchable, indexed) instead of
@@ -408,6 +406,24 @@ func handleQueueMessage(content *map[string]interface{}) error {
 	return services.CreateIncident("", content, &overwrite) // teamID as empty string
 }
 
+// newRedisClient builds the Redis client both subsystems share. It reuses
+// handlerRedisOptions for the addr/password/TLS settings, then returns either a
+// single-node client (the default) or a cluster-aware client when the operator
+// opts in via redis.cluster / REDIS_CLUSTER. Both concrete types satisfy
+// redis.UniversalClient, so callers thread the interface and single-key
+// Get/Set follow MOVED/ASK redirects transparently in cluster mode.
+func newRedisClient(rc c.RedisConfig) redis.UniversalClient {
+	opts := handlerRedisOptions(rc)
+	if rc.ClusterEnabled() {
+		return redis.NewClusterClient(&redis.ClusterOptions{
+			Addrs:     []string{opts.Addr},
+			Password:  opts.Password,
+			TLSConfig: opts.TLSConfig,
+		})
+	}
+	return redis.NewClient(opts)
+}
+
 func handlerRedisOptions(rc c.RedisConfig) *redis.Options {
 	redisOptions := &redis.Options{
 		Addr:     rc.Host + ":" + strconv.Itoa(rc.Port),

pkg/agent/brain_log_test.go

@@ -379,3 +379,39 @@ func TestLogBrain_AutoPromoteAfter_AlreadyKnownStaysKnown(t *testing.T) {
 		t.Fatalf("catalog verdict = %q, want known (an already-known pattern must stay known)", c.Get("p").Verdict)
 	}
 }
+
+// (f) The operator's real "first fetch" scenario: an Elasticsearch source's
+// FIRST batch lands a pattern whose sighting count is already far above the
+// threshold (e.g. 3304 >= 100) in a single tick. It must classify known
+// immediately — Promote flips the stored verdict and LogReadiness reports Ready
+// — not sit "still learning" forever. This pins the count clause of isLogKnown
+// on the first-batch path so a large first pull is never stuck learning once the
+// threshold is the (non-zero) default.
+func TestLogBrain_AutoPromoteAfter_FirstBatchAboveThresholdIsKnown(t *testing.T) {
+	cat := config.AgentCatalogConfig{
+		AutoPromoteAfter: 100, // the embedded default an omitted key resolves to
+		SpikeMultiplier:  0,   // isolate the count-promotion path from spike
+	}
+	b, c := newLogBrainForTest(t, cat)
+
+	v := classifyOnce(t, b, logObs("p", 3304)) // single first-fetch batch
+	if got := c.Get("p").Count; got != 3304 {
+		t.Fatalf("count = %d, want 3304", got)
+	}
+	if v.Class != core.VerdictKnownPattern {
+		t.Fatalf("first-batch verdict = %v, want known (3304 >= 100)", v.Class)
+	}
+	if c.Get("p").Verdict != "known" {
+		t.Fatalf("catalog verdict = %q, want known (Promote must fire on the first batch)", c.Get("p").Verdict)
+	}
+	// The read-side readiness the UI consumes must agree with the classifier:
+	// Ready, with the threshold as the target — never the Needed=0 "no target"
+	// state for a pattern already past the (non-zero) threshold.
+	r := LogReadiness(c.Get("p"), cat.AutoPromoteAfter, 30*time.Second)
+	if !r.Ready {
+		t.Fatalf("LogReadiness.Ready = false, want true (3304 >= 100)")
+	}
+	if r.Needed != 100 {
+		t.Fatalf("LogReadiness.Needed = %d, want 100 (threshold target, not the no-target sentinel)", r.Needed)
+	}
+}

pkg/agent/cursors.go

@@ -13,15 +13,15 @@ import (
 // falls back to in-memory storage when Redis is unavailable so that
 // development setups don't require Redis just to try training mode.
 type CursorStore struct {
-	rdb       *redis.Client
+	rdb       redis.UniversalClient
 	keyPrefix string
 
 	mu  sync.RWMutex
 	mem map[string]time.Time
 }
 
 // NewCursorStore returns a CursorStore. Pass `rdb=nil` for in-memory only.
-func NewCursorStore(rdb *redis.Client) *CursorStore {
+func NewCursorStore(rdb redis.UniversalClient) *CursorStore {
 	return &CursorStore{
 		rdb:       rdb,
 		keyPrefix: "versus:agent:cursor:",
@@ -77,14 +77,30 @@ func (s *CursorStore) Reset(ctx context.Context) error {
 	if s.rdb == nil {
 		return nil
 	}
+	// In cluster mode a plain SCAN only walks the node the call routes to,
+	// so sweep every master shard; in single-node mode there is exactly one
+	// backend and the single deleteCursorKeys pass covers it.
+	if cc, ok := s.rdb.(*redis.ClusterClient); ok {
+		return cc.ForEachMaster(ctx, func(ctx context.Context, m *redis.Client) error {
+			return s.deleteCursorKeys(ctx, m)
+		})
+	}
+	return s.deleteCursorKeys(ctx, s.rdb)
+}
+
+// deleteCursorKeys scans one Redis backend for every cursor key under the
+// prefix and deletes them ONE AT A TIME. Single-key Del always routes to the
+// key's owning slot, so it is correct on a *redis.ClusterClient too — a
+// multi-key Del across different hash slots would fail with CROSSSLOT.
+func (s *CursorStore) deleteCursorKeys(ctx context.Context, c redis.Cmdable) error {
 	var cursor uint64
 	for {
-		keys, next, err := s.rdb.Scan(ctx, cursor, s.keyPrefix+"*", 100).Result()
+		keys, next, err := c.Scan(ctx, cursor, s.keyPrefix+"*", 100).Result()
 		if err != nil {
 			return err
 		}
-		if len(keys) > 0 {
-			if err := s.rdb.Del(ctx, keys...).Err(); err != nil {
+		for _, k := range keys {
+			if err := c.Del(ctx, k).Err(); err != nil {
 				return err
 			}
 		}

pkg/config/config.go

@@ -236,6 +236,16 @@ type RedisConfig struct {
 	DB                 int    `mapstructure:"db"`
 	TLS                *bool  `mapstructure:"tls"`
 	InsecureSkipVerify bool   `mapstructure:"insecure_skip_verify"`
+	Cluster            *bool  `mapstructure:"cluster"`
+}
+
+// ClusterEnabled reports whether the Redis client should be built in
+// cluster mode (redis.NewClusterClient) so it follows MOVED/ASK redirects
+// across shards — required for AWS ElastiCache Valkey/Redis in cluster mode.
+// Single-node is the default: a nil flag (key omitted) preserves today's
+// behaviour. Set redis.cluster: true (or REDIS_CLUSTER=true) to opt in.
+func (r RedisConfig) ClusterEnabled() bool {
+	return r.Cluster != nil && *r.Cluster
 }
 
 // TLSEnabled reports whether the Redis client should dial over TLS. TLS is
@@ -256,6 +266,16 @@ func redisTLSFromEnv(v string) *bool {
 	return &enabled
 }
 
+// redisClusterFromEnv resolves a non-empty REDIS_CLUSTER value to a cluster
+// flag, fail-safe OFF: only an explicit on-value (true/1/yes/on, any case)
+// enables cluster mode; every other value (a typo, empty, off, ...) keeps the
+// single-node default so an operator never accidentally switches topology.
+func redisClusterFromEnv(v string) *bool {
+	on := map[string]bool{"true": true, "1": true, "yes": true, "on": true}
+	enabled := on[strings.ToLower(strings.TrimSpace(v))]
+	return &enabled
+}
+
 var (
 	cfg     *Config
 	cfgOnce sync.Once
@@ -352,6 +372,14 @@ func loadConfigFromPath(path string) (*Config, error) {
 		loaded.Redis.TLS = redisTLSFromEnv(val)
 	}
 
+	// Redis env override: REDIS_CLUSTER toggles cluster-mode client
+	// construction. Fail-safe OFF: only an explicit on-value (true/1/yes/on,
+	// any case) enables cluster mode; unset or any other value keeps the
+	// single-node default so topology is never switched by accident.
+	if val := os.Getenv("REDIS_CLUSTER"); val != "" {
+		loaded.Redis.Cluster = redisClusterFromEnv(val)
+	}
+
 	// Storage env overrides
 	if t := os.Getenv("STORAGE_TYPE"); t != "" {
 		loaded.Storage.Type = t

pkg/config/redis_test.go

@@ -62,3 +62,63 @@ func TestRedisTLSFromEnv(t *testing.T) {
 		})
 	}
 }
+
+// TestRedisConfigClusterEnabled covers the cluster opt-in: single-node is the
+// default when the flag is omitted, an explicit true enables cluster mode.
+func TestRedisConfigClusterEnabled(t *testing.T) {
+	tru := true
+	fls := false
+
+	tests := []struct {
+		name    string
+		cluster *bool
+		want    bool
+	}{
+		{"omitted defaults to single-node", nil, false},
+		{"explicit true", &tru, true},
+		{"explicit false", &fls, false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			rc := RedisConfig{Cluster: tc.cluster}
+			if got := rc.ClusterEnabled(); got != tc.want {
+				t.Fatalf("ClusterEnabled() = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// TestRedisClusterFromEnv covers the fail-safe-OFF parse: only an explicit
+// on-value (true/1/yes/on, any case) enables cluster mode; every other value
+// — an off-value, a typo, or garbage — keeps the single-node default. Unset
+// is handled by the caller (no override) and stays off via ClusterEnabled().
+func TestRedisClusterFromEnv(t *testing.T) {
+	tests := []struct {
+		value string
+		want  bool // resolved ClusterEnabled()
+	}{
+		{"true", true},
+		{"TRUE", true},
+		{"1", true},
+		{"yes", true},
+		{"on", true},
+		{" true ", true},
+		{"false", false},
+		{"0", false},
+		{"no", false},
+		{"off", false},
+		{"ture", false}, // typo stays single-node
+		{"garbage", false},
+		{"", false},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.value, func(t *testing.T) {
+			got := RedisConfig{Cluster: redisClusterFromEnv(tc.value)}.ClusterEnabled()
+			if got != tc.want {
+				t.Fatalf("redisClusterFromEnv(%q) → ClusterEnabled() = %v, want %v", tc.value, got, tc.want)
+			}
+		})
+	}
+}

pkg/core/oncall.go

@@ -23,7 +23,7 @@ var CreateOnCallProvider func(cfg *config.Config, awsClient *ssmincidents.Client
 // OnCallWorkflow coordinates on-call escalation with a single provider
 type OnCallWorkflow struct {
 	provider    OnCallProvider
-	redisClient *redis.Client
+	redisClient redis.UniversalClient
 }
 
 // Global instance for singleton access
@@ -33,7 +33,7 @@ var (
 )
 
 // NewOnCallWorkflow creates a new on-call workflow with the given provider
-func NewOnCallWorkflow(redisClient *redis.Client, provider OnCallProvider) *OnCallWorkflow {
+func NewOnCallWorkflow(redisClient redis.UniversalClient, provider OnCallProvider) *OnCallWorkflow {
 	return &OnCallWorkflow{
 		provider:    provider,
 		redisClient: redisClient,
@@ -42,7 +42,7 @@ func NewOnCallWorkflow(redisClient *redis.Client, provider OnCallProvider) *OnCa
 
 // InitOnCallWorkflow initializes the global singleton instance
 // This is called once from main.go with the Redis client and AWS client
-func InitOnCallWorkflow(awsClient *ssmincidents.Client, redisClient *redis.Client) {
+func InitOnCallWorkflow(awsClient *ssmincidents.Client, redisClient redis.UniversalClient) {
 	once.Do(func() {
 		cfg := config.GetConfig()