Merge pull request #8502 from rabbitmq/move-clustering-steps-to-rabbit_db_cluster

rabbit_db: Further split Mnesia-specific and non-Mnesia-specific clustering steps

Author: dumbbell

deps/rabbit/BUILD.bazel

@@ -553,7 +553,7 @@ rabbitmq_suite(
 
 rabbitmq_integration_suite(
     name = "peer_discovery_classic_config_SUITE",
-    size = "medium",
+    size = "large",
     flaky = True,
 )
 

deps/rabbit/src/rabbit_db.erl

@@ -42,12 +42,23 @@ init() ->
     ?LOG_DEBUG(
        "DB: this node is virgin: ~ts", [IsVirgin],
        #{domain => ?RMQLOG_DOMAIN_DB}),
+
     ensure_dir_exists(),
-    case init_using_mnesia() of
+    rabbit_peer_discovery:log_configured_backend(),
+    rabbit_peer_discovery:maybe_init(),
+
+    pre_init(IsVirgin),
+
+    Ret = run(
+            #{mnesia => fun init_using_mnesia/0}),
+    case Ret of
         ok ->
             ?LOG_DEBUG(
                "DB: initialization successeful",
                #{domain => ?RMQLOG_DOMAIN_DB}),
+
+            post_init(IsVirgin),
+
             ok;
         Error ->
             ?LOG_DEBUG(
@@ -56,6 +67,17 @@ init() ->
             Error
     end.
 
+pre_init(IsVirgin) ->
+    Members = rabbit_db_cluster:members(),
+    OtherMembers = rabbit_nodes:nodes_excl_me(Members),
+    rabbit_db_cluster:ensure_feature_flags_are_in_sync(OtherMembers, IsVirgin).
+
+post_init(false = _IsVirgin) ->
+    rabbit_peer_discovery:maybe_register();
+post_init(true = _IsVirgin) ->
+    %% Registration handled by rabbit_peer_discovery.
+    ok.
+
 init_using_mnesia() ->
     ?LOG_DEBUG(
       "DB: initialize Mnesia",
@@ -69,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().
 
@@ -82,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().
 
@@ -98,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().
 
@@ -115,8 +140,11 @@ force_load_on_next_boot_using_mnesia() ->
 %% @see is_virgin_node/1.
 
 is_virgin_node() ->
-    ThisNode = node(),
-    is_virgin_node(ThisNode).
+    run(
+      #{mnesia => fun is_virgin_node_using_mnesia/0}).
+
+is_virgin_node_using_mnesia() ->
+    rabbit_mnesia:is_virgin_node().
 
 -spec is_virgin_node(Node) -> IsVirgin | undefined when
       Node :: node(),
@@ -129,14 +157,11 @@ is_virgin_node() ->
 %% @returns `true' if the node is virgin, `false' if it is not, or `undefined'
 %% if the given node is remote and we couldn't determine it.
 
+is_virgin_node(Node) when Node =:= node() ->
+    is_virgin_node();
 is_virgin_node(Node) when is_atom(Node) ->
-    is_virgin_node_with_mnesia(Node).
-
-is_virgin_node_with_mnesia(Node) when Node =:= node() ->
-    rabbit_mnesia:is_virgin_node();
-is_virgin_node_with_mnesia(Node) ->
     try
-        erpc:call(Node, rabbit_mnesia, is_virgin_node, [], ?TIMEOUT)
+        erpc:call(Node, ?MODULE, is_virgin_node, [], ?TIMEOUT)
     catch
         _:_ ->
             undefined
@@ -149,7 +174,8 @@ is_virgin_node_with_mnesia(Node) ->
 %% @returns the directory path.
 
 dir() ->
-    mnesia_dir().
+    run(
+      #{mnesia => fun mnesia_dir/0}).
 
 mnesia_dir() ->
     rabbit_mnesia:dir().
@@ -183,9 +209,9 @@ ensure_dir_exists() ->
 run(Funs)
   when is_map(Funs) andalso is_map_key(mnesia, Funs) ->
     #{mnesia := MnesiaFun} = Funs,
-    run_with_mnesia(MnesiaFun).
+    run_using_mnesia(MnesiaFun).
 
-run_with_mnesia(Fun) ->
+run_using_mnesia(Fun) ->
     Fun().
 
 list_in_mnesia(Table, Match) ->

deps/rabbit/src/rabbit_db_cluster.erl

@@ -11,13 +11,15 @@
 
 -include_lib("rabbit_common/include/logging.hrl").
 
--export([join/2,
+-export([ensure_feature_flags_are_in_sync/2,
+         join/2,
          forget_member/2]).
 -export([change_node_type/1]).
 -export([is_clustered/0,
          members/0,
          disc_members/0,
          node_type/0,
+         check_compatibility/1,
          check_consistency/0,
          cli_cluster_status/0]).
 
@@ -35,6 +37,36 @@
 %% Cluster formation.
 %% -------------------------------------------------------------------
 
+ensure_feature_flags_are_in_sync(Nodes, NodeIsVirgin) ->
+    Ret = rabbit_feature_flags:sync_feature_flags_with_cluster(
+            Nodes, NodeIsVirgin),
+    case Ret of
+        ok              -> ok;
+        {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(),
@@ -45,21 +77,41 @@
 
 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).
@@ -75,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).
@@ -89,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().
@@ -99,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
@@ -127,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).
@@ -139,16 +195,37 @@ 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().
 
+-spec check_compatibility(RemoteNode) -> ok | {error, Reason} when
+      RemoteNode :: node(),
+      Reason :: any().
+%% @doc Ensures the given remote node is compatible with the node calling this
+%% function.
+
+check_compatibility(RemoteNode) ->
+    case rabbit_feature_flags:check_node_compatibility(RemoteNode) of
+        ok ->
+            rabbit_db:run(
+              #{mnesia =>
+                fun() -> check_compatibility_using_mnesia(RemoteNode) end});
+        Error ->
+            Error
+    end.
+
+check_compatibility_using_mnesia(RemoteNode) ->
+    rabbit_mnesia:check_mnesia_consistency(RemoteNode).
+
 -spec check_consistency() -> ok.
 %% @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().
@@ -161,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().