WIP; Fix Khepri clustering, take #3

dumbbell · dumbbell · commit d008e4bbaf34 · 2022-04-21T17:27:59.000+02:00
diff --git a/deps/rabbit/src/rabbit_core_ff.erl b/deps/rabbit/src/rabbit_core_ff.erl
@@ -364,20 +364,23 @@ add_nodes_to_khepri_cluster(_FeatureName, _KhepriCluster, []) ->
     ok.
 
 add_node_to_khepri_cluster(FeatureName, KhepriCluster, Node) ->
+    ?assertNotEqual([], KhepriCluster),
     case lists:member(Node, KhepriCluster) of
         true ->
             ?LOG_DEBUG(
-               "Feature flag `~s`:   this node (~s) is already a member of "
+               "Feature flag `~s`:   node ~p is already a member of "
                "the largest cluster: ~p",
                [FeatureName, Node, KhepriCluster]),
             ok;
         false ->
-            [KhepriNode | _] = KhepriCluster,
             ?LOG_DEBUG(
-               "Feature flag `~s`:   adding this node (~s) to the largest "
+               "Feature flag `~s`:   adding node ~p to the largest "
                "Khepri cluster found among Mnesia nodes: ~p",
                [FeatureName, Node, KhepriCluster]),
-            ok = rabbit_khepri:join_cluster(KhepriNode)
+            case rabbit_khepri:add_member(Node, KhepriCluster) of
+                ok                   -> ok;
+                {ok, already_member} -> ok
+            end
     end.
 
 find_largest_khepri_cluster(FeatureName) ->
diff --git a/deps/rabbit/src/rabbit_khepri.erl b/deps/rabbit/src/rabbit_khepri.erl
@@ -15,8 +15,7 @@
 
 -export([setup/0,
          setup/1,
-         join_cluster/1,
-         add_member/1,
+         add_member/2,
          remove_member/1,
          members/0,
          locally_known_members/0,
@@ -56,7 +55,8 @@
          is_enabled/1,
          nodes_if_khepri_enabled/0,
          try_mnesia_or_khepri/2]).
--export([priv_reset/0]).
+-export([do_add_member/1,
+         priv_reset/0]).
 
 -ifdef(TEST).
 -export([force_metadata_store/1,
@@ -126,37 +126,43 @@ wait_for_leader(Fun, Timeout) ->
             Error
     end.
 
-join_cluster(RemoteNode) when RemoteNode =/= node() ->
-    ThisNode = node(),
-    ?LOG_INFO(
-      "Khepri clustering: Attempt to add node ~p to Khepri cluster through "
-      "node ~p",
-      [ThisNode, RemoteNode],
-      #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+add_member(JoiningNode, JoinedNode) when JoinedNode =:= node() ->
+    Ret = do_add_member(JoiningNode),
+    post_add_member(JoiningNode, JoinedNode, Ret);
+add_member(JoiningNode, JoinedNode) when is_atom(JoinedNode) ->
     Ret = rabbit_misc:rpc_call(
-            RemoteNode, rabbit_khepri, add_member, [ThisNode]),
-    case Ret of
-        ok ->
-            ?LOG_INFO(
-              "Khepri clustering: Node ~p successfully added to Khepri "
-              "cluster through node ~p",
-              [ThisNode, RemoteNode],
-              #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-            ok;
-        Error ->
+            JoinedNode, rabbit_khepri, do_add_member, [JoiningNode]),
+    post_add_member(JoiningNode, JoinedNode, Ret);
+add_member(JoiningNode, [_ | _] = Cluster) ->
+    case lists:member(JoiningNode, Cluster) of
+        false ->
+            JoinedNode = pick_node_in_cluster(Cluster),
             ?LOG_INFO(
-              "Khepri clustering: Failed to add node ~p to Khepri cluster "
-              "through ~p: ~p",
-              [ThisNode, RemoteNode, Error],
-              #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-            Error
-    end;
-join_cluster(ThisNode) when ThisNode =:= node() ->
-    ok.
+               "Khepri clustering: Attempt to add node ~p to cluster ~0p "
+               "through node ~p",
+               [JoiningNode, Cluster, JoinedNode],
+               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+            %% Recurse with a single node taken in the `Cluster' list.
+            add_member(JoiningNode, JoinedNode);
+        true ->
+            ?LOG_DEBUG(
+               "Khepri clustering: Node ~p is already a member of cluster ~p",
+               [JoiningNode, Cluster]),
+            ok
+    end.
+
+pick_node_in_cluster(Cluster) when is_list(Cluster) ->
+    ?assertNotEqual([], Cluster),
+    ThisNode = node(),
+    case lists:member(ThisNode, Cluster) of
+        true  -> ThisNode;
+        false -> hd(Cluster)
+    end.
 
-add_member(NewNode) when NewNode =/= node() ->
+do_add_member(NewNode) when NewNode =/= node() ->
     ?LOG_DEBUG(
-       "Trying to add node ~s to Khepri cluster \"~s\" from node ~s",
+       "Khepri clustering: Trying to add node ~p to cluster \"~s\" through "
+       "node ~p",
        [NewNode, ?RA_CLUSTER_NAME, node()],
        #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
 
@@ -169,24 +175,20 @@ add_member(NewNode) when NewNode =/= node() ->
             pong = net_adm:ping(NewNode),
 
             ?LOG_DEBUG(
-               "Resetting Khepri on remote node ~s",
+               "Khepri clustering: Resetting Khepri on remote node ~p",
                [NewNode],
                #{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.
             RemoteIsRunning = rabbit:is_running(NewNode),
             RemoteAlreadyBeingDrained =
             rabbit_maintenance:is_being_drained_consistent_read(NewNode),
             NeedToReviveRemote =
-            RemoteIsRunning andalso RemoteAlreadyBeingDrained,
-            case RemoteIsRunning of
-                true ->
-                    ok = rabbit_misc:rpc_call(
-                           NewNode, rabbit_maintenance, drain, []);
-                false ->
-                    ok
-            end,
+            RemoteIsRunning andalso not RemoteAlreadyBeingDrained,
+            maybe_drain_node(NewNode, RemoteIsRunning),
+
             Ret1 = rabbit_misc:rpc_call(
                      NewNode, rabbit_khepri, priv_reset, []),
             case Ret1 of
@@ -199,46 +201,74 @@ add_member(NewNode) when NewNode =/= node() ->
                     Ret2 = khepri:add_member(
                              ?RA_SYSTEM, ?RA_CLUSTER_NAME, ?RA_FRIENDLY_NAME,
                              NewNode),
+
                     %% Revive the remote node if it was running and not under
                     %% maintenance before we changed the cluster membership.
-                    case NeedToReviveRemote of
-                        true ->
-                            ok = rabbit_misc:rpc_call(
-                                   NewNode, rabbit_maintenance, revive, []);
-                        false ->
-                            ok
-                    end,
-                    case Ret2 of
-                        ok ->
-                            ?LOG_DEBUG(
-                               "Node ~s added to Khepri cluster \"~s\"",
-                               [NewNode, ?RA_CLUSTER_NAME],
-                               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-                            ok;
-                        {error, _} = Error ->
-                            ?LOG_ERROR(
-                               "Failed to add remote node ~s to Khepri "
-                               "cluster \"~s\": ~p",
-                               [NewNode, ?RA_CLUSTER_NAME, Error],
-                               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-                            Error
-                    end;
+                    maybe_revive_node(NewNode, NeedToReviveRemote),
+
+                    Ret2;
                 Error ->
                     ?LOG_ERROR(
-                       "Failed to reset Khepri on remote node ~s: ~p",
+                       "Khepri clustering: Failed to reset Khepri on node "
+                       "~p: ~p",
                        [NewNode, Error],
                        #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+
+                    %% Revive the remote node if it was running and not under
+                    %% maintenance before we changed the cluster membership.
+                    maybe_revive_node(NewNode, NeedToReviveRemote),
+
                     Error
             end;
         true ->
-            ?LOG_INFO(
-               "Asked to add node ~s to Khepri cluster \"~s\" but already a "
-               "member of it: ~p",
-               [NewNode, ?RA_CLUSTER_NAME, lists:sort(CurrentNodes)],
-               #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
-            ok
+            {error, {already_member, CurrentNodes}}
     end.
 
+maybe_drain_node(Node, true) ->
+    ok = rabbit_misc:rpc_call(Node, rabbit_maintenance, drain, []);
+maybe_drain_node(_Node, false) ->
+    ok.
+
+maybe_revive_node(Node, true) ->
+    ok = rabbit_misc:rpc_call(Node, rabbit_maintenance, revive, []);
+maybe_revive_node(_Node, false) ->
+    ok.
+
+post_add_member(JoiningNode, JoinedNode, ok) ->
+    ?LOG_INFO(
+       "Khepri clustering: Node ~p successfully added to cluster \"~s\" "
+       "through node ~p",
+       [JoiningNode, ?RA_CLUSTER_NAME, JoinedNode],
+       #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+    ok;
+post_add_member(
+  JoiningNode, _JoinedNode, {error, {already_member, Cluster}}) ->
+    ?LOG_INFO(
+       "Khepri clustering: Asked to add node ~p to cluster \"~s\" "
+       "but already a member of it: ~p",
+       [JoiningNode, ?RA_CLUSTER_NAME, lists:sort(Cluster)],
+       #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+    {ok, already_member};
+post_add_member(
+  JoiningNode, JoinedNode,
+  {badrpc, {'EXIT', {undef, [{rabbit_khepri, Function, _, _}]}}} = Error)
+  when Function =:= do_add_member orelse
+       Function =:= priv_reset ->
+    ?LOG_INFO(
+       "Khepri clustering: Can't add node ~p to cluster \"~s\"; "
+       "Khepri unavailable on node ~p: ~p",
+       [JoiningNode, ?RA_CLUSTER_NAME, JoinedNode, Error],
+       #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+    %% TODO: Should we return an error and let the caller decide?
+    ok;
+post_add_member(JoiningNode, JoinedNode, Error) ->
+    ?LOG_INFO(
+       "Khepri clustering: Failed to add node ~p to cluster \"~s\" "
+       "through ~p: ~p",
+       [JoiningNode, ?RA_CLUSTER_NAME, JoinedNode, Error],
+       #{domain => ?RMQLOG_DOMAIN_GLOBAL}),
+    Error.
+
 remove_member(NodeToRemove) when NodeToRemove =/= node() ->
     ?LOG_DEBUG(
        "Trying to remove node ~s from Khepri cluster \"~s\" on node ~s",
diff --git a/deps/rabbit/src/rabbit_mnesia.erl b/deps/rabbit/src/rabbit_mnesia.erl
@@ -216,8 +216,9 @@ join_discovered_peers_with_retries(TryNodes, NodeType, RetriesLeft, DelayInterva
 
 join_cluster(DiscoveryNode, NodeType) ->
     case join_mnesia_cluster(DiscoveryNode, NodeType) of
-        ok    -> join_khepri_cluster(DiscoveryNode);
-        Other -> Other
+        ok                   -> join_khepri_cluster(DiscoveryNode);
+        {ok, already_member} -> join_khepri_cluster(DiscoveryNode);
+        Other                -> Other
     end.
 
 join_mnesia_cluster(DiscoveryNode, NodeType) ->
@@ -265,16 +266,8 @@ join_mnesia_cluster(DiscoveryNode, NodeType) ->
     end.
 
 join_khepri_cluster(DiscoveryNode) ->
-    case rabbit_khepri:join_cluster(DiscoveryNode) of
-        {badrpc, {'EXIT', {undef, [{rabbit_khepri, add_member, _, _}]}}} ->
-            rabbit_log:debug(
-              "Skip Khepri cluster expansion; Khepri is unavailable on "
-              "remote node ~p",
-              [DiscoveryNode]),
-            ok;
-        Other ->
-            Other
-    end.
+    ThisNode = node(),
+    rabbit_khepri:add_member(ThisNode, [DiscoveryNode]).
 
 %% return node to its virgin state, where it is not member of any
 %% cluster, has no cluster configuration, no local database, and no
@@ -929,7 +922,17 @@ stop_mnesia() ->
     stopped = mnesia:stop(),
     ensure_mnesia_not_running().
 
-change_extra_db_nodes(ClusterNodes0, CheckOtherNodes) ->
+change_extra_db_nodes(ClusterNodes, CheckOtherNodes) ->
+    Nodes = change_extra_mnesia_nodes(ClusterNodes, CheckOtherNodes),
+    %% FIXME: Need to cluster Khepri at this point? I don't think so... I keep
+    %% the code but it does nothing (`false' condition).
+    case false andalso rabbit_khepri:is_enabled() of
+        true  -> _ = change_extra_khepri_nodes(ClusterNodes, CheckOtherNodes);
+        false -> ok
+    end,
+    Nodes.
+
+change_extra_mnesia_nodes(ClusterNodes0, CheckOtherNodes) ->
     ClusterNodes = nodes_excl_me(ClusterNodes0),
     case {mnesia:change_config(extra_db_nodes, ClusterNodes), ClusterNodes} of
         {{ok, []}, [_|_]} when CheckOtherNodes ->
@@ -939,6 +942,26 @@ change_extra_db_nodes(ClusterNodes0, CheckOtherNodes) ->
             Nodes
     end.
 
+change_extra_khepri_nodes(ClusterNodes, CheckOtherNodes) ->
+    ThisNode = node(),
+    _ = rabbit_khepri:add_member(ThisNode, ClusterNodes),
+    ActualNodes = rabbit_khepri:locally_known_nodes(),
+    case CheckOtherNodes of
+        true ->
+            UnclusteredNodes = ClusterNodes -- ActualNodes,
+            case UnclusteredNodes of
+                [] ->
+                    ActualNodes;
+                _ ->
+                    rabbit_log:error("UnclusteredNodes = ~p~n", [UnclusteredNodes]),
+                    throw({error,
+                           {failed_to_cluster_with, ClusterNodes,
+                            "Khepri could not connect to any nodes."}})
+            end;
+        false ->
+            ActualNodes
+    end.
+
 check_consistency(Node, OTP, Rabbit, ProtocolVersion) ->
     rabbit_misc:sequence_error(
       [check_mnesia_or_otp_consistency(Node, ProtocolVersion, OTP),
