Skip to content

Commit 669f0f2

Browse files
michaelklishinmichaelklishin
authored
Merge pull request #9443 from rabbitmq/mergify/bp/v3.12.x/pr-9442
forget_cluster_node: handle errors while shrinking streams and/or QQs (backport #9442)
2 parents 6468d78 + 78b49ae commit 669f0f2

File tree

6 files changed

+318
-6
lines changed

6 files changed

+318
-6
lines changed

‎deps/rabbit/BUILD.bazel

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -332,6 +332,9 @@ rabbitmq_integration_suite(
332332
rabbitmq_integration_suite(
333333
name = "clustering_management_SUITE",
334334
size = "large",
335+
additional_beam = [
336+
":test_clustering_utils_beam",
337+
],
335338
flaky = True,
336339
shard_count = 18,
337340
sharding_method = "case",
@@ -432,6 +435,9 @@ rabbitmq_integration_suite(
432435
rabbitmq_integration_suite(
433436
name = "feature_flags_SUITE",
434437
size = "large",
438+
additional_beam = [
439+
":test_clustering_utils_beam",
440+
],
435441
flaky = True,
436442
shard_count = 5,
437443
runtime_deps = [
@@ -1120,6 +1126,14 @@ rabbitmq_integration_suite(
11201126
],
11211127
)
11221128

1129+
rabbitmq_integration_suite(
1130+
name = "cli_forget_cluster_node_SUITE",
1131+
size = "medium",
1132+
additional_beam = [
1133+
":test_clustering_utils_beam",
1134+
],
1135+
)
1136+
11231137
assert_suites()
11241138

11251139
filegroup(
@@ -1215,6 +1229,7 @@ eunit(
12151229
":test_rabbit_ha_test_producer_beam",
12161230
":test_test_util_beam",
12171231
":test_test_rabbit_event_handler_beam",
1232+
":test_clustering_utils_beam",
12181233
],
12191234
target = ":test_erlang_app",
12201235
)

‎deps/rabbit/app.bzl

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1951,3 +1951,21 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
19511951
erlc_opts = "//:test_erlc_opts",
19521952
deps = ["//deps/amqp_client:erlang_app"],
19531953
)
1954+
1955+
erlang_bytecode(
1956+
name = "cli_forget_cluster_node_SUITE_beam_files",
1957+
testonly = True,
1958+
srcs = ["test/cli_forget_cluster_node_SUITE.erl"],
1959+
outs = ["test/cli_forget_cluster_node_SUITE.beam"],
1960+
app_name = "rabbit",
1961+
erlc_opts = "//:test_erlc_opts",
1962+
deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
1963+
)
1964+
erlang_bytecode(
1965+
name = "test_clustering_utils_beam",
1966+
testonly = True,
1967+
srcs = ["test/clustering_utils.erl"],
1968+
outs = ["test/clustering_utils.beam"],
1969+
app_name = "rabbit",
1970+
erlc_opts = "//:test_erlc_opts",
1971+
)

‎deps/rabbit/src/rabbit_stream_queue.erl

Lines changed: 28 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -42,7 +42,8 @@
4242
-export([set_retention_policy/3]).
4343
-export([restart_stream/3,
4444
add_replica/3,
45-
delete_replica/3]).
45+
delete_replica/3,
46+
delete_all_replicas/1]).
4647
-export([format_osiris_event/2]).
4748
-export([update_stream_conf/2]).
4849
-export([readers/1]).
@@ -861,6 +862,32 @@ delete_replica(VHost, Name, Node) ->
861862
E
862863
end.
863864

865+
delete_all_replicas(Node) ->
866+
rabbit_log:info("Asked to remove all stream replicas from node ~ts", [Node]),
867+
Streams = rabbit_amqqueue:list_by_type(stream),
868+
Errors =
869+
lists:foldl(fun(Q, Acc) ->
870+
QName = amqqueue:get_name(Q),
871+
rabbit_log:info("~ts: removing replica on node ~w",
872+
[rabbit_misc:rs(QName), Node]),
873+
#{name := StreamId} = amqqueue:get_type_state(Q),
874+
{ok, Reply, _} = rabbit_stream_coordinator:delete_replica(StreamId, Node),
875+
case Reply of
876+
ok ->
877+
Acc;
878+
Err ->
879+
rabbit_log:warning("~ts: failed to remove replica on node ~w, error: ~w",
880+
[rabbit_misc:rs(QName), Node, Err]),
881+
[{QName, Err} | Acc]
882+
end
883+
end, [], Streams),
884+
case Errors of
885+
[] ->
886+
ok;
887+
_ ->
888+
{error, Errors}
889+
end.
890+
864891
make_stream_conf(Q) ->
865892
QName = amqqueue:get_name(Q),
866893
Name = stream_name(QName),

‎deps/rabbit/test/cli_forget_cluster_node_SUITE.erl

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
%% This Source Code Form is subject to the terms of the Mozilla Public
2+
%% License, v. 2.0. If a copy of the MPL was not distributed with this
3+
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
4+
%%
5+
%% Copyright (c) 2007-2023 VMware, Inc. or its affiliates. All rights reserved.
6+
%%
7+
8+
-module(cli_forget_cluster_node_SUITE).
9+
10+
-include_lib("common_test/include/ct.hrl").
11+
-include_lib("amqp_client/include/amqp_client.hrl").
12+
-include_lib("eunit/include/eunit.hrl").
13+
-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
14+
15+
-compile(export_all).
16+
17+
-import(clustering_utils, [
18+
assert_cluster_status/2,
19+
assert_clustered/1,
20+
assert_not_clustered/1
21+
]).
22+
23+
all() ->
24+
[
25+
{group, cluster_size_3}
26+
].
27+
28+
groups() ->
29+
[
30+
{cluster_size_3, [], [
31+
forget_cluster_node_with_quorum_queues,
32+
forget_cluster_node_with_last_quorum_member
33+
]}
34+
].
35+
36+
suite() ->
37+
[
38+
%% If a testcase hangs, no need to wait for 30 minutes.
39+
{timetrap, {minutes, 5}}
40+
].
41+
42+
%% -------------------------------------------------------------------
43+
%% Testsuite setup/teardown.
44+
%% -------------------------------------------------------------------
45+
46+
init_per_suite(Config) ->
47+
rabbit_ct_helpers:log_environment(),
48+
Config1 = rabbit_ct_helpers:merge_app_env(
49+
Config, {rabbit, [
50+
{mnesia_table_loading_retry_limit, 2},
51+
{mnesia_table_loading_retry_timeout,1000}
52+
]}),
53+
rabbit_ct_helpers:run_setup_steps(Config1).
54+
55+
end_per_suite(Config) ->
56+
rabbit_ct_helpers:run_teardown_steps(Config).
57+
58+
init_per_group(cluster_size_3, Config) ->
59+
rabbit_ct_helpers:set_config(Config, [{rmq_nodes_count, 3},
60+
{rmq_nodes_clustered, true}]).
61+
62+
end_per_group(_, Config) ->
63+
Config.
64+
65+
init_per_testcase(Testcase, Config) ->
66+
rabbit_ct_helpers:testcase_started(Config, Testcase),
67+
ClusterSize = ?config(rmq_nodes_count, Config),
68+
TestNumber = rabbit_ct_helpers:testcase_number(Config, ?MODULE, Testcase),
69+
Config1 = rabbit_ct_helpers:set_config(Config, [
70+
{rmq_nodename_suffix, Testcase},
71+
{tcp_ports_base, {skip_n_nodes, TestNumber * ClusterSize}},
72+
{keep_pid_file_on_exit, true}
73+
]),
74+
rabbit_ct_helpers:run_steps(Config1,
75+
rabbit_ct_broker_helpers:setup_steps() ++
76+
rabbit_ct_client_helpers:setup_steps()).
77+
78+
end_per_testcase(Testcase, Config) ->
79+
Config1 = rabbit_ct_helpers:run_steps(Config,
80+
rabbit_ct_client_helpers:teardown_steps() ++
81+
rabbit_ct_broker_helpers:teardown_steps()),
82+
rabbit_ct_helpers:testcase_finished(Config1, Testcase).
83+
84+
%% -------------------------------------------------------------------
85+
%% Test cases
86+
%% -------------------------------------------------------------------
87+
forget_cluster_node_with_quorum_queues(Config) ->
88+
[Rabbit, Hare, Bunny] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
89+
90+
assert_clustered([Rabbit, Hare, Bunny]),
91+
92+
Ch = rabbit_ct_client_helpers:open_channel(Config, Rabbit),
93+
QQ1 = <<"quorum-queue-1">>,
94+
QQ2 = <<"quorum-queue-2">>,
95+
declare(Ch, QQ1, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
96+
declare(Ch, QQ2, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
97+
98+
?awaitMatch(Members when length(Members) == 3, get_quorum_members(Rabbit, QQ1), 30000),
99+
?awaitMatch(Members when length(Members) == 3, get_quorum_members(Rabbit, QQ2), 30000),
100+
101+
?assertEqual(ok, rabbit_control_helper:command(stop_app, Bunny)),
102+
?assertEqual(ok, forget_cluster_node(Rabbit, Bunny)),
103+
assert_cluster_status({[Rabbit, Hare], [Rabbit, Hare], [Rabbit, Hare]},
104+
[Rabbit, Hare]),
105+
?awaitMatch(Members when length(Members) == 2, get_quorum_members(Rabbit, QQ1), 30000),
106+
?awaitMatch(Members when length(Members) == 2, get_quorum_members(Rabbit, QQ2), 30000).
107+
108+
forget_cluster_node_with_last_quorum_member(Config) ->
109+
[Rabbit, Hare, Bunny] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
110+
111+
assert_clustered([Rabbit, Hare, Bunny]),
112+
113+
Ch = rabbit_ct_client_helpers:open_channel(Config, Bunny),
114+
QQ1 = <<"quorum-queue-1">>,
115+
QQ2 = <<"quorum-queue-2">>,
116+
declare(Ch, QQ1, [{<<"x-queue-type">>, longstr, <<"quorum">>},
117+
{<<"x-quorum-initial-group-size">>, long, 1}]),
118+
declare(Ch, QQ2, [{<<"x-queue-type">>, longstr, <<"quorum">>},
119+
{<<"x-quorum-initial-group-size">>, long, 1}]),
120+
121+
?awaitMatch(Members when length(Members) == 1, get_quorum_members(Rabbit, QQ1), 30000),
122+
?awaitMatch(Members when length(Members) == 1, get_quorum_members(Rabbit, QQ2), 30000),
123+
124+
?assertEqual(ok, rabbit_control_helper:command(stop_app, Bunny)),
125+
?assertMatch({error, 69, _}, forget_cluster_node(Rabbit, Bunny)),
126+
assert_cluster_status({[Rabbit, Hare], [Rabbit, Hare], [Rabbit, Hare]},
127+
[Rabbit, Hare]),
128+
?awaitMatch(Members when length(Members) == 1, get_quorum_members(Rabbit, QQ1), 30000),
129+
?awaitMatch(Members when length(Members) == 1, get_quorum_members(Rabbit, QQ2), 30000).
130+
131+
forget_cluster_node(Node, Removee) ->
132+
rabbit_control_helper:command(forget_cluster_node, Node, [atom_to_list(Removee)],
133+
[]).
134+
135+
get_quorum_members(Server, Q) ->
136+
Info = rpc:call(Server, rabbit_quorum_queue, infos, [rabbit_misc:r(<<"/">>, queue, Q)]),
137+
proplists:get_value(members, Info).
138+
139+
declare(Ch, Q, Args) ->
140+
amqp_channel:call(Ch, #'queue.declare'{queue = Q,
141+
durable = true,
142+
auto_delete = false,
143+
arguments = Args}).

‎deps/rabbit/test/clustering_utils.erl

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
%% This Source Code Form is subject to the terms of the Mozilla Public
2+
%% License, v. 2.0. If a copy of the MPL was not distributed with this
3+
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
4+
%%
5+
%% Copyright (c) 2018-2023 VMware, Inc. or its affiliates. All rights reserved.
6+
%%
7+
-module(clustering_utils).
8+
9+
-export([
10+
assert_cluster_status/2,
11+
assert_clustered/1,
12+
assert_not_clustered/1
13+
]).
14+
15+
-define(LOOP_RECURSION_DELAY, 100).
16+
17+
assert_cluster_status({All, Disc, Running}, Nodes) ->
18+
assert_cluster_status({All, Running, All, Disc, Running}, Nodes);
19+
assert_cluster_status(Status0, Nodes) ->
20+
Status = sort_cluster_status(Status0),
21+
AllNodes = case Status of
22+
{undef, undef, All, _, _} ->
23+
%% Support mixed-version clusters
24+
All;
25+
{All, _, _, _, _} ->
26+
All
27+
end,
28+
wait_for_cluster_status(Status, AllNodes, Nodes).
29+
30+
wait_for_cluster_status(Status, AllNodes, Nodes) ->
31+
Max = 10000 / ?LOOP_RECURSION_DELAY,
32+
wait_for_cluster_status(0, Max, Status, AllNodes, Nodes).
33+
34+
wait_for_cluster_status(N, Max, Status, _AllNodes, Nodes) when N >= Max ->
35+
erlang:error({cluster_status_max_tries_failed,
36+
[{nodes, Nodes},
37+
{expected_status, Status},
38+
{max_tried, Max},
39+
{status, sort_cluster_status(cluster_status(hd(Nodes)))}]});
40+
wait_for_cluster_status(N, Max, Status, AllNodes, Nodes) ->
41+
case lists:all(fun (Node) ->
42+
verify_status_equal(Node, Status, AllNodes)
43+
end, Nodes) of
44+
true -> ok;
45+
false -> timer:sleep(?LOOP_RECURSION_DELAY),
46+
wait_for_cluster_status(N + 1, Max, Status, AllNodes, Nodes)
47+
end.
48+
49+
verify_status_equal(Node, Status, AllNodes) ->
50+
NodeStatus = sort_cluster_status(cluster_status(Node)),
51+
IsClustered = case rpc:call(Node, rabbit_db_cluster, is_clustered, []) of
52+
{badrpc, {'EXIT', {undef, _}}} ->
53+
rpc:call(Node, rabbit_mnesia, is_clustered, []);
54+
Ret ->
55+
Ret
56+
end,
57+
(AllNodes =/= [Node]) =:= IsClustered andalso equal(Status, NodeStatus).
58+
59+
equal({_, _, A, B, C}, {undef, undef, A, B, C}) ->
60+
true;
61+
equal({_, _, _, _, _}, {undef, undef, _, _, _}) ->
62+
false;
63+
equal(Status0, Status1) ->
64+
Status0 == Status1.
65+
66+
cluster_status(Node) ->
67+
AllMembers = rpc:call(Node, rabbit_nodes, list_members, []),
68+
RunningMembers = rpc:call(Node, rabbit_nodes, list_running, []),
69+
70+
AllDbNodes = case rpc:call(Node, rabbit_db_cluster, members, []) of
71+
{badrpc, {'EXIT', {undef, _}}} ->
72+
rpc:call(Node, rabbit_mnesia, cluster_nodes, [all]);
73+
Ret ->
74+
Ret
75+
end,
76+
DiscDbNodes = rpc:call(Node, rabbit_mnesia, cluster_nodes, [disc]),
77+
RunningDbNodes = rpc:call(Node, rabbit_mnesia, cluster_nodes, [running]),
78+
79+
{AllMembers,
80+
RunningMembers,
81+
AllDbNodes,
82+
DiscDbNodes,
83+
RunningDbNodes}.
84+
85+
sort_cluster_status({{badrpc, {'EXIT', {undef, _}}}, {badrpc, {'EXIT', {undef, _}}}, AllM, DiscM, RunningM}) ->
86+
{undef, undef, lists:sort(AllM), lists:sort(DiscM), lists:sort(RunningM)};
87+
sort_cluster_status({All, Running, AllM, DiscM, RunningM}) ->
88+
{lists:sort(All), lists:sort(Running), lists:sort(AllM), lists:sort(DiscM), lists:sort(RunningM)}.
89+
90+
assert_clustered(Nodes) ->
91+
assert_cluster_status({Nodes, Nodes, Nodes, Nodes, Nodes}, Nodes).
92+
93+
assert_not_clustered(Node) ->
94+
assert_cluster_status({[Node], [Node], [Node], [Node], [Node]}, [Node]).
95+

‎deps/rabbitmq_cli/lib/rabbitmq/cli/ctl/commands/forget_cluster_node_command.ex

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -60,13 +60,27 @@ defmodule RabbitMQ.CLI.Ctl.Commands.ForgetClusterNodeCommand do
6060
error
6161

6262
:ok ->
63-
case :rabbit_misc.rpc_call(node_name, :rabbit_quorum_queue, :shrink_all, [atom_name]) do
64-
{:error, _} ->
65-
{:error,
66-
"RabbitMQ failed to shrink some of the quorum queues on node #{node_to_remove}"}
63+
qq_shrink_result =
64+
:rabbit_misc.rpc_call(node_name, :rabbit_quorum_queue, :shrink_all, [atom_name])
6765

68-
_ ->
66+
is_ok_fun = fn
67+
{_, {:ok, _}} -> true
68+
{_, {:error, _, _}} -> false
69+
end
70+
71+
case Enum.empty?(qq_shrink_result) do
72+
true ->
6973
:ok
74+
75+
false ->
76+
case Enum.any?(qq_shrink_result, is_ok_fun) do
77+
false ->
78+
{:error,
79+
"RabbitMQ failed to shrink some of the quorum queues on node #{node_to_remove}"}
80+
81+
true ->
82+
:ok
83+
end
7084
end
7185

7286
other ->

0 commit comments

Comments
 (0)