summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorMichael Klishin <michael@novemberain.com>2016-12-27 16:08:33 +0300
committerGitHub <noreply@github.com>2016-12-27 16:08:33 +0300
commit30ebac8cc103cf032d5e7bdd1b5f74407f4b986f (patch)
tree030a352ad4a45a7aa3f3a529cea1ff72cea6abc0
parent9468daed03453545d3d9a6c0f66b2def77f9d42d (diff)
parent0463c59f046dad645cc4f83b9c2289df755af85a (diff)
downloadrabbitmq-server-git-30ebac8cc103cf032d5e7bdd1b5f74407f4b986f.tar.gz
Merge pull request #1067 from rabbitmq/rabbitmq-server-1058
Getting rid of rabbit_control_main
Diffstat
-rw-r--r--src/rabbit_control_main.erl1110
-rw-r--r--src/rabbit_mnesia_rename.erl16
-rw-r--r--src/rabbit_runtime_parameters.erl2
-rw-r--r--test/clustering_management_SUITE.erl28
-rw-r--r--test/eager_sync_SUITE.erl18
-rw-r--r--test/per_vhost_connection_limit_SUITE.erl2
-rw-r--r--test/rabbitmqctl_integration_SUITE.erl86
-rw-r--r--test/unit_inbroker_SUITE.erl858
8 files changed, 114 insertions, 2006 deletions
diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
deleted file mode 100644
index d96c1dd476..0000000000
--- a/src/rabbit_control_main.erl
+++ /dev/null
@@ -1,1110 +0,0 @@
-%% 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) 2007-2016 Pivotal Software, Inc. All rights reserved.
-%%
-
--module(rabbit_control_main).
--include("rabbit.hrl").
--include("rabbit_cli.hrl").
--include("rabbit_misc.hrl").
-
--export([start/0, stop/0, parse_arguments/2, action/5, action/6,
- sync_queue/1, cancel_sync_queue/1, become/1,
- purge_queue/1]).
-
--import(rabbit_misc, [rpc_call/4, rpc_call/5]).
-
--define(EXTERNAL_CHECK_INTERVAL, 1000).
-
--define(GLOBAL_DEFS(Node), [?QUIET_DEF, ?NODE_DEF(Node), ?TIMEOUT_DEF]).
-
--define(COMMANDS,
- [stop,
- stop_app,
- start_app,
- wait,
- reset,
- force_reset,
- rotate_logs,
- hipe_compile,
-
- {join_cluster, [?RAM_DEF]},
- change_cluster_node_type,
- update_cluster_nodes,
- {forget_cluster_node, [?OFFLINE_DEF]},
- rename_cluster_node,
- force_boot,
- cluster_status,
- {sync_queue, [?VHOST_DEF]},
- {cancel_sync_queue, [?VHOST_DEF]},
- {purge_queue, [?VHOST_DEF]},
-
- add_user,
- delete_user,
- change_password,
- clear_password,
- authenticate_user,
- set_user_tags,
- list_users,
-
- add_vhost,
- delete_vhost,
- list_vhosts,
- {set_permissions, [?VHOST_DEF]},
- {clear_permissions, [?VHOST_DEF]},
- {list_permissions, [?VHOST_DEF]},
- list_user_permissions,
-
- {set_parameter, [?VHOST_DEF]},
- {clear_parameter, [?VHOST_DEF]},
- {list_parameters, [?VHOST_DEF]},
-
- set_global_parameter,
- clear_global_parameter,
- list_global_parameters,
-
- {set_policy, [?VHOST_DEF, ?PRIORITY_DEF, ?APPLY_TO_DEF]},
- {clear_policy, [?VHOST_DEF]},
- {set_operator_policy, [?VHOST_DEF, ?PRIORITY_DEF, ?APPLY_TO_DEF]},
- {clear_operator_policy, [?VHOST_DEF]},
- {list_policies, [?VHOST_DEF]},
- {list_operator_policies, [?VHOST_DEF]},
-
- {set_vhost_limits, [?VHOST_DEF]},
- {clear_vhost_limits, [?VHOST_DEF]},
- {list_queues, [?VHOST_DEF, ?OFFLINE_DEF, ?ONLINE_DEF, ?LOCAL_DEF]},
- {list_exchanges, [?VHOST_DEF]},
- {list_bindings, [?VHOST_DEF]},
- {list_connections, [?VHOST_DEF]},
- list_channels,
- {list_consumers, [?VHOST_DEF]},
- status,
- environment,
- report,
- set_cluster_name,
- eval,
- node_health_check,
-
- close_connection,
- {trace_on, [?VHOST_DEF]},
- {trace_off, [?VHOST_DEF]},
- set_vm_memory_high_watermark,
- set_disk_free_limit,
- help,
- {encode, [?DECODE_DEF, ?CIPHER_DEF, ?HASH_DEF, ?ITERATIONS_DEF, ?LIST_CIPHERS_DEF, ?LIST_HASHES_DEF]}
- ]).
-
--define(GLOBAL_QUERIES,
- [{"Connections", rabbit_networking, connection_info_all,
- connection_info_keys},
- {"Channels", rabbit_channel, info_all, info_keys}]).
-
--define(VHOST_QUERIES,
- [{"Queues", rabbit_amqqueue, info_all, info_keys},
- {"Exchanges", rabbit_exchange, info_all, info_keys},
- {"Bindings", rabbit_binding, info_all, info_keys},
- {"Consumers", rabbit_amqqueue, consumers_all, consumer_info_keys},
- {"Permissions", rabbit_auth_backend_internal, list_vhost_permissions,
- vhost_perms_info_keys},
- {"Policies", rabbit_policy, list_formatted, info_keys},
- {"Parameters", rabbit_runtime_parameters, list_formatted, info_keys}]).
-
--define(COMMANDS_NOT_REQUIRING_APP,
- [stop, stop_app, start_app, wait, reset, force_reset, rotate_logs,
- join_cluster, change_cluster_node_type, update_cluster_nodes,
- forget_cluster_node, rename_cluster_node, cluster_status, status,
- environment, eval, force_boot, help, hipe_compile, encode]).
-
-%% [Command | {Command, DefaultTimeoutInMilliSeconds}]
--define(COMMANDS_WITH_TIMEOUT,
- [list_user_permissions, list_policies, list_queues, list_exchanges,
- list_bindings, list_connections, list_channels, list_consumers,
- list_vhosts, list_parameters, list_global_parameters,
- purge_queue,
- {node_health_check, 70000}]).
-
-%%----------------------------------------------------------------------------
-
--spec start() -> no_return().
--spec stop() -> 'ok'.
--spec action
- (atom(), node(), [string()], [{string(), any()}],
- fun ((string(), [any()]) -> 'ok')) ->
- 'ok'.
-
--spec action
- (atom(), node(), [string()], [{string(), any()}],
- fun ((string(), [any()]) -> 'ok'), timeout()) ->
- 'ok'.
-
-%%----------------------------------------------------------------------------
-
-start() ->
- rabbit_cli:main(
- fun (Args, NodeStr) ->
- parse_arguments(Args, NodeStr)
- end,
- fun (Command, Node, Args, Opts) ->
- Quiet = proplists:get_bool(?QUIET_OPT, Opts),
- Inform = case Quiet of
- true -> fun (_Format, _Args1) -> ok end;
- false -> fun (Format, Args1) ->
- io:format(Format ++ " ...~n", Args1)
- end
- end,
- try
- T = case get_timeout(Command, Opts) of
- {ok, Timeout} ->
- Timeout;
- {error, _} ->
- %% since this is an error with user input, ignore the quiet
- %% setting
- io:format("Failed to parse provided timeout value, using ~s~n", [?RPC_TIMEOUT]),
- ?RPC_TIMEOUT
- end,
- do_action(Command, Node, Args, Opts, Inform, T)
- catch _:E -> E
- end
- end, rabbit_ctl_usage).
-
-parse_arguments(CmdLine, NodeStr) ->
- rabbit_cli:parse_arguments(
- ?COMMANDS, ?GLOBAL_DEFS(NodeStr), ?NODE_OPT, CmdLine).
-
-print_report(Node, {Descr, Module, InfoFun, KeysFun}) ->
- io:format("~s:~n", [Descr]),
- print_report0(Node, {Module, InfoFun, KeysFun}, []).
-
-print_report(Node, {Descr, Module, InfoFun, KeysFun}, VHostArg) ->
- io:format("~s on ~s:~n", [Descr, VHostArg]),
- print_report0(Node, {Module, InfoFun, KeysFun}, VHostArg).
-
-print_report0(Node, {Module, InfoFun, KeysFun}, VHostArg) ->
- case rpc_call(Node, Module, InfoFun, VHostArg) of
- [_|_] = Results -> InfoItems = rpc_call(Node, Module, KeysFun, []),
- display_row([atom_to_list(I) || I <- InfoItems]),
- display_info_list(Results, InfoItems);
- _ -> ok
- end,
- io:nl().
-
-get_timeout(Command, Opts) ->
- Default = case proplists:lookup(Command, ?COMMANDS_WITH_TIMEOUT) of
- none ->
- infinity;
- {Command, true} ->
- ?RPC_TIMEOUT;
- {Command, D} ->
- D
- end,
- Result = case proplists:get_value(?TIMEOUT_OPT, Opts, Default) of
- use_default ->
- parse_timeout(Default);
- Value ->
- parse_timeout(Value)
- end,
- Result.
-
-
-parse_number(N) when is_list(N) ->
- try list_to_integer(N) of
- Val -> Val
- catch error:badarg ->
- %% could have been a float, give it
- %% another shot
- list_to_float(N)
- end.
-
-parse_timeout("infinity") ->
- {ok, infinity};
-parse_timeout(infinity) ->
- {ok, infinity};
-parse_timeout(N) when is_list(N) ->
- try parse_number(N) of
- M ->
- Y = case M >= 0 of
- true -> round(M) * 1000;
- false -> ?RPC_TIMEOUT
- end,
- {ok, Y}
- catch error:badarg ->
- {error, infinity}
- end;
-parse_timeout(N) ->
- {ok, N}.
-
-announce_timeout(infinity, _Inform) ->
- %% no-op
- ok;
-announce_timeout(Timeout, Inform) when is_number(Timeout) ->
- Inform("Timeout: ~w seconds", [Timeout/1000]),
- ok.
-
-stop() ->
- ok.
-
-%%----------------------------------------------------------------------------
-
-do_action(Command, Node, Args, Opts, Inform, Timeout) ->
- case lists:member(Command, ?COMMANDS_NOT_REQUIRING_APP) of
- false ->
- case ensure_app_running(Node) of
- ok ->
- case proplists:lookup(Command, ?COMMANDS_WITH_TIMEOUT) of
- {Command, _} ->
- announce_timeout(Timeout, Inform),
- action(Command, Node, Args, Opts, Inform, Timeout);
- none ->
- action(Command, Node, Args, Opts, Inform)
- end;
- E -> E
- end;
- true ->
- action(Command, Node, Args, Opts, Inform)
- end.
-
-action(stop, Node, Args, _Opts, Inform) ->
- Inform("Stopping and halting node ~p", [Node]),
- Res = call(Node, {rabbit, stop_and_halt, []}),
- case {Res, Args} of
- {ok, [PidFile]} -> wait_for_process_death(
- read_pid_file(PidFile, false));
- {ok, [_, _| _]} -> exit({badarg, Args});
- _ -> ok
- end,
- Res;
-
-action(stop_app, Node, [], _Opts, Inform) ->
- Inform("Stopping rabbit application on node ~p", [Node]),
- call(Node, {rabbit, stop, []});
-
-action(start_app, Node, [], _Opts, Inform) ->
- Inform("Starting node ~p", [Node]),
- call(Node, {rabbit, start, []});
-
-action(reset, Node, [], _Opts, Inform) ->
- Inform("Resetting node ~p", [Node]),
- require_mnesia_stopped(Node,
- fun() ->
- call(Node, {rabbit_mnesia, reset, []})
- end);
-
-action(force_reset, Node, [], _Opts, Inform) ->
- Inform("Forcefully resetting node ~p", [Node]),
- require_mnesia_stopped(Node,
- fun() ->
- call(Node, {rabbit_mnesia, force_reset, []})
- end);
-
-action(join_cluster, Node, [ClusterNodeS], Opts, Inform) ->
- ClusterNode = list_to_atom(ClusterNodeS),
- NodeType = case proplists:get_bool(?RAM_OPT, Opts) of
- true -> ram;
- false -> disc
- end,
- Inform("Clustering node ~p with ~p", [Node, ClusterNode]),
- require_mnesia_stopped(Node,
- fun() ->
- rpc_call(Node, rabbit_mnesia, join_cluster, [ClusterNode, NodeType])
- end);
-
-action(change_cluster_node_type, Node, ["ram"], _Opts, Inform) ->
- Inform("Turning ~p into a ram node", [Node]),
- require_mnesia_stopped(Node,
- fun() ->
- rpc_call(Node, rabbit_mnesia, change_cluster_node_type, [ram])
- end);
-action(change_cluster_node_type, Node, [Type], _Opts, Inform)
- when Type =:= "disc" orelse Type =:= "disk" ->
- Inform("Turning ~p into a disc node", [Node]),
- require_mnesia_stopped(Node,
- fun() ->
- rpc_call(Node, rabbit_mnesia, change_cluster_node_type, [disc])
- end);
-
-action(update_cluster_nodes, Node, [ClusterNodeS], _Opts, Inform) ->
- ClusterNode = list_to_atom(ClusterNodeS),
- Inform("Updating cluster nodes for ~p from ~p", [Node, ClusterNode]),
- require_mnesia_stopped(Node,
- fun() ->
- rpc_call(Node, rabbit_mnesia, update_cluster_nodes, [ClusterNode])
- end);
-
-action(forget_cluster_node, Node, [ClusterNodeS], Opts, Inform) ->
- ClusterNode = list_to_atom(ClusterNodeS),
- RemoveWhenOffline = proplists:get_bool(?OFFLINE_OPT, Opts),
- Inform("Removing node ~p from cluster", [ClusterNode]),
- case RemoveWhenOffline of
- true -> become(Node),
- rabbit_mnesia:forget_cluster_node(ClusterNode, true);
- false -> rpc_call(Node, rabbit_mnesia, forget_cluster_node,
- [ClusterNode, false])
- end;
-
-action(rename_cluster_node, Node, NodesS, _Opts, Inform) ->
- Nodes = split_list([list_to_atom(N) || N <- NodesS]),
- Inform("Renaming cluster nodes:~n~s~n",
- [lists:flatten([rabbit_misc:format(" ~s -> ~s~n", [F, T]) ||
- {F, T} <- Nodes])]),
- rabbit_mnesia_rename:rename(Node, Nodes);
-
-action(force_boot, Node, [], _Opts, Inform) ->
- Inform("Forcing boot for Mnesia dir ~s", [mnesia:system_info(directory)]),
- case rabbit:is_running(Node) of
- false -> rabbit_mnesia:force_load_next_boot();
- true -> {error, rabbit_running}
- end;
-
-action(sync_queue, Node, [Q], Opts, Inform) ->
- VHost = proplists:get_value(?VHOST_OPT, Opts),
- QName = rabbit_misc:r(list_to_binary(VHost), queue, list_to_binary(Q)),
- Inform("Synchronising ~s", [rabbit_misc:rs(QName)]),
- rpc_call(Node, rabbit_control_main, sync_queue, [QName]);
-
-action(cancel_sync_queue, Node, [Q], Opts, Inform) ->
- VHost = proplists:get_value(?VHOST_OPT, Opts),
- QName = rabbit_misc:r(list_to_binary(VHost), queue, list_to_binary(Q)),
- Inform("Stopping synchronising ~s", [rabbit_misc:rs(QName)]),
- rpc_call(Node, rabbit_control_main, cancel_sync_queue, [QName]);
-
-action(wait, Node, [PidFile], _Opts, Inform) ->
- Inform("Waiting for ~p", [Node]),
- wait_for_application(Node, PidFile, rabbit_and_plugins, Inform);
-action(wait, Node, [PidFile, App], _Opts, Inform) ->
- Inform("Waiting for ~p on ~p", [App, Node]),
- wait_for_application(Node, PidFile, list_to_atom(App), Inform);
-
-action(status, Node, [], _Opts, Inform) ->
- Inform("Status of node ~p", [Node]),
- display_call_result(Node, {rabbit, status, []});
-
-action(cluster_status, Node, [], _Opts, Inform) ->
- Inform("Cluster status of node ~p", [Node]),
- Status = unsafe_rpc(Node, rabbit_mnesia, status, []),
- io:format("~p~n", [Status ++ [{alarms,
- [alarms_by_node(Name) || Name <- nodes_in_cluster(Node)]}]]),
- ok;
-
-action(environment, Node, _App, _Opts, Inform) ->
- Inform("Application environment of node ~p", [Node]),
- display_call_result(Node, {rabbit, environment, []});
-
-action(rotate_logs, Node, [], _Opts, Inform) ->
- Inform("Rotating logs for node ~p", [Node]),
- call(Node, {rabbit, rotate_logs, []});
-
-action(hipe_compile, _Node, [TargetDir], _Opts, _Inform) ->
- ok = application:load(rabbit),
- case rabbit_hipe:can_hipe_compile() of
- true ->
- {ok, _, _} = rabbit_hipe:compile_to_directory(TargetDir),
- ok;
- false ->
- {error, "HiPE compilation is not supported"}
- end;
-
-action(close_connection, Node, [PidStr, Explanation], _Opts, Inform) ->
- Inform("Closing connection \"~s\"", [PidStr]),
- rpc_call(Node, rabbit_networking, close_connection,
- [rabbit_misc:string_to_pid(PidStr), Explanation]);
-
-action(add_user, Node, Args = [Username, _Password], _Opts, Inform) ->
- Inform("Creating user \"~s\"", [Username]),
- call(Node, {rabbit_auth_backend_internal, add_user, Args});
-
-action(delete_user, Node, Args = [_Username], _Opts, Inform) ->
- Inform("Deleting user \"~s\"", Args),
- call(Node, {rabbit_auth_backend_internal, delete_user, Args});
-
-action(change_password, Node, Args = [Username, _Newpassword], _Opts, Inform) ->
- Inform("Changing password for user \"~s\"", [Username]),
- call(Node, {rabbit_auth_backend_internal, change_password, Args});
-
-action(clear_password, Node, Args = [Username], _Opts, Inform) ->
- Inform("Clearing password for user \"~s\"", [Username]),
- call(Node, {rabbit_auth_backend_internal, clear_password, Args});
-
-action(authenticate_user, Node, Args = [Username, _Password], _Opts, Inform) ->
- Inform("Authenticating user \"~s\"", [Username]),
- call(Node, {rabbit_access_control, check_user_pass_login, Args});
-
-action(set_user_tags, Node, [Username | TagsStr], _Opts, Inform) ->
- Tags = [list_to_atom(T) || T <- TagsStr],
- Inform("Setting tags for user \"~s\" to ~p", [Username, Tags]),
- rpc_call(Node, rabbit_auth_backend_internal, set_tags,
- [list_to_binary(Username), Tags]);
-
-action(add_vhost, Node, Args = [_VHostPath], _Opts, Inform) ->
- Inform("Creating vhost \"~s\"", Args),
- call(Node, {rabbit_vhost, add, Args});
-
-action(delete_vhost, Node, Args = [_VHostPath], _Opts, Inform) ->
- Inform("Deleting vhost \"~s\"", Args),
- call(Node, {rabbit_vhost, delete, Args});
-
-action(trace_on, Node, [], Opts, Inform) ->
- VHost = proplists:get_value(?VHOST_OPT, Opts),
- Inform("Starting tracing for vhost \"~s\"", [VHost]),
- rpc_call(Node, rabbit_trace, start, [list_to_binary(VHost)]);
-
-action(trace_off, Node, [], Opts, Inform) ->
- VHost = proplists:get_value(?VHOST_OPT, Opts),
- Inform("Stopping tracing for vhost \"~s\"", [VHost]),
- rpc_call(Node, rabbit_trace, stop, [list_to_binary(VHost)]);
-
-action(set_vm_memory_high_watermark, Node, [Arg], _Opts, Inform) ->
- Frac = list_to_float(case string:chr(Arg, $.) of
- 0 -> Arg ++ ".0";
- _ -> Arg
- end),
- Inform("Setting memory threshold on ~p to ~p", [Node, Frac]),
- rpc_call(Node, vm_memory_monitor, set_vm_memory_high_watermark, [Frac]);
-
-action(set_vm_memory_high_watermark, Node, ["absolute", Arg], _Opts, Inform) ->
- case rabbit_resource_monitor_misc:parse_information_unit(Arg) of
- {ok, Limit} ->
- Inform("Setting memory threshold on ~p to ~p bytes", [Node, Limit]),
- rpc_call(Node, vm_memory_monitor, set_vm_memory_high_watermark,
- [{absolute, Limit}]);
- {error, parse_error} ->
- {error_string, rabbit_misc:format(
- "Unable to parse absolute memory limit value ~p", [Arg])}
- end;
-
-action(set_disk_free_limit, Node, [Arg], _Opts, Inform) ->
- case rabbit_resource_monitor_misc:parse_information_unit(Arg) of
- {ok, Limit} ->
- Inform("Setting disk free limit on ~p to ~p bytes", [Node, Limit]),
- rpc_call(Node, rabbit_disk_monitor, set_disk_free_limit, [Limit]);
- {error, parse_error} ->
- {error_string, rabbit_misc:format(
- "Unable to parse disk free limit value ~p", [Arg])}
- end;
-
-action(set_disk_free_limit, Node, ["mem_relative", Arg], _Opts, Inform) ->
- Frac = list_to_float(case string:chr(Arg, $.) of
- 0 -> Arg ++ ".0";
- _ -> Arg
- end),
- Inform("Setting disk free limit on ~p to ~p of total RAM", [Node, Frac]),
- rpc_call(Node,
- rabbit_disk_monitor,
- set_disk_free_limit,
- [{mem_relative, Frac}]);
-
-
-action(set_permissions, Node, [Username, CPerm, WPerm, RPerm], Opts, Inform) ->
- VHost = proplists:get_value(?VHOST_OPT, Opts),
- Inform("Setting permissions for user \"~s\" in vhost \"~s\"",
- [Username, VHost]),
- call(Node, {rabbit_auth_backend_internal, set_permissions,
- [Username, VHost, CPerm, WPerm, RPerm]});
-
-action(clear_permissions, Node, [Username], Opts, Inform) ->
- VHost = proplists:get_value(?VHOST_OPT, Opts),
- Inform("Clearing permissions for user \"~s\" in vhost \"~s\"",
- [Username, VHost]),
- call(Node, {rabbit_auth_backend_internal, clear_permissions,
- [Username, VHost]});
-
-action(set_parameter, Node, [Component, Key, Value], Opts, Inform) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Setting runtime parameter ~p for component ~p to ~p",
- [Key, Component, Value]),
- rpc_call(
- Node, rabbit_runtime_parameters, parse_set,
- [VHostArg, list_to_binary(Component), list_to_binary(Key), Value, none]);
-
-action(clear_parameter, Node, [Component, Key], Opts, Inform) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Clearing runtime parameter ~p for component ~p", [Key, Component]),
- rpc_call(Node, rabbit_runtime_parameters, clear, [VHostArg,
- list_to_binary(Component),
- list_to_binary(Key)]);
-
-action(set_global_parameter, Node, [Key, Value], _Opts, Inform) ->
- Inform("Setting global runtime parameter ~p to ~p", [Key, Value]),
- rpc_call(
- Node, rabbit_runtime_parameters, parse_set_global,
- [rabbit_data_coercion:to_atom(Key), rabbit_data_coercion:to_binary(Value)]
- );
-
-action(clear_global_parameter, Node, [Key], _Opts, Inform) ->
- Inform("Clearing global runtime parameter ~p", [Key]),
- rpc_call(
- Node, rabbit_runtime_parameters, clear_global,
- [rabbit_data_coercion:to_atom(Key)]
- );
-
-action(set_policy, Node, [Key, Pattern, Defn], Opts, Inform) ->
- Msg = "Setting policy ~p for pattern ~p to ~p with priority ~p",
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- PriorityArg = proplists:get_value(?PRIORITY_OPT, Opts),
- ApplyToArg = list_to_binary(proplists:get_value(?APPLY_TO_OPT, Opts)),
- Inform(Msg, [Key, Pattern, Defn, PriorityArg]),
- Res = rpc_call(
- Node, rabbit_policy, parse_set,
- [VHostArg, list_to_binary(Key), list_to_binary(Pattern), list_to_binary(Defn), list_to_binary(PriorityArg), ApplyToArg]),
- case Res of
- {error, Format, Args} when is_list(Format) andalso is_list(Args) ->
- {error_string, rabbit_misc:format(Format, Args)};
- _ ->
- Res
- end;
-
-action(clear_policy, Node, [Key], Opts, Inform) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Clearing policy ~p", [Key]),
- rpc_call(Node, rabbit_policy, delete, [VHostArg, list_to_binary(Key)]);
-
-action(set_operator_policy, Node, [Key, Pattern, Defn], Opts, Inform) ->
- Msg = "Setting operator policy override ~p for pattern ~p to ~p with priority ~p",
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- PriorityArg = proplists:get_value(?PRIORITY_OPT, Opts),
- ApplyToArg = list_to_binary(proplists:get_value(?APPLY_TO_OPT, Opts)),
- Inform(Msg, [Key, Pattern, Defn, PriorityArg]),
- Res = rpc_call(
- Node, rabbit_policy, parse_set_op,
- [VHostArg, list_to_binary(Key), list_to_binary(Pattern), list_to_binary(Defn), list_to_binary(PriorityArg), ApplyToArg]),
- case Res of
- {error, Format, Args} when is_list(Format) andalso is_list(Args) ->
- {error_string, rabbit_misc:format(Format, Args)};
- _ ->
- Res
- end;
-
-action(clear_operator_policy, Node, [Key], Opts, Inform) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Clearing operator policy ~p", [Key]),
- rpc_call(Node, rabbit_policy, delete_op, [VHostArg, list_to_binary(Key)]);
-
-action(set_vhost_limits, Node, [Defn], Opts, Inform) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Setting vhost limits for vhost ~p", [VHostArg]),
- rpc_call(Node, rabbit_vhost_limit, parse_set, [VHostArg, Defn]),
- ok;
-
-action(clear_vhost_limits, Node, [], Opts, Inform) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Clearing vhost ~p limits", [VHostArg]),
- rpc_call(Node, rabbit_vhost_limit, clear, [VHostArg]);
-
-action(report, Node, _Args, _Opts, Inform) ->
- Inform("Reporting server status on ~p~n~n", [erlang:universaltime()]),
- [begin ok = action(Action, N, [], [], Inform), io:nl() end ||
- N <- unsafe_rpc(Node, rabbit_mnesia, cluster_nodes, [running]),
- Action <- [status, cluster_status, environment]],
- VHosts = unsafe_rpc(Node, rabbit_vhost, list, []),
- [print_report(Node, Q) || Q <- ?GLOBAL_QUERIES],
- [print_report(Node, Q, [V]) || Q <- ?VHOST_QUERIES, V <- VHosts],
- ok;
-
-action(set_cluster_name, Node, [Name], _Opts, Inform) ->
- Inform("Setting cluster name to ~s", [Name]),
- rpc_call(Node, rabbit_nodes, set_cluster_name, [list_to_binary(Name)]);
-
-action(eval, Node, [Expr], _Opts, _Inform) ->
- case erl_scan:string(Expr) of
- {ok, Scanned, _} ->
- case erl_parse:parse_exprs(Scanned) of
- {ok, Parsed} -> {value, Value, _} =
- unsafe_rpc(
- Node, erl_eval, exprs, [Parsed, []]),
- io:format("~p~n", [Value]),
- ok;
- {error, E} -> {error_string, format_parse_error(E)}
- end;
- {error, E, _} ->
- {error_string, format_parse_error(E)}
- end;
-
-action(help, _Node, _Args, _Opts, _Inform) ->
- io:format("~s", [rabbit_ctl_usage:usage()]);
-
-action(encode, _Node, Args, Opts, _Inform) ->
- ListCiphers = lists:member({?LIST_CIPHERS_OPT, true}, Opts),
- ListHashes = lists:member({?LIST_HASHES_OPT, true}, Opts),
- Decode = lists:member({?DECODE_OPT, true}, Opts),
- Cipher = list_to_atom(proplists:get_value(?CIPHER_OPT, Opts)),
- Hash = list_to_atom(proplists:get_value(?HASH_OPT, Opts)),
- Iterations = list_to_integer(proplists:get_value(?ITERATIONS_OPT, Opts)),
-
- {_, Msg} = rabbit_control_pbe:encode(ListCiphers, ListHashes, Decode, Cipher, Hash, Iterations, Args),
- io:format(Msg ++ "~n");
-
-action(Command, Node, Args, Opts, Inform) ->
- %% For backward compatibility, run commands accepting a timeout with
- %% the default timeout.
- action(Command, Node, Args, Opts, Inform, ?RPC_TIMEOUT).
-
-action(purge_queue, _Node, [], _Opts, _Inform, _Timeout) ->
- {error, "purge_queue takes queue name as an argument"};
-
-action(purge_queue, Node, [Q], Opts, Inform, Timeout) ->
- VHost = proplists:get_value(?VHOST_OPT, Opts),
- QRes = rabbit_misc:r(list_to_binary(VHost), queue, list_to_binary(Q)),
- Inform("Purging ~s", [rabbit_misc:rs(QRes)]),
- rpc_call(Node, rabbit_control_main, purge_queue, [QRes], Timeout);
-
-action(list_users, Node, [], _Opts, Inform, Timeout) ->
- Inform("Listing users", []),
- call_emitter(Node, {rabbit_auth_backend_internal, list_users, []},
- rabbit_auth_backend_internal:user_info_keys(),
- [{timeout, Timeout}, to_bin_utf8]);
-
-action(list_permissions, Node, [], Opts, Inform, Timeout) ->
- VHost = proplists:get_value(?VHOST_OPT, Opts),
- Inform("Listing permissions in vhost \"~s\"", [VHost]),
- call_emitter(Node, {rabbit_auth_backend_internal, list_vhost_permissions, [VHost]},
- rabbit_auth_backend_internal:vhost_perms_info_keys(),
- [{timeout, Timeout}, to_bin_utf8, is_escaped]);
-
-action(list_parameters, Node, [], Opts, Inform, Timeout) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Listing runtime parameters", []),
- call_emitter(Node, {rabbit_runtime_parameters, list_formatted, [VHostArg]},
- rabbit_runtime_parameters:info_keys(),
- [{timeout, Timeout}]);
-
-action(list_global_parameters, Node, [], _Opts, Inform, Timeout) ->
- Inform("Listing global runtime parameters", []),
- call_emitter(Node, {rabbit_runtime_parameters, list_global_formatted, []},
- rabbit_runtime_parameters:global_info_keys(),
- [{timeout, Timeout}]);
-
-action(list_policies, Node, [], Opts, Inform, Timeout) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Listing policies", []),
- call_emitter(Node, {rabbit_policy, list_formatted, [VHostArg]},
- rabbit_policy:info_keys(),
- [{timeout, Timeout}]);
-
-action(list_operator_policies, Node, [], Opts, Inform, Timeout) ->
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Inform("Listing policies", []),
- call_emitter(Node, {rabbit_policy, list_formatted_op, [VHostArg]},
- rabbit_policy:info_keys(),
- [{timeout, Timeout}]);
-
-
-action(list_vhosts, Node, Args, _Opts, Inform, Timeout) ->
- Inform("Listing vhosts", []),
- ArgAtoms = default_if_empty(Args, [name]),
- call_emitter(Node, {rabbit_vhost, info_all, []}, ArgAtoms,
- [{timeout, Timeout}, to_bin_utf8]);
-
-action(list_user_permissions, _Node, _Args = [], _Opts, _Inform, _Timeout) ->
- {error_string,
- "list_user_permissions expects a username argument, but none provided."};
-action(list_user_permissions, Node, Args = [_Username], _Opts, Inform, Timeout) ->
- Inform("Listing permissions for user ~p", Args),
- call_emitter(Node, {rabbit_auth_backend_internal, list_user_permissions, Args},
- rabbit_auth_backend_internal:user_perms_info_keys(),
- [{timeout, Timeout}, to_bin_utf8, is_escaped]);
-
-action(list_queues, Node, Args, Opts, Inform, 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]),
-
- %% Data for emission
- Nodes = nodes_in_cluster(Node, Timeout),
- ChunksOpt = {chunks, get_number_of_chunks(Filter, Nodes)},
- TimeoutOpt = {timeout, Timeout},
- EmissionRef = make_ref(),
- EmissionRefOpt = {ref, EmissionRef},
-
- case Filter of
- all ->
- start_emission(Node, {rabbit_amqqueue, emit_info_all,
- [Nodes, VHostArg, ArgAtoms]},
- [TimeoutOpt, EmissionRefOpt]),
- start_emission(Node, {rabbit_amqqueue, emit_info_down,
- [VHostArg, ArgAtoms]},
- [TimeoutOpt, EmissionRefOpt]);
- online ->
- start_emission(Node, {rabbit_amqqueue, emit_info_all,
- [Nodes, VHostArg, ArgAtoms]},
- [TimeoutOpt, EmissionRefOpt]);
- offline ->
- start_emission(Node, {rabbit_amqqueue, emit_info_down,
- [VHostArg, ArgAtoms]},
- [TimeoutOpt, EmissionRefOpt]);
- local ->
- start_emission(Node, {rabbit_amqqueue, emit_info_local,
- [VHostArg, ArgAtoms]},
- [TimeoutOpt, EmissionRefOpt])
- end,
- display_emission_result(EmissionRef, ArgAtoms, [ChunksOpt, TimeoutOpt]);
- {error, ErrStr} ->
- {error_string, ErrStr}
- end;
-
-action(list_exchanges, Node, Args, Opts, Inform, Timeout) ->
- Inform("Listing exchanges", []),
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- ArgAtoms = default_if_empty(Args, [name, type]),
- call_emitter(Node, {rabbit_exchange, info_all, [VHostArg, ArgAtoms]},
- ArgAtoms, [{timeout, Timeout}]);
-
-action(list_bindings, Node, Args, Opts, Inform, Timeout) ->
- Inform("Listing bindings", []),
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- ArgAtoms = default_if_empty(Args, [source_name, source_kind,
- destination_name, destination_kind,
- routing_key, arguments]),
- call_emitter(Node, {rabbit_binding, info_all, [VHostArg, ArgAtoms]},
- ArgAtoms, [{timeout, Timeout}]);
-
-action(list_connections, Node, Args, _Opts, Inform, Timeout) ->
- Inform("Listing connections", []),
- ArgAtoms = default_if_empty(Args, [user, peer_host, peer_port, state]),
- Nodes = nodes_in_cluster(Node, Timeout),
- call_emitter(Node, {rabbit_networking, emit_connection_info_all, [Nodes, ArgAtoms]},
- ArgAtoms, [{timeout, Timeout}, {chunks, length(Nodes)}]);
-
-action(list_channels, Node, Args, _Opts, Inform, Timeout) ->
- Inform("Listing channels", []),
- ArgAtoms = default_if_empty(Args, [pid, user, consumer_count,
- messages_unacknowledged]),
- Nodes = nodes_in_cluster(Node, Timeout),
- call_emitter(Node, {rabbit_channel, emit_info_all, [Nodes, ArgAtoms]}, ArgAtoms,
- [{timeout, Timeout}, {chunks, length(Nodes)}]);
-
-action(list_consumers, Node, _Args, Opts, Inform, Timeout) ->
- Inform("Listing consumers", []),
- VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
- Nodes = nodes_in_cluster(Node, Timeout),
- call_emitter(Node, {rabbit_amqqueue, emit_consumers_all, [Nodes, VHostArg]},
- rabbit_amqqueue:consumer_info_keys(),
- [{timeout, Timeout}, {chunks, length(Nodes)}]);
-
-action(node_health_check, Node, _Args, _Opts, Inform, Timeout) ->
- Inform("Checking health of node ~p", [Node]),
- case rabbit_health_check:node(Node, Timeout) of
- ok ->
- io:format("Health check passed~n"),
- ok;
- Other ->
- Other
- end.
-
-format_parse_error({_Line, Mod, Err}) -> lists:flatten(Mod:format_error(Err)).
-
-sync_queue(Q) ->
- rabbit_mirror_queue_misc:sync_queue(Q).
-
-cancel_sync_queue(Q) ->
- rabbit_mirror_queue_misc:cancel_sync_queue(Q).
-
-purge_queue(Q) ->
- rabbit_amqqueue:with(
- Q, fun(Q1) ->
- rabbit_amqqueue:purge(Q1),
- ok
- end).
-
-%%----------------------------------------------------------------------------
-
-require_mnesia_stopped(Node, Fun) ->
- case Fun() of
- {error, mnesia_unexpectedly_running} ->
- {error_string, rabbit_misc:format(
- " Mnesia is still running on node ~p.
- Please stop the node with rabbitmqctl stop_app first.", [Node])};
- Other -> Other
- end.
-
-wait_for_application(Node, PidFile, Application, Inform) ->
- Pid = read_pid_file(PidFile, true),
- Inform("pid is ~s", [Pid]),
- wait_for_application(Node, Pid, Application).
-
-wait_for_application(Node, Pid, rabbit_and_plugins) ->
- wait_for_startup(Node, Pid);
-wait_for_application(Node, Pid, Application) ->
- while_process_is_alive(
- Node, Pid, fun() -> rabbit_nodes:is_running(Node, Application) end).
-
-wait_for_startup(Node, Pid) ->
- while_process_is_alive(
- Node, Pid, fun() -> rpc:call(Node, rabbit, await_startup, []) =:= ok end).
-
-while_process_is_alive(Node, Pid, Activity) ->
- case rabbit_misc:is_os_process_alive(Pid) of
- true -> case Activity() of
- true -> ok;
- false -> timer:sleep(?EXTERNAL_CHECK_INTERVAL),
- while_process_is_alive(Node, Pid, Activity)
- end;
- false -> {error, process_not_running}
- end.
-
-wait_for_process_death(Pid) ->
- case rabbit_misc:is_os_process_alive(Pid) of
- true -> timer:sleep(?EXTERNAL_CHECK_INTERVAL),
- wait_for_process_death(Pid);
- false -> ok
- end.
-
-read_pid_file(PidFile, Wait) ->
- case {file:read_file(PidFile), Wait} of
- {{ok, Bin}, _} ->
- S = binary_to_list(Bin),
- {match, [PidS]} = re:run(S, "[^\\s]+",
- [{capture, all, list}]),
- try list_to_integer(PidS)
- catch error:badarg ->
- exit({error, {garbage_in_pid_file, PidFile}})
- end,
- PidS;
- {{error, enoent}, true} ->
- timer:sleep(?EXTERNAL_CHECK_INTERVAL),
- read_pid_file(PidFile, Wait);
- {{error, _} = E, _} ->
- exit({error, {could_not_read_pid, E}})
- end.
-
-become(BecomeNode) ->
- error_logger:tty(false),
- case net_adm:ping(BecomeNode) of
- pong -> exit({node_running, BecomeNode});
- pang -> ok = net_kernel:stop(),
- io:format(" * Impersonating node: ~s...", [BecomeNode]),
- {ok, _} = rabbit_cli:start_distribution(BecomeNode),
- io:format(" done~n", []),
- Dir = mnesia:system_info(directory),
- io:format(" * Mnesia directory : ~s~n", [Dir])
- end.
-
-%%----------------------------------------------------------------------------
-
-default_if_empty(List, Default) when is_list(List) ->
- if List == [] -> Default;
- true -> [list_to_atom(X) || X <- List]
- end.
-
-display_info_message_row(IsEscaped, Result, InfoItemKeys) ->
- display_row([format_info_item(
- case proplists:lookup(X, Result) of
- none when is_list(Result), length(Result) > 0 ->
- exit({error, {bad_info_key, X}});
- none -> Result;
- {X, Value} -> Value
- end, IsEscaped) || X <- InfoItemKeys]).
-
-display_info_message(IsEscaped, InfoItemKeys) ->
- fun ([], _) ->
- ok;
- ([FirstResult|_] = List, _) when is_list(FirstResult) ->
- lists:foreach(fun(Result) ->
- display_info_message_row(IsEscaped, Result, InfoItemKeys)
- end,
- List),
- ok;
- (Result, _) ->
- display_info_message_row(IsEscaped, Result, InfoItemKeys),
- ok
- end.
-
-display_info_list(Results, InfoItemKeys) when is_list(Results) ->
- lists:foreach(
- fun (Result) -> display_row(
- [format_info_item(proplists:get_value(X, Result), true)
- || X <- InfoItemKeys])
- end, lists:sort(Results)),
- ok;
-display_info_list(Other, _) ->
- Other.
-
-display_row(Row) ->
- io:fwrite(string:join(Row, "\t")),
- io:nl().
-
--define(IS_U8(X), (X >= 0 andalso X =< 255)).
--define(IS_U16(X), (X >= 0 andalso X =< 65535)).
-
-format_info_item(#resource{name = Name}, IsEscaped) ->
- escape(Name, IsEscaped);
-format_info_item({N1, N2, N3, N4} = Value, _IsEscaped) when
- ?IS_U8(N1), ?IS_U8(N2), ?IS_U8(N3), ?IS_U8(N4) ->
- rabbit_misc:ntoa(Value);
-format_info_item({K1, K2, K3, K4, K5, K6, K7, K8} = Value, _IsEscaped) when
- ?IS_U16(K1), ?IS_U16(K2), ?IS_U16(K3), ?IS_U16(K4),
- ?IS_U16(K5), ?IS_U16(K6), ?IS_U16(K7), ?IS_U16(K8) ->
- rabbit_misc:ntoa(Value);
-format_info_item(Value, _IsEscaped) when is_pid(Value) ->
- rabbit_misc:pid_to_string(Value);
-format_info_item(Value, IsEscaped) when is_binary(Value) ->
- escape(Value, IsEscaped);
-format_info_item(Value, IsEscaped) when is_atom(Value) ->
- escape(atom_to_list(Value), IsEscaped);
-format_info_item([{TableEntryKey, TableEntryType, _TableEntryValue} | _] =
- Value, IsEscaped) when is_binary(TableEntryKey) andalso
- is_atom(TableEntryType) ->
- io_lib:format("~1000000000000p", [prettify_amqp_table(Value, IsEscaped)]);
-format_info_item([T | _] = Value, IsEscaped)
- when is_tuple(T) orelse is_pid(T) orelse is_binary(T) orelse is_atom(T) orelse
- is_list(T) ->
- "[" ++
- lists:nthtail(2, lists:append(
- [", " ++ format_info_item(E, IsEscaped)
- || E <- Value])) ++ "]";
-format_info_item({Key, Value}, IsEscaped) ->
- "{" ++ io_lib:format("~p", [Key]) ++ ", " ++
- format_info_item(Value, IsEscaped) ++ "}";
-format_info_item(Value, _IsEscaped) ->
- io_lib:format("~w", [Value]).
-
-display_call_result(Node, MFA) ->
- case call(Node, MFA) of
- {badrpc, _} = Res -> throw(Res);
- Res -> io:format("~p~n", [Res]),
- ok
- end.
-
-unsafe_rpc(Node, Mod, Fun, Args) ->
- unsafe_rpc(Node, Mod, Fun, Args, ?RPC_TIMEOUT).
-
-unsafe_rpc(Node, Mod, Fun, Args, Timeout) ->
- case rpc_call(Node, Mod, Fun, Args, Timeout) of
- {badrpc, _} = Res -> throw(Res);
- Normal -> Normal
- end.
-
-ensure_app_running(Node) ->
- case call(Node, {rabbit, is_running, []}) of
- true -> ok;
- false -> {error_string,
- rabbit_misc:format(
- "rabbit application is not running on node ~s.~n"
- " * Suggestion: start it with \"rabbitmqctl start_app\" "
- "and try again", [Node])};
- Other -> Other
- end.
-
-call(Node, {Mod, Fun, Args}) ->
- rpc_call(Node, Mod, Fun, lists:map(fun list_to_binary_utf8/1, Args)).
-
-call_emitter(Node, {Mod, Fun, Args}, InfoKeys, Opts) ->
- Ref = start_emission(Node, {Mod, Fun, Args}, Opts),
- display_emission_result(Ref, InfoKeys, Opts).
-
-start_emission(Node, {Mod, Fun, Args}, Opts) ->
- ToBinUtf8 = proplists:get_value(to_bin_utf8, Opts, false),
- Timeout = proplists:get_value(timeout, Opts, infinity),
- Ref = proplists:get_value(ref, Opts, make_ref()),
- rabbit_control_misc:spawn_emitter_caller(
- Node, Mod, Fun, prepare_call_args(Args, ToBinUtf8),
- Ref, self(), Timeout),
- Ref.
-
-display_emission_result(Ref, InfoKeys, Opts) ->
- IsEscaped = proplists:get_value(is_escaped, Opts, false),
- Chunks = proplists:get_value(chunks, Opts, 1),
- Timeout = proplists:get_value(timeout, Opts, infinity),
- EmissionStatus = rabbit_control_misc:wait_for_info_messages(
- self(), Ref, display_info_message(IsEscaped, InfoKeys), ok, Timeout, Chunks),
- emission_to_action_result(EmissionStatus).
-
-%% Convert rabbit_control_misc:wait_for_info_messages/6 return value
-%% into form expected by rabbit_cli:main/3.
-emission_to_action_result({ok, ok}) ->
- ok;
-emission_to_action_result({error, Error}) ->
- Error.
-
-prepare_call_args(Args, ToBinUtf8) ->
- case ToBinUtf8 of
- true -> valid_utf8_args(Args);
- false -> Args
- end.
-
-valid_utf8_args(Args) ->
- lists:map(fun list_to_binary_utf8/1, Args).
-
-list_to_binary_utf8(L) ->
- B = list_to_binary(L),
- case rabbit_binary_parser:validate_utf8(B) of
- ok -> B;
- error -> throw({error, {not_utf_8, L}})
- end.
-
-%% escape does C-style backslash escaping of non-printable ASCII
-%% characters. We don't escape characters above 127, since they may
-%% form part of UTF-8 strings.
-
-escape(Atom, IsEscaped) when is_atom(Atom) ->
- escape(atom_to_list(Atom), IsEscaped);
-escape(Bin, IsEscaped) when is_binary(Bin) ->
- escape(binary_to_list(Bin), IsEscaped);
-escape(L, false) when is_list(L) ->
- escape_char(lists:reverse(L), []);
-escape(L, true) when is_list(L) ->
- L.
-
-escape_char([$\\ | T], Acc) ->
- escape_char(T, [$\\, $\\ | Acc]);
-escape_char([X | T], Acc) when X >= 32, X /= 127 ->
- escape_char(T, [X | Acc]);
-escape_char([X | T], Acc) ->
- escape_char(T, [$\\, $0 + (X bsr 6), $0 + (X band 8#070 bsr 3),
- $0 + (X band 7) | Acc]);
-escape_char([], Acc) ->
- Acc.
-
-prettify_amqp_table(Table, IsEscaped) ->
- [{escape(K, IsEscaped), prettify_typed_amqp_value(T, V, IsEscaped)}
- || {K, T, V} <- Table].
-
-prettify_typed_amqp_value(longstr, Value, IsEscaped) ->
- escape(Value, IsEscaped);
-prettify_typed_amqp_value(table, Value, IsEscaped) ->
- prettify_amqp_table(Value, IsEscaped);
-prettify_typed_amqp_value(array, Value, IsEscaped) ->
- [prettify_typed_amqp_value(T, V, IsEscaped) || {T, V} <- Value];
-prettify_typed_amqp_value(_Type, Value, _IsEscaped) ->
- Value.
-
-split_list([]) -> [];
-split_list([_]) -> exit(even_list_needed);
-split_list([A, B | T]) -> [{A, B} | split_list(T)].
-
-nodes_in_cluster(Node) ->
- unsafe_rpc(Node, rabbit_mnesia, cluster_nodes, [running], ?RPC_TIMEOUT).
-
-nodes_in_cluster(Node, Timeout) ->
- unsafe_rpc(Node, rabbit_mnesia, cluster_nodes, [running], Timeout).
-
-alarms_by_node(Name) ->
- case rpc_call(Name, rabbit, status, []) of
- {badrpc,nodedown} -> {Name, [nodedown]};
- Status ->
- {_, As} = lists:keyfind(alarms, 1, Status),
- {Name, As}
- end.
-
-get_number_of_chunks(all, Nodes) ->
- length(Nodes) + 1;
-get_number_of_chunks(online, Nodes) ->
- length(Nodes);
-get_number_of_chunks(offline, _) ->
- 1;
-get_number_of_chunks(local, _) ->
- 1.
diff --git a/src/rabbit_mnesia_rename.erl b/src/rabbit_mnesia_rename.erl
index 2d7e0f56b6..bcaaf117ff 100644
--- a/src/rabbit_mnesia_rename.erl
+++ b/src/rabbit_mnesia_rename.erl
@@ -70,13 +70,13 @@ rename(Node, NodeMapList) ->
ok = rabbit_mnesia:copy_db(mnesia_copy_dir()),
%% And make the actual changes
- rabbit_control_main:become(FromNode),
+ become(FromNode),
take_backup(before_backup_name()),
convert_backup(NodeMap, before_backup_name(), after_backup_name()),
ok = rabbit_file:write_term_file(rename_config_name(),
[{FromNode, ToNode}]),
convert_config_files(NodeMap),
- rabbit_control_main:become(ToNode),
+ become(ToNode),
restore_backup(after_backup_name()),
ok
after
@@ -267,3 +267,15 @@ transform_table(Table, Map, Key) ->
[Term] = mnesia:read(Table, Key, write),
ok = mnesia:write(Table, update_term(Map, Term), write),
transform_table(Table, Map, mnesia:next(Table, Key)).
+
+become(BecomeNode) ->
+ error_logger:tty(false),
+ case net_adm:ping(BecomeNode) of
+ pong -> exit({node_running, BecomeNode});
+ pang -> ok = net_kernel:stop(),
+ io:format(" * Impersonating node: ~s...", [BecomeNode]),
+ {ok, _} = rabbit_cli:start_distribution(BecomeNode),
+ io:format(" done~n", []),
+ Dir = mnesia:system_info(directory),
+ io:format(" * Mnesia directory : ~s~n", [Dir])
+ end.
diff --git a/src/rabbit_runtime_parameters.erl b/src/rabbit_runtime_parameters.erl
index 94018a5b54..cee5408f0a 100644
--- a/src/rabbit_runtime_parameters.erl
+++ b/src/rabbit_runtime_parameters.erl
@@ -27,7 +27,7 @@
%%
%% The most obvious use case for runtime parameters is policies but
%% there are others:
-%%
+%%
%% * Plugin-specific parameters that only make sense at runtime,
%% e.g. Federation and Shovel link settings
%% * Exchange and queue decorators
diff --git a/test/clustering_management_SUITE.erl b/test/clustering_management_SUITE.erl
index eac5fa3683..9b605d07ae 100644
--- a/test/clustering_management_SUITE.erl
+++ b/test/clustering_management_SUITE.erl
@@ -644,6 +644,7 @@ assert_not_clustered(Node) ->
assert_failure(Fun) ->
case catch Fun() of
+ {error, Code, Reason} -> Reason;
{error, Reason} -> Reason;
{error_string, Reason} -> Reason;
{badrpc, {'EXIT', Reason}} -> Reason;
@@ -652,35 +653,35 @@ assert_failure(Fun) ->
end.
stop_app(Node) ->
- control_action(stop_app, Node).
+ rabbit_control_helper:command(stop_app, Node).
start_app(Node) ->
- control_action(start_app, Node).
+ rabbit_control_helper:command(start_app, Node).
join_cluster(Node, To) ->
join_cluster(Node, To, false).
join_cluster(Node, To, Ram) ->
- control_action(join_cluster, Node, [atom_to_list(To)], [{"--ram", Ram}]).
+ rabbit_control_helper:command_with_output(join_cluster, Node, [atom_to_list(To)], [{"--ram", Ram}]).
reset(Node) ->
- control_action(reset, Node).
+ rabbit_control_helper:command(reset, Node).
force_reset(Node) ->
- control_action(force_reset, Node).
+ rabbit_control_helper:command(force_reset, Node).
forget_cluster_node(Node, Removee, RemoveWhenOffline) ->
- control_action(forget_cluster_node, Node, [atom_to_list(Removee)],
+ rabbit_control_helper:command(forget_cluster_node, Node, [atom_to_list(Removee)],
[{"--offline", RemoveWhenOffline}]).
forget_cluster_node(Node, Removee) ->
forget_cluster_node(Node, Removee, false).
change_cluster_node_type(Node, Type) ->
- control_action(change_cluster_node_type, Node, [atom_to_list(Type)]).
+ rabbit_control_helper:command(change_cluster_node_type, Node, [atom_to_list(Type)]).
update_cluster_nodes(Node, DiscoveryNode) ->
- control_action(update_cluster_nodes, Node, [atom_to_list(DiscoveryNode)]).
+ rabbit_control_helper:command(update_cluster_nodes, Node, [atom_to_list(DiscoveryNode)]).
stop_join_start(Node, ClusterTo, Ram) ->
ok = stop_app(Node),
@@ -695,17 +696,6 @@ stop_reset_start(Node) ->
ok = reset(Node),
ok = start_app(Node).
-control_action(Command, Node) ->
- control_action(Command, Node, [], []).
-
-control_action(Command, Node, Args) ->
- control_action(Command, Node, Args, []).
-
-control_action(Command, Node, Args, Opts) ->
- rpc:call(Node, rabbit_control_main, action,
- [Command, Node, Args, Opts,
- fun io:format/2]).
-
declare(Ch, Name) ->
Res = amqp_channel:call(Ch, #'queue.declare'{durable = true,
queue = Name}),
diff --git a/test/eager_sync_SUITE.erl b/test/eager_sync_SUITE.erl
index 93b308b6c5..70d7080269 100644
--- a/test/eager_sync_SUITE.erl
+++ b/test/eager_sync_SUITE.erl
@@ -23,7 +23,7 @@
-define(QNAME, <<"ha.two.test">>).
-define(QNAME_AUTO, <<"ha.auto.test">>).
--define(MESSAGE_COUNT, 2000).
+-define(MESSAGE_COUNT, 200000).
all() ->
[
@@ -135,9 +135,11 @@ eager_sync_cancel(Config) ->
amqp_channel:call(ACh, #'queue.declare'{queue = ?QNAME,
durable = true}),
{ok, not_syncing} = sync_cancel(C, ?QNAME), %% Idempotence
- eager_sync_cancel_test2(Config, A, B, C, Ch).
+ eager_sync_cancel_test2(Config, A, B, C, Ch, 100).
-eager_sync_cancel_test2(Config, A, B, C, Ch) ->
+eager_sync_cancel_test2(_, _, _, _, _, 0) ->
+ error(no_more_attempts_left);
+eager_sync_cancel_test2(Config, A, B, C, Ch, Attempts) ->
%% Sync then cancel
rabbit_ct_client_helpers:publish(Ch, ?QNAME, ?MESSAGE_COUNT),
restart(Config, A),
@@ -158,12 +160,12 @@ eager_sync_cancel_test2(Config, A, B, C, Ch) ->
%% Damn. Syncing finished between wait_for_syncing/3 and
%% sync_cancel/2 above. Start again.
amqp_channel:call(Ch, #'queue.purge'{queue = ?QNAME}),
- eager_sync_cancel_test2(Config, A, B, C, Ch)
+ eager_sync_cancel_test2(Config, A, B, C, Ch, Attempts - 1)
end;
synced_already ->
%% Damn. Syncing finished before wait_for_syncing/3. Start again.
amqp_channel:call(Ch, #'queue.purge'{queue = ?QNAME}),
- eager_sync_cancel_test2(Config, A, B, C, Ch)
+ eager_sync_cancel_test2(Config, A, B, C, Ch, Attempts - 1)
end.
eager_sync_auto(Config) ->
@@ -240,8 +242,8 @@ wait_for_sync(Node, QName) ->
sync_detection_SUITE:wait_for_sync_status(true, Node, QName).
action(Node, Action, QName) ->
- rabbit_ct_broker_helpers:control_action(
- Action, Node, [binary_to_list(QName)], [{"-p", "/"}]).
+ rabbit_control_helper:command_with_output(
+ Action, Node, [binary_to_list(QName)], [{"-p", "/"}]).
queue(Node, QName) ->
QNameRes = rabbit_misc:r(<<"/">>, queue, QName),
@@ -273,6 +275,6 @@ state(Node, QName) ->
%% in order to pass, because a SyncBatchSize >= ?MESSAGE_COUNT will
%% always finish before the test is able to cancel the sync.
set_app_sync_batch_size(Node) ->
- rabbit_ct_broker_helpers:control_action(
+ rabbit_control_helper:command(
eval, Node,
["application:set_env(rabbit, mirroring_sync_batch_size, 1)."]).
diff --git a/test/per_vhost_connection_limit_SUITE.erl b/test/per_vhost_connection_limit_SUITE.erl
index 592e57c41a..efc5ca830e 100644
--- a/test/per_vhost_connection_limit_SUITE.erl
+++ b/test/per_vhost_connection_limit_SUITE.erl
@@ -795,7 +795,7 @@ set_vhost_connection_limit(Config, VHost, Count) ->
set_vhost_connection_limit(Config, NodeIndex, VHost, Count) ->
Node = rabbit_ct_broker_helpers:get_node_config(
Config, NodeIndex, nodename),
- rabbit_ct_broker_helpers:control_action(
+ ok = rabbit_ct_broker_helpers:control_action(
set_vhost_limits, Node,
["{\"max-connections\": " ++ integer_to_list(Count) ++ "}"],
[{"-p", binary_to_list(VHost)}]).
diff --git a/test/rabbitmqctl_integration_SUITE.erl b/test/rabbitmqctl_integration_SUITE.erl
index ef85472f48..bbc4447102 100644
--- a/test/rabbitmqctl_integration_SUITE.erl
+++ b/test/rabbitmqctl_integration_SUITE.erl
@@ -31,7 +31,6 @@
-export([list_queues_local/1
,list_queues_offline/1
,list_queues_online/1
- ,manage_global_parameters/1
]).
all() ->
@@ -46,8 +45,7 @@ groups() ->
[list_queues_local
,list_queues_online
,list_queues_offline
- ]},
- {global_parameters, [], [manage_global_parameters]}
+ ]}
].
init_per_suite(Config) ->
@@ -63,13 +61,6 @@ init_per_group(list_queues, Config0) ->
Config1 = declare_some_queues(Config),
rabbit_ct_broker_helpers:stop_node(Config1, NumNodes - 1),
Config1;
-init_per_group(global_parameters,Config) ->
- Config1 = rabbit_ct_helpers:set_config(Config, [
- {rmq_nodename_suffix, ?MODULE}
- ]),
- rabbit_ct_helpers:run_setup_steps(Config1,
- rabbit_ct_broker_helpers:setup_steps() ++
- rabbit_ct_client_helpers:setup_steps());
init_per_group(_, Config) ->
Config.
@@ -144,75 +135,6 @@ list_queues_offline(Config) ->
assert_ctl_queues(Config, 1, ["--offline"], OfflineQueues),
ok.
-manage_global_parameters(Config) ->
- 0 = length(global_parameters(Config)),
- Parameter1Key = global_param1,
- GlobalParameter1ValueAsString = "{\"a\":\"b\", \"c\":\"d\"}",
- ok = control_action(Config, set_global_parameter,
- [atom_to_list(Parameter1Key),
- GlobalParameter1ValueAsString
- ]),
-
- 1 = length(global_parameters(Config)),
-
- GlobalParameter1Value = rabbit_ct_broker_helpers:rpc(
- Config, 0,
- rabbit_runtime_parameters, value_global,
- [Parameter1Key]
- ),
-
- [{<<"a">>,<<"b">>}, {<<"c">>,<<"d">>}] = GlobalParameter1Value,
-
- Parameter2Key = global_param2,
- GlobalParameter2ValueAsString = "{\"e\":\"f\", \"g\":\"h\"}",
- ok = control_action(Config, set_global_parameter,
- [atom_to_list(Parameter2Key),
- GlobalParameter2ValueAsString
- ]),
-
- 2 = length(global_parameters(Config)),
-
- GlobalParameter2Value = rabbit_ct_broker_helpers:rpc(
- Config, 0,
- rabbit_runtime_parameters, value_global,
- [Parameter2Key]
- ),
-
- [{<<"e">>,<<"f">>}, {<<"g">>,<<"h">>}] = GlobalParameter2Value,
-
-
- GlobalParameter1Value2AsString = "{\"a\":\"z\", \"c\":\"d\"}",
- ok = control_action(Config, set_global_parameter,
- [atom_to_list(Parameter1Key),
- GlobalParameter1Value2AsString
- ]),
-
- 2 = length(global_parameters(Config)),
-
- GlobalParameter1Value2 = rabbit_ct_broker_helpers:rpc(
- Config, 0,
- rabbit_runtime_parameters, value_global,
- [Parameter1Key]
- ),
-
- [{<<"a">>,<<"z">>}, {<<"c">>,<<"d">>}] = GlobalParameter1Value2,
-
- ok = control_action(Config, clear_global_parameter,
- [atom_to_list(Parameter1Key)]
- ),
-
- 1 = length(global_parameters(Config)),
-
- not_found = rabbit_ct_broker_helpers:rpc(
- Config, 0,
- rabbit_runtime_parameters, value_global,
- [Parameter1Key]
- ),
-
- ok = control_action(Config, list_global_parameters, []),
-
- ok.
-
%%----------------------------------------------------------------------------
%% Helpers
%%----------------------------------------------------------------------------
@@ -234,11 +156,7 @@ run_list_queues(Config, Node, Args) ->
control_action(Config, Command, Args) ->
Node = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
- rabbit_control_main:action(
- Command, Node, Args, [],
- fun (Format, Args1) ->
- io:format(Format ++ " ...~n", Args1)
- end).
+ rabbit_ct_broker_helpers:control_action(Command, Node, Args, []).
global_parameters(Config) ->
rabbit_ct_broker_helpers:rpc(
diff --git a/test/unit_inbroker_SUITE.erl b/test/unit_inbroker_SUITE.erl
index 98fb6ac3c2..af86371fc2 100644
--- a/test/unit_inbroker_SUITE.erl
+++ b/test/unit_inbroker_SUITE.erl
@@ -70,7 +70,8 @@
all() ->
[
{group, parallel_tests},
- {group, non_parallel_tests},
+ {group, non_parallel_tests}
+ ,
{group, backing_queue_tests},
{group, cluster_tests},
@@ -88,24 +89,12 @@ groups() ->
credit_flow_settings,
dynamic_mirroring,
gen_server2_with_state,
- list_operations_timeout_pass,
mcall,
{password_hashing, [], [
password_hashing,
change_password
]},
- {policy_validation, [parallel, {repeat, 20}], [
- ha_policy_validation,
- policy_validation,
- policy_opts_validation,
- queue_master_location_policy_validation,
- queue_modes_policy_validation,
- vhost_removed_while_updating_policy
- ]},
- runtime_parameters,
- set_disk_free_limit_command,
- topic_matching,
- user_management
+ topic_matching
]},
{non_parallel_tests, [], [
app_management, %% Restart RabbitMQ.
@@ -115,9 +104,7 @@ groups() ->
head_message_timestamp_statistics, %% Expect specific statistics.
log_management, %% Check log files.
log_management_during_startup, %% Check log files.
- memory_high_watermark, %% Trigger alarm.
- externally_rotated_logs_are_automatically_reopened, %% Check log files.
- server_status %% Trigger alarm.
+ externally_rotated_logs_are_automatically_reopened %% Check log files.
]},
{backing_queue_tests, [], [
msg_store,
@@ -265,25 +252,41 @@ app_management(Config) ->
?MODULE, app_management1, [Config]).
app_management1(_Config) ->
- control_action(wait, [os:getenv("RABBITMQ_PID_FILE")]),
+ wait_for_application(rabbit),
%% Starting, stopping and diagnostics. Note that we don't try
%% 'report' when the rabbit app is stopped and that we enable
%% tracing for the duration of this function.
- ok = control_action(trace_on, []),
- ok = control_action(stop_app, []),
- ok = control_action(stop_app, []),
- ok = control_action(status, []),
- ok = control_action(cluster_status, []),
- ok = control_action(environment, []),
- ok = control_action(start_app, []),
- ok = control_action(start_app, []),
- ok = control_action(status, []),
- ok = control_action(report, []),
- ok = control_action(cluster_status, []),
- ok = control_action(environment, []),
- ok = control_action(trace_off, []),
+ ok = rabbit_trace:start(<<"/">>),
+ ok = rabbit:stop(),
+ ok = rabbit:stop(),
+ ok = no_exceptions(rabbit, status, []),
+ ok = no_exceptions(rabbit, environment, []),
+ ok = rabbit:start(),
+ ok = rabbit:start(),
+ ok = no_exceptions(rabbit, status, []),
+ ok = no_exceptions(rabbit, environment, []),
+ ok = rabbit_trace:stop(<<"/">>),
passed.
+no_exceptions(Mod, Fun, Args) ->
+ try erlang:apply(Mod, Fun, Args) of _ -> ok
+ catch Type:Ex -> {Type, Ex}
+ end.
+
+wait_for_application(Application) ->
+ wait_for_application(Application, 5000).
+
+wait_for_application(_, Time) when Time =< 0 ->
+ {error, timeout};
+wait_for_application(Application, Time) ->
+ Interval = 100,
+ case lists:keyfind(Application, 1, application:which_applications()) of
+ false ->
+ timer:sleep(Interval),
+ wait_for_application(Application, Time - Interval);
+ _ -> ok
+ end.
+
%% -------------------------------------------------------------------
%% Message store.
%% -------------------------------------------------------------------
@@ -1852,7 +1855,7 @@ log_management1(_Config) ->
ok = test_logs_working([LogFile]),
%% simple log rotation
- ok = control_action(rotate_logs, []),
+ ok = rabbit:rotate_logs(),
%% FIXME: rabbit:rotate_logs/0 is asynchronous due to a limitation
%% in Lager. Therefore, we have no choice but to wait an arbitrary
%% amount of time.
@@ -1862,53 +1865,53 @@ log_management1(_Config) ->
%% log rotation on empty files
ok = clean_logs([LogFile], Suffix),
- ok = control_action(rotate_logs, []),
+ ok = rabbit:rotate_logs(),
timer:sleep(2000),
[{error, enoent}, true] = non_empty_files([LogFile ++ Suffix, LogFile]),
%% logs with suffix are not writable
- ok = control_action(rotate_logs, []),
+ ok = rabbit:rotate_logs(),
timer:sleep(2000),
ok = make_files_non_writable([LogFile ++ Suffix]),
- ok = control_action(rotate_logs, []),
+ ok = rabbit:rotate_logs(),
timer:sleep(2000),
ok = test_logs_working([LogFile]),
%% rotate when original log files are not writable
ok = make_files_non_writable([LogFile]),
- ok = control_action(rotate_logs, []),
+ ok = rabbit:rotate_logs(),
timer:sleep(2000),
%% logging directed to tty (first, remove handlers)
- ok = control_action(stop_app, []),
+ ok = rabbit:stop(),
ok = clean_logs([LogFile], Suffix),
ok = application:set_env(rabbit, lager_handler, tty),
application:unset_env(lager, handlers),
application:unset_env(lager, extra_sinks),
- ok = control_action(start_app, []),
+ ok = rabbit:start(),
timer:sleep(200),
rabbit_log:info("test info"),
[{error, enoent}] = empty_files([LogFile]),
%% rotate logs when logging is turned off
- ok = control_action(stop_app, []),
+ ok = rabbit:stop(),
ok = clean_logs([LogFile], Suffix),
ok = application:set_env(rabbit, lager_handler, false),
application:unset_env(lager, handlers),
application:unset_env(lager, extra_sinks),
- ok = control_action(start_app, []),
+ ok = rabbit:start(),
timer:sleep(200),
rabbit_log:error("test error"),
timer:sleep(200),
[{error, enoent}] = empty_files([LogFile]),
%% cleanup
- ok = control_action(stop_app, []),
+ ok = rabbit:stop(),
ok = clean_logs([LogFile], Suffix),
ok = application:set_env(rabbit, lager_handler, LogFile),
application:unset_env(lager, handlers),
application:unset_env(lager, extra_sinks),
- ok = control_action(start_app, []),
+ ok = rabbit:start(),
ok = test_logs_working([LogFile]),
passed.
@@ -1921,80 +1924,76 @@ log_management_during_startup1(_Config) ->
Suffix = ".0",
%% start application with simple tty logging
- ok = control_action(stop_app, []),
+ ok = rabbit:stop(),
ok = clean_logs([LogFile], Suffix),
ok = application:set_env(rabbit, lager_handler, tty),
application:unset_env(lager, handlers),
application:unset_env(lager, extra_sinks),
- ok = control_action(start_app, []),
+ ok = rabbit:start(),
%% start application with logging to non-existing directory
NonExistent = "/tmp/non-existent/test.log",
delete_file(NonExistent),
delete_file(filename:dirname(NonExistent)),
- ok = control_action(stop_app, []),
+ ok = rabbit:stop(),
ok = application:set_env(rabbit, lager_handler, NonExistent),
application:unset_env(lager, handlers),
application:unset_env(lager, extra_sinks),
- ok = control_action(start_app, []),
+ ok = rabbit:start(),
%% start application with logging to directory with no
%% write permissions
- ok = control_action(stop_app, []),
+ ok = rabbit:stop(),
NoPermission1 = "/var/empty/test.log",
delete_file(NoPermission1),
delete_file(filename:dirname(NoPermission1)),
- ok = control_action(stop_app, []),
+ ok = rabbit:stop(),
ok = application:set_env(rabbit, lager_handler, NoPermission1),
application:unset_env(lager, handlers),
application:unset_env(lager, extra_sinks),
- ok = case control_action(start_app, []) of
- ok -> exit({got_success_but_expected_failure,
- log_rotation_no_write_permission_dir_test});
- {badrpc,
- {'EXIT', {error, {cannot_log_to_file, _, Reason1}}}}
- when Reason1 =:= enoent orelse Reason1 =:= eacces -> ok;
- {badrpc,
- {'EXIT',
- {error, {cannot_log_to_file, _,
- {cannot_create_parent_dirs, _, Reason1}}}}}
- when Reason1 =:= eperm orelse
- Reason1 =:= eacces orelse
- Reason1 =:= enoent-> ok
- end,
+ ok = try rabbit:start() of
+ ok -> exit({got_success_but_expected_failure,
+ log_rotation_no_write_permission_dir_test})
+ catch
+ _:{error, {cannot_log_to_file, _, Reason1}}
+ when Reason1 =:= enoent orelse Reason1 =:= eacces -> ok;
+ _:{error, {cannot_log_to_file, _,
+ {cannot_create_parent_dirs, _, Reason1}}}
+ when Reason1 =:= eperm orelse
+ Reason1 =:= eacces orelse
+ Reason1 =:= enoent-> ok
+ end,
%% start application with logging to a subdirectory which
%% parent directory has no write permissions
NoPermission2 = "/var/empty/non-existent/test.log",
delete_file(NoPermission2),
delete_file(filename:dirname(NoPermission2)),
- case control_action(stop_app, []) of
+ case rabbit:stop() of
ok -> ok;
{error, lager_not_running} -> ok
end,
ok = application:set_env(rabbit, lager_handler, NoPermission2),
application:unset_env(lager, handlers),
application:unset_env(lager, extra_sinks),
- ok = case control_action(start_app, []) of
- ok -> exit({got_success_but_expected_failure,
- log_rotatation_parent_dirs_test});
- {badrpc,
- {'EXIT', {error, {cannot_log_to_file, _, Reason2}}}}
- when Reason2 =:= enoent orelse Reason2 =:= eacces -> ok;
- {badrpc,
- {'EXIT',
- {error, {cannot_log_to_file, _,
- {cannot_create_parent_dirs, _, Reason2}}}}}
- when Reason2 =:= eperm orelse
- Reason2 =:= eacces orelse
- Reason2 =:= enoent-> ok
- end,
+ ok = try rabbit:start() of
+ ok -> exit({got_success_but_expected_failure,
+ log_rotatation_parent_dirs_test})
+ catch
+ _:{error, {cannot_log_to_file, _, Reason2}}
+ when Reason2 =:= enoent orelse Reason2 =:= eacces -> ok;
+ _:{error, {cannot_log_to_file, _,
+ {cannot_create_parent_dirs, _, Reason2}}}
+ when Reason2 =:= eperm orelse
+ Reason2 =:= eacces orelse
+ Reason2 =:= enoent-> ok
+ end,
%% cleanup
ok = application:set_env(rabbit, lager_handler, LogFile),
application:unset_env(lager, handlers),
application:unset_env(lager, extra_sinks),
- ok = control_action(start_app, []),
+ ok = rabbit:start(),
passed.
externally_rotated_logs_are_automatically_reopened(Config) ->
@@ -2185,599 +2184,6 @@ change_password1(_Config) ->
%% rabbitmqctl.
%% -------------------------------------------------------------------
-list_operations_timeout_pass(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, list_operations_timeout_pass1, [Config]).
-
-list_operations_timeout_pass1(Config) ->
- %% create a few things so there is some useful information to list
- {_Writer1, Limiter1, Ch1} = rabbit_ct_broker_helpers:test_channel(),
- {_Writer2, Limiter2, Ch2} = rabbit_ct_broker_helpers:test_channel(),
-
- [Q, Q2] = [Queue || Name <- [<<"list_operations_timeout_pass-q1">>,
- <<"list_operations_timeout_pass-q2">>],
- {new, Queue = #amqqueue{}} <-
- [rabbit_amqqueue:declare(
- rabbit_misc:r(<<"/">>, queue, Name),
- false, false, [], none)]],
-
- ok = rabbit_amqqueue:basic_consume(
- Q, true, Ch1, Limiter1, false, 0, <<"ctag1">>, true, [],
- undefined),
- ok = rabbit_amqqueue:basic_consume(
- Q2, true, Ch2, Limiter2, false, 0, <<"ctag2">>, true, [],
- undefined),
-
- %% list users
- ok = control_action(add_user,
- ["list_operations_timeout_pass-user",
- "list_operations_timeout_pass-password"]),
- {error, {user_already_exists, _}} =
- control_action(add_user,
- ["list_operations_timeout_pass-user",
- "list_operations_timeout_pass-password"]),
- ok = control_action_t(list_users, [], ?TIMEOUT_LIST_OPS_PASS),
-
- %% list parameters
- ok = dummy_runtime_parameters:register(),
- ok = control_action(set_parameter, ["test", "good", "123"]),
- ok = control_action_t(list_parameters, [], ?TIMEOUT_LIST_OPS_PASS),
- ok = control_action(clear_parameter, ["test", "good"]),
- dummy_runtime_parameters:unregister(),
-
- %% list vhosts
- ok = control_action(add_vhost, ["/list_operations_timeout_pass-vhost"]),
- {error, {vhost_already_exists, _}} =
- control_action(add_vhost, ["/list_operations_timeout_pass-vhost"]),
- ok = control_action_t(list_vhosts, [], ?TIMEOUT_LIST_OPS_PASS),
-
- %% list permissions
- ok = control_action(set_permissions,
- ["list_operations_timeout_pass-user", ".*", ".*", ".*"],
- [{"-p", "/list_operations_timeout_pass-vhost"}]),
- ok = control_action_t(list_permissions, [],
- [{"-p", "/list_operations_timeout_pass-vhost"}],
- ?TIMEOUT_LIST_OPS_PASS),
-
- %% list user permissions
- ok = control_action_t(list_user_permissions,
- ["list_operations_timeout_pass-user"],
- ?TIMEOUT_LIST_OPS_PASS),
-
- %% list policies
- ok = control_action_opts(
- ["set_policy", "list_operations_timeout_pass-policy", ".*",
- "{\"ha-mode\":\"all\"}"]),
- ok = control_action_t(list_policies, [], ?TIMEOUT_LIST_OPS_PASS),
- ok = control_action(clear_policy, ["list_operations_timeout_pass-policy"]),
-
- %% list queues
- ok = info_action_t(list_queues,
- rabbit_amqqueue:info_keys(), false,
- ?TIMEOUT_LIST_OPS_PASS),
-
- %% list exchanges
- ok = info_action_t(list_exchanges,
- rabbit_exchange:info_keys(), true,
- ?TIMEOUT_LIST_OPS_PASS),
-
- %% list bindings
- ok = info_action_t(list_bindings,
- rabbit_binding:info_keys(), true,
- ?TIMEOUT_LIST_OPS_PASS),
-
- %% list connections
- H = ?config(rmq_hostname, Config),
- P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
- {ok, C1} = gen_tcp:connect(H, P, [binary, {active, false}]),
- gen_tcp:send(C1, <<"AMQP", 0, 0, 9, 1>>),
- {ok, <<1,0,0>>} = gen_tcp:recv(C1, 3, 100),
-
- {ok, C2} = gen_tcp:connect(H, P, [binary, {active, false}]),
- gen_tcp:send(C2, <<"AMQP", 0, 0, 9, 1>>),
- {ok, <<1,0,0>>} = gen_tcp:recv(C2, 3, 100),
-
- ok = info_action_t(
- list_connections, rabbit_networking:connection_info_keys(), false,
- ?TIMEOUT_LIST_OPS_PASS),
-
- %% list consumers
- ok = info_action_t(
- list_consumers, rabbit_amqqueue:consumer_info_keys(), false,
- ?TIMEOUT_LIST_OPS_PASS),
-
- %% list channels
- ok = info_action_t(
- list_channels, rabbit_channel:info_keys(), false,
- ?TIMEOUT_LIST_OPS_PASS),
-
- %% do some cleaning up
- ok = control_action(delete_user, ["list_operations_timeout_pass-user"]),
- {error, {no_such_user, _}} =
- control_action(delete_user, ["list_operations_timeout_pass-user"]),
-
- ok = control_action(delete_vhost, ["/list_operations_timeout_pass-vhost"]),
- {error, {no_such_vhost, _}} =
- control_action(delete_vhost, ["/list_operations_timeout_pass-vhost"]),
-
- %% close_connection
- Conns = rabbit_ct_broker_helpers:get_connection_pids([C1, C2]),
- [ok, ok] = [ok = control_action(
- close_connection, [rabbit_misc:pid_to_string(ConnPid), "go away"])
- || ConnPid <- Conns],
-
- %% cleanup queues
- [{ok, _} = rabbit_amqqueue:delete(QR, false, false) || QR <- [Q, Q2]],
-
- [begin
- unlink(Chan),
- ok = rabbit_channel:shutdown(Chan)
- end || Chan <- [Ch1, Ch2]],
- passed.
-
-user_management(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, user_management1, [Config]).
-
-user_management1(_Config) ->
-
- %% lots if stuff that should fail
- {error, {no_such_user, _}} =
- control_action(delete_user,
- ["user_management-user"]),
- {error, {no_such_user, _}} =
- control_action(change_password,
- ["user_management-user", "user_management-password"]),
- {error, {no_such_vhost, _}} =
- control_action(delete_vhost,
- ["/user_management-vhost"]),
- {error, {no_such_user, _}} =
- control_action(set_permissions,
- ["user_management-user", ".*", ".*", ".*"]),
- {error, {no_such_user, _}} =
- control_action(clear_permissions,
- ["user_management-user"]),
- {error, {no_such_user, _}} =
- control_action(list_user_permissions,
- ["user_management-user"]),
- {error, {no_such_vhost, _}} =
- control_action(list_permissions, [],
- [{"-p", "/user_management-vhost"}]),
- {error, {invalid_regexp, _, _}} =
- control_action(set_permissions,
- ["guest", "+foo", ".*", ".*"]),
- {error, {no_such_user, _}} =
- control_action(set_user_tags,
- ["user_management-user", "bar"]),
-
- %% user creation
- ok = control_action(add_user,
- ["user_management-user", "user_management-password"]),
- {error, {user_already_exists, _}} =
- control_action(add_user,
- ["user_management-user", "user_management-password"]),
- ok = control_action(clear_password,
- ["user_management-user"]),
- ok = control_action(change_password,
- ["user_management-user", "user_management-newpassword"]),
-
- TestTags = fun (Tags) ->
- Args = ["user_management-user" | [atom_to_list(T) || T <- Tags]],
- ok = control_action(set_user_tags, Args),
- {ok, #internal_user{tags = Tags}} =
- rabbit_auth_backend_internal:lookup_user(
- <<"user_management-user">>),
- ok = control_action(list_users, [])
- end,
- TestTags([foo, bar, baz]),
- TestTags([administrator]),
- TestTags([]),
-
- %% user authentication
- ok = control_action(authenticate_user,
- ["user_management-user", "user_management-newpassword"]),
- {refused, _User, _Format, _Params} =
- control_action(authenticate_user,
- ["user_management-user", "user_management-password"]),
-
- %% vhost creation
- ok = control_action(add_vhost,
- ["/user_management-vhost"]),
- {error, {vhost_already_exists, _}} =
- control_action(add_vhost,
- ["/user_management-vhost"]),
- ok = control_action(list_vhosts, []),
-
- %% user/vhost mapping
- ok = control_action(set_permissions,
- ["user_management-user", ".*", ".*", ".*"],
- [{"-p", "/user_management-vhost"}]),
- ok = control_action(set_permissions,
- ["user_management-user", ".*", ".*", ".*"],
- [{"-p", "/user_management-vhost"}]),
- ok = control_action(set_permissions,
- ["user_management-user", ".*", ".*", ".*"],
- [{"-p", "/user_management-vhost"}]),
- ok = control_action(list_permissions, [],
- [{"-p", "/user_management-vhost"}]),
- ok = control_action(list_permissions, [],
- [{"-p", "/user_management-vhost"}]),
- ok = control_action(list_user_permissions,
- ["user_management-user"]),
-
- %% user/vhost unmapping
- ok = control_action(clear_permissions,
- ["user_management-user"], [{"-p", "/user_management-vhost"}]),
- ok = control_action(clear_permissions,
- ["user_management-user"], [{"-p", "/user_management-vhost"}]),
-
- %% vhost deletion
- ok = control_action(delete_vhost,
- ["/user_management-vhost"]),
- {error, {no_such_vhost, _}} =
- control_action(delete_vhost,
- ["/user_management-vhost"]),
-
- %% deleting a populated vhost
- ok = control_action(add_vhost,
- ["/user_management-vhost"]),
- ok = control_action(set_permissions,
- ["user_management-user", ".*", ".*", ".*"],
- [{"-p", "/user_management-vhost"}]),
- {new, _} = rabbit_amqqueue:declare(
- rabbit_misc:r(<<"/user_management-vhost">>, queue,
- <<"user_management-vhost-queue">>),
- true, false, [], none),
- ok = control_action(delete_vhost,
- ["/user_management-vhost"]),
-
- %% user deletion
- ok = control_action(delete_user,
- ["user_management-user"]),
- {error, {no_such_user, _}} =
- control_action(delete_user,
- ["user_management-user"]),
-
- passed.
-
-runtime_parameters(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, runtime_parameters1, [Config]).
-
-runtime_parameters1(_Config) ->
- dummy_runtime_parameters:register(),
- Good = fun(L) -> ok = control_action(set_parameter, L) end,
- Bad = fun(L) -> {error_string, _} = control_action(set_parameter, L) end,
-
- %% Acceptable for bijection
- Good(["test", "good", "\"ignore\""]),
- Good(["test", "good", "123"]),
- Good(["test", "good", "true"]),
- Good(["test", "good", "false"]),
- Good(["test", "good", "null"]),
- Good(["test", "good", "{\"key\": \"value\"}"]),
-
- %% Invalid json
- Bad(["test", "good", "atom"]),
- Bad(["test", "good", "{\"foo\": \"bar\""]),
- Bad(["test", "good", "{foo: \"bar\"}"]),
-
- %% Test actual validation hook
- Good(["test", "maybe", "\"good\""]),
- Bad(["test", "maybe", "\"bad\""]),
- Good(["test", "admin", "\"ignore\""]), %% ctl means 'user' -> none
-
- ok = control_action(list_parameters, []),
-
- ok = control_action(clear_parameter, ["test", "good"]),
- ok = control_action(clear_parameter, ["test", "maybe"]),
- ok = control_action(clear_parameter, ["test", "admin"]),
- {error_string, _} =
- control_action(clear_parameter, ["test", "neverexisted"]),
-
- %% We can delete for a component that no longer exists
- Good(["test", "good", "\"ignore\""]),
- dummy_runtime_parameters:unregister(),
- ok = control_action(clear_parameter, ["test", "good"]),
- passed.
-
-policy_validation(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, policy_validation1, [Config]).
-
-policy_validation1(_Config) ->
- PolicyName = "runtime_parameters-policy",
- dummy_runtime_parameters:register_policy_validator(),
- SetPol = fun (Key, Val) ->
- control_action_opts(
- ["set_policy", PolicyName, ".*",
- rabbit_misc:format("{\"~s\":~p}", [Key, Val])])
- end,
- OK = fun (Key, Val) ->
- ok = SetPol(Key, Val),
- true = does_policy_exist(PolicyName,
- [{definition, [{list_to_binary(Key), Val}]}])
- end,
-
- OK("testeven", []),
- OK("testeven", [1, 2]),
- OK("testeven", [1, 2, 3, 4]),
- OK("testpos", [2, 5, 5678]),
-
- {error_string, _} = SetPol("testpos", [-1, 0, 1]),
- {error_string, _} = SetPol("testeven", [ 1, 2, 3]),
-
- ok = control_action(clear_policy, [PolicyName]),
- dummy_runtime_parameters:unregister_policy_validator(),
- passed.
-
-policy_opts_validation(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, policy_opts_validation1, [Config]).
-
-policy_opts_validation1(_Config) ->
- PolicyName = "policy_opts_validation-policy",
- Set = fun (Extra) -> control_action_opts(
- ["set_policy", PolicyName,
- ".*", "{\"ha-mode\":\"all\"}"
- | Extra]) end,
- OK = fun (Extra, Props) ->
- ok = Set(Extra),
- true = does_policy_exist(PolicyName, Props)
- end,
- Fail = fun (Extra) ->
- case Set(Extra) of
- {error_string, _} -> ok;
- no_command when Extra =:= ["--priority"] -> ok;
- no_command when Extra =:= ["--apply-to"] -> ok;
- {'EXIT',
- {function_clause,
- [{rabbit_control_main,action, _, _} | _]}}
- when Extra =:= ["--offline"] -> ok
- end
- end,
-
- OK ([], [{priority, 0}, {'apply-to', <<"all">>}]),
-
- OK (["--priority", "0"], [{priority, 0}]),
- OK (["--priority", "3"], [{priority, 3}]),
- Fail(["--priority", "banana"]),
- Fail(["--priority"]),
-
- OK (["--apply-to", "all"], [{'apply-to', <<"all">>}]),
- OK (["--apply-to", "queues"], [{'apply-to', <<"queues">>}]),
- Fail(["--apply-to", "bananas"]),
- Fail(["--apply-to"]),
-
- OK (["--priority", "3", "--apply-to", "queues"], [{priority, 3}, {'apply-to', <<"queues">>}]),
- Fail(["--priority", "banana", "--apply-to", "queues"]),
- Fail(["--priority", "3", "--apply-to", "bananas"]),
-
- Fail(["--offline"]),
-
- ok = control_action(clear_policy, [PolicyName]),
- passed.
-
-ha_policy_validation(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, ha_policy_validation1, [Config]).
-
-ha_policy_validation1(_Config) ->
- PolicyName = "ha_policy_validation-policy",
- Set = fun (JSON) -> control_action_opts(
- ["set_policy", PolicyName,
- ".*", JSON]) end,
- OK = fun (JSON, Def) ->
- ok = Set(JSON),
- true = does_policy_exist(PolicyName, [{definition, Def}])
- end,
- Fail = fun (JSON) -> {error_string, _} = Set(JSON) end,
-
- OK ("{\"ha-mode\":\"all\"}", [{<<"ha-mode">>, <<"all">>}]),
- Fail("{\"ha-mode\":\"made_up\"}"),
-
- Fail("{\"ha-mode\":\"nodes\"}"),
- Fail("{\"ha-mode\":\"nodes\",\"ha-params\":2}"),
- Fail("{\"ha-mode\":\"nodes\",\"ha-params\":[\"a\",2]}"),
- OK ("{\"ha-mode\":\"nodes\",\"ha-params\":[\"a\",\"b\"]}",
- [{<<"ha-mode">>, <<"nodes">>}, {<<"ha-params">>, [<<"a">>, <<"b">>]}]),
- Fail("{\"ha-params\":[\"a\",\"b\"]}"),
-
- Fail("{\"ha-mode\":\"exactly\"}"),
- Fail("{\"ha-mode\":\"exactly\",\"ha-params\":[\"a\",\"b\"]}"),
- OK ("{\"ha-mode\":\"exactly\",\"ha-params\":2}",
- [{<<"ha-mode">>, <<"exactly">>}, {<<"ha-params">>, 2}]),
- Fail("{\"ha-params\":2}"),
-
- OK ("{\"ha-mode\":\"all\",\"ha-sync-mode\":\"manual\"}",
- [{<<"ha-mode">>, <<"all">>}, {<<"ha-sync-mode">>, <<"manual">>}]),
- OK ("{\"ha-mode\":\"all\",\"ha-sync-mode\":\"automatic\"}",
- [{<<"ha-mode">>, <<"all">>}, {<<"ha-sync-mode">>, <<"automatic">>}]),
- Fail("{\"ha-mode\":\"all\",\"ha-sync-mode\":\"made_up\"}"),
- Fail("{\"ha-sync-mode\":\"manual\"}"),
- Fail("{\"ha-sync-mode\":\"automatic\"}"),
-
- ok = control_action(clear_policy, [PolicyName]),
- passed.
-
-queue_master_location_policy_validation(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, queue_master_location_policy_validation1, [Config]).
-
-queue_master_location_policy_validation1(_Config) ->
- PolicyName = "queue_master_location_policy_validation-policy",
- Set = fun (JSON) ->
- control_action_opts(
- ["set_policy", PolicyName, ".*", JSON])
- end,
- OK = fun (JSON, Def) ->
- ok = Set(JSON),
- true = does_policy_exist(PolicyName, [{definition, Def}])
- end,
- Fail = fun (JSON) -> {error_string, _} = Set(JSON) end,
-
- OK ("{\"queue-master-locator\":\"min-masters\"}",
- [{<<"queue-master-locator">>, <<"min-masters">>}]),
- OK ("{\"queue-master-locator\":\"client-local\"}",
- [{<<"queue-master-locator">>, <<"client-local">>}]),
- OK ("{\"queue-master-locator\":\"random\"}",
- [{<<"queue-master-locator">>, <<"random">>}]),
- Fail("{\"queue-master-locator\":\"made_up\"}"),
-
- ok = control_action(clear_policy, [PolicyName]),
- passed.
-
-queue_modes_policy_validation(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, queue_modes_policy_validation1, [Config]).
-
-queue_modes_policy_validation1(_Config) ->
- PolicyName = "queue_modes_policy_validation-policy",
- Set = fun (JSON) ->
- control_action_opts(
- ["set_policy", PolicyName, ".*", JSON])
- end,
- OK = fun (JSON, Def) ->
- ok = Set(JSON),
- true = does_policy_exist(PolicyName, [{definition, Def}])
- end,
- Fail = fun (JSON) -> {error_string, _} = Set(JSON) end,
-
- OK ("{\"queue-mode\":\"lazy\"}",
- [{<<"queue-mode">>, <<"lazy">>}]),
- OK ("{\"queue-mode\":\"default\"}",
- [{<<"queue-mode">>, <<"default">>}]),
- Fail("{\"queue-mode\":\"wrong\"}"),
-
- ok = control_action(clear_policy, [PolicyName]),
- passed.
-
-vhost_removed_while_updating_policy(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, vhost_removed_while_updating_policy1, [Config]).
-
-vhost_removed_while_updating_policy1(_Config) ->
- VHost = "/vhost_removed_while_updating_policy-vhost",
- PolicyName = "vhost_removed_while_updating_policy-policy",
-
- ok = control_action(add_vhost, [VHost]),
- ok = control_action_opts(
- ["set_policy", "-p", VHost, PolicyName, ".*", "{\"ha-mode\":\"all\"}"]),
- true = does_policy_exist(PolicyName, []),
-
- %% Removing the vhost triggers the deletion of the policy. Once
- %% the policy and the vhost are actually removed, RabbitMQ calls
- %% update_policies() which lists policies on the given vhost. This
- %% obviously fails because the vhost is gone, but the call should
- %% still succeed.
- ok = control_action(delete_vhost, [VHost]),
- false = does_policy_exist(PolicyName, []),
-
- passed.
-
-does_policy_exist(PolicyName, Props) ->
- PolicyNameBin = list_to_binary(PolicyName),
- Policies = lists:filter(
- fun(Policy) ->
- lists:member({name, PolicyNameBin}, Policy)
- end, rabbit_policy:list()),
- case Policies of
- [Policy] -> check_policy_props(Policy, Props);
- [] -> false;
- _ -> false
- end.
-
-check_policy_props(Policy, [Prop | Rest]) ->
- case lists:member(Prop, Policy) of
- true -> check_policy_props(Policy, Rest);
- false -> false
- end;
-check_policy_props(_Policy, []) ->
- true.
-
-server_status(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, server_status1, [Config]).
-
-server_status1(Config) ->
- %% create a few things so there is some useful information to list
- {_Writer, Limiter, Ch} = rabbit_ct_broker_helpers:test_channel(),
- [Q, Q2] = [Queue || {Name, Owner} <- [{<<"server_status-q1">>, none},
- {<<"server_status-q2">>, self()}],
- {new, Queue = #amqqueue{}} <-
- [rabbit_amqqueue:declare(
- rabbit_misc:r(<<"/">>, queue, Name),
- false, false, [], Owner)]],
- ok = rabbit_amqqueue:basic_consume(
- Q, true, Ch, Limiter, false, 0, <<"ctag">>, true, [], undefined),
-
- %% list queues
- ok = info_action(list_queues,
- rabbit_amqqueue:info_keys(), true),
-
- %% as we have no way to collect output of
- %% info_action/3 call, the only way we
- %% can test individual queueinfoitems is by directly calling
- %% rabbit_amqqueue:info/2
- [{exclusive, false}] = rabbit_amqqueue:info(Q, [exclusive]),
- [{exclusive, true}] = rabbit_amqqueue:info(Q2, [exclusive]),
-
- %% list exchanges
- ok = info_action(list_exchanges,
- rabbit_exchange:info_keys(), true),
-
- %% list bindings
- ok = info_action(list_bindings,
- rabbit_binding:info_keys(), true),
- %% misc binding listing APIs
- [_|_] = rabbit_binding:list_for_source(
- rabbit_misc:r(<<"/">>, exchange, <<"">>)),
- [_] = rabbit_binding:list_for_destination(
- rabbit_misc:r(<<"/">>, queue, <<"server_status-q1">>)),
- [_] = rabbit_binding:list_for_source_and_destination(
- rabbit_misc:r(<<"/">>, exchange, <<"">>),
- rabbit_misc:r(<<"/">>, queue, <<"server_status-q1">>)),
-
- %% list connections
- H = ?config(rmq_hostname, Config),
- P = rabbit_ct_broker_helpers:get_node_config(Config, 0, tcp_port_amqp),
- {ok, C} = gen_tcp:connect(H, P, []),
- gen_tcp:send(C, <<"AMQP", 0, 0, 9, 1>>),
- timer:sleep(100),
- ok = info_action(list_connections,
- rabbit_networking:connection_info_keys(), false),
- %% close_connection
- [ConnPid] = rabbit_ct_broker_helpers:get_connection_pids([C]),
- ok = control_action(close_connection,
- [rabbit_misc:pid_to_string(ConnPid), "go away"]),
-
- %% list channels
- ok = info_action(list_channels, rabbit_channel:info_keys(), false),
-
- %% list consumers
- ok = control_action(list_consumers, []),
-
- %% set vm memory high watermark
- HWM = vm_memory_monitor:get_vm_memory_high_watermark(),
- ok = control_action(set_vm_memory_high_watermark, ["1"]),
- ok = control_action(set_vm_memory_high_watermark, ["1.0"]),
- %% this will trigger an alarm
- ok = control_action(set_vm_memory_high_watermark, ["0.0"]),
- %% reset
- ok = control_action(set_vm_memory_high_watermark, [float_to_list(HWM)]),
-
- %% eval
- {error_string, _} = control_action(eval, ["\""]),
- {error_string, _} = control_action(eval, ["a("]),
- ok = control_action(eval, ["a."]),
-
- %% cleanup
- [{ok, _} = rabbit_amqqueue:delete(QR, false, false) || QR <- [Q, Q2]],
-
- unlink(Ch),
- ok = rabbit_channel:shutdown(Ch),
-
- passed.
amqp_connection_refusal(Config) ->
passed = rabbit_ct_broker_helpers:rpc(Config, 0,
@@ -3395,39 +2801,6 @@ configurable_server_properties1(_Config) ->
application:set_env(rabbit, server_properties, ServerProperties),
passed.
-memory_high_watermark(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, memory_high_watermark1, [Config]).
-
-memory_high_watermark1(_Config) ->
- %% set vm memory high watermark
- HWM = vm_memory_monitor:get_vm_memory_high_watermark(),
- %% this will trigger an alarm
- ok = control_action(set_vm_memory_high_watermark,
- ["absolute", "2000"]),
- [{{resource_limit,memory,_},[]}] = rabbit_alarm:get_alarms(),
- %% reset
- ok = control_action(set_vm_memory_high_watermark,
- [float_to_list(HWM)]),
-
- passed.
-
-set_disk_free_limit_command(Config) ->
- passed = rabbit_ct_broker_helpers:rpc(Config, 0,
- ?MODULE, set_disk_free_limit_command1, [Config]).
-
-set_disk_free_limit_command1(_Config) ->
- ok = control_action(set_disk_free_limit,
- ["2000kiB"]),
- 2048000 = rabbit_disk_monitor:get_disk_free_limit(),
- ok = control_action(set_disk_free_limit,
- ["mem_relative", "1.1"]),
- ExpectedLimit = 1.1 * vm_memory_monitor:get_total_memory(),
- % Total memory is unstable, so checking order
- true = ExpectedLimit/rabbit_disk_monitor:get_disk_free_limit() < 1.2,
- true = ExpectedLimit/rabbit_disk_monitor:get_disk_free_limit() > 0.98,
- ok = control_action(set_disk_free_limit, ["50MB"]),
- passed.
disk_monitor(Config) ->
passed = rabbit_ct_broker_helpers:rpc(Config, 0,
@@ -3715,83 +3088,6 @@ expect_event(Tag, Key, Type) ->
%% rabbitmqctl helpers.
%% ---------------------------------------------------------------------------
-control_action(Command, Args) ->
- control_action(Command, node(), Args, default_options()).
-
-control_action(Command, Args, NewOpts) ->
- control_action(Command, node(), Args,
- expand_options(default_options(), NewOpts)).
-
-control_action(Command, Node, Args, Opts) ->
- case catch rabbit_control_main:action(
- Command, Node, Args, Opts,
- fun (Format, Args1) ->
- io:format(Format ++ " ...~n", Args1)
- end) of
- ok ->
- io:format("done.~n"),
- ok;
- {ok, Result} ->
- rabbit_control_misc:print_cmd_result(Command, Result),
- ok;
- Other ->
- io:format("failed: ~p~n", [Other]),
- Other
- end.
-
-control_action_t(Command, Args, Timeout) when is_number(Timeout) ->
- control_action_t(Command, node(), Args, default_options(), Timeout).
-
-control_action_t(Command, Args, NewOpts, Timeout) when is_number(Timeout) ->
- control_action_t(Command, node(), Args,
- expand_options(default_options(), NewOpts),
- Timeout).
-
-control_action_t(Command, Node, Args, Opts, Timeout) when is_number(Timeout) ->
- case catch rabbit_control_main:action(
- Command, Node, Args, Opts,
- fun (Format, Args1) ->
- io:format(Format ++ " ...~n", Args1)
- end, Timeout) of
- ok ->
- io:format("done.~n"),
- ok;
- {ok, Result} ->
- rabbit_control_misc:print_cmd_result(Command, Result),
- ok;
- Other ->
- io:format("failed: ~p~n", [Other]),
- Other
- end.
-
-control_action_opts(Raw) ->
- NodeStr = atom_to_list(node()),
- case rabbit_control_main:parse_arguments(Raw, NodeStr) of
- {ok, {Cmd, Opts, Args}} ->
- case control_action(Cmd, node(), Args, Opts) of
- ok -> ok;
- Error -> Error
- end;
- Error ->
- Error
- end.
-
-info_action(Command, Args, CheckVHost) ->
- ok = control_action(Command, []),
- if CheckVHost -> ok = control_action(Command, [], ["-p", "/"]);
- true -> ok
- end,
- ok = control_action(Command, lists:map(fun atom_to_list/1, Args)),
- {bad_argument, dummy} = control_action(Command, ["dummy"]),
- ok.
-
-info_action_t(Command, Args, CheckVHost, Timeout) when is_number(Timeout) ->
- if CheckVHost -> ok = control_action_t(Command, [], ["-p", "/"], Timeout);
- true -> ok
- end,
- ok = control_action_t(Command, lists:map(fun atom_to_list/1, Args), Timeout),
- ok.
-
default_options() -> [{"-p", "/"}, {"-q", "false"}].
expand_options(As, Bs) ->
View Lorry Controller Status