Merge pull request #2349 from rabbitmq/rabbitmq-server-2321

Node drain (maintenance mode) operations

2 files changed, 310 insertions, 8 deletions

diff --git a/test/maintenance_mode_SUITE.erl b/test/maintenance_mode_SUITE.erl
new file mode 100644
index 0000000000..6ec3f61ad5
--- /dev/null
+++ b/test/maintenance_mode_SUITE.erl
@@ -0,0 +1,227 @@
+%% This Source Code Form is subject to the terms of the Mozilla Public
+%% License, v. 2.0. If a copy of the MPL was not distributed with this
+%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
+%%
+%% Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+%%
+
+-module(maintenance_mode_SUITE).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("amqp_client/include/amqp_client.hrl").
+-include_lib("eunit/include/eunit.hrl").
+
+-compile(export_all).
+
+all() ->
+    [
+      {group, cluster_size_3}
+    ].
+
+groups() ->
+    [
+      {cluster_size_3, [], [
+          maintenance_mode_status,
+          listener_suspension_status,
+          client_connection_closure,
+          classic_mirrored_queue_leadership_transfer,
+          quorum_queue_leadership_transfer
+        ]}
+    ].
+
+%% -------------------------------------------------------------------
+%% Setup and teardown.
+%% -------------------------------------------------------------------
+
+init_per_suite(Config) ->
+    rabbit_ct_helpers:log_environment(),
+    rabbit_ct_helpers:run_setup_steps(Config).
+
+end_per_suite(Config) ->
+    rabbit_ct_helpers:run_teardown_steps(Config).
+
+init_per_group(cluster_size_3, Config) ->
+    rabbit_ct_helpers:set_config(Config, [
+        {rmq_nodes_count, 3}
+      ]).
+
+end_per_group(_, Config) ->
+    Config.
+
+init_per_testcase(Testcase, Config) ->
+    rabbit_ct_helpers:testcase_started(Config, Testcase),
+    ClusterSize = ?config(rmq_nodes_count, Config),
+    TestNumber = rabbit_ct_helpers:testcase_number(Config, ?MODULE, Testcase),
+    Config1 = rabbit_ct_helpers:set_config(Config, [
+        {rmq_nodes_clustered, true},
+        {rmq_nodename_suffix, Testcase},
+        {tcp_ports_base, {skip_n_nodes, TestNumber * ClusterSize}}
+      ]),
+    rabbit_ct_helpers:run_steps(Config1,
+      rabbit_ct_broker_helpers:setup_steps() ++
+      rabbit_ct_client_helpers:setup_steps() ++ [
+        fun rabbit_ct_broker_helpers:set_ha_policy_all/1
+      ]).
+
+end_per_testcase(Testcase, Config) ->
+    Config1 = rabbit_ct_helpers:run_steps(Config,
+      rabbit_ct_client_helpers:teardown_steps() ++
+      rabbit_ct_broker_helpers:teardown_steps()),
+    rabbit_ct_helpers:testcase_finished(Config1, Testcase).
+
+%% -------------------------------------------------------------------
+%% Test Cases
+%% -------------------------------------------------------------------
+
+maintenance_mode_status(Config) ->
+    Nodes = [A, B, C] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    
+    [begin
+         ?assertNot(rabbit_ct_broker_helpers:is_being_drained_local_read(Config, Node)),
+         ?assertNot(rabbit_ct_broker_helpers:is_being_drained_consistent_read(Config, Node))
+     end || Node <- Nodes],
+    
+    [begin
+        [begin
+             ?assertNot(rabbit_ct_broker_helpers:is_being_drained_consistent_read(Config, TargetNode, NodeToCheck))
+        end || NodeToCheck <- Nodes]
+     end || TargetNode <- Nodes],
+
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, B),
+    rabbit_ct_helpers:await_condition(
+        fun () -> rabbit_ct_broker_helpers:is_being_drained_local_read(Config, B) end,
+        10000),
+    
+    [begin
+         ?assert(rabbit_ct_broker_helpers:is_being_drained_consistent_read(Config, TargetNode, B))
+     end || TargetNode <- Nodes],
+    
+    ?assertEqual(
+        lists:usort([A, C]),
+        lists:usort(rabbit_ct_broker_helpers:rpc(Config, B,
+                        rabbit_maintenance, primary_replica_transfer_candidate_nodes, []))),
+    
+    rabbit_ct_broker_helpers:unmark_as_being_drained(Config, B),
+    rabbit_ct_helpers:await_condition(
+        fun () -> not rabbit_ct_broker_helpers:is_being_drained_local_read(Config, B) end,
+        10000),
+    
+    [begin
+         ?assertNot(rabbit_ct_broker_helpers:is_being_drained_local_read(Config, TargetNode, B)),
+         ?assertNot(rabbit_ct_broker_helpers:is_being_drained_consistent_read(Config, TargetNode, B))
+     end || TargetNode <- Nodes],
+    
+    ?assertEqual(
+        lists:usort([A, C]),
+        lists:usort(rabbit_ct_broker_helpers:rpc(Config, B,
+                        rabbit_maintenance, primary_replica_transfer_candidate_nodes, []))),
+    
+    ok.
+
+
+listener_suspension_status(Config) ->
+    Nodes = [A | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    ct:pal("Picked node ~s for maintenance tests...", [A]),
+    
+    rabbit_ct_helpers:await_condition(
+        fun () -> not rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
+
+    [begin
+         ?assertNot(rabbit_ct_broker_helpers:is_being_drained_consistent_read(Config, Node))
+     end || Node <- Nodes],
+
+    Conn1 = rabbit_ct_client_helpers:open_connection(Config, A),
+    ?assert(is_pid(Conn1)),
+    rabbit_ct_client_helpers:close_connection(Conn1),
+
+    rabbit_ct_broker_helpers:drain_node(Config, A),
+    rabbit_ct_helpers:await_condition(
+        fun () -> rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
+
+    ?assertEqual({error, econnrefused}, rabbit_ct_client_helpers:open_unmanaged_connection(Config, A)),
+
+    rabbit_ct_broker_helpers:revive_node(Config, A),
+    rabbit_ct_helpers:await_condition(
+        fun () -> not rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
+
+    Conn3 = rabbit_ct_client_helpers:open_connection(Config, A),
+    ?assert(is_pid(Conn3)),
+    rabbit_ct_client_helpers:close_connection(Conn3),
+
+    ok.
+
+
+client_connection_closure(Config) ->
+    [A | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    ct:pal("Picked node ~s for maintenance tests...", [A]),
+
+    rabbit_ct_helpers:await_condition(
+        fun () -> not rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
+
+    Conn1 = rabbit_ct_client_helpers:open_connection(Config, A),
+    ?assert(is_pid(Conn1)),
+    ?assertEqual(1, length(rabbit_ct_broker_helpers:rpc(Config, A, rabbit_networking, local_connections, []))),
+
+    rabbit_ct_broker_helpers:drain_node(Config, A),
+    ?assertEqual(0, length(rabbit_ct_broker_helpers:rpc(Config, A, rabbit_networking, local_connections, []))),
+
+    rabbit_ct_broker_helpers:revive_node(Config, A).
+
+
+classic_mirrored_queue_leadership_transfer(Config) ->
+    [A | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    ct:pal("Picked node ~s for maintenance tests...", [A]),
+
+    rabbit_ct_helpers:await_condition(
+        fun () -> not rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
+
+    PolicyPattern = <<"^cq.mirrored">>,
+    rabbit_ct_broker_helpers:set_ha_policy(Config, A, PolicyPattern, <<"all">>),
+
+    Conn = rabbit_ct_client_helpers:open_connection(Config, A),
+    {ok, Ch} = amqp_connection:open_channel(Conn),
+    QName = <<"cq.mirrored.1">>,
+    amqp_channel:call(Ch, #'queue.declare'{queue = QName, durable = true}),
+
+    ?assertEqual(1, length(rabbit_ct_broker_helpers:rpc(Config, A, rabbit_amqqueue, list_local, [<<"/">>]))),
+
+    rabbit_ct_broker_helpers:drain_node(Config, A),
+    rabbit_ct_helpers:await_condition(
+        fun () -> rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
+
+    ?assertEqual(0, length(rabbit_ct_broker_helpers:rpc(Config, A, rabbit_amqqueue, list_local, [<<"/">>]))),
+
+    rabbit_ct_broker_helpers:revive_node(Config, A),
+    %% rabbit_ct_broker_helpers:set_ha_policy/4 uses pattern for policy name
+    rabbit_ct_broker_helpers:clear_policy(Config, A, PolicyPattern).
+
+quorum_queue_leadership_transfer(Config) ->
+    [A | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    ct:pal("Picked node ~s for maintenance tests...", [A]),
+
+    rabbit_ct_helpers:await_condition(
+        fun () -> not rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
+
+    Conn = rabbit_ct_client_helpers:open_connection(Config, A),
+    {ok, Ch} = amqp_connection:open_channel(Conn),
+    QName = <<"qq.1">>,
+    amqp_channel:call(Ch, #'queue.declare'{queue = QName, durable = true, arguments = [
+        {<<"x-queue-type">>, longstr, <<"quorum">>}
+    ]}),
+    
+    %% we cannot assert on the number of local leaders here: declaring a QQ on node A
+    %% does not guarantee that the leader will be hosted on node A
+    
+    rabbit_ct_broker_helpers:drain_node(Config, A),
+    rabbit_ct_helpers:await_condition(
+        fun () -> rabbit_ct_broker_helpers:is_being_drained_local_read(Config, A) end, 10000),
+    
+    %% quorum queue leader election is asynchronous
+    rabbit_ct_helpers:await_condition(
+        fun () ->
+            LocalLeaders = rabbit_ct_broker_helpers:rpc(Config, A,
+                            rabbit_amqqueue, list_local_leaders, []),
+            length(LocalLeaders) =:= 0
+        end, 20000),
+
+    rabbit_ct_broker_helpers:revive_node(Config, A).
diff --git a/test/queue_master_location_SUITE.erl b/test/queue_master_location_SUITE.erl
index c8ea773682..6a346c5618 100644
--- a/test/queue_master_location_SUITE.erl
+++ b/test/queue_master_location_SUITE.erl
@@ -34,7 +34,8 @@
 
 all() ->
     [
-      {group, cluster_size_3}
+      {group, cluster_size_3},
+      {group, maintenance_mode}
     ].
 
 groups() ->
@@ -51,11 +52,19 @@ groups() ->
           calculate_min_master_with_bindings,
           calculate_random,
           calculate_client_local
-        ]}
+        ]},
+        
+      {maintenance_mode, [], [
+          declare_with_min_masters_and_some_nodes_under_maintenance,
+          declare_with_min_masters_and_all_nodes_under_maintenance,
+          
+          declare_with_random_and_some_nodes_under_maintenance,
+          declare_with_random_and_all_nodes_under_maintenance
+      ]}
     ].
 
 %% -------------------------------------------------------------------
-%% Testsuite setup/teardown.
+%% Test suite setup/teardown
 %% -------------------------------------------------------------------
 
 init_per_suite(Config) ->
@@ -67,7 +76,12 @@ end_per_suite(Config) ->
 
 init_per_group(cluster_size_3, Config) ->
     rabbit_ct_helpers:set_config(Config, [
-        {rmq_nodes_count, 3} %% Replaced with a list of node names later.
+        %% Replaced with a list of node names later
+        {rmq_nodes_count, 3}
+      ]);
+init_per_group(maintenance_mode, Config) ->
+    rabbit_ct_helpers:set_config(Config, [
+        {rmq_nodes_count, 3}
       ]).
 
 end_per_group(_, Config) ->
@@ -98,7 +112,7 @@ end_per_testcase(Testcase, Config) ->
     rabbit_ct_helpers:testcase_finished(Config1, Testcase).
 
 %% -------------------------------------------------------------------
-%% Testcases.
+%% Test cases
 %% -------------------------------------------------------------------
 
 %%
@@ -199,12 +213,71 @@ declare_config(Config) ->
     setup_test_environment(Config),
     set_location_config(Config, <<"min-masters">>),
     QueueName = rabbit_misc:r(<<"/">>, queue, Q = <<"qm.test">>),
-    declare(Config, QueueName, false, false, _Args=[], none),
+    declare(Config, QueueName, false, false, _Args = [], none),
     verify_min_master(Config, Q),
     unset_location_config(Config),
     ok.
 
 %%
+%% Maintenance mode effects
+%%
+
+declare_with_min_masters_and_some_nodes_under_maintenance(Config) ->
+    set_location_policy(Config, ?POLICY, <<"min-masters">>),
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, 0),
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, 1),
+    
+    QName = <<"qm.tests.min_masters.maintenance.case1">>,
+    Resource = rabbit_misc:r(<<"/">>, queue, QName),
+    Record = declare(Config, Resource, false, false, _Args = [], none),
+    %% the only node that's not being drained
+    ?assertEqual(rabbit_ct_broker_helpers:get_node_config(Config, 2, nodename),
+                 node(amqqueue:get_pid(Record))),
+
+    rabbit_ct_broker_helpers:unmark_as_being_drained(Config, 0),
+    rabbit_ct_broker_helpers:unmark_as_being_drained(Config, 1).
+
+declare_with_min_masters_and_all_nodes_under_maintenance(Config) ->
+    declare_with_all_nodes_under_maintenance(Config, <<"min-masters">>).
+
+declare_with_random_and_some_nodes_under_maintenance(Config) ->
+    set_location_policy(Config, ?POLICY, <<"random">>),
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, 0),
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, 2),
+    
+    QName = <<"qm.tests.random.maintenance.case1">>,
+    Resource = rabbit_misc:r(<<"/">>, queue, QName),
+    Record = declare(Config, Resource, false, false, _Args = [], none),
+    %% the only node that's not being drained
+    ?assertEqual(rabbit_ct_broker_helpers:get_node_config(Config, 1, nodename),
+                 node(amqqueue:get_pid(Record))),
+
+    rabbit_ct_broker_helpers:unmark_as_being_drained(Config, 0),
+    rabbit_ct_broker_helpers:unmark_as_being_drained(Config, 2).
+
+declare_with_random_and_all_nodes_under_maintenance(Config) ->
+    declare_with_all_nodes_under_maintenance(Config, <<"random">>).
+
+declare_with_all_nodes_under_maintenance(Config, Locator) ->
+    set_location_policy(Config, ?POLICY, Locator),
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, 0),
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, 1),
+    rabbit_ct_broker_helpers:mark_as_being_drained(Config, 2),
+    
+    QName = rabbit_data_coercion:to_binary(
+        rabbit_misc:format("qm.tests.~s.maintenance.case2", [Locator])),
+    Resource = rabbit_misc:r(<<"/">>, queue, QName),
+    Record = declare(Config, Resource, false, false, _Args = [], none),
+    %% when queue master locator returns no node, the node that handles
+    %% the declaration method will be used as a fallback
+    ?assertEqual(rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+                 node(amqqueue:get_pid(Record))),
+
+    rabbit_ct_broker_helpers:unmark_as_being_drained(Config, 0),
+    rabbit_ct_broker_helpers:unmark_as_being_drained(Config, 1),
+    rabbit_ct_broker_helpers:unmark_as_being_drained(Config, 2).
+
+%%
 %% Test 'calculations'
 %%
 
@@ -333,8 +406,10 @@ unset_location_config(Config) ->
 
 declare(Config, QueueName, Durable, AutoDelete, Args0, Owner) ->
     Args1 = [QueueName, Durable, AutoDelete, Args0, Owner, <<"acting-user">>],
-    {new, Queue} = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, declare, Args1),
-    Queue.
+    case rabbit_ct_broker_helpers:rpc(Config, 0, rabbit_amqqueue, declare, Args1) of
+        {new, Queue} -> Queue;
+        Other        -> Other
+    end.
 
 bind(Config, QueueName, RoutingKey) ->
     ExchangeName = rabbit_misc:r(QueueName, exchange, <<"amq.direct">>),