%% -*- erlang-indent-level: 2 -*- %%%------------------------------------------------------------------- %%% File : system_monitor.erl %%% Description : Monitor for some system parameters. %%% %%% Created : 20 Dec 2011 by Thomas Jarvstrand <> %%%------------------------------------------------------------------- %% @private -module(system_monitor). -behaviour(gen_server). %%-------------------------------------------------------------------- %% Include files %%-------------------------------------------------------------------- -include_lib("system_monitor/include/system_monitor.hrl"). -include_lib("hut/include/hut.hrl"). %% API -export([start_link/0]). -export([reset/0]). -export([ report_full_status/0 , check_process_count/0 , self_monitor/0 , start_top/0 , stop_top/0 ]). %% gen_server callbacks -export([ init/1 , handle_call/3 , handle_cast/2 , handle_info/2 , terminate/2 ]). -include_lib("hut/include/hut.hrl"). -define(SERVER, ?MODULE). -define(TICK_INTERVAL, 1000). -record(state, { monitors = [] , timer_ref }). %% System monitor is started early, some application may be %% unavalable -define(MAYBE(Prog), try Prog catch _:_ -> undefined end). %%==================================================================== %% API %%==================================================================== %%-------------------------------------------------------------------- %% @doc Starts the server %%-------------------------------------------------------------------- -spec start_link() -> {ok, pid()} | ignore | {error, term()}. start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). %%-------------------------------------------------------------------- %% @doc Start printing erlang top to console %%-------------------------------------------------------------------- -spec start_top() -> ok. start_top() -> application:set_env(?APP, top_printing, group_leader()). %%-------------------------------------------------------------------- %% @doc Stop printing erlang top to console %%-------------------------------------------------------------------- -spec stop_top() -> ok. stop_top() -> application:set_env(?APP, top_printing, false). %%-------------------------------------------------------------------- %% @doc Reset monitors %%-------------------------------------------------------------------- -spec reset() -> ok. reset() -> gen_server:cast(?SERVER, reset). %%==================================================================== %% gen_server callbacks %%==================================================================== init([]) -> {ok, Timer} = timer:send_interval(?TICK_INTERVAL, {self(), tick}), {ok, #state{ monitors = init_monitors() , timer_ref = Timer }}. handle_call(_Request, _From, State) -> {reply, {error, unknown_call}, State}. handle_cast(reset, State) -> {noreply, State#state{monitors = init_monitors()}}; handle_cast(_Msg, State) -> {noreply, State}. handle_info({Self, tick}, State) when Self =:= self() -> Monitors = [case Ticks - 1 of 0 -> try apply(Module, Function, []) catch EC:Error:Stack -> error_logger:warning_msg( "system_monitor ~p crashed:~n~p:~p~nStacktrace: ~p~n", [{Module, Function}, EC, Error, Stack]) end, {Module, Function, F, TicksReset, TicksReset}; TicksDecremented -> {Module, Function, F, TicksReset, TicksDecremented} end || {Module, Function, F, TicksReset, Ticks} <- State#state.monitors], {noreply, State#state{monitors = Monitors}}; handle_info(_Info, State) -> {noreply, State}. -spec terminate(term(), #state{}) -> any(). terminate(_Reason, State) -> %% Possibly, one last check. [apply(?MODULE, Monitor, []) || {Monitor, true, _TicksReset, _Ticks} <- State#state.monitors]. %%============================================================================== %% Internal functions %%============================================================================== %%------------------------------------------------------------------------------ %% @doc Returns the list of initiated monitors. %%------------------------------------------------------------------------------ -spec init_monitors() -> [{module(), function(), boolean(), pos_integer(), pos_integer()}]. init_monitors() -> [{Module, Function, F, Ticks, Ticks} || {Module, Function, F, Ticks} <- monitors()]. %%------------------------------------------------------------------------------ %% @doc Returns the list of monitors. The format is %% {FunctionName, RunMonitorAtTerminate, NumberOfTicks}. %% RunMonitorAtTerminate determines whether the monitor is to be run in %% the terminate gen_server callback. %% ... and NumberOfTicks is the number of ticks between invocations of %% the monitor in question. So, if NumberOfTicks is 3600, the monitor is %% to be run once every hour, as there is a tick every second. %%------------------------------------------------------------------------------ -spec monitors() -> [{module(), function(), boolean(), pos_integer()}]. monitors() -> {ok, AdditionalMonitors} = application:get_env(system_monitor, status_checks), {ok, TopInterval} = application:get_env(?APP, top_sample_interval), [ {?MODULE, check_process_count, true, 2} , {?MODULE, self_monitor, false, 5} , {?MODULE, report_full_status, false, TopInterval div 1000} ] ++ AdditionalMonitors. %%------------------------------------------------------------------------------ %% @doc %% Monitor mailbox size of system_monitor_kafka process %% %% Check message queue length of this process and kill it when it's growing %% uncontrollably. It is needed because this process doesn't have backpressure %% by design %% @end %%------------------------------------------------------------------------------ self_monitor() -> message_queue_sentinel(system_monitor_kafka, 3000). -spec message_queue_sentinel(atom() | pid(), integer()) -> ok. message_queue_sentinel(Name, Limit) when is_atom(Name) -> case whereis(Name) of Pid when is_pid(Pid) -> message_queue_sentinel(Pid, Limit); _ -> ok end; message_queue_sentinel(Pid, Limit) when is_pid(Pid) -> case process_info(Pid, [message_queue_len, current_function]) of [{message_queue_len, Len}, {current_function, Fun}] when Len >= Limit -> ?log( warning , "Abnormal message queue length (~p). " "Process ~p (~p) will be terminated." , [Len, Pid, Fun] , #{domain => [system_monitor]} ), exit(Pid, kill); _ -> ok end. %%------------------------------------------------------------------------------ %% Monitor for number of processes %%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------ %% @doc Check the number of processes and log an aggregate summary of the %% process info if the count is above Threshold. %%------------------------------------------------------------------------------ -spec check_process_count() -> ok. check_process_count() -> {ok, MaxProcs} = application:get_env(?APP, top_max_procs), case erlang:system_info(process_count) of Count when Count > MaxProcs div 5 -> ?log( warning , "Abnormal process count (~p).~n" , [Count] , #{domain => [system_monitor]} ); _ -> ok end. %%------------------------------------------------------------------------------ %% @doc Report top processes %%------------------------------------------------------------------------------ -spec report_full_status() -> ok. report_full_status() -> %% `TS' variable should be used consistently in all following %% reports for this time interval, so it can be used as a key to %% lookup the relevant events {TS, ProcTop} = system_monitor_top:get_proc_top(), push_to_kafka(ProcTop), report_app_top(TS), %% Node status report goes last, and it "seals" the report for this %% time interval: NodeReport = case application:get_env(?APP, node_status_fun) of {ok, {Module, Function}} -> try Module:Function() catch _:_ -> <<>> end; _ -> <<>> end, system_monitor_kafka:produce({node_role, node(), TS, iolist_to_binary(NodeReport)}). %%------------------------------------------------------------------------------ %% @doc Calculate reductions per application. %%------------------------------------------------------------------------------ -spec report_app_top(erlang:timestamp()) -> ok. report_app_top(TS) -> AppReds = system_monitor_top:get_abs_app_top(), present_results(app_top, reductions, AppReds, TS), AppMem = system_monitor_top:get_app_memory(), present_results(app_top, memory, AppMem, TS), AppProcs = system_monitor_top:get_app_processes(), present_results(app_top, processes, AppProcs, TS), #{ current_function := CurrentFunction , initial_call := InitialCall } = system_monitor_top:get_function_top(), present_results(fun_top, current_function, CurrentFunction, TS), present_results(fun_top, initial_call, InitialCall, TS), ok. %%-------------------------------------------------------------------- %% @doc Push app_top or fun_top information to kafka %%-------------------------------------------------------------------- present_results(Record, Tag, Values, TS) -> {ok, Thresholds} = application:get_env(?APP, top_significance_threshold), Threshold = maps:get(Tag, Thresholds, 0), Node = node(), [system_monitor_kafka:produce({Record, Node, TS, Key, Tag, Val}) || {Key, Val} <- Values, Val > Threshold]. %%-------------------------------------------------------------------- %% @doc Push plain records to Kafka %%-------------------------------------------------------------------- -spec push_to_kafka([term()]) -> ok. push_to_kafka(L) -> lists:foreach(fun system_monitor_kafka:produce/1, L).