diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/rabbit.erl | 4 | ||||
| -rw-r--r-- | src/rabbit_alarm.erl | 55 | ||||
| -rw-r--r-- | src/rabbit_amqqueue.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 4 | ||||
| -rw-r--r-- | src/rabbit_control.erl | 171 | ||||
| -rw-r--r-- | src/rabbit_exchange.erl | 12 | ||||
| -rw-r--r-- | src/rabbit_heartbeat.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_load.erl | 6 | ||||
| -rw-r--r-- | src/rabbit_log.erl | 14 | ||||
| -rw-r--r-- | src/rabbit_memsup_linux.erl | 150 | ||||
| -rw-r--r-- | src/rabbit_misc.erl | 9 | ||||
| -rw-r--r-- | src/rabbit_mnesia.erl | 4 | ||||
| -rw-r--r-- | src/rabbit_multi.erl | 24 | ||||
| -rw-r--r-- | src/rabbit_networking.erl | 4 | ||||
| -rw-r--r-- | src/rabbit_sasl_report_file_h.erl | 5 | ||||
| -rw-r--r-- | src/rabbit_tests.erl | 2 |
16 files changed, 355 insertions, 113 deletions
diff --git a/src/rabbit.erl b/src/rabbit.erl index a33c5b7bcb..195dd729bd 100644 --- a/src/rabbit.erl +++ b/src/rabbit.erl @@ -57,8 +57,8 @@ -spec(rotate_logs/1 :: (file_suffix()) -> 'ok' | {'error', any()}). -spec(status/0 :: () -> [{running_applications, [{atom(), string(), string()}]} | - {nodes, [node()]} | - {running_nodes, [node()]}]). + {nodes, [erlang_node()]} | + {running_nodes, [erlang_node()]}]). -spec(log_location/1 :: ('sasl' | 'kernel') -> log_location()). -endif. diff --git a/src/rabbit_alarm.erl b/src/rabbit_alarm.erl index d9c1c45042..6d65b3a407 100644 --- a/src/rabbit_alarm.erl +++ b/src/rabbit_alarm.erl @@ -50,22 +50,57 @@ %%---------------------------------------------------------------------------- start() -> + ok = alarm_handler:add_alarm_handler(?MODULE), + case whereis(memsup) of + undefined -> + Mod = case os:type() of + %% memsup doesn't take account of buffers or + %% cache when considering "free" memory - + %% therefore on Linux we can get memory alarms + %% very easily without any pressure existing on + %% memory at all. Therefore we need to use our + %% own simple memory monitor. + %% + {unix, linux} -> rabbit_memsup_linux; + + %% Start memsup programmatically rather than via + %% the rabbitmq-server script. This is not quite + %% the right thing to do as os_mon checks to see + %% if memsup is available before starting it, + %% but as memsup is available everywhere (even + %% on VXWorks) it should be ok. + %% + %% One benefit of the programmatic startup is + %% that we can add our alarm_handler before + %% memsup is running, thus ensuring that we + %% notice memory alarms that go off on startup. + %% + _ -> memsup + end, + %% This is based on os_mon:childspec(memsup, true) + {ok, _} = supervisor:start_child( + os_mon_sup, + {memsup, {Mod, start_link, []}, + permanent, 2000, worker, [Mod]}), + ok; + _ -> + ok + end, %% The default memsup check interval is 1 minute, which is way too - %% long - rabbit can gobble up all memory in a matter of - %% seconds. Unfortunately the memory_check_interval configuration - %% parameter and memsup:set_check_interval/1 function only provide - %% a granularity of minutes. So we have to peel off one layer of - %% the API to get to the underlying layer which operates at the + %% long - rabbit can gobble up all memory in a matter of seconds. + %% Unfortunately the memory_check_interval configuration parameter + %% and memsup:set_check_interval/1 function only provide a + %% granularity of minutes. So we have to peel off one layer of the + %% API to get to the underlying layer which operates at the %% granularity of milliseconds. %% %% Note that the new setting will only take effect after the first %% check has completed, i.e. after one minute. So if rabbit eats %% all the memory within the first minute after startup then we %% are out of luck. - ok = os_mon:call(memsup, {set_check_interval, ?MEMSUP_CHECK_INTERVAL}, - infinity), - - ok = alarm_handler:add_alarm_handler(?MODULE). + ok = os_mon:call(memsup, + {set_check_interval, ?MEMSUP_CHECK_INTERVAL}, + infinity). stop() -> ok = alarm_handler:delete_alarm_handler(?MODULE). @@ -118,7 +153,7 @@ code_change(_OldVsn, State, _Extra) -> {ok, State}. %%---------------------------------------------------------------------------- - + alert(Alert, Alertees) -> dict:fold(fun (Pid, {M, F, A}, Acc) -> ok = erlang:apply(M, F, A ++ [Pid, Alert]), diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index e2174e1100..694e525e07 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -96,7 +96,7 @@ -spec(basic_cancel/4 :: (amqqueue(), pid(), ctag(), any()) -> 'ok'). -spec(notify_sent/2 :: (pid(), pid()) -> 'ok'). -spec(internal_delete/1 :: (queue_name()) -> 'ok' | not_found()). --spec(on_node_down/1 :: (node()) -> 'ok'). +-spec(on_node_down/1 :: (erlang_node()) -> 'ok'). -spec(pseudo_queue/2 :: (binary(), pid()) -> amqqueue()). -endif. diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index a9addd5293..186f43898a 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -749,7 +749,9 @@ handle_info({'DOWN', _MonitorRef, process, DownPid, _Reason}, State) -> handle_ch_down(DownPid, State); handle_info(timeout, State) -> - {noreply, State, hibernate}; + %% TODO: Once we drop support for R11B-5, we can change this to + %% {noreply, State, hibernate}; + proc_lib:hibernate(gen_server, enter_loop, [?MODULE, [], State]); handle_info(Info, State) -> ?LOGDEBUG("Info in queue: ~p~n", [Info]), diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl index d4906d8f1e..ae346236e1 100644 --- a/src/rabbit_control.erl +++ b/src/rabbit_control.erl @@ -26,43 +26,67 @@ -module(rabbit_control). -include("rabbit.hrl"). --export([start/0, stop/0, action/3]). +-export([start/0, stop/0, action/4]). + +-record(params, {quiet, node, command, args}). -define(RPC_TIMEOUT, 30000). start() -> - case init:get_plain_arguments() of - [] -> + FullCommand = init:get_plain_arguments(), + #params{quiet = Quiet, node = Node, command = Command, args = Args} = + parse_args(FullCommand, #params{quiet = false, + node = rabbit_misc:localnode(rabbit)}), + Inform = case Quiet of + true -> fun(_Format, _Args1) -> ok end; + false -> fun(Format, Args1) -> + io:format(Format ++ " ...~n", Args1) + end + end, + %% The reason we don't use a try/catch here is that rpc:call turns + %% thrown errors into normal return values + case catch action(Command, Node, Args, Inform) of + ok -> + case Quiet of + true -> ok; + false -> io:format("...done.~n") + end, + init:stop(); + {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} -> + error("invalid command '~s'", + [lists:flatten( + rabbit_misc:intersperse( + " ", [atom_to_list(Command) | Args]))]), usage(); - FullCommand -> - {Node, Command, Args} = parse_args(FullCommand), - case catch action(Command, Node, Args) of - ok -> - io:format("done.~n"), - init:stop(); - {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} -> - io:format("Invalid command ~p~n", [FullCommand]), - usage(); - Other -> - io:format("~nrabbit_control action ~p failed:~n~p~n", [Command, Other]), - halt(2) - end + {error, Reason} -> + error("~p", [Reason]), + halt(2); + Other -> + error("~p", [Other]), + halt(2) end. -parse_args(["-n", NodeS, Command | Args]) -> +error(Format, Args) -> + rabbit_misc:format_stderr("Error: " ++ Format ++ "~n", Args). + +parse_args(["-n", NodeS | Args], Params) -> Node = case lists:member($@, NodeS) of true -> list_to_atom(NodeS); false -> rabbit_misc:localnode(list_to_atom(NodeS)) end, - {Node, list_to_atom(Command), Args}; -parse_args([Command | Args]) -> - {rabbit_misc:localnode(rabbit), list_to_atom(Command), Args}. + parse_args(Args, Params#params{node = Node}); +parse_args(["-q" | Args], Params) -> + parse_args(Args, Params#params{quiet = true}); +parse_args([Command | Args], Params) -> + Params#params{command = list_to_atom(Command), args = Args}; +parse_args([], _) -> + usage(). stop() -> ok. usage() -> - io:format("Usage: rabbitmqctl [-n <node>] <command> [<arg> ...] + io:format("Usage: rabbitmqctl [-q] [-n <node>] <command> [<arg> ...] Available commands: @@ -94,6 +118,9 @@ Available commands: list_bindings list_connections <ConnectionInfoItem> [<ConnectionInfoItem> ...] +Quiet output mode is selected with the \"-q\" flag. Informational messages +are suppressed when quiet mode is in effect. + <node> should be the name of the master node of the RabbitMQ cluster. It defaults to the node named \"rabbit\" on the local host. On a host named \"server.example.com\", the master node will usually be rabbit@server (unless @@ -120,101 +147,101 @@ user, peer_address and peer_port. "), halt(1). -action(stop, Node, []) -> - io:format("Stopping and halting node ~p ...", [Node]), +action(stop, Node, [], Inform) -> + Inform("Stopping and halting node ~p", [Node]), call(Node, {rabbit, stop_and_halt, []}); -action(stop_app, Node, []) -> - io:format("Stopping node ~p ...", [Node]), +action(stop_app, Node, [], Inform) -> + Inform("Stopping node ~p", [Node]), call(Node, {rabbit, stop, []}); -action(start_app, Node, []) -> - io:format("Starting node ~p ...", [Node]), +action(start_app, Node, [], Inform) -> + Inform("Starting node ~p", [Node]), call(Node, {rabbit, start, []}); -action(reset, Node, []) -> - io:format("Resetting node ~p ...", [Node]), +action(reset, Node, [], Inform) -> + Inform("Resetting node ~p", [Node]), call(Node, {rabbit_mnesia, reset, []}); -action(force_reset, Node, []) -> - io:format("Forcefully resetting node ~p ...", [Node]), +action(force_reset, Node, [], Inform) -> + Inform("Forcefully resetting node ~p", [Node]), call(Node, {rabbit_mnesia, force_reset, []}); -action(cluster, Node, ClusterNodeSs) -> +action(cluster, Node, ClusterNodeSs, Inform) -> ClusterNodes = lists:map(fun list_to_atom/1, ClusterNodeSs), - io:format("Clustering node ~p with ~p ...", + Inform("Clustering node ~p with ~p", [Node, ClusterNodes]), rpc_call(Node, rabbit_mnesia, cluster, [ClusterNodes]); -action(status, Node, []) -> - io:format("Status of node ~p ...", [Node]), +action(status, Node, [], Inform) -> + Inform("Status of node ~p", [Node]), Res = call(Node, {rabbit, status, []}), - io:format("~n~p~n", [Res]), + io:format("~p~n", [Res]), ok; -action(rotate_logs, Node, []) -> - io:format("Reopening logs for node ~p ...", [Node]), +action(rotate_logs, Node, [], Inform) -> + Inform("Reopening logs for node ~p", [Node]), call(Node, {rabbit, rotate_logs, [""]}); -action(rotate_logs, Node, Args = [Suffix]) -> - io:format("Rotating logs to files with suffix ~p ...", [Suffix]), +action(rotate_logs, Node, Args = [Suffix], Inform) -> + Inform("Rotating logs to files with suffix ~p", [Suffix]), call(Node, {rabbit, rotate_logs, Args}); -action(add_user, Node, Args = [Username, _Password]) -> - io:format("Creating user ~p ...", [Username]), +action(add_user, Node, Args = [Username, _Password], Inform) -> + Inform("Creating user ~p", [Username]), call(Node, {rabbit_access_control, add_user, Args}); -action(delete_user, Node, Args = [_Username]) -> - io:format("Deleting user ~p ...", Args), +action(delete_user, Node, Args = [_Username], Inform) -> + Inform("Deleting user ~p", Args), call(Node, {rabbit_access_control, delete_user, Args}); -action(change_password, Node, Args = [Username, _Newpassword]) -> - io:format("Changing password for user ~p ...", [Username]), +action(change_password, Node, Args = [Username, _Newpassword], Inform) -> + Inform("Changing password for user ~p", [Username]), call(Node, {rabbit_access_control, change_password, Args}); -action(list_users, Node, []) -> - io:format("Listing users ..."), +action(list_users, Node, [], Inform) -> + Inform("Listing users", []), display_list(call(Node, {rabbit_access_control, list_users, []})); -action(add_vhost, Node, Args = [_VHostPath]) -> - io:format("Creating vhost ~p ...", Args), +action(add_vhost, Node, Args = [_VHostPath], Inform) -> + Inform("Creating vhost ~p", Args), call(Node, {rabbit_access_control, add_vhost, Args}); -action(delete_vhost, Node, Args = [_VHostPath]) -> - io:format("Deleting vhost ~p ...", Args), +action(delete_vhost, Node, Args = [_VHostPath], Inform) -> + Inform("Deleting vhost ~p", Args), call(Node, {rabbit_access_control, delete_vhost, Args}); -action(list_vhosts, Node, []) -> - io:format("Listing vhosts ..."), +action(list_vhosts, Node, [], Inform) -> + Inform("Listing vhosts", []), display_list(call(Node, {rabbit_access_control, list_vhosts, []})); -action(map_user_vhost, Node, Args = [_Username, _VHostPath]) -> - io:format("Mapping user ~p to vhost ~p ...", Args), +action(map_user_vhost, Node, Args = [_Username, _VHostPath], Inform) -> + Inform("Mapping user ~p to vhost ~p", Args), call(Node, {rabbit_access_control, map_user_vhost, Args}); -action(unmap_user_vhost, Node, Args = [_Username, _VHostPath]) -> - io:format("Unmapping user ~p from vhost ~p ...", Args), +action(unmap_user_vhost, Node, Args = [_Username, _VHostPath], Inform) -> + Inform("Unmapping user ~p from vhost ~p", Args), call(Node, {rabbit_access_control, unmap_user_vhost, Args}); -action(list_user_vhosts, Node, Args = [_Username]) -> - io:format("Listing vhosts for user ~p...", Args), +action(list_user_vhosts, Node, Args = [_Username], Inform) -> + Inform("Listing vhosts for user ~p", Args), display_list(call(Node, {rabbit_access_control, list_user_vhosts, Args})); -action(list_vhost_users, Node, Args = [_VHostPath]) -> - io:format("Listing users for vhosts ~p...", Args), +action(list_vhost_users, Node, Args = [_VHostPath], Inform) -> + Inform("Listing users for vhosts ~p", Args), display_list(call(Node, {rabbit_access_control, list_vhost_users, Args})); -action(list_queues, Node, Args) -> - io:format("Listing queues ...~n"), +action(list_queues, Node, Args, Inform) -> + Inform("Listing queues", []), ArgAtoms = [list_to_atom(X) || X <- default_if_empty(Args, ["name", "messages"])], display_info_list(rpc_call(Node, rabbit_amqqueue, info_all, [ArgAtoms]), ArgAtoms); -action(list_exchanges, Node, Args) -> - io:format("Listing exchanges ...~n"), +action(list_exchanges, Node, Args, Inform) -> + Inform("Listing exchanges", []), ArgAtoms = [list_to_atom(X) || X <- default_if_empty(Args, ["name", "type"])], display_info_list(rpc_call(Node, rabbit_exchange, info_all, [ArgAtoms]), ArgAtoms); -action(list_bindings, Node, []) -> - io:format("Listing bindings ...~n"), +action(list_bindings, Node, [], Inform) -> + Inform("Listing bindings", []), lists:map( fun({#resource{name = ExchangeName, virtual_host = VirtualHost}, #resource{name = QueueName, virtual_host = VirtualHost}, @@ -225,8 +252,8 @@ action(list_bindings, Node, []) -> rpc_call(Node, rabbit_exchange, list_bindings, [])), ok; -action(list_connections, Node, Args) -> - io:format("Listing connections ...~n"), +action(list_connections, Node, Args, Inform) -> + Inform("Listing connections", []), ArgAtoms = [list_to_atom(X) || X <- default_if_empty(Args, ["user", "peer_address", "peer_port"])], display_info_list(rpc_call(Node, rabbit_networking, connection_info_all, [ArgAtoms]), ArgAtoms). @@ -261,10 +288,10 @@ display_info_list(Results, InfoItemArgs) when is_list(Results) -> display_list(L) when is_list(L) -> lists:foreach(fun (I) -> - io:format("~n~s", [binary_to_list(I)]) + io:format("~s~n", [binary_to_list(I)]) end, lists:sort(L)), - io:nl(); + ok; display_list(Other) -> Other. call(Node, {Mod, Fun, Args}) -> diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl index 273cf3d380..9a3caad107 100644 --- a/src/rabbit_exchange.erl +++ b/src/rabbit_exchange.erl @@ -302,9 +302,17 @@ exchanges_for_queue(QueueName) -> has_bindings(ExchangeName) -> MatchHead = #route{binding = #binding{exchange_name = ExchangeName, - queue_name = '$1', _ = '_'}}, - continue(mnesia:select(route, [{MatchHead, [], ['$1']}], 1, read)). + try + continue(mnesia:select(route, [{MatchHead, [], ['$_']}], 1, read)) + catch exit:{aborted, {badarg, _}} -> + %% work around OTP-7025, which was fixed in R12B-1, by + %% falling back on a less efficient method + case mnesia:match_object(MatchHead) of + [] -> false; + [_|_] -> true + end + end. continue('$end_of_table') -> false; continue({[_|_], _}) -> true; diff --git a/src/rabbit_heartbeat.erl b/src/rabbit_heartbeat.erl index 6e825578d3..b05a4655e5 100644 --- a/src/rabbit_heartbeat.erl +++ b/src/rabbit_heartbeat.erl @@ -67,7 +67,7 @@ heartbeater(Sock, TimeoutMillisec, StatName, Threshold, Handler, MonitorRef) -> {'DOWN', MonitorRef, process, _Object, _Info} -> ok; Other -> exit({unexpected_message, Other}) after TimeoutMillisec -> - case prim_inet:getstat(Sock, [StatName]) of + case inet:getstat(Sock, [StatName]) of {ok, [{StatName, NewStatVal}]} -> if NewStatVal =/= StatVal -> F({NewStatVal, 0}); diff --git a/src/rabbit_load.erl b/src/rabbit_load.erl index 8deec8ebee..2154bf72ae 100644 --- a/src/rabbit_load.erl +++ b/src/rabbit_load.erl @@ -34,11 +34,11 @@ -ifdef(use_specs). --type(node() :: atom()). --type(load() :: {{non_neg_integer(), float()}, node()}). +-type(erlang_node() :: atom()). +-type(load() :: {{non_neg_integer(), float()}, erlang_node()}). -spec(local_load/0 :: () -> load()). -spec(remote_loads/0 :: () -> [load()]). --spec(pick/0 :: () -> node()). +-spec(pick/0 :: () -> erlang_node()). -endif. diff --git a/src/rabbit_log.erl b/src/rabbit_log.erl index a8f839f0a6..b4729230e6 100644 --- a/src/rabbit_log.erl +++ b/src/rabbit_log.erl @@ -32,7 +32,7 @@ -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). --export([debug/1, debug/2, info/1, info/2, +-export([debug/1, debug/2, message/4, info/1, info/2, warning/1, warning/2, error/1, error/2]). -import(io). @@ -67,6 +67,10 @@ debug(Fmt) -> debug(Fmt, Args) when is_list(Args) -> gen_server:cast(?SERVER, {debug, Fmt, Args}). +message(Direction, Channel, MethodRecord, Content) -> + gen_server:cast(?SERVER, + {message, Direction, Channel, MethodRecord, Content}). + info(Fmt) -> gen_server:cast(?SERVER, {info, Fmt}). @@ -100,6 +104,14 @@ handle_cast({debug, Fmt, Args}, State) -> io:format("debug:: "), io:format(Fmt, Args), error_logger:info_msg("debug:: " ++ Fmt, Args), {noreply, State}; +handle_cast({message, Direction, Channel, MethodRecord, Content}, State) -> + io:format("~s ch~p ~p~n", + [case Direction of + in -> "-->"; + out -> "<--" end, + Channel, + {MethodRecord, Content}]), + {noreply, State}; handle_cast({info, Fmt}, State) -> error_logger:info_msg(Fmt), {noreply, State}; diff --git a/src/rabbit_memsup_linux.erl b/src/rabbit_memsup_linux.erl new file mode 100644 index 0000000000..b77ffcabb4 --- /dev/null +++ b/src/rabbit_memsup_linux.erl @@ -0,0 +1,150 @@ +%% 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 Developers of the Original Code are LShift Ltd., +%% Cohesive Financial Technologies LLC., and Rabbit Technologies Ltd. +%% +%% Portions created by LShift Ltd., Cohesive Financial Technologies +%% LLC., and Rabbit Technologies Ltd. are Copyright (C) 2007-2008 +%% LShift Ltd., Cohesive Financial Technologies LLC., and Rabbit +%% Technologies Ltd.; +%% +%% All Rights Reserved. +%% +%% Contributor(s): ______________________________________. +%% + +-module(rabbit_memsup_linux). + +-behaviour(gen_server). + +-export([start_link/0]). + +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, + terminate/2, code_change/3]). + +-export([update/0]). + +-define(SERVER, memsup). %% must be the same as the standard memsup + +-define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000). + +-record(state, {memory_fraction, alarmed, timeout, timer}). + +%%---------------------------------------------------------------------------- + +-ifdef(use_specs). + +-spec(start_link/0 :: () -> {'ok', pid()} | 'ignore' | {'error', any()}). +-spec(update/0 :: () -> 'ok'). + +-endif. + +%%---------------------------------------------------------------------------- + +start_link() -> + gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). + + +update() -> + gen_server:cast(?SERVER, update). + +%%---------------------------------------------------------------------------- + +init(_Args) -> + Fraction = os_mon:get_env(memsup, system_memory_high_watermark), + TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL), + {ok, #state{alarmed = false, + memory_fraction = Fraction, + timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL, + timer = TRef}}. + +start_timer(Timeout) -> + {ok, TRef} = timer:apply_interval(Timeout, ?MODULE, update, []), + TRef. + +%% Export the same API as the real memsup. Note that +%% get_sysmem_high_watermark gives an int in the range 0 - 100, while +%% set_sysmem_high_watermark takes a float in the range 0.0 - 1.0. +handle_call(get_sysmem_high_watermark, _From, State) -> + {reply, trunc(100 * State#state.memory_fraction), State}; + +handle_call({set_sysmem_high_watermark, Float}, _From, State) -> + {reply, ok, State#state{memory_fraction = Float}}; + +handle_call(get_check_interval, _From, State) -> + {reply, State#state.timeout, State}; + +handle_call({set_check_interval, Timeout}, _From, State) -> + {ok, cancel} = timer:cancel(State#state.timer), + {reply, ok, State#state{timeout = Timeout, timer = start_timer(Timeout)}}; + +handle_call(_Request, _From, State) -> + {noreply, State}. + +handle_cast(update, State = #state{alarmed = Alarmed, + memory_fraction = MemoryFraction}) -> + File = read_proc_file("/proc/meminfo"), + Lines = string:tokens(File, "\n"), + Dict = dict:from_list(lists:map(fun parse_line/1, Lines)), + MemTotal = dict:fetch('MemTotal', Dict), + MemUsed = MemTotal + - dict:fetch('MemFree', Dict) + - dict:fetch('Buffers', Dict) + - dict:fetch('Cached', Dict), + NewAlarmed = MemUsed / MemTotal > MemoryFraction, + case {Alarmed, NewAlarmed} of + {false, true} -> + alarm_handler:set_alarm({system_memory_high_watermark, []}); + {true, false} -> + alarm_handler:clear_alarm(system_memory_high_watermark); + _ -> + ok + end, + {noreply, State#state{alarmed = NewAlarmed}}; + +handle_cast(_Request, State) -> + {noreply, State}. + +handle_info(_Info, State) -> + {noreply, State}. + +terminate(_Reason, _State) -> + ok. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%%---------------------------------------------------------------------------- + +-define(BUFFER_SIZE, 1024). + +%% file:read_file does not work on files in /proc as it seems to get +%% the size of the file first and then read that many bytes. But files +%% in /proc always have length 0, we just have to read until we get +%% eof. +read_proc_file(File) -> + {ok, IoDevice} = file:open(File, [read, raw]), + Res = read_proc_file(IoDevice, []), + file:close(IoDevice), + lists:flatten(lists:reverse(Res)). + +read_proc_file(IoDevice, Acc) -> + case file:read(IoDevice, ?BUFFER_SIZE) of + {ok, Res} -> read_proc_file(IoDevice, [Res | Acc]); + eof -> Acc + end. + +%% A line looks like "FooBar: 123456 kB" +parse_line(Line) -> + [Name, Value | _] = string:tokens(Line, ": "), + {list_to_atom(Name), list_to_integer(Value)}. diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl index 3c8da720ff..18570171f7 100644 --- a/src/rabbit_misc.erl +++ b/src/rabbit_misc.erl @@ -43,6 +43,7 @@ -export([guid/0, string_guid/1, binstring_guid/1]). -export([dirty_read_all/1, dirty_foreach_key/2, dirty_dump_log/1]). -export([append_file/2]). +-export([format_stderr/2]). -import(mnesia). -import(lists). @@ -86,7 +87,7 @@ -spec(with_user_and_vhost/3 :: (username(), vhost(), thunk(A)) -> A). -spec(execute_mnesia_transaction/1 :: (thunk(A)) -> A). -spec(ensure_ok/2 :: ('ok' | {'error', any()}, atom()) -> 'ok'). --spec(localnode/1 :: (atom()) -> node()). +-spec(localnode/1 :: (atom()) -> erlang_node()). -spec(tcp_name/3 :: (atom(), ip_address(), ip_port()) -> atom()). -spec(intersperse/2 :: (A, [A]) -> [A]). -spec(upmap/2 :: (fun ((A) -> B), [A]) -> [B]). @@ -99,6 +100,7 @@ 'ok' | 'aborted'). -spec(dirty_dump_log/1 :: (string()) -> 'ok' | {'error', any()}). -spec(append_file/2 :: (string(), string()) -> 'ok' | {'error', any()}). +-spec(format_stderr/2 :: (string(), [any()]) -> 'true'). -endif. @@ -377,3 +379,8 @@ append_file(File, _, Suffix) -> {ok, Data} -> file:write_file([File, Suffix], Data, [append]); Error -> Error end. + +format_stderr(Fmt, Args) -> + Port = open_port({fd, 0, 2}, [out]), + port_command(Port, io_lib:format(Fmt, Args)), + port_close(Port). diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl index 9b67135def..8d34d2855b 100644 --- a/src/rabbit_mnesia.erl +++ b/src/rabbit_mnesia.erl @@ -40,11 +40,11 @@ -ifdef(use_specs). --spec(status/0 :: () -> [{'nodes' | 'running_nodes', [node()]}]). +-spec(status/0 :: () -> [{'nodes' | 'running_nodes', [erlang_node()]}]). -spec(ensure_mnesia_dir/0 :: () -> 'ok'). -spec(init/0 :: () -> 'ok'). -spec(is_db_empty/0 :: () -> bool()). --spec(cluster/1 :: ([node()]) -> 'ok'). +-spec(cluster/1 :: ([erlang_node()]) -> 'ok'). -spec(reset/0 :: () -> 'ok'). -spec(force_reset/0 :: () -> 'ok'). -spec(create_tables/0 :: () -> 'ok'). diff --git a/src/rabbit_multi.erl b/src/rabbit_multi.erl index c6a7e9200d..b99dfbc1c1 100644 --- a/src/rabbit_multi.erl +++ b/src/rabbit_multi.erl @@ -46,18 +46,22 @@ start() -> io:format("done.~n"), init:stop(); {'EXIT', {function_clause, [{?MODULE, action, _} | _]}} -> - io:format("Invalid command ~p~n", [FullCommand]), + error("invalid command '~s'", + [lists:flatten( + rabbit_misc:intersperse(" ", FullCommand))]), usage(); timeout -> - io:format("timeout starting some nodes.~n"), + error("timeout starting some nodes.", []), halt(1); Other -> - io:format("~nrabbit_multi action ~p failed:~n~p~n", - [Command, Other]), + error("~p", [Other]), halt(2) end end. +error(Format, Args) -> + rabbit_misc:format_stderr("Error: " ++ Format ++ "~n", Args). + parse_args([Command | Args]) -> {list_to_atom(Command), Args}. @@ -223,8 +227,7 @@ write_pids_file(Pids) -> {ok, Device} -> Device; {error, Reason} -> - throw({error, {cannot_create_pids_file, - FileName, Reason}}) + throw({cannot_create_pids_file, FileName, Reason}) end, try ok = io:write(Handle, Pids), @@ -233,8 +236,7 @@ write_pids_file(Pids) -> case file:close(Handle) of ok -> ok; {error, Reason1} -> - throw({error, {cannot_create_pids_file, - FileName, Reason1}}) + throw({cannot_create_pids_file, FileName, Reason1}) end end, ok. @@ -244,8 +246,7 @@ delete_pids_file() -> case file:delete(FileName) of ok -> ok; {error, enoent} -> ok; - {error, Reason} -> throw({error, {cannot_delete_pids_file, - FileName, Reason}}) + {error, Reason} -> throw({cannot_delete_pids_file, FileName, Reason}) end. read_pids_file() -> @@ -253,8 +254,7 @@ read_pids_file() -> case file:consult(FileName) of {ok, [Pids]} -> Pids; {error, enoent} -> []; - {error, Reason} -> throw({error, {cannot_read_pids_file, - FileName, Reason}}) + {error, Reason} -> throw({cannot_read_pids_file, FileName, Reason}) end. kill_wait(Pid, TimeLeft, Forceful) when TimeLeft < 0 -> diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl index 2ed57d7783..04d4c7d453 100644 --- a/src/rabbit_networking.erl +++ b/src/rabbit_networking.erl @@ -48,13 +48,13 @@ -spec(start_tcp_listener/2 :: (host(), ip_port()) -> 'ok'). -spec(stop_tcp_listener/2 :: (host(), ip_port()) -> 'ok'). -spec(active_listeners/0 :: () -> [listener()]). --spec(node_listeners/1 :: (node()) -> [listener()]). +-spec(node_listeners/1 :: (erlang_node()) -> [listener()]). -spec(connections/0 :: () -> [connection()]). -spec(connection_info/1 :: (connection()) -> [info()]). -spec(connection_info/2 :: (connection(), [info_key()]) -> [info()]). -spec(connection_info_all/0 :: () -> [[info()]]). -spec(connection_info_all/1 :: ([info_key()]) -> [[info()]]). --spec(on_node_down/1 :: (node()) -> 'ok'). +-spec(on_node_down/1 :: (erlang_node()) -> 'ok'). -spec(check_tcp_listener_address/3 :: (atom(), host(), ip_port()) -> {ip_address(), atom()}). diff --git a/src/rabbit_sasl_report_file_h.erl b/src/rabbit_sasl_report_file_h.erl index 3374d63d0b..10e2ee6eb9 100644 --- a/src/rabbit_sasl_report_file_h.erl +++ b/src/rabbit_sasl_report_file_h.erl @@ -71,8 +71,9 @@ handle_call(Event, State) -> terminate(Reason, State) -> sasl_report_file_h:terminate(Reason, State). -code_change(OldVsn, State, Extra) -> - sasl_report_file_h:code_change(OldVsn, State, Extra). +code_change(_OldVsn, State, _Extra) -> + %% There is no sasl_report_file_h:code_change/3 + {ok, State}. %%---------------------------------------------------------------------- diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index fff02d7363..db78bbcc50 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -465,7 +465,7 @@ test_user_management() -> control_action(Command, Args) -> control_action(Command, node(), Args). control_action(Command, Node, Args) -> - case catch rabbit_control:action(Command, Node, Args) of + case catch rabbit_control:action(Command, Node, Args, fun io:format/2) of ok -> io:format("done.~n"), ok; |