slipenkomk/valkey
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Generate correct slot information in cluster shards command on primary failure (#790)

Browse Source 
Fix #784 

Prior to the change, `CLUSTER SHARDS` command processing might pick a
failed primary node which won't have the slot coverage information and
the slots `output` in turn would be empty. This change finds an
appropriate node which has the slot coverage information served by a
given shard and correctly displays it as part of `CLUSTER SHARDS`
output.

 Before:
 
 ```
 1) 1) "slots"
   2)  (empty array)
   3) "nodes"
   4) 1)  1) "id"
          2) "2936f22a490095a0a851b7956b0a88f2b67a5d44"
          ...
          9) "role"
         10) "master"
         ...
         13) "health"
         14) "fail"
 ```
 
 After:
 

 ```
 1) 1) "slots"
   2) 1) 0
       2) 5461
   3) "nodes"
   4) 1)  1) "id"
          2) "2936f22a490095a0a851b7956b0a88f2b67a5d44"
          ...
          9) "role"
         10) "master"
         ...
         13) "health"
         14) "fail"
 ```

---------

Signed-off-by: Harkrishn Patro <harkrisp@amazon.com>
This commit is contained in:
Harkrishn Patro 2024-07-19 09:32:39 -07:00 committed by GitHub
parent 35a1888333
commit 816accea76
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 90 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
src/cluster_legacy.c
View File

@ -5662,36 +5662,6 @@ void addNodeDetailsToShardReply(client *c, clusterNode *node) {
setDeferredMapLen(c, node_replylen, reply_count);
}
/* Add the shard reply of a single shard based off the given primary node. */
void addShardReplyForClusterShards(client *c, list *nodes) {
serverAssert(listLength(nodes) > 0);
clusterNode *n = listNodeValue(listFirst(nodes));
addReplyMapLen(c, 2);
addReplyBulkCString(c, "slots");
/* Use slot_info_pairs from the primary only */
n = clusterNodeGetPrimary(n);
if (n->slot_info_pairs != NULL) {
serverAssert((n->slot_info_pairs_count % 2) == 0);
addReplyArrayLen(c, n->slot_info_pairs_count);
for (int i = 0; i < n->slot_info_pairs_count; i++) addReplyLongLong(c, (unsigned long)n->slot_info_pairs[i]);
} else {
/* If no slot info pair is provided, the node owns no slots */
addReplyArrayLen(c, 0);
}
addReplyBulkCString(c, "nodes");
addReplyArrayLen(c, listLength(nodes));
listIter li;
listRewind(nodes, &li);
for (listNode *ln = listNext(&li); ln != NULL; ln = listNext(&li)) {
clusterNode *n = listNodeValue(ln);
addNodeDetailsToShardReply(c, n);
clusterFreeNodesSlotsInfo(n);
}
}
/* Add to the output buffer of the given client, an array of slot (start, end)
* pair owned by the shard, also the primary and set of replica(s) along with
* information about each node. */
@ -5701,7 +5671,41 @@ void clusterCommandShards(client *c) {
clusterGenNodesSlotsInfo(0);
dictIterator *di = dictGetSafeIterator(server.cluster->shards);
for (dictEntry *de = dictNext(di); de != NULL; de = dictNext(di)) {
addShardReplyForClusterShards(c, dictGetVal(de));
list *nodes = dictGetVal(de);
serverAssert(listLength(nodes) > 0);
addReplyMapLen(c, 2);
addReplyBulkCString(c, "slots");
/* Find a node which has the slot information served by this shard. */
clusterNode *n = NULL;
listIter li;
listRewind(nodes, &li);
for (listNode *ln = listNext(&li); ln != NULL; ln = listNext(&li)) {
n = listNodeValue(ln);
if (n->slot_info_pairs) {
break;
}
}
if (n && n->slot_info_pairs != NULL) {
serverAssert((n->slot_info_pairs_count % 2) == 0);
addReplyArrayLen(c, n->slot_info_pairs_count);
for (int i = 0; i < n->slot_info_pairs_count; i++) {
addReplyLongLong(c, (unsigned long)n->slot_info_pairs[i]);
}
} else {
/* If no slot info pair is provided, the node owns no slots */
addReplyArrayLen(c, 0);
}
addReplyBulkCString(c, "nodes");
addReplyArrayLen(c, listLength(nodes));
listRewind(nodes, &li);
for (listNode *ln = listNext(&li); ln != NULL; ln = listNext(&li)) {
clusterNode *n = listNodeValue(ln);
addNodeDetailsToShardReply(c, n);
clusterFreeNodesSlotsInfo(n);
}
}
dictReleaseIterator(di);
}

55
tests/unit/cluster/cluster-shards.tcl Normal file
View File

@ -0,0 +1,55 @@
# Get the node info with the specific node_id from the
# given reference node. Valid type options are "node" and "shard"
proc get_node_info_from_shard {id reference {type node}} {
set shards_response [R $reference CLUSTER SHARDS]
foreach shard_response $shards_response {
set nodes [dict get $shard_response nodes]
foreach node $nodes {
if {[dict get $node id] eq $id} {
if {$type eq "node"} {
return $node
} elseif {$type eq "shard"} {
return $shard_response
} else {
return {}
}
}
}
}
# No shard found, return nothing
return {}
}
start_cluster 3 3 {tags {external:skip cluster}} {
set primary_node 0
set replica_node 3
set validation_node 4
set node_0_id ""
set shard_0_slot_coverage {0 5461}
test "Cluster should start ok" {
wait_for_cluster_state ok
}
test "Cluster shards response is ok for shard 0" {
set node_0_id [R $primary_node CLUSTER MYID]
assert_equal $shard_0_slot_coverage [dict get [get_node_info_from_shard $node_0_id $validation_node "shard"] "slots"]
}
test "Kill a node and tell the replica to immediately takeover" {
pause_process [srv $primary_node pid]
R $replica_node CLUSTER failover force
}
test "Verify health as fail for killed node" {
wait_for_condition 50 100 {
"fail" eq [dict get [get_node_info_from_shard $node_0_id $validation_node "node"] "health"]
} else {
fail "New primary never detected the node failed"
}
}
test "CLUSTER SHARDS slot response is non-empty when primary node fails" {
assert_equal $shard_0_slot_coverage [dict get [get_node_info_from_shard $node_0_id $validation_node "shard"] "slots"]
}
}