Skip to content

Commit

Permalink
Merge pull request #911 from binarin/rabbitmq-server-851
Browse files Browse the repository at this point in the history
Add support for listing only local queues
  • Loading branch information
dcorbacho authored Aug 15, 2016
2 parents 9c08237 + 2482c9f commit c2bd22c
Show file tree
Hide file tree
Showing 7 changed files with 257 additions and 44 deletions.
42 changes: 38 additions & 4 deletions docs/rabbitmqctl.1.xml
Original file line number Diff line number Diff line change
Expand Up @@ -749,9 +749,7 @@

<varlistentry>
<term>
<cmdsynopsis>
<command>authenticate_user</command> <arg choice="req"><replaceable>username</replaceable></arg> <arg choice="req"><replaceable>password</replaceable></arg>
</cmdsynopsis>
<cmdsynopsis><command>authenticate_user</command> <arg choice="req"><replaceable>username</replaceable></arg> <arg choice="req"><replaceable>password</replaceable></arg></cmdsynopsis>
</term>
<listitem>
<variablelist>
Expand Down Expand Up @@ -1237,13 +1235,49 @@

<variablelist>
<varlistentry role="usage-has-option-list">
<term><cmdsynopsis><command>list_queues</command> <arg choice="opt">-p <replaceable>vhost</replaceable></arg> <arg choice="opt" role="usage-option-list"><replaceable>queueinfoitem</replaceable> ...</arg></cmdsynopsis></term>
<term>
<cmdsynopsis><command>list_queues</command> <arg choice="opt">-p <replaceable>vhost</replaceable></arg> <group choice="opt"><arg>--offline</arg><arg>--online</arg><arg>--local</arg></group> <arg choice="opt" role="usage-option-list"><replaceable>queueinfoitem</replaceable> ...</arg></cmdsynopsis>
</term>
<listitem>
<para>
Returns queue details. Queue details of the <command>/</command> virtual host
are returned if the "-p" flag is absent. The "-p" flag can be used to
override this default.
</para>
<para>
Displayed queues can be filtered by their status or
location using one of the following mutually exclusive
options:
<variablelist>
<varlistentry>
<term><cmdsynopsis><arg choice="opt">--offline</arg></cmdsynopsis></term>
<listitem>
<para>
List only those durable queues that are not
currently running - i.e. they are located on
inaccessible nodes.
</para>
</listitem>
</varlistentry>
<varlistentry>
<term><cmdsynopsis><arg choice="opt">--online</arg></cmdsynopsis></term>
<listitem>
<para>
List queues that are currently live.
</para>
</listitem>
</varlistentry>
<varlistentry>
<term><cmdsynopsis><arg choice="opt">--local</arg></cmdsynopsis></term>
<listitem>
<para>
List only those queues whose master process is
located on the current node.
</para>
</listitem>
</varlistentry>
</variablelist>
</para>
<para>
The <command>queueinfoitem</command> parameter is used to indicate which queue
information items to include in the results. The column order in the
Expand Down
10 changes: 9 additions & 1 deletion docs/usage.xsl
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
encoding="UTF-8"
indent="no"/>
<xsl:strip-space elements="*"/>
<xsl:preserve-space elements="cmdsynopsis arg" />
<xsl:preserve-space elements="cmdsynopsis arg group" />

<xsl:template match="/">
<!-- Pull out cmdsynopsis to show the command usage line. -->%% Generated, do not edit!
Expand Down Expand Up @@ -68,6 +68,14 @@ usage() -> %QUOTE%Usage:
<!-- Don't show anything else in command usage -->
<xsl:template match="text()" mode="command-usage"/>

<xsl:template match="group[@choice='opt']">
<xsl:text>[</xsl:text>
<xsl:for-each select="arg">
<xsl:apply-templates/>
<xsl:if test="not(position() = last())"><xsl:text>|</xsl:text></xsl:if>
</xsl:for-each>
<xsl:text>]</xsl:text>
</xsl:template>
<xsl:template match="arg[@choice='opt']">[<xsl:apply-templates/>]</xsl:template>
<xsl:template match="replaceable">&lt;<xsl:value-of select="."/>&gt;</xsl:template>

Expand Down
2 changes: 2 additions & 0 deletions include/rabbit_cli.hrl
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
-define(RAM_OPT, "--ram").
-define(OFFLINE_OPT, "--offline").
-define(ONLINE_OPT, "--online").
-define(LOCAL_OPT, "--local").


-define(NODE_DEF(Node), {?NODE_OPT, {option, Node}}).
Expand All @@ -46,6 +47,7 @@
-define(RAM_DEF, {?RAM_OPT, flag}).
-define(OFFLINE_DEF, {?OFFLINE_OPT, flag}).
-define(ONLINE_DEF, {?ONLINE_OPT, flag}).
-define(LOCAL_DEF, {?LOCAL_OPT, flag}).

%% Subset of standartized exit codes from sysexits.h, see
%% https://github.com/rabbitmq/rabbitmq-server/issues/396 for discussion.
Expand Down
35 changes: 18 additions & 17 deletions src/rabbit_cli.erl
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
-include("rabbit_cli.hrl").

-export([main/3, start_distribution/0, start_distribution/1,
parse_arguments/4, filter_opts/2,
parse_arguments/4, mutually_exclusive_flags/3,
rpc_call/4, rpc_call/5, rpc_call/7]).

%%----------------------------------------------------------------------------
Expand All @@ -42,8 +42,7 @@
[{string(), optdef()}], string(), [string()]) ->
parse_result().

-spec filter_opts([{option_name(), option_value()}], [option_name()]) ->
[boolean()].
-spec mutually_exclusive_flags([{option_name(), option_value()}], term(), [{option_name(), term()}]) -> {ok, term()} | {error, string()}.

-spec rpc_call(node(), atom(), atom(), [any()]) -> any().
-spec rpc_call(node(), atom(), atom(), [any()], number()) -> any().
Expand Down Expand Up @@ -266,20 +265,22 @@ process_opts(Defs, C, [A | As], Found, KVs, Outs) ->
{none, _, _} -> no_command
end.

%% When we have a set of flags that are used for filtering, we want by
%% default to include every such option in our output. But if a user
%% explicitly specified any such flag, we want to include only items
%% which he has requested.
filter_opts(CurrentOptionValues, AllOptionNames) ->
Explicit = lists:map(fun(OptName) ->
proplists:get_bool(OptName, CurrentOptionValues)
end,
AllOptionNames),
case lists:member(true, Explicit) of
true ->
Explicit;
false ->
lists:duplicate(length(AllOptionNames), true)
mutually_exclusive_flags(CurrentOptionValues, Default, FlagsAndValues) ->
PresentFlags = lists:filtermap(fun({OptName, _} = _O) ->
proplists:get_bool(OptName, CurrentOptionValues)
end,
FlagsAndValues),
case PresentFlags of
[] ->
{ok, Default};
[{_, Value}] ->
{ok, Value};
_ ->
Names = [ [$', N, $'] || {N, _} <- PresentFlags ],
CommaSeparated = string:join(lists:droplast(Names), ", "),
AndOneMore = lists:last(Names),
Msg = io_lib:format("Options ~s and ~s are mutually exclusive", [CommaSeparated, AndOneMore]),
{error, lists:flatten(Msg)}
end.

%%----------------------------------------------------------------------------
Expand Down
21 changes: 14 additions & 7 deletions src/rabbit_control_main.erl
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@
{clear_policy, [?VHOST_DEF]},
{list_policies, [?VHOST_DEF]},

{list_queues, [?VHOST_DEF, ?OFFLINE_DEF, ?ONLINE_DEF]},
{list_queues, [?VHOST_DEF, ?OFFLINE_DEF, ?ONLINE_DEF, ?LOCAL_DEF]},
{list_exchanges, [?VHOST_DEF]},
{list_bindings, [?VHOST_DEF]},
{list_connections, [?VHOST_DEF]},
Expand Down Expand Up @@ -632,12 +632,19 @@ action(list_user_permissions, Node, Args = [_Username], _Opts, Inform, Timeout)
true);

action(list_queues, Node, Args, Opts, Inform, Timeout) ->
[Online, Offline] = rabbit_cli:filter_opts(Opts, [?ONLINE_OPT, ?OFFLINE_OPT]),
Inform("Listing queues", []),
VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
ArgAtoms = default_if_empty(Args, [name, messages]),
call(Node, {rabbit_amqqueue, info_all, [VHostArg, ArgAtoms, Online, Offline]},
ArgAtoms, Timeout);
case rabbit_cli:mutually_exclusive_flags(
Opts, all, [{?ONLINE_OPT, online}
,{?OFFLINE_OPT, offline}
,{?LOCAL_OPT, local}]) of
{ok, Filter} ->
Inform("Listing queues", []),
VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
ArgAtoms = default_if_empty(Args, [name, messages]),
call(Node, {rabbit_amqqueue, info_all, [VHostArg, ArgAtoms, Filter]},
ArgAtoms, Timeout);
{error, ErrStr} ->
{error_string, ErrStr}
end;

action(list_exchanges, Node, Args, Opts, Inform, Timeout) ->
Inform("Listing exchanges", []),
Expand Down
146 changes: 146 additions & 0 deletions test/rabbitmqctl_integration_SUITE.erl
Original file line number Diff line number Diff line change
@@ -0,0 +1,146 @@
%% The contents of this file are subject to the Mozilla Public License
%% Version 1.1 (the "License"); you may not use this file except in
%% compliance with the License. You may obtain a copy of the License
%% at http://www.mozilla.org/MPL/
%%
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and
%% limitations under the License.
%%
%% The Original Code is RabbitMQ.
%%
%% The Initial Developer of the Original Code is GoPivotal, Inc.
%% Copyright (c) 2016 Pivotal Software, Inc. All rights reserved.
%%
-module(rabbitmqctl_integration_SUITE).

-include_lib("common_test/include/ct.hrl").
-include_lib("amqp_client/include/amqp_client.hrl").

-export([all/0
,groups/0
,init_per_suite/1
,end_per_suite/1
,init_per_group/2
,end_per_group/2
,init_per_testcase/2
,end_per_testcase/2
]).

-export([list_queues_local/1
,list_queues_offline/1
,list_queues_online/1
]).

all() ->
[{group, list_queues}].

groups() ->
[{list_queues, [],
[list_queues_local
,list_queues_online
,list_queues_offline
]}].

init_per_suite(Config) ->
rabbit_ct_helpers:log_environment(),
rabbit_ct_helpers:run_setup_steps(Config).

end_per_suite(Config) ->
rabbit_ct_helpers:run_teardown_steps(Config).

init_per_group(list_queues, Config0) ->
NumNodes = 3,
Config = create_n_node_cluster(Config0, NumNodes),
Config1 = declare_some_queues(Config),
rabbit_ct_broker_helpers:stop_node(Config1, NumNodes - 1),
Config1;
init_per_group(_, Config) ->
Config.

create_n_node_cluster(Config0, NumNodes) ->
Config1 = rabbit_ct_helpers:set_config(
Config0, [{rmq_nodes_count, NumNodes},
{rmq_nodes_clustered, true}]),
rabbit_ct_helpers:run_steps(Config1,
rabbit_ct_broker_helpers:setup_steps() ++
rabbit_ct_client_helpers:setup_steps()).

declare_some_queues(Config) ->
Nodes = rabbit_ct_helpers:get_config(Config, rmq_nodes),
PerNodeQueues = [ declare_some_queues(Config, NodeNum)
|| NodeNum <- lists:seq(0, length(Nodes)-1) ],
rabbit_ct_helpers:set_config(Config, {per_node_queues, PerNodeQueues}).

declare_some_queues(Config, NodeNum) ->
{Conn, Chan} = rabbit_ct_client_helpers:open_connection_and_channel(Config, NodeNum),
NumQueues = 5,
Queues = [ list_to_binary(io_lib:format("queue-~b-on-node-~b", [QueueNum, NodeNum]))
|| QueueNum <- lists:seq(1, NumQueues) ],
lists:foreach(fun (QueueName) ->
#'queue.declare_ok'{} = amqp_channel:call(Chan, #'queue.declare'{queue = QueueName, durable = true})
end, Queues),
rabbit_ct_client_helpers:close_connection_and_channel(Conn, Chan),
Queues.

end_per_group(list_queues, Config0) ->
Config1 = case rabbit_ct_helpers:get_config(Config0, save_config) of
undefined -> Config0;
C -> C
end,
rabbit_ct_helpers:run_steps(Config1,
rabbit_ct_client_helpers:teardown_steps() ++
rabbit_ct_broker_helpers:teardown_steps());
end_per_group(_, Config) ->
Config.

init_per_testcase(Testcase, Config0) ->
rabbit_ct_helpers:testcase_started(Config0, Testcase).

end_per_testcase(Testcase, Config0) ->
rabbit_ct_helpers:testcase_finished(Config0, Testcase).

%%----------------------------------------------------------------------------
%% Test cases
%%----------------------------------------------------------------------------
list_queues_local(Config) ->
Node1Queues = lists:sort(lists:nth(1, ?config(per_node_queues, Config))),
Node2Queues = lists:sort(lists:nth(2, ?config(per_node_queues, Config))),
assert_ctl_queues(Config, 0, ["--local"], Node1Queues),
assert_ctl_queues(Config, 1, ["--local"], Node2Queues),
ok.

list_queues_online(Config) ->
Node1Queues = lists:sort(lists:nth(1, ?config(per_node_queues, Config))),
Node2Queues = lists:sort(lists:nth(2, ?config(per_node_queues, Config))),
OnlineQueues = Node1Queues ++ Node2Queues,
assert_ctl_queues(Config, 0, ["--online"], OnlineQueues),
assert_ctl_queues(Config, 1, ["--online"], OnlineQueues),
ok.

list_queues_offline(Config) ->
Node3Queues = lists:sort(lists:nth(3, ?config(per_node_queues, Config))),
OfflineQueues = Node3Queues,
assert_ctl_queues(Config, 0, ["--offline"], OfflineQueues),
assert_ctl_queues(Config, 1, ["--offline"], OfflineQueues),
ok.

%%----------------------------------------------------------------------------
%% Helpers
%%----------------------------------------------------------------------------
assert_ctl_queues(Config, Node, Args, Expected0) ->
Expected = lists:sort(Expected0),
Got0 = run_list_queues(Config, Node, Args),
Got = lists:sort(lists:map(fun hd/1, Got0)),
case Got of
Expected ->
ok;
_ ->
ct:pal(error, "Listing queues on node ~p failed. Expected:~n~p~n~nGot:~n~p~n~n",
[Node, Expected, Got]),
exit({list_queues_unexpected_on, Node, Expected, Got})
end.

run_list_queues(Config, Node, Args) ->
rabbit_ct_broker_helpers:rabbitmqctl_list(Config, Node, ["list_queues"] ++ Args ++ ["name"]).
Loading

0 comments on commit c2bd22c

Please sign in to comment.