rabbit_mnesia:all_clustered_nodes/0 does not return disc nodes only. Duh. But we can do better anyway: allow any disc node to do the upgrade.

2 files changed, 70 insertions, 34 deletions

diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index e63e5de220..47e68c87b4 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -35,7 +35,8 @@
 -export([ensure_mnesia_dir/0, dir/0, status/0, init/0, is_db_empty/0,
          cluster/1, force_cluster/1, reset/0, force_reset/0,
          is_clustered/0, running_clustered_nodes/0, all_clustered_nodes/0,
-         empty_ram_only_tables/0, copy_db/1, create_cluster_nodes_config/1]).
+         empty_ram_only_tables/0, copy_db/1,
+         create_cluster_nodes_config/1, read_cluster_nodes_config/0]).
 
 -export([table_names/0]).
 
@@ -71,6 +72,8 @@
 -spec(empty_ram_only_tables/0 :: () -> 'ok').
 -spec(create_tables/0 :: () -> 'ok').
 -spec(copy_db/1 :: (file:filename()) ->  rabbit_types:ok_or_error(any())).
+-spec(create_cluster_nodes_config/1 :: ([node()]) ->  'ok').
+-spec(read_cluster_nodes_config/0 :: () ->  [node()]).
 
 -endif.
 
diff --git a/src/rabbit_upgrade.erl b/src/rabbit_upgrade.erl
index 23dd416afe..dcbffd03bc 100644
--- a/src/rabbit_upgrade.erl
+++ b/src/rabbit_upgrade.erl
@@ -22,7 +22,8 @@
 -module(rabbit_upgrade).
 
 -export([maybe_upgrade_mnesia/0, maybe_upgrade/1]).
--export([read_version/0, write_version/0, desired_version/0]).
+-export([read_version/0, write_version/0, desired_version/0,
+         desired_version/1]).
 
 -include("rabbit.hrl").
 
@@ -43,6 +44,7 @@
 -spec(read_version/0 :: () -> rabbit_types:ok_or_error2(version(), any())).
 -spec(write_version/0 :: () -> 'ok').
 -spec(desired_version/0 :: () -> version()).
+-spec(desired_version/1 :: (scope()) -> [step()]).
 
 -endif.
 
@@ -52,13 +54,10 @@ maybe_upgrade_mnesia() ->
     rabbit:prepare(),
     case upgrades_required(mnesia) of
         [_|_] = Upgrades ->
-            DiscNodes = rabbit_mnesia:all_clustered_nodes(),
-            Upgrader = upgrader(DiscNodes),
-            case node() of
-                Upgrader ->
-                    primary_upgrade(Upgrades, DiscNodes);
-                _ ->
-                    non_primary_upgrade(Upgrader, DiscNodes)
+            Nodes = rabbit_mnesia:all_clustered_nodes(),
+            case am_i_upgrader(Nodes) of
+                true  -> primary_upgrade(Upgrades, Nodes);
+                false -> non_primary_upgrade(Nodes)
             end;
         [] ->
             ok;
@@ -66,12 +65,57 @@ maybe_upgrade_mnesia() ->
             ok
     end.
 
-upgrader(Nodes) ->
-    [Upgrader|_] = lists:usort(Nodes),
-    Upgrader.
+am_i_upgrader(Nodes) ->
+    Running = nodes_running(Nodes),
+    case Running of
+        [] ->
+            case am_i_disc_node() of
+                true ->
+                    true;
+                false ->
+                    die("Cluster upgrade needed but this is a ram node.~n   "
+                        "Please start any of the disc nodes first.", [])
+            end;
+        [Another|_] ->
+            ClusterVersion =
+                case rpc:call(Another,
+                              rabbit_upgrade, desired_version, [mnesia]) of
+                    {badrpc, {'EXIT', {undef, _}}} -> unknown_old_version;
+                    {badrpc, Reason}               -> {unknown, Reason};
+                    V                              -> V
+                end,
+            case desired_version(mnesia) of
+                ClusterVersion ->
+                    %% The other node(s) have upgraded already, I am not the
+                    %% upgrader
+                    false;
+                MyVersion ->
+                    %% The other node(s) are running an unexpected version.
+                    die("Cluster upgrade needed but other nodes are "
+                        "running ~p~n"
+                        "and I want ~p", [ClusterVersion, MyVersion])
+            end
+    end.
+
+am_i_disc_node() ->
+    %% The cluster config does not list all disc nodes, but it will list us
+    %% if we're one.
+    case rabbit_mnesia:read_cluster_nodes_config() of
+        []        -> true;
+        DiscNodes -> lists:member(node(), DiscNodes)
+    end.
 
-primary_upgrade(Upgrades, DiscNodes) ->
-    Others = DiscNodes -- [node()],
+die(Msg, Args) ->
+    %% We don't throw or exit here since that gets thrown
+    %% straight out into do_boot, generating an erl_crash.dump
+    %% and displaying any error message in a confusing way.
+    error_logger:error_msg(Msg, Args),
+    io:format("~n~n** " ++ Msg ++ " **~n~n~n", Args),
+    error_logger:logfile(close),
+    halt(1).
+
+primary_upgrade(Upgrades, Nodes) ->
+    Others = Nodes -- [node()],
     apply_upgrades(
       mnesia,
       Upgrades,
@@ -87,26 +131,15 @@ primary_upgrade(Upgrades, DiscNodes) ->
 force_tables() ->
     [mnesia:force_load_table(T) || T <- rabbit_mnesia:table_names()].
 
-non_primary_upgrade(Upgrader, DiscNodes) ->
-    case node_running(Upgrader) of
-        false ->
-            Msg = "~n~n * Cluster upgrade needed. Please start node ~s "
-                "first. * ~n~n~n",
-            Args = [Upgrader],
-            %% We don't throw or exit here since that gets thrown
-            %% straight out into do_boot, generating an erl_crash.dump
-            %% and displaying any error message in a confusing way.
-            error_logger:error_msg(Msg, Args),
-            io:format(Msg, Args),
-            error_logger:logfile(close),
-            halt(1);
-        true ->
-            rabbit_misc:ensure_ok(mnesia:delete_schema([node()]),
-                                  cannot_delete_schema),
-            ok = rabbit_mnesia:create_cluster_nodes_config(DiscNodes),
-            write_version(mnesia),
-            ok
-    end.
+non_primary_upgrade(Nodes) ->
+    rabbit_misc:ensure_ok(mnesia:delete_schema([node()]),
+                          cannot_delete_schema),
+    ok = rabbit_mnesia:create_cluster_nodes_config(Nodes),
+    write_version(mnesia),
+    ok.
+
+nodes_running(Nodes) ->
+    [N || N <- Nodes, node_running(N)].
 
 node_running(Node) ->
     case rpc:call(Node, application, which_applications, []) of