Commit ca0b6cae authored by Harkrishn Patro's avatar Harkrishn Patro Committed by Oran Agra
Browse files

Propagate message to a node only if the cluster link is healthy. (#11752)

Currently while a sharded pubsub message publish tries to propagate the message across the cluster, a NULL check is missing for clusterLink. clusterLink could be NULL if the link is causing memory beyond the set threshold cluster-link-sendbuf-limit and server terminates the link.

This change introduces two things:

Avoids the engine crashes on the publishing node if a message is tried to be sent to a node and the link is NULL.
Adds a debugging tool CLUSTERLINK KILL to terminate the clusterLink between two nodes.

(cherry picked from commit fd397568)
parent 5aaa1a27
...@@ -2774,6 +2774,9 @@ void clusterReadHandler(connection *conn) { ...@@ -2774,6 +2774,9 @@ void clusterReadHandler(connection *conn) {
* the link to be invalidated, so it is safe to call this function * the link to be invalidated, so it is safe to call this function
* from event handlers that will do stuff with the same link later. */ * from event handlers that will do stuff with the same link later. */
void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) { void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
if (!link) {
return;
}
if (sdslen(link->sndbuf) == 0 && msglen != 0) if (sdslen(link->sndbuf) == 0 && msglen != 0)
connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1); connSetWriteHandlerWithBarrier(link->conn, clusterWriteHandler, 1);
...@@ -2800,7 +2803,6 @@ void clusterBroadcastMessage(void *buf, size_t len) { ...@@ -2800,7 +2803,6 @@ void clusterBroadcastMessage(void *buf, size_t len) {
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
clusterNode *node = dictGetVal(de); clusterNode *node = dictGetVal(de);
if (!node->link) continue;
if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE)) if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
continue; continue;
clusterSendMessage(node->link,buf,len); clusterSendMessage(node->link,buf,len);
...@@ -3241,11 +3243,11 @@ void clusterPropagatePublish(robj *channel, robj *message, int sharded) { ...@@ -3241,11 +3243,11 @@ void clusterPropagatePublish(robj *channel, robj *message, int sharded) {
listRewind(nodes_for_slot, &li); listRewind(nodes_for_slot, &li);
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
clusterNode *node = listNodeValue(ln); clusterNode *node = listNodeValue(ln);
if (node != myself) { if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
continue;
clusterSendPublish(node->link, channel, message, CLUSTERMSG_TYPE_PUBLISHSHARD); clusterSendPublish(node->link, channel, message, CLUSTERMSG_TYPE_PUBLISHSHARD);
} }
} }
}
listRelease(nodes_for_slot); listRelease(nodes_for_slot);
} }
......
...@@ -399,5 +399,6 @@ void slotToChannelAdd(sds channel); ...@@ -399,5 +399,6 @@ void slotToChannelAdd(sds channel);
void slotToChannelDel(sds channel); void slotToChannelDel(sds channel);
void clusterUpdateMyselfHostname(void); void clusterUpdateMyselfHostname(void);
void clusterUpdateMyselfAnnouncedPorts(void); void clusterUpdateMyselfAnnouncedPorts(void);
void freeClusterLink(clusterLink *link);
#endif /* __CLUSTER_H */ #endif /* __CLUSTER_H */
...@@ -34,6 +34,7 @@ ...@@ -34,6 +34,7 @@
#include "crc64.h" #include "crc64.h"
#include "bio.h" #include "bio.h"
#include "quicklist.h" #include "quicklist.h"
#include "cluster.h"
#include <arpa/inet.h> #include <arpa/inet.h>
#include <signal.h> #include <signal.h>
...@@ -489,7 +490,9 @@ void debugCommand(client *c) { ...@@ -489,7 +490,9 @@ void debugCommand(client *c) {
" In case NEVER is provided the last observed peak will never be reset", " In case NEVER is provided the last observed peak will never be reset",
" In case RESET is provided the peak reset time will be restored to the default value", " In case RESET is provided the peak reset time will be restored to the default value",
"REPLYBUFFER RESIZING <0|1>", "REPLYBUFFER RESIZING <0|1>",
" Enable or disable the replay buffer resize cron job", " Enable or disable the reply buffer resize cron job",
"CLUSTERLINK KILL <to|from|all> <node-id>",
" Kills the link based on the direction to/from (both) with the provided node." ,
NULL NULL
}; };
addReplyHelp(c, help); addReplyHelp(c, help);
...@@ -995,6 +998,33 @@ NULL ...@@ -995,6 +998,33 @@ NULL
return; return;
} }
addReply(c, shared.ok); addReply(c, shared.ok);
} else if(!strcasecmp(c->argv[1]->ptr,"CLUSTERLINK") &&
!strcasecmp(c->argv[2]->ptr,"KILL") &&
c->argc == 5) {
if (!server.cluster_enabled) {
addReplyError(c, "Debug option only available for cluster mode enabled setup!");
return;
}
/* Find the node. */
clusterNode *n = clusterLookupNode(c->argv[4]->ptr, sdslen(c->argv[4]->ptr));
if (!n) {
addReplyErrorFormat(c,"Unknown node %s", (char*)c->argv[4]->ptr);
return;
}
/* Terminate the link based on the direction or all. */
if (!strcasecmp(c->argv[3]->ptr,"from")) {
freeClusterLink(n->inbound_link);
} else if (!strcasecmp(c->argv[3]->ptr,"to")) {
freeClusterLink(n->link);
} else if (!strcasecmp(c->argv[3]->ptr,"all")) {
freeClusterLink(n->link);
freeClusterLink(n->inbound_link);
} else {
addReplyErrorFormat(c, "Unknown direction %s", (char*) c->argv[3]->ptr);
}
addReply(c,shared.ok);
} else { } else {
addReplySubcommandSyntaxError(c); addReplySubcommandSyntaxError(c);
return; return;
......
...@@ -97,6 +97,7 @@ set ::all_tests { ...@@ -97,6 +97,7 @@ set ::all_tests {
unit/replybufsize unit/replybufsize
unit/cluster-scripting unit/cluster-scripting
unit/cluster/misc unit/cluster/misc
unit/cluster/links
} }
# Index to the next test to run in the ::all_tests list. # Index to the next test to run in the ::all_tests list.
set ::next_test 0 set ::next_test 0
...@@ -199,11 +200,16 @@ proc r {args} { ...@@ -199,11 +200,16 @@ proc r {args} {
[srv $level "client"] {*}$args [srv $level "client"] {*}$args
} }
# Returns a Redis instance by index.
proc Rn {n} {
set level [expr -1*$n]
return [srv $level "client"]
}
# Provide easy access to a client for an inner server. Requires a positive # Provide easy access to a client for an inner server. Requires a positive
# index, unlike r which uses an optional negative index. # index, unlike r which uses an optional negative index.
proc R {n args} { proc R {n args} {
set level [expr -1*$n] [Rn $n] {*}$args
[srv $level "client"] {*}$args
} }
proc reconnect {args} { proc reconnect {args} {
......
proc number_of_peers {id} {
expr [llength $::servers] - 1
}
proc number_of_links {id} {
llength [R $id cluster links]
}
proc publish_messages {server num_msgs msg_size} {
for {set i 0} {$i < $num_msgs} {incr i} {
$server PUBLISH channel [string repeat "x" $msg_size]
}
}
start_cluster 1 2 {tags {external:skip cluster}} {
set primary_id 0
set replica1_id 1
set primary [Rn $primary_id]
set replica1 [Rn $replica1_id]
test "Broadcast message across a cluster shard while a cluster link is down" {
set replica1_node_id [$replica1 CLUSTER MYID]
set channelname ch3
# subscribe on replica1
set subscribeclient1 [redis_deferring_client -1]
$subscribeclient1 deferred 1
$subscribeclient1 SSUBSCRIBE $channelname
$subscribeclient1 read
# subscribe on replica2
set subscribeclient2 [redis_deferring_client -2]
$subscribeclient2 deferred 1
$subscribeclient2 SSUBSCRIBE $channelname
$subscribeclient2 read
# Verify number of links with cluster stable state
assert_equal [expr [number_of_peers $primary_id]*2] [number_of_links $primary_id]
# Disconnect the cluster between primary and replica1 and publish a message.
$primary MULTI
$primary DEBUG CLUSTERLINK KILL TO $replica1_node_id
$primary SPUBLISH $channelname hello
set res [$primary EXEC]
# Verify no client exists on the primary to receive the published message.
assert_equal $res {OK 0}
# Wait for all the cluster links are healthy
wait_for_condition 50 100 {
[number_of_peers $primary_id]*2 == [number_of_links $primary_id]
} else {
fail "All peer links couldn't be established"
}
# Publish a message afterwards.
$primary SPUBLISH $channelname world
# Verify replica1 has received only (world) / hello is lost.
assert_equal "smessage ch3 world" [$subscribeclient1 read]
# Verify replica2 has received both messages (hello/world)
assert_equal "smessage ch3 hello" [$subscribeclient2 read]
assert_equal "smessage ch3 world" [$subscribeclient2 read]
} {} {needs:debug}
}
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment