rabbit_khepri: Improve compatibility with rabbit_mnesia's behavior

Author: dumbbell

deps/rabbit/src/rabbit_khepri.erl

@@ -121,7 +121,7 @@ add_member(JoiningNode, [_ | _] = Cluster) ->
             ?LOG_DEBUG(
                "Khepri clustering: Node ~p is already a member of cluster ~p",
                [JoiningNode, Cluster]),
-            ok
+            {ok, already_member}
     end.
 
 pick_node_in_cluster(Cluster) when is_list(Cluster) ->
@@ -146,36 +146,49 @@ do_join(RemoteNode) when RemoteNode =/= node() ->
     ok = setup(),
     khepri:info(?RA_CLUSTER_NAME),
 
-    %% We don't verify the cluster membership before adding this node to the
-    %% remote cluster because such a check would not be atomic: the membership
-    %% could well change between the check and the actual join.
-
-    ?LOG_DEBUG(
-       "Adding this node (~p) to Khepri cluster \"~s\" through "
-       "node ~p",
-       [ThisNode, ?RA_CLUSTER_NAME, RemoteNode],
-       #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-
     %% Ensure the remote node is reachable before we add it.
     pong = net_adm:ping(RemoteNode),
 
-    %% If the remote node to add is running RabbitMQ, we need to put
-    %% it in maintenance mode at least. We remember that state to
-    %% revive the node only if it was fully running before this code.
-    IsRunning = rabbit:is_running(ThisNode),
-    AlreadyBeingDrained =
-    rabbit_maintenance:is_being_drained_consistent_read(ThisNode),
-    NeedToRevive = IsRunning andalso not AlreadyBeingDrained,
-    maybe_drain_node(IsRunning),
+    %% We verify the cluster membership before adding `ThisNode' to
+    %% `RemoteNode''s cluster. We do it mostly to keep the same behavior as
+    %% what we do with Mnesia. Otherwise, the interest is limited given the
+    %% check and the actual join are not atomic.
+
+    ClusteredNodes = rabbit_misc:rpc_call(
+                       RemoteNode, rabbit_khepri, locally_known_nodes, []),
+    case lists:member(ThisNode, ClusteredNodes) of
+        false ->
+            ?LOG_DEBUG(
+               "Adding this node (~p) to Khepri cluster \"~s\" through "
+               "node ~p",
+               [ThisNode, ?RA_CLUSTER_NAME, RemoteNode],
+               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+
+            %% If the remote node to add is running RabbitMQ, we need to put it
+            %% in maintenance mode at least. We remember that state to revive
+            %% the node only if it was fully running before this code.
+            IsRunning = rabbit:is_running(ThisNode),
+            AlreadyBeingDrained =
+            rabbit_maintenance:is_being_drained_consistent_read(ThisNode),
+            NeedToRevive = IsRunning andalso not AlreadyBeingDrained,
+            maybe_drain_node(IsRunning),
 
-    %% Joining a cluster includes a reset of the local Khepri store.
-    Ret = khepri_cluster:join(?RA_CLUSTER_NAME, RemoteNode),
+            %% Joining a cluster includes a reset of the local Khepri store.
+            Ret = khepri_cluster:join(?RA_CLUSTER_NAME, RemoteNode),
 
-    %% Revive the remote node if it was running and not under
-    %% maintenance before we changed the cluster membership.
-    maybe_revive_node(NeedToRevive),
+            %% Revive the remote node if it was running and not under
+            %% maintenance before we changed the cluster membership.
+            maybe_revive_node(NeedToRevive),
 
-    Ret.
+            Ret;
+        true ->
+            ?LOG_DEBUG(
+               "This node (~p) is already part of the Khepri cluster \"~s\" "
+               "like node ~p",
+               [ThisNode, ?RA_CLUSTER_NAME, RemoteNode],
+               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+            {ok, already_member}
+    end.
 
 maybe_drain_node(true) ->
     ok = rabbit_maintenance:drain();
@@ -238,7 +251,12 @@ remove_member(NodeToRemove) when NodeToRemove =/= node() ->
                "Removing remote node ~s from Khepri cluster \"~s\"",
                [NodeToRemove, ?RA_CLUSTER_NAME],
                #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-            ok = ensure_ra_system_started(),
+
+            %% We need the Khepri store to run on the node to remove, to be
+            %% able to reset it.
+            ok = rabbit_misc:rpc_call(
+                    NodeToRemove, ?MODULE, setup, []),
+
             Ret = rabbit_misc:rpc_call(
                     NodeToRemove, khepri_cluster, reset, [?RA_CLUSTER_NAME]),
             case Ret of

deps/rabbitmq_ct_helpers/src/rabbit_ct_broker_helpers.erl

@@ -1001,7 +1001,8 @@ stop_rabbitmq_nodes(Config) ->
           fun(NodeConfig) ->
                   stop_rabbitmq_node(Config, NodeConfig)
           end),
-    find_crashes_in_logs(NodeConfigs),
+    IgnoredCrashes = ["** force_vhost_failure"],
+    find_crashes_in_logs(NodeConfigs, IgnoredCrashes),
     proplists:delete(rmq_nodes, Config).
 
 stop_rabbitmq_node(Config, NodeConfig) ->
@@ -1022,68 +1023,83 @@ stop_rabbitmq_node(Config, NodeConfig) ->
     end,
     NodeConfig.
 
-find_crashes_in_logs(NodeConfigs) ->
+find_crashes_in_logs(NodeConfigs, IgnoredCrashes) ->
     ct:pal(
       "Looking up any crash reports in the nodes' log files. If we find "
       "some, they will appear below:"),
     CrashesCount = lists:foldl(
                      fun(NodeConfig, Total) ->
-                             Count = count_crashes_in_logs(NodeConfig),
+                             Count = count_crashes_in_logs(
+                                       NodeConfig, IgnoredCrashes),
                              Total + Count
                      end, 0, NodeConfigs),
     ct:pal("Found ~b crash report(s)", [CrashesCount]),
     ?assertEqual(0, CrashesCount).
 
-count_crashes_in_logs(NodeConfig) ->
+count_crashes_in_logs(NodeConfig, IgnoredCrashes) ->
     LogLocations = ?config(log_locations, NodeConfig),
     lists:foldl(
       fun(LogLocation, Total) ->
-              Count = count_crashes_in_log(LogLocation),
+              Count = count_crashes_in_log(LogLocation, IgnoredCrashes),
               Total + Count
       end, 0, LogLocations).
 
-count_crashes_in_log(LogLocation) ->
+count_crashes_in_log(LogLocation, IgnoredCrashes) ->
     case file:read_file(LogLocation) of
-        {ok, Content} -> count_crashes_in_content(Content);
+        {ok, Content} -> count_crashes_in_content(Content, IgnoredCrashes);
         _             -> 0
     end.
 
-count_crashes_in_content(Content) ->
+count_crashes_in_content(Content, IgnoredCrashes) ->
     ReOpts = [multiline],
     Lines = re:split(Content, "^", ReOpts),
-    count_gen_server_terminations(Lines).
+    count_gen_server_terminations(Lines, IgnoredCrashes).
 
-count_gen_server_terminations(Lines) ->
-    count_gen_server_terminations(Lines, 0).
+count_gen_server_terminations(Lines, IgnoredCrashes) ->
+    count_gen_server_terminations(Lines, 0, IgnoredCrashes).
 
-count_gen_server_terminations([Line | Rest], Count) ->
+count_gen_server_terminations([Line | Rest], Count, IgnoredCrashes) ->
     ReOpts = [{capture, all_but_first, list}],
     Ret = re:run(
             Line,
             "(<[0-9.]+> )[*]{2} Generic server .+ terminating$",
             ReOpts),
     case Ret of
         {match, [Prefix]} ->
-            capture_gen_server_termination(Rest, Prefix, [Line], Count + 1);
+            capture_gen_server_termination(
+              Rest, Prefix, [Line], Count, IgnoredCrashes);
         nomatch ->
-            count_gen_server_terminations(Rest, Count)
+            count_gen_server_terminations(Rest, Count, IgnoredCrashes)
     end;
-count_gen_server_terminations([], Count) ->
+count_gen_server_terminations([], Count, _IgnoredCrashes) ->
     Count.
 
-capture_gen_server_termination([Line | Rest] = Lines, Prefix, Acc, Count) ->
-    ReOpts = [{capture, none}],
-    Ret = re:run(Line, Prefix ++ "( |\\*|$)", ReOpts),
+capture_gen_server_termination(
+  [Line | Rest] = Lines, Prefix, Acc, Count, IgnoredCrashes) ->
+    ReOpts = [{capture, all_but_first, list}],
+    Ret = re:run(Line, Prefix ++ "( .*|\\*.*|)$", ReOpts),
     case Ret of
-        match ->
-            capture_gen_server_termination(Rest, Prefix, [Line | Acc], Count);
+        {match, [Suffix]} ->
+            case lists:member(Suffix, IgnoredCrashes) of
+                false ->
+                    capture_gen_server_termination(
+                      Rest, Prefix, [Line | Acc], Count, IgnoredCrashes);
+                true ->
+                    count_gen_server_terminations(
+                      Lines, Count, IgnoredCrashes)
+            end;
         nomatch ->
-            ct:pal("gen_server termination:~n~n~s", [lists:reverse(Acc)]),
-            count_gen_server_terminations(Lines, Count)
+            found_gen_server_termiation(
+              lists:reverse(Acc), Lines, Count, IgnoredCrashes)
     end;
-capture_gen_server_termination([] = Rest, _Prefix, Acc, Count) ->
-    ct:pal("gen_server termination:~n~n~s", [lists:reverse(Acc)]),
-    count_gen_server_terminations(Rest, Count).
+capture_gen_server_termination(
+  [] = Rest, _Prefix, Acc, Count, IgnoredCrashes) ->
+    found_gen_server_termiation(
+      lists:reverse(Acc), Rest, Count, IgnoredCrashes).
+
+found_gen_server_termiation(Message, Lines, Count, IgnoredCrashes) ->
+    ct:pal("gen_server termination:~n~n~s", [Message]),
+    count_gen_server_terminations(Lines, Count + 1, IgnoredCrashes).
 
 %% -------------------------------------------------------------------
 %% Helpers for partition simulation
@@ -1332,6 +1348,8 @@ delete_vhost(Config, Node, VHost) ->
 delete_vhost(Config, Node, VHost, Username) ->
     catch rpc(Config, Node, rabbit_vhost, delete, [VHost, Username]).
 
+-define(FORCE_VHOST_FAILURE_REASON, force_vhost_failure).
+
 force_vhost_failure(Config, VHost) -> force_vhost_failure(Config, 0, VHost).
 
 force_vhost_failure(Config, Node, VHost) ->
@@ -1345,7 +1363,8 @@ force_vhost_failure(Config, Node, VHost, Attempts) ->
             try
                 MessageStorePid = get_message_store_pid(Config, Node, VHost),
                 rpc(Config, Node,
-                    erlang, exit, [MessageStorePid, force_vhost_failure]),
+                    erlang, exit,
+                    [MessageStorePid, ?FORCE_VHOST_FAILURE_REASON]),
                 %% Give it a time to fail
                 timer:sleep(300),
                 force_vhost_failure(Config, Node, VHost, Attempts - 1)