rabbit_db_cluster: Move generic clustering steps from rabbit_mnesia

dumbbell · dumbbell · commit 15f208eed892 · 2023-07-07T09:42:32.000+02:00
[Why]
When a single node or a cluster is initialized, we go through a few
steps which are not Mnesia-specific or even related. For instance, we
verify that both ends are compatible w.r.t. feature flags.

When we will introduce Khepri, we will have to go through the same
generic steps. Therefore, it makes sense to drive those steps from
`rabbit_db_cluster` and only call into `rabbit_mnesia` when needed.

[How]
The generic code is moved from `rabbit_mnesia` to `rabbit_db_cluster`.
diff --git a/deps/rabbit/src/rabbit_db.erl b/deps/rabbit/src/rabbit_db.erl
@@ -91,11 +91,12 @@ init_using_mnesia() ->
 %% @doc Resets the database and the node.
 
 reset() ->
-    reset_using_mnesia().
+    run(
+      #{mnesia => fun reset_using_mnesia/0}).
 
 reset_using_mnesia() ->
-    ?LOG_DEBUG(
-      "DB: resetting node",
+    ?LOG_INFO(
+      "DB: resetting node (using Mnesia)",
       #{domain => ?RMQLOG_DOMAIN_DB}),
     rabbit_mnesia:reset().
 
@@ -104,11 +105,12 @@ reset_using_mnesia() ->
 %% @doc Resets the database and the node.
 
 force_reset() ->
-    force_reset_using_mnesia().
+    run(
+      #{mnesia => fun force_reset_using_mnesia/0}).
 
 force_reset_using_mnesia() ->
     ?LOG_DEBUG(
-      "DB: resetting node forcefully",
+      "DB: resetting node forcefully (using Mnesia)",
       #{domain => ?RMQLOG_DOMAIN_DB}),
     rabbit_mnesia:force_reset().
 
@@ -120,11 +122,12 @@ force_reset_using_mnesia() ->
 %% state, like if critical members are MIA.
 
 force_load_on_next_boot() ->
-    force_load_on_next_boot_using_mnesia().
+    run(
+      #{mnesia => fun force_load_on_next_boot_using_mnesia/0}).
 
 force_load_on_next_boot_using_mnesia() ->
     ?LOG_DEBUG(
-      "DB: resetting node forcefully",
+      "DB: force load on next boot (using Mnesia)",
       #{domain => ?RMQLOG_DOMAIN_DB}),
     rabbit_mnesia:force_load_next_boot().
 
diff --git a/deps/rabbit/src/rabbit_db_cluster.erl b/deps/rabbit/src/rabbit_db_cluster.erl
@@ -45,6 +45,28 @@ ensure_feature_flags_are_in_sync(Nodes, NodeIsVirgin) ->
         {error, Reason} -> throw({error, {incompatible_feature_flags, Reason}})
     end.
 
+-spec can_join(RemoteNode) -> Ret when
+      RemoteNode :: node(),
+      Ret :: Ok | Error,
+      Ok :: {ok, [node()]} | {ok, already_member},
+      Error :: {error, {inconsistent_cluster, string()}}.
+
+can_join(RemoteNode) ->
+    ?LOG_INFO(
+       "DB: checking if `~ts` can join cluster using remote node `~ts`",
+       [node(), RemoteNode],
+       #{domain => ?RMQLOG_DOMAIN_DB}),
+    case rabbit_feature_flags:check_node_compatibility(RemoteNode) of
+        ok ->
+            rabbit_db:run(
+              #{mnesia => fun() -> can_join_using_mnesia(RemoteNode) end});
+        Error ->
+            Error
+    end.
+
+can_join_using_mnesia(RemoteNode) ->
+    rabbit_mnesia:can_join_cluster(RemoteNode).
+
 -spec join(RemoteNode, NodeType) -> Ret when
       RemoteNode :: node(),
       NodeType :: rabbit_db_cluster:node_type(),
@@ -55,21 +77,41 @@ ensure_feature_flags_are_in_sync(Nodes, NodeIsVirgin) ->
 
 join(RemoteNode, NodeType)
   when is_atom(RemoteNode) andalso ?IS_NODE_TYPE(NodeType) ->
-    ?LOG_DEBUG(
-      "DB: joining cluster using remote node `~ts`", [RemoteNode],
-      #{domain => ?RMQLOG_DOMAIN_DB}),
-    join_using_mnesia(RemoteNode, NodeType).
+    case can_join(RemoteNode) of
+        {ok, ClusterNodes} when is_list(ClusterNodes) ->
+            rabbit_db:reset(),
+
+            ?LOG_INFO(
+               "DB: joining cluster using remote nodes:~n~tp", [ClusterNodes],
+               #{domain => ?RMQLOG_DOMAIN_DB}),
+            Ret = rabbit_db:run(
+                    #{mnesia =>
+                      fun() -> join_using_mnesia(ClusterNodes, NodeType) end}),
+            case Ret of
+                ok ->
+                    rabbit_node_monitor:notify_joined_cluster(),
+                    ok;
+                {error, _} = Error ->
+                    Error
+            end;
+        {ok, already_member} ->
+            {ok, already_member};
+        {error, _} = Error ->
+            Error
+    end.
 
-join_using_mnesia(RemoteNode, NodeType) ->
-    rabbit_mnesia:join_cluster(RemoteNode, NodeType).
+join_using_mnesia(ClusterNodes, NodeType) when is_list(ClusterNodes) ->
+    rabbit_mnesia:join_cluster(ClusterNodes, NodeType).
 
 -spec forget_member(Node, RemoveWhenOffline) -> ok when
       Node :: node(),
       RemoveWhenOffline :: boolean().
 %% @doc Removes `Node' from the cluster.
 
 forget_member(Node, RemoveWhenOffline) ->
-    forget_member_using_mnesia(Node, RemoveWhenOffline).
+    rabbit_db:run(
+      #{mnesia =>
+        fun() -> forget_member_using_mnesia(Node, RemoveWhenOffline) end}).
 
 forget_member_using_mnesia(Node, RemoveWhenOffline) ->
     rabbit_mnesia:forget_cluster_node(Node, RemoveWhenOffline).
@@ -85,7 +127,8 @@ forget_member_using_mnesia(Node, RemoveWhenOffline) ->
 %% Node types may not all be valid with all databases.
 
 change_node_type(NodeType) ->
-    change_node_type_using_mnesia(NodeType).
+    rabbit_db:run(
+      #{mnesia => fun() -> change_node_type_using_mnesia(NodeType) end}).
 
 change_node_type_using_mnesia(NodeType) ->
     rabbit_mnesia:change_cluster_node_type(NodeType).
@@ -99,7 +142,8 @@ change_node_type_using_mnesia(NodeType) ->
 %% @doc Indicates if this node is clustered with other nodes or not.
 
 is_clustered() ->
-    is_clustered_using_mnesia().
+    rabbit_db:run(
+      #{mnesia => fun is_clustered_using_mnesia/0}).
 
 is_clustered_using_mnesia() ->
     rabbit_mnesia:is_clustered().
@@ -109,7 +153,8 @@ is_clustered_using_mnesia() ->
 %% @doc Returns the list of cluster members.
 
 members() ->
-    members_using_mnesia().
+    rabbit_db:run(
+      #{mnesia => fun members_using_mnesia/0}).
 
 members_using_mnesia() ->
     case rabbit_mnesia:is_running() andalso rabbit_table:is_present() of
@@ -137,7 +182,8 @@ members_using_mnesia() ->
 %% @private
 
 disc_members() ->
-    disc_members_using_mnesia().
+    rabbit_db:run(
+      #{mnesia => fun disc_members_using_mnesia/0}).
 
 disc_members_using_mnesia() ->
     rabbit_mnesia:cluster_nodes(disc).
@@ -149,7 +195,8 @@ disc_members_using_mnesia() ->
 %% Node types may not all be relevant with all databases.
 
 node_type() ->
-    node_type_using_mnesia().
+    rabbit_db:run(
+      #{mnesia => fun node_type_using_mnesia/0}).
 
 node_type_using_mnesia() ->
     rabbit_mnesia:node_type().
@@ -177,7 +224,8 @@ check_compatibility_using_mnesia(RemoteNode) ->
 %% @doc Ensures the cluster is consistent.
 
 check_consistency() ->
-    check_consistency_using_mnesia().
+    rabbit_db:run(
+      #{mnesia => fun check_consistency_using_mnesia/0}).
 
 check_consistency_using_mnesia() ->
     rabbit_mnesia:check_cluster_consistency().
@@ -190,7 +238,8 @@ check_consistency_using_mnesia() ->
 %% command.
 
 cli_cluster_status() ->
-    cli_cluster_status_using_mnesia().
+    rabbit_db:run(
+      #{mnesia => fun cli_cluster_status_using_mnesia/0}).
 
 cli_cluster_status_using_mnesia() ->
     rabbit_mnesia:status().
diff --git a/deps/rabbit/src/rabbit_mnesia.erl b/deps/rabbit/src/rabbit_mnesia.erl
@@ -13,6 +13,7 @@
 
 -export([%% Main interface
          init/0,
+         can_join_cluster/1,
          join_cluster/2,
          reset/0,
          force_reset/0,
@@ -153,10 +154,10 @@ create_cluster_callback(RemoteNode, NodeType) ->
 %% all in the same cluster, we simply pick the first online node and
 %% we cluster to its cluster.
 
--spec join_cluster(node(), rabbit_db_cluster:node_type())
-                        -> ok | {ok, already_member} | {error, {inconsistent_cluster, string()}}.
+-spec can_join_cluster(node())
+                        -> {ok, [node()]} | {ok, already_member} | {error, {inconsistent_cluster, string()}}.
 
-join_cluster(DiscoveryNode, NodeType) ->
+can_join_cluster(DiscoveryNode) ->
     ensure_mnesia_not_running(),
     ensure_mnesia_dir(),
     case is_only_clustered_disc_node() of
@@ -167,28 +168,8 @@ join_cluster(DiscoveryNode, NodeType) ->
     case rabbit_nodes:me_in_nodes(ClusterNodes) of
         false ->
             case check_cluster_consistency(DiscoveryNode, false) of
-                {ok, _} ->
-                    %% reset the node. this simplifies things and it
-                    %% will be needed in this case - we're joining a new
-                    %% cluster with new nodes which are not in synch
-                    %% with the current node. It also lifts the burden
-                    %% of resetting the node from the user.
-                    reset_gracefully(),
-
-                    NodeType1 = case is_node_type_permitted(NodeType) of
-                                    false -> disc;
-                                    true  -> NodeType
-                                end,
-
-                    %% Join the cluster
-                    rabbit_log:info("Clustering with ~tp as ~tp node",
-                                    [ClusterNodes, NodeType1]),
-                    ok = init_db_with_mnesia(ClusterNodes, NodeType1,
-                                             true, true, _Retry = true),
-                    rabbit_node_monitor:notify_joined_cluster(),
-                    ok;
-                {error, Reason} ->
-                    {error, Reason}
+                {ok, _} -> {ok, ClusterNodes};
+                Error   -> Error
             end;
         true ->
             %% DiscoveryNode thinks that we are part of a cluster, but
@@ -204,6 +185,38 @@ join_cluster(DiscoveryNode, NodeType) ->
             end
     end.
 
+-spec join_cluster
+([node()], rabbit_db_cluster:node_type()) ->
+    ok | {error, any()};
+(node(), rabbit_db_cluster:node_type()) ->
+    ok | {ok, already_member} | {error, {inconsistent_cluster, string()}}.
+
+join_cluster(ClusterNodes, NodeType) when is_list(ClusterNodes) ->
+    %% Join the cluster.
+    NodeType1 = case is_node_type_permitted(NodeType) of
+                    false -> disc;
+                    true  -> NodeType
+                end,
+    rabbit_log:info("Clustering with ~tp as ~tp node",
+                    [ClusterNodes, NodeType1]),
+    ok = init_db_with_mnesia(ClusterNodes, NodeType1,
+                             true, true, _Retry = true),
+    rabbit_node_monitor:notify_joined_cluster(),
+    ok;
+join_cluster(DiscoveryNode, NodeType) when is_atom(DiscoveryNode) ->
+    %% Code to remain compatible with `change_cluster_node_type/1' and older
+    %% CLI.
+    case can_join_cluster(DiscoveryNode) of
+        {ok, ClusterNodes} when is_list(ClusterNodes) ->
+            ok = reset_gracefully(),
+            ok = join_cluster(ClusterNodes, NodeType),
+            ok;
+        {ok, already_member} ->
+            {ok, already_member};
+        Error ->
+            Error
+    end.
+
 %% return node to its virgin state, where it is not member of any
 %% cluster, has no cluster configuration, no local database, and no
 %% persisted messages
diff --git a/deps/rabbitmq_cli/test/ctl/join_cluster_command_test.exs b/deps/rabbitmq_cli/test/ctl/join_cluster_command_test.exs
@@ -93,7 +93,7 @@ defmodule JoinClusterCommandTest do
     stop_rabbitmq_app()
 
     assert match?(
-             {:badrpc_multi, _, [_]},
+             {:error, {:aborted_feature_flags_compat_check, {:error, {:erpc, :noconnection}}}},
              @command.run([:jake@thedog], context[:opts])
            )
 

Original file line number	Diff line number	Diff line change
`@@ -93,7 +93,7 @@ defmodule JoinClusterCommandTest do`
`93`	`93`	`stop_rabbitmq_app()`
`94`	`94`
`95`	`95`	`assert match?(`
`96`		`- {:badrpc_multi, _, [_]},`
	`96`	`+ {:error, {:aborted_feature_flags_compat_check, {:error, {:erpc, :noconnection}}}},`
`97`	`97`	`@command.run([:jake@thedog], context[:opts])`
`98`	`98`	`)`
`99`	`99`