Merge pull request #2165 from rabbitmq/mgmt-less-improvements

Add format/1 for classic queues and totals/type_specific info for all queue types

(cherry picked from commit ddc9878)

Author: michaelklishin

src/rabbit_amqqueue.erl

@@ -54,6 +54,7 @@
 -export([mark_local_durable_queues_stopped/1]).
 
 -export([rebalance/3]).
+-export([collect_info_all/2]).
 
 %% internal
 -export([internal_declare/2, internal_delete/2, run_backing_queue/3,
@@ -1097,7 +1098,7 @@ is_unresponsive(Q, Timeout) when ?amqqueue_is_quorum(Q) ->
     end.
 
 format(Q) when ?amqqueue_is_quorum(Q) -> rabbit_quorum_queue:format(Q);
-format(_) -> [].
+format(Q) -> rabbit_amqqueue_process:format(Q).
 
 -spec info(amqqueue:amqqueue()) -> rabbit_types:infos().
 
@@ -1128,7 +1129,7 @@ info_down(Q, DownReason) ->
     info_down(Q, rabbit_amqqueue_process:info_keys(), DownReason).
 
 info_down(Q, Items, DownReason) ->
-    [{Item, i_down(Item, Q, DownReason)} || Item <- Items].
+    [{Item, i_down(Item, Q, DownReason)} || Item <- Items, Item =/= totals, Item =/= type_specific].
 
 i_down(name,               Q, _) -> amqqueue:get_name(Q);
 i_down(durable,            Q, _) -> amqqueue:is_durable(Q);
@@ -1165,6 +1166,26 @@ emit_info_all(Nodes, VHostPath, Items, Ref, AggregatorPid) ->
     Pids = [ spawn_link(Node, rabbit_amqqueue, emit_info_local, [VHostPath, Items, Ref, AggregatorPid]) || Node <- Nodes ],
     rabbit_control_misc:await_emitters_termination(Pids).
 
+collect_info_all(VHostPath, Items) ->
+    Nodes = rabbit_mnesia:cluster_nodes(running),
+    Ref = make_ref(),
+    Pids = [ spawn_link(Node, rabbit_amqqueue, emit_info_local, [VHostPath, Items, Ref, self()]) || Node <- Nodes ],
+    rabbit_control_misc:await_emitters_termination(Pids),
+    wait_for_queues(Ref, length(Pids), []).
+
+wait_for_queues(Ref, N, Acc) ->
+    receive
+        {Ref, finished} when N == 1 ->
+            Acc;
+        {Ref, finished} ->
+            wait_for_queues(Ref, N - 1, Acc);
+        {Ref, Items, continue} ->
+            wait_for_queues(Ref, N, [Items | Acc])
+    after
+        1000 ->
+            Acc
+    end.
+
 emit_info_down(VHostPath, Items, Ref, AggregatorPid) ->
     rabbit_control_misc:emitting_map_with_exit_handler(
       AggregatorPid, Ref, fun(Q) -> info_down(Q, Items, down) end,

src/rabbit_amqqueue_process.erl

@@ -32,6 +32,7 @@
 -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2,
          handle_info/2, handle_pre_hibernate/1, prioritise_call/4,
          prioritise_cast/3, prioritise_info/3, format_message_queue/2]).
+-export([format/1]).
 
 %% Queue's state
 -record(q, {
@@ -1066,7 +1067,16 @@ stop(State) -> stop(noreply, State).
 stop(noreply, State) -> {stop, normal, State};
 stop(Reply,   State) -> {stop, normal, Reply, State}.
 
-infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
+infos(Items, #q{q = Q} = State) ->
+    lists:foldr(fun(totals, Acc) ->
+                        [{messages_ready, i(messages_ready, State)},
+                         {messages, i(messages, State)},
+                         {messages_unacknowledged, i(messages_unacknowledged, State)}] ++ Acc;
+                   (type_specific, Acc) ->
+                        format(Q) ++ Acc;
+                   (Item, Acc) ->
+                        [{Item, i(Item, State)} | Acc]
+                end, [], Items).
 
 i(name,        #q{q = Q}) -> amqqueue:get_name(Q);
 i(durable,     #q{q = Q}) -> amqqueue:is_durable(Q);
@@ -1761,6 +1771,18 @@ handle_pre_hibernate(State = #q{backing_queue = BQ,
 
 format_message_queue(Opt, MQ) -> rabbit_misc:format_message_queue(Opt, MQ).
 
+format(Q) when ?is_amqqueue(Q) ->
+    case rabbit_mirror_queue_misc:is_mirrored(Q) of
+        false ->
+            [{node, node(amqqueue:get_pid(Q))}];
+        true ->
+            Slaves = amqqueue:get_slave_pids(Q),
+            SSlaves = amqqueue:get_sync_slave_pids(Q),
+            [{slave_nodes, [node(S) || S <- Slaves]},
+             {synchronised_slave_nodes, [node(S) || S <- SSlaves]},
+             {node, node(amqqueue:get_pid(Q))}]
+    end.
+
 log_delete_exclusive({ConPid, _ConRef}, State) ->
     log_delete_exclusive(ConPid, State);
 log_delete_exclusive(ConPid, #q{ q = Q }) ->

src/rabbit_quorum_queue.erl

@@ -592,7 +592,13 @@ infos(QName) ->
 -spec info(amqqueue:amqqueue(), rabbit_types:info_keys()) -> rabbit_types:infos().
 
 info(Q, Items) ->
-    [{Item, i(Item, Q)} || Item <- Items].
+    lists:foldr(fun(totals, Acc) ->
+                        i_totals(Q) ++ Acc;
+                   (type_specific, Acc) ->
+                        format(Q) ++ Acc;
+                   (Item, Acc) ->
+                        [{Item, i(Item, Q)} | Acc]
+                end, [], Items).
 
 -spec stat(amqqueue:amqqueue()) -> {'ok', non_neg_integer(), non_neg_integer()}.
 
@@ -973,6 +979,19 @@ find_quorum_queues(VHost) ->
                                 amqqueue:qnode(Q) == Node]))
       end).
 
+i_totals(Q) when ?is_amqqueue(Q) ->
+    QName = amqqueue:get_name(Q),
+    case ets:lookup(queue_coarse_metrics, QName) of
+        [{_, MR, MU, M, _}] ->
+            [{messages_ready, MR},
+             {messages_unacknowledged, MU},
+             {messages, M}];
+        [] ->
+            [{messages_ready, 0},
+             {messages_unacknowledged, 0},
+             {messages, 0}]
+    end.
+
 i(name,        Q) when ?is_amqqueue(Q) -> amqqueue:get_name(Q);
 i(durable,     Q) when ?is_amqqueue(Q) -> amqqueue:is_durable(Q);
 i(auto_delete, Q) when ?is_amqqueue(Q) -> amqqueue:is_auto_delete(Q);