Merge pull request #1372 from rabbitmq/rabbitmq-server-1371

Take ha-mode into account in choosing queue master

4 files changed, 33 insertions, 8 deletions

diff --git a/src/rabbit_mirror_queue_misc.erl b/src/rabbit_mirror_queue_misc.erl
index 375a0366dd..333aa2cfb1 100644
--- a/src/rabbit_mirror_queue_misc.erl
+++ b/src/rabbit_mirror_queue_misc.erl
@@ -19,8 +19,9 @@
 
 -export([remove_from_queue/3, on_node_up/0, add_mirrors/3,
          report_deaths/4, store_updated_slaves/1,
-         initial_queue_node/2, suggested_queue_nodes/1,
-         is_mirrored/1, update_mirrors/2, update_mirrors/1, validate_policy/1,
+         initial_queue_node/2, suggested_queue_nodes/1, actual_queue_nodes/1,
+         is_mirrored/1, is_mirrored_ha_nodes/1,
+         update_mirrors/2, update_mirrors/1, validate_policy/1,
          maybe_auto_sync/1, maybe_drop_master_after_sync/1,
          sync_batch_size/1, log_info/3, log_warning/3]).
 
@@ -31,6 +32,8 @@
 
 -include("rabbit.hrl").
 
+-define(HA_NODES_MODULE, rabbit_mirror_queue_mode_nodes).
+
 -rabbit_boot_step(
    {?MODULE,
     [{description, "HA policy validation"},
@@ -355,6 +358,12 @@ is_mirrored(Q) ->
         _        -> false
     end.
 
+is_mirrored_ha_nodes(Q) ->
+    case module(Q) of
+        {ok, ?HA_NODES_MODULE} -> true;
+        _ -> false
+    end.
+
 actual_queue_nodes(#amqqueue{pid             = MPid,
                              slave_pids      = SPids,
                              sync_slave_pids = SSPids}) ->
diff --git a/src/rabbit_queue_location_min_masters.erl b/src/rabbit_queue_location_min_masters.erl
index a1e73ee5d5..5d6f568d4a 100644
--- a/src/rabbit_queue_location_min_masters.erl
+++ b/src/rabbit_queue_location_min_masters.erl
@@ -37,8 +37,8 @@ description() ->
     [{description,
       <<"Locate queue master node from cluster node with least bound queues">>}].
 
-queue_master_location(#amqqueue{}) ->
-    Cluster            = rabbit_queue_master_location_misc:all_nodes(),
+queue_master_location(#amqqueue{} = Q) ->
+    Cluster            = rabbit_queue_master_location_misc:all_nodes(Q),
     VHosts             = rabbit_vhost:list(),
     BoundQueueMasters  = get_bound_queue_masters_per_vhost(VHosts, []),
     {_Count, MinMaster}= get_min_master(Cluster, BoundQueueMasters),
diff --git a/src/rabbit_queue_location_random.erl b/src/rabbit_queue_location_random.erl
index 3a3a2360b7..d103d3f126 100644
--- a/src/rabbit_queue_location_random.erl
+++ b/src/rabbit_queue_location_random.erl
@@ -37,8 +37,8 @@ description() ->
     [{description,
       <<"Locate queue master node from cluster in a random manner">>}].
 
-queue_master_location(#amqqueue{}) ->
-    Cluster    = rabbit_queue_master_location_misc:all_nodes(),
+queue_master_location(#amqqueue{} = Q) ->
+    Cluster    = rabbit_queue_master_location_misc:all_nodes(Q),
     RandomPos  = erlang:phash2(time_compat:monotonic_time(), length(Cluster)),
     MasterNode = lists:nth(RandomPos + 1, Cluster),
     {ok, MasterNode}.
diff --git a/src/rabbit_queue_master_location_misc.erl b/src/rabbit_queue_master_location_misc.erl
index 0b9efdd6fc..c2cc489ec2 100644
--- a/src/rabbit_queue_master_location_misc.erl
+++ b/src/rabbit_queue_master_location_misc.erl
@@ -24,7 +24,7 @@
          get_location_mod_by_config/1,
          get_location_mod_by_args/1,
          get_location_mod_by_policy/1,
-         all_nodes/0]).
+         all_nodes/1]).
 
 lookup_master(QueueNameBin, VHostPath) when is_binary(QueueNameBin),
                                             is_binary(VHostPath) ->
@@ -92,4 +92,20 @@ get_location_mod_by_config(#amqqueue{}) ->
         _ -> {error, "queue_master_locator undefined"}
     end.
 
-all_nodes()  -> rabbit_mnesia:cluster_nodes(running).
+all_nodes(Queue = #amqqueue{}) ->
+    handle_is_mirrored_ha_nodes(rabbit_mirror_queue_misc:is_mirrored_ha_nodes(Queue), Queue).
+
+handle_is_mirrored_ha_nodes(false, _Queue) ->
+    % Note: ha-mode is NOT 'nodes' - it is either exactly or all, which means
+    % that any node in the cluster is eligible to be the new queue master node
+    rabbit_nodes:all_running();
+handle_is_mirrored_ha_nodes(true, Queue) ->
+    % Note: ha-mode is 'nodes', which explicitly specifies allowed nodes.
+    % We must use suggested_queue_nodes to get that list of nodes as the
+    % starting point for finding the queue master location
+    handle_suggested_queue_nodes(rabbit_mirror_queue_misc:suggested_queue_nodes(Queue)).
+
+handle_suggested_queue_nodes({_MNode, []}) ->
+    rabbit_nodes:all_running();
+handle_suggested_queue_nodes({MNode, SNodes}) ->
+    [MNode | SNodes].