From d9de6d989c7b8482f410e75d6791cdc1dc3ec8db Mon Sep 17 00:00:00 2001 From: Ayanda Dube Date: Fri, 6 Sep 2024 16:49:54 +0100 Subject: [PATCH 1/5] Shutdown peer QQ FSMs on connected nodes on force-shrink execution for cluster wide consistency, ensuring only the leader is active/running (cherry picked from commit b675ce29f022bc9d46f20ef32e065b0bb9684c8b) --- deps/rabbit/src/rabbit_quorum_queue.erl | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl index 18cc10f55ef8..cd71d250b7b8 100644 --- a/deps/rabbit/src/rabbit_quorum_queue.erl +++ b/deps/rabbit/src/rabbit_quorum_queue.erl @@ -1376,6 +1376,7 @@ delete_member(Q, Node) when ?amqqueue_is_quorum(Q) -> _ = rabbit_amqqueue:update(QName, Fun), case ra:force_delete_server(?RA_SYSTEM, ServerId) of ok -> + rabbit_log:info("Deleted a replica of quorum ~ts on node ~ts", [rabbit_misc:rs(QName), Node]), ok; {error, {badrpc, nodedown}} -> ok; @@ -1957,6 +1958,7 @@ force_shrink_member_to_current_member(VHost, Name) -> case rabbit_amqqueue:lookup(QName) of {ok, Q} when ?is_amqqueue(Q) -> {RaName, _} = amqqueue:get_pid(Q), + OtherNodes = lists:delete(Node, get_nodes(Q)), ok = ra_server_proc:force_shrink_members_to_current_member({RaName, Node}), Fun = fun (Q0) -> TS0 = amqqueue:get_type_state(Q0), @@ -1964,6 +1966,7 @@ force_shrink_member_to_current_member(VHost, Name) -> amqqueue:set_type_state(Q, TS) end, _ = rabbit_amqqueue:update(QName, Fun), + _ = [ra:force_delete_server(?RA_SYSTEM, {RaName, N}) || N <- OtherNodes], rabbit_log:warning("Disaster recovery procedure: shrinking finished"); _ -> rabbit_log:warning("Disaster recovery procedure: shrinking failed, queue ~p not found at vhost ~p", [Name, VHost]), @@ -1976,6 +1979,7 @@ force_all_queues_shrink_member_to_current_member() -> _ = [begin QName = amqqueue:get_name(Q), {RaName, _} = amqqueue:get_pid(Q), + OtherNodes = lists:delete(Node, get_nodes(Q)), rabbit_log:warning("Disaster recovery procedure: shrinking queue ~p", [QName]), ok = ra_server_proc:force_shrink_members_to_current_member({RaName, Node}), Fun = fun (QQ) -> @@ -1983,7 +1987,8 @@ force_all_queues_shrink_member_to_current_member() -> TS = TS0#{nodes => [Node]}, amqqueue:set_type_state(QQ, TS) end, - _ = rabbit_amqqueue:update(QName, Fun) + _ = rabbit_amqqueue:update(QName, Fun), + _ = [ra:force_delete_server(?RA_SYSTEM, {RaName, N}) || N <- OtherNodes] end || Q <- rabbit_amqqueue:list(), amqqueue:get_type(Q) == ?MODULE], rabbit_log:warning("Disaster recovery procedure: shrinking finished"), ok. From 10dbde1f7102fb53d99c9c27ffc6ad7ff0094f8b Mon Sep 17 00:00:00 2001 From: Ayanda Dube Date: Tue, 1 Oct 2024 15:16:16 +0100 Subject: [PATCH 2/5] QQ tests for force-shrink to current member operations (cherry picked from commit 60ee35ea7e269dca3eecc84a68fd1a5feaa64ec2) --- deps/rabbit/test/quorum_queue_SUITE.erl | 83 ++++++++++++++++++++++++- 1 file changed, 82 insertions(+), 1 deletion(-) diff --git a/deps/rabbit/test/quorum_queue_SUITE.erl b/deps/rabbit/test/quorum_queue_SUITE.erl index 06341e37b851..775555eac2cb 100644 --- a/deps/rabbit/test/quorum_queue_SUITE.erl +++ b/deps/rabbit/test/quorum_queue_SUITE.erl @@ -92,7 +92,9 @@ groups() -> format, add_member_2, single_active_consumer_priority_take_over, - single_active_consumer_priority + single_active_consumer_priority, + force_shrink_member_to_current_member, + force_all_queues_shrink_member_to_current_member ] ++ all_tests()}, {cluster_size_5, [], [start_queue, @@ -1152,6 +1154,85 @@ single_active_consumer_priority(Config) -> rpc:call(Server0, ra, local_query, [RaNameQ3, QueryFun])), ok. +force_shrink_member_to_current_member(Config) -> + [Server0, Server1, Server2] = + rabbit_ct_broker_helpers:get_node_configs(Config, nodename), + + Ch = rabbit_ct_client_helpers:open_channel(Config, Server0), + QQ = ?config(queue_name, Config), + ?assertEqual({'queue.declare_ok', QQ, 0, 0}, + declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])), + + RaName = ra_name(QQ), + rabbit_ct_client_helpers:publish(Ch, QQ, 3), + wait_for_messages_ready([Server0], RaName, 3), + + {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]), + #{nodes := Nodes0} = amqqueue:get_type_state(Q0), + ?assertEqual(3, length(Nodes0)), + + rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, + force_shrink_member_to_current_member, [<<"/">>, QQ]), + + wait_for_messages_ready([Server0], RaName, 3), + + {ok, Q1} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]), + #{nodes := Nodes1} = amqqueue:get_type_state(Q1), + ?assertEqual(1, length(Nodes1)), + + %% grow queues back to all nodes + [rpc:call(Server0, rabbit_quorum_queue, grow, [S, <<"/">>, <<".*">>, all]) || S <- [Server1, Server2]], + + wait_for_messages_ready([Server0], RaName, 3), + {ok, Q2} = rpc:call(Server0, rabbit_amqqueue, lookup, [QQ, <<"/">>]), + #{nodes := Nodes2} = amqqueue:get_type_state(Q2), + ?assertEqual(3, length(Nodes2)). + +force_all_queues_shrink_member_to_current_member(Config) -> + [Server0, Server1, Server2] = + rabbit_ct_broker_helpers:get_node_configs(Config, nodename), + + Ch = rabbit_ct_client_helpers:open_channel(Config, Server0), + QQ = ?config(queue_name, Config), + AQ = ?config(alt_queue_name, Config), + ?assertEqual({'queue.declare_ok', QQ, 0, 0}, + declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])), + ?assertEqual({'queue.declare_ok', AQ, 0, 0}, + declare(Ch, AQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])), + + QQs = [QQ, AQ], + + [begin + RaName = ra_name(Q), + rabbit_ct_client_helpers:publish(Ch, Q, 3), + wait_for_messages_ready([Server0], RaName, 3), + {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]), + #{nodes := Nodes0} = amqqueue:get_type_state(Q0), + ?assertEqual(3, length(Nodes0)) + end || Q <- QQs], + + rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, + force_all_queues_shrink_member_to_current_member, []), + + [begin + RaName = ra_name(Q), + wait_for_messages_ready([Server0], RaName, 3), + {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]), + #{nodes := Nodes0} = amqqueue:get_type_state(Q0), + ?assertEqual(1, length(Nodes0)) + end || Q <- QQs], + + %% grow queues back to all nodes + [rpc:call(Server0, rabbit_quorum_queue, grow, [S, <<"/">>, <<".*">>, all]) || S <- [Server1, Server2]], + + [begin + RaName = ra_name(Q), + wait_for_messages_ready([Server0], RaName, 3), + {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, <<"/">>]), + #{nodes := Nodes0} = amqqueue:get_type_state(Q0), + ?assertEqual(3, length(Nodes0)) + end || Q <- QQs]. + priority_queue_fifo(Config) -> %% testing: if hi priority messages are published before lo priority %% messages they are always consumed first (fifo) From b03637f8ecbcbf8085100c809cde59abad760d70 Mon Sep 17 00:00:00 2001 From: Ayanda Dube Date: Wed, 2 Oct 2024 11:28:41 +0100 Subject: [PATCH 3/5] Implement force_vhost_queues_shrink_member_to_current_member/1 (cherry picked from commit c26aa3b1c76dd82fb9e67852b6a2f030a92bf7cc) --- deps/rabbit/src/rabbit_quorum_queue.erl | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl index cd71d250b7b8..2b29759b3940 100644 --- a/deps/rabbit/src/rabbit_quorum_queue.erl +++ b/deps/rabbit/src/rabbit_quorum_queue.erl @@ -74,6 +74,7 @@ -export([validate_policy/1, merge_policy_value/3]). -export([force_shrink_member_to_current_member/2, + force_vhost_queues_shrink_member_to_current_member/1, force_all_queues_shrink_member_to_current_member/0]). %% for backwards compatibility @@ -1973,8 +1974,17 @@ force_shrink_member_to_current_member(VHost, Name) -> {error, not_found} end. +force_vhost_queues_shrink_member_to_current_member(VHost) when is_binary(VHost) -> + rabbit_log:warning("Disaster recovery procedure: shrinking all quorum queues in vhost ~tp to a single node cluster", [VHost]), + ListQQs = fun() -> rabbit_amqqueue:list(VHost) end, + force_all_queues_shrink_member_to_current_member(ListQQs). + force_all_queues_shrink_member_to_current_member() -> rabbit_log:warning("Disaster recovery procedure: shrinking all quorum queues to a single node cluster"), + ListQQs = fun() -> rabbit_amqqueue:list() end, + force_all_queues_shrink_member_to_current_member(ListQQs). + +force_all_queues_shrink_member_to_current_member(ListQQFun) when is_function(ListQQFun) -> Node = node(), _ = [begin QName = amqqueue:get_name(Q), @@ -1989,7 +1999,7 @@ force_all_queues_shrink_member_to_current_member() -> end, _ = rabbit_amqqueue:update(QName, Fun), _ = [ra:force_delete_server(?RA_SYSTEM, {RaName, N}) || N <- OtherNodes] - end || Q <- rabbit_amqqueue:list(), amqqueue:get_type(Q) == ?MODULE], + end || Q <- ListQQFun(), amqqueue:get_type(Q) == ?MODULE], rabbit_log:warning("Disaster recovery procedure: shrinking finished"), ok. From c9d97e61de062ca05b060b7ab327837e54e226d2 Mon Sep 17 00:00:00 2001 From: Ayanda Dube Date: Wed, 2 Oct 2024 12:28:18 +0100 Subject: [PATCH 4/5] Add test for QQ force_vhost_queues_shrink_member_to_current_member/1 (cherry picked from commit de0c0dbd89b7c278a1145833cbeda6b7d3de34eb) --- deps/rabbit/test/quorum_queue_SUITE.erl | 69 ++++++++++++++++++++++++- 1 file changed, 68 insertions(+), 1 deletion(-) diff --git a/deps/rabbit/test/quorum_queue_SUITE.erl b/deps/rabbit/test/quorum_queue_SUITE.erl index 775555eac2cb..deaf095409d9 100644 --- a/deps/rabbit/test/quorum_queue_SUITE.erl +++ b/deps/rabbit/test/quorum_queue_SUITE.erl @@ -94,7 +94,8 @@ groups() -> single_active_consumer_priority_take_over, single_active_consumer_priority, force_shrink_member_to_current_member, - force_all_queues_shrink_member_to_current_member + force_all_queues_shrink_member_to_current_member, + force_vhost_queues_shrink_member_to_current_member ] ++ all_tests()}, {cluster_size_5, [], [start_queue, @@ -1233,6 +1234,72 @@ force_all_queues_shrink_member_to_current_member(Config) -> ?assertEqual(3, length(Nodes0)) end || Q <- QQs]. +force_vhost_queues_shrink_member_to_current_member(Config) -> + [Server0, Server1, Server2] = + rabbit_ct_broker_helpers:get_node_configs(Config, nodename), + + Ch0 = rabbit_ct_client_helpers:open_channel(Config, Server0), + QQ = ?config(queue_name, Config), + AQ = ?config(alt_queue_name, Config), + ?assertEqual({'queue.declare_ok', QQ, 0, 0}, + declare(Ch0, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])), + ?assertEqual({'queue.declare_ok', AQ, 0, 0}, + declare(Ch0, AQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])), + + QQs = [QQ, AQ], + + VHost1 = <<"/">>, + VHost2 = <<"another-vhost">>, + VHosts = [VHost1, VHost2], + + User = ?config(rmq_username, Config), + ok = rabbit_ct_broker_helpers:add_vhost(Config, Server0, VHost2, User), + ok = rabbit_ct_broker_helpers:set_full_permissions(Config, User, VHost2), + Conn1 = rabbit_ct_client_helpers:open_unmanaged_connection(Config, Server0, VHost2), + {ok, Ch1} = amqp_connection:open_channel(Conn1), + ?assertEqual({'queue.declare_ok', QQ, 0, 0}, + declare(Ch1, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])), + ?assertEqual({'queue.declare_ok', AQ, 0, 0}, + declare(Ch1, AQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}])), + + [rabbit_ct_client_helpers:publish(Ch, Q, 3) || Q <- QQs, Ch <- [Ch0, Ch1]], + + [begin + QQRes = rabbit_misc:r(VHost, queue, Q), + {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]), + wait_for_messages_ready([Server0], RaName, 3), + {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]), + #{nodes := Nodes0} = amqqueue:get_type_state(Q0), + ?assertEqual(3, length(Nodes0)) + end || Q <- QQs, VHost <- VHosts], + + rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_quorum_queue, + force_vhost_queues_shrink_member_to_current_member, [VHost2]), + + [begin + QQRes = rabbit_misc:r(VHost, queue, Q), + {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]), + wait_for_messages_ready([Server0], RaName, 3), + {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]), + #{nodes := Nodes0} = amqqueue:get_type_state(Q0), + case VHost of + VHost1 -> ?assertEqual(3, length(Nodes0)); + VHost2 -> ?assertEqual(1, length(Nodes0)) + end + end || Q <- QQs, VHost <- VHosts], + + %% grow queues back to all nodes in VHost2 only + [rpc:call(Server0, rabbit_quorum_queue, grow, [S, VHost2, <<".*">>, all]) || S <- [Server1, Server2]], + + [begin + QQRes = rabbit_misc:r(VHost, queue, Q), + {ok, RaName} = rpc:call(Server0, rabbit_queue_type_util, qname_to_internal_name, [QQRes]), + wait_for_messages_ready([Server0], RaName, 3), + {ok, Q0} = rpc:call(Server0, rabbit_amqqueue, lookup, [Q, VHost]), + #{nodes := Nodes0} = amqqueue:get_type_state(Q0), + ?assertEqual(3, length(Nodes0)) + end || Q <- QQs, VHost <- VHosts]. + priority_queue_fifo(Config) -> %% testing: if hi priority messages are published before lo priority %% messages they are always consumed first (fifo) From 16170d093bac1df4766ba36e17c50f281ba2466c Mon Sep 17 00:00:00 2001 From: Ayanda Dube Date: Thu, 3 Oct 2024 10:57:11 +0100 Subject: [PATCH 5/5] Update QQ force-shrink logging (cherry picked from commit dd5ec3ccc0715ac47c2dc2f82191263bfc860204) --- deps/rabbit/src/rabbit_quorum_queue.erl | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl index 2b29759b3940..6d4eb2cae820 100644 --- a/deps/rabbit/src/rabbit_quorum_queue.erl +++ b/deps/rabbit/src/rabbit_quorum_queue.erl @@ -1953,9 +1953,10 @@ notify_decorators(QName, F, A) -> is_stateful() -> true. force_shrink_member_to_current_member(VHost, Name) -> - rabbit_log:warning("Disaster recovery procedure: shrinking ~p queue at vhost ~p to a single node cluster", [Name, VHost]), Node = node(), QName = rabbit_misc:r(VHost, queue, Name), + QNameFmt = rabbit_misc:rs(QName), + rabbit_log:warning("Shrinking ~ts to a single node: ~ts", [QNameFmt, Node]), case rabbit_amqqueue:lookup(QName) of {ok, Q} when ?is_amqqueue(Q) -> {RaName, _} = amqqueue:get_pid(Q), @@ -1968,19 +1969,19 @@ force_shrink_member_to_current_member(VHost, Name) -> end, _ = rabbit_amqqueue:update(QName, Fun), _ = [ra:force_delete_server(?RA_SYSTEM, {RaName, N}) || N <- OtherNodes], - rabbit_log:warning("Disaster recovery procedure: shrinking finished"); + rabbit_log:warning("Shrinking ~ts finished", [QNameFmt]); _ -> - rabbit_log:warning("Disaster recovery procedure: shrinking failed, queue ~p not found at vhost ~p", [Name, VHost]), + rabbit_log:warning("Shrinking failed, ~ts not found", [QNameFmt]), {error, not_found} end. force_vhost_queues_shrink_member_to_current_member(VHost) when is_binary(VHost) -> - rabbit_log:warning("Disaster recovery procedure: shrinking all quorum queues in vhost ~tp to a single node cluster", [VHost]), + rabbit_log:warning("Shrinking all quorum queues in vhost '~ts' to a single node: ~ts", [VHost, node()]), ListQQs = fun() -> rabbit_amqqueue:list(VHost) end, force_all_queues_shrink_member_to_current_member(ListQQs). force_all_queues_shrink_member_to_current_member() -> - rabbit_log:warning("Disaster recovery procedure: shrinking all quorum queues to a single node cluster"), + rabbit_log:warning("Shrinking all quorum queues to a single node: ~ts", [node()]), ListQQs = fun() -> rabbit_amqqueue:list() end, force_all_queues_shrink_member_to_current_member(ListQQs). @@ -1990,7 +1991,7 @@ force_all_queues_shrink_member_to_current_member(ListQQFun) when is_function(Lis QName = amqqueue:get_name(Q), {RaName, _} = amqqueue:get_pid(Q), OtherNodes = lists:delete(Node, get_nodes(Q)), - rabbit_log:warning("Disaster recovery procedure: shrinking queue ~p", [QName]), + rabbit_log:warning("Shrinking queue ~ts to a single node: ~ts", [rabbit_misc:rs(QName), Node]), ok = ra_server_proc:force_shrink_members_to_current_member({RaName, Node}), Fun = fun (QQ) -> TS0 = amqqueue:get_type_state(QQ), @@ -2000,7 +2001,7 @@ force_all_queues_shrink_member_to_current_member(ListQQFun) when is_function(Lis _ = rabbit_amqqueue:update(QName, Fun), _ = [ra:force_delete_server(?RA_SYSTEM, {RaName, N}) || N <- OtherNodes] end || Q <- ListQQFun(), amqqueue:get_type(Q) == ?MODULE], - rabbit_log:warning("Disaster recovery procedure: shrinking finished"), + rabbit_log:warning("Shrinking finished"), ok. is_minority(All, Up) ->