Skip to content

Commit 30a0a48

Browse files
DhruvilK7rueian
DhruvilK7
and
rueian
committed
feat(cluster): add opt-in PreferClusterShards to use CLUSTER SHARDS before version 8 (#181)
* feat(cluster): add opt-in PreferClusterShards to use CLUSTER SHARDS before version 8 By default the client refreshes cluster topology with CLUSTER SHARDS only on servers version 8 and above, and uses CLUSTER SLOTS below that. This was done because CLUSTER SHARDS had a server side bug on older versions. The bug is fixed in Valkey 7.2.6 and Redis 8.0, so users on those versions can now opt in to CLUSTER SHARDS earlier by setting PreferClusterShards. This keeps the health field from CLUSTER SHARDS during rolling engine upgrades, which avoids routing commands to nodes that are still loading. The option is off by default so existing behavior does not change. It only takes effect on servers version 7 and above, so it never sends CLUSTER SHARDS to a server that does not support the command. Signed-off-by: DhruvilK7 <dhruvil.kakadiya@zomato.com> * Apply batched suggestions from code review Co-authored-by: Rueian <rueiancsie@gmail.com> Signed-off-by: Rueian <rueiancsie@gmail.com> --------- Signed-off-by: DhruvilK7 <dhruvil.kakadiya@zomato.com> Signed-off-by: Rueian <rueiancsie@gmail.com> Co-authored-by: Rueian <rueiancsie@gmail.com> Signed-off-by: Rueian <rueiancsie@gmail.com>
1 parent 4d7ae1a commit 30a0a48

3 files changed

Lines changed: 57 additions & 11 deletions

File tree

‎cluster.go‎

Lines changed: 15 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -198,19 +198,19 @@ func clusterRefreshAutoMaxDelay(n int) time.Duration {
198198
}
199199

200200
type clusterslots struct {
201-
addr string
202-
reply RedisResult
203-
ver int
201+
addr string
202+
reply RedisResult
203+
useShards bool
204204
}
205205

206206
func (s clusterslots) parse(tls bool) map[string]group {
207-
if s.ver < 8 {
207+
if !s.useShards {
208208
return parseSlots(s.reply.val, s.addr)
209209
}
210210
return parseShards(s.reply.val, s.addr, tls)
211211
}
212212

213-
func getClusterSlots(c conn, timeout time.Duration) clusterslots {
213+
func getClusterSlots(c conn, timeout time.Duration, preferShards bool) clusterslots {
214214
var ctx context.Context
215215
var cancel context.CancelFunc
216216
if timeout > 0 {
@@ -220,10 +220,13 @@ func getClusterSlots(c conn, timeout time.Duration) clusterslots {
220220
ctx = context.Background()
221221
}
222222
v := c.Version()
223-
if v < 8 {
224-
return clusterslots{reply: c.Do(ctx, cmds.SlotCmd), addr: c.Addr(), ver: v}
223+
// CLUSTER SHARDS on >= 8 always; below that only when opted in, and never
224+
// below 7 where the command does not exist. See ClusterOption.PreferClusterShards.
225+
useShards := v >= 8 || (preferShards && v >= 7)
226+
if !useShards {
227+
return clusterslots{reply: c.Do(ctx, cmds.SlotCmd), addr: c.Addr()}
225228
}
226-
return clusterslots{reply: c.Do(ctx, cmds.ShardsCmd), addr: c.Addr(), ver: v}
229+
return clusterslots{reply: c.Do(ctx, cmds.ShardsCmd), addr: c.Addr(), useShards: true}
227230
}
228231

229232
func (c *clusterClient) _refresh() (err error) {
@@ -429,15 +432,16 @@ func (c *clusterClient) clusterRefreshConns() []conn {
429432

430433
func (c *clusterClient) refreshConns(pending []conn, batchDelay time.Duration) (result clusterslots, err error) {
431434
results := make(chan clusterslots, len(pending))
435+
preferShards := c.opt.ClusterOption.PreferClusterShards
432436
for i := 0; i < len(pending); i++ {
433437
if i&3 == 0 { // batch CLUSTER SLOTS/CLUSTER SHARDS for every 4 connections
434438
if i > 0 && batchDelay > 0 {
435439
time.Sleep(batchDelay)
436440
}
437441
for j := i; j < i+4 && j < len(pending); j++ {
438-
go func(c conn, timeout time.Duration) {
439-
results <- getClusterSlots(c, timeout)
440-
}(pending[j], c.opt.ConnWriteTimeout)
442+
go func(c conn, timeout time.Duration, preferShards bool) {
443+
results <- getClusterSlots(c, timeout, preferShards)
444+
}(pending[j], c.opt.ConnWriteTimeout, preferShards)
441445
}
442446
}
443447
result = <-results

‎cluster_test.go‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6295,6 +6295,40 @@ func TestConnectToNonAvailableCluster(t *testing.T) {
62956295
wg.Wait()
62966296
}
62976297

6298+
func TestGetClusterSlotsPreferShards(t *testing.T) {
6299+
for _, tc := range []struct {
6300+
version int
6301+
preferShards bool
6302+
wantCmd string
6303+
wantShards bool
6304+
}{
6305+
{version: 8, preferShards: false, wantCmd: "CLUSTER SHARDS", wantShards: true}, // >= 8 always shards
6306+
{version: 7, preferShards: false, wantCmd: "CLUSTER SLOTS", wantShards: false}, // default keeps 7.x on slots
6307+
{version: 7, preferShards: true, wantCmd: "CLUSTER SHARDS", wantShards: true}, // opt-in enables shards on 7
6308+
{version: 6, preferShards: true, wantCmd: "CLUSTER SLOTS", wantShards: false}, // floor: no shards below 7
6309+
{version: 5, preferShards: true, wantCmd: "CLUSTER SLOTS", wantShards: false}, // RESP2 fallback stays slots
6310+
} {
6311+
t.Run(fmt.Sprintf("v%d_prefer%v", tc.version, tc.preferShards), func(t *testing.T) {
6312+
var got string
6313+
c := &mockConn{
6314+
VersionFn: func() int { return tc.version },
6315+
AddrFn: func() string { return "127.0.0.1:0" },
6316+
DoFn: func(cmd Completed) RedisResult {
6317+
got = strings.Join(cmd.Commands(), " ")
6318+
return NewResult(slicemsg('*', []RedisMessage{}), nil)
6319+
},
6320+
}
6321+
res := getClusterSlots(c, 0, tc.preferShards)
6322+
if got != tc.wantCmd {
6323+
t.Fatalf("version %d preferShards %v: sent %q, want %q", tc.version, tc.preferShards, got, tc.wantCmd)
6324+
}
6325+
if res.useShards != tc.wantShards {
6326+
t.Fatalf("version %d preferShards %v: useShards %v, want %v", tc.version, tc.preferShards, res.useShards, tc.wantShards)
6327+
}
6328+
})
6329+
}
6330+
}
6331+
62986332
func TestClusterTopologyRefreshment(t *testing.T) {
62996333
defer ShouldNotLeak(SetupLeakDetection())
63006334

‎rueidis.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -337,6 +337,14 @@ type ClusterOption struct {
337337

338338
// PreferInitAddressRefresh only uses ClientOption.InitAddress nodes during cluster topology refresh.
339339
PreferInitAddressRefresh bool
340+
341+
// PreferClusterShards uses CLUSTER SHARDS instead of CLUSTER SLOTS to refresh the cluster topology
342+
// on servers with version >= 7 (by default CLUSTER SHARDS is only used on version >= 8).
343+
// CLUSTER SHARDS reports the per-node "health" field, which lets the client drop nodes that are not
344+
// online (e.g. LOADING during a failover) instead of routing to them.
345+
// Enable this ONLY if your engine's CLUSTER SHARDS is fixed: Valkey >= 7.2.6 or Redis >= 8.0.
346+
// Do NOT enable it on Redis 7.x: its CLUSTER SHARDS returns wrong topology after a failover.
347+
PreferClusterShards bool
340348
}
341349

342350
// StandaloneOption is the options for the standalone client.

0 commit comments

Comments
 (0)