Merge pull request #6738 from rabbitmq/coordinator-quorum-check

Check if stream coordinator is quorum critial

Author: mkuratczyk

deps/rabbit/src/rabbit_upgrade_preparation.erl

@@ -32,15 +32,39 @@ await_online_synchronised_mirrors(Timeout) ->
 %% Implementation
 %%
 
+online_members(Component) ->
+    lists:filter(fun erlang:is_pid/1,
+                 rabbit_misc:append_rpc_all_nodes(rabbit_nodes:all_running(),
+                                                  erlang, whereis, [Component])).
+
+endangered_critical_components() ->
+    CriticalComponents = [rabbit_stream_coordinator],
+    Nodes = rabbit_nodes:all(),
+    lists:filter(fun (Component) ->
+                         NumAlive = length(online_members(Component)),
+                         ServerIds = lists:zip(lists:duplicate(length(Nodes), Component), Nodes),
+                         case ra:members(ServerIds) of
+                             {error, _E} ->
+                                 %% we've asked all nodes about it; if we didn't get an answer,
+                                 %% the component is probably not running at all
+                                 %% (eg. rabbit_stream_coordinator is only started when the
+                                 %% first straem is declared)
+                                 false;
+                             {ok, Members, _Leader} ->
+                                 NumAlive =< (length(Members) div 2) + 1
+                         end
+                 end,
+                 CriticalComponents).
+
 do_await_safe_online_quorum(0) ->
     false;
 do_await_safe_online_quorum(IterationsLeft) ->
     EndangeredQueues = lists:append(
                          rabbit_quorum_queue:list_with_minimum_quorum(),
                          rabbit_stream_queue:list_with_minimum_quorum()),
-    case EndangeredQueues of
-        []  -> true;
-        List when is_list(List) ->
+    case EndangeredQueues =:= [] andalso endangered_critical_components() =:= [] of
+        true -> true;
+        false ->
             timer:sleep(?SAMPLING_INTERVAL),
             do_await_safe_online_quorum(IterationsLeft - 1)
     end.
@@ -69,4 +93,10 @@ list_with_minimum_quorum_for_cli() ->
            <<"virtual_host">> => amqqueue:get_vhost(Q),
            <<"type">> => amqqueue:get_type(Q)
           }
-     end || Q <- EndangeredQueues].
+     end || Q <- EndangeredQueues] ++
+    [#{
+           <<"readable_name">> => C,
+           <<"name">> => C,
+           <<"virtual_host">> => "-",
+           <<"type">> => process
+      } || C <- endangered_critical_components()].

deps/rabbit/test/rabbit_stream_queue_SUITE.erl

@@ -360,9 +360,9 @@ declare_queue(Config) ->
     %% Test declare an existing queue
     ?assertEqual({'queue.declare_ok', Q, 0, 0},
                 declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}])),
-    
+
     ?assertMatch([_], find_queue_info(Config, [])),
-    
+
     %% Test declare an existing queue with different arguments
     ?assertExit(_, declare(Ch, Q, [])),
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
@@ -1075,7 +1075,7 @@ consume_and_nack(Config) ->
                                                       multiple     = false,
                                                       requeue      = true}),
             %% Nack will throw a not implemented exception. As it is a cast operation,
-            %% we'll detect the conneciton/channel closure on the next call. 
+            %% we'll detect the conneciton/channel closure on the next call.
             %% Let's try to redeclare and see what happens
             ?assertExit({{shutdown, {connection_closing, {server_initiated_close, 540, _}}}, _},
                         declare(Ch1, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}]))
@@ -1299,7 +1299,7 @@ consume_and_reject(Config) ->
             ok = amqp_channel:cast(Ch1, #'basic.reject'{delivery_tag = DeliveryTag,
                                                       requeue      = true}),
             %% Reject will throw a not implemented exception. As it is a cast operation,
-            %% we'll detect the conneciton/channel closure on the next call. 
+            %% we'll detect the conneciton/channel closure on the next call.
             %% Let's try to redeclare and see what happens
             ?assertExit({{shutdown, {connection_closing, {server_initiated_close, 540, _}}}, _},
                         declare(Ch1, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}]))
@@ -1710,7 +1710,12 @@ max_length_bytes(Config) ->
 
     Payload = << <<"1">> || _ <- lists:seq(1, 100) >>,
 
-    publish_confirm(Ch, Q, [Payload || _ <- lists:seq(1, 500)]), %% 100 bytes/msg * 500 = 50000 bytes
+    %% 100 bytes/msg * 500 = 50000 bytes
+    publish_confirm(Ch, Q, [Payload || _ <- lists:seq(1, 100)]),
+    publish_confirm(Ch, Q, [Payload || _ <- lists:seq(1, 100)]),
+    publish_confirm(Ch, Q, [Payload || _ <- lists:seq(1, 100)]),
+    publish_confirm(Ch, Q, [Payload || _ <- lists:seq(1, 100)]),
+    publish_confirm(Ch, Q, [Payload || _ <- lists:seq(1, 100)]),
     ensure_retention_applied(Config, Server),
 
     %% We don't yet have reliable metrics, as the committed offset doesn't work

deps/rabbit/test/upgrade_preparation_SUITE.erl

@@ -26,6 +26,9 @@ groups() ->
      ]},
      {stream, [], [
          await_quorum_plus_one_stream
+     ]},
+     {stream_coordinator, [], [
+         await_quorum_plus_one_stream_coordinator
      ]}
     ].
 
@@ -107,6 +110,23 @@ await_quorum_plus_one_stream(Config) ->
     ok = rabbit_ct_broker_helpers:start_node(Config, B),
     ?assert(await_quorum_plus_one(Config, 0)).
 
+await_quorum_plus_one_stream_coordinator(Config) ->
+    catch delete_queues(),
+    [A, B, _C] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, A),
+    declare(Ch, <<"st.1">>, [{<<"x-queue-type">>, longstr, <<"stream">>}]),
+    timer:sleep(100),
+    ?assert(await_quorum_plus_one(Config, 0)),
+    delete(Ch, <<"st.1">>),
+    %% no queues/streams beyond this point
+
+    ok = rabbit_ct_broker_helpers:stop_node(Config, B),
+    %% this should fail because the corrdinator has only 2 running nodes
+    ?assertNot(await_quorum_plus_one(Config, 0)),
+
+    ok = rabbit_ct_broker_helpers:start_node(Config, B),
+    ?assert(await_quorum_plus_one(Config, 0)).
+
 %%
 %% Implementation
 %%
@@ -120,6 +140,9 @@ declare(Ch, Q, Args) ->
                                            auto_delete = false,
                                            arguments = Args}).
 
+delete(Ch, Q) ->
+    amqp_channel:call(Ch, #'queue.delete'{queue = Q}).
+
 delete_queues() ->
     [rabbit_amqqueue:delete(Q, false, false, <<"tests">>) || Q <- rabbit_amqqueue:list()].
 

deps/rabbitmq_cli/lib/rabbitmq/cli/upgrade/commands/await_online_quorum_plus_one_command.ex

@@ -57,7 +57,7 @@ defmodule RabbitMQ.CLI.Upgrade.Commands.AwaitOnlineQuorumPlusOneCommand do
 
           false ->
             {:error,
-             "time is up, no quorum + 1 online replicas came online for at least some quorum queues"}
+             "time is up, no quorum + 1 online replicas came online for at least some quorum queues or streams"}
         end
 
       other ->
@@ -103,11 +103,11 @@ defmodule RabbitMQ.CLI.Upgrade.Commands.AwaitOnlineQuorumPlusOneCommand do
   def help_section, do: :upgrade
 
   def description() do
-    "Waits for all quorum queues to have an above minimum online quorum. " <>
-      "This makes sure that no queues would lose their quorum if the target node is shut down"
+    "Waits for all quorum queues and streams to have an above minimum online quorum. " <>
+      "This makes sure that no queues/streams would lose their quorum if the target node is shut down"
   end
 
   def banner([], %{timeout: timeout}) do
-    "Will wait for a quorum + 1 of nodes to be online for all quorum queues for #{round(timeout / 1000)} seconds..."
+    "Will wait for a quorum + 1 of nodes to be online for all quorum queues and streams for #{round(timeout / 1000)} seconds..."
   end
 end