%%% @copyright (C) 2019-2023, Maxim Fedorov %%% @doc %%% Logs monitoring events for the entire cluster, to file or device. %%% Requires {@link erlperf_history} service running, fails otherwise. %%% Uses completely different to {@link erlperf_monitor} approach; instead of waiting %%% for new samples to come, cluster monitor just outputs existing %%% samples periodically. %%% %%% Example primary node: %%% ``` %%% rebar3 shell --sname primary %%% (primary@ubuntu22)1> erlperf_history:start_link(). %%% {ok,<0.211.0>} %%% (primary@ubuntu22)2> erlperf_cluster_monitor:start_link(). %%% {ok,<0.216.0>} %%% ''' %%% %%% Example benchmarking node: %%% ``` %%% rebar3 shell --sname bench1 %%% (bench1@ubuntu22)1> net_kernel:connect_node('primary@ubuntu22'). %%% true %%% (bench1@ubuntu22)2> erlperf:run(rand, uniform, []). %%% ''' %%% %%% As soon as the new benchmarking jon on the node `bench' is started, it is %%% reported in the cluster monitoring output. %%% @end -module(erlperf_cluster_monitor). -author("maximfca@gmail.com"). -behaviour(gen_server). %% API -export([ start_link/0, start_link/3 ]). %% gen_server callbacks -export([ init/1, handle_call/3, handle_cast/2, handle_info/2 ]). %% Handler: just like gen_event handler. %% If you do need gen_event handler, make a fun of it. -type handler() :: {module(), atom(), term()} | file:filename_all() | {fd, io:device()} | io:device(). %% Specifies monitoring output device. %% %% It could be an output {@link io:device()} (such as %% {@link erlang:group_leader/0}, `user' or `standard_error'), a file name, or a %% `{Module, Function, UserState}' tuple. In the latter case, instead of printing, cluster monitor %% calls the specified function, which must have arity of 2, accepting filtered %% {@link erlperf_monitor:monitor_sample()} as the first argument, and `Userstate' as the second, %% returning next `UserState'. %% Take a sample every second -define(DEFAULT_INTERVAL, 1000). -define(KNOWN_FIELDS, [time, node, sched_util, dcpu, dio, processes, ports, ets, memory_total, memory_processes, memory_binary, memory_ets, jobs]). %% @equiv start_link(erlang:group_leader(), 1000, undefined) -spec start_link() -> {ok, Pid :: pid()} | {error, Reason :: term()}. start_link() -> start_link(erlang:group_leader(), ?DEFAULT_INTERVAL, undefined). %% @doc %% Starts cluster-wide monitor process, and links it to the caller. %% %% Intended to be used in a supervisor `ChildSpec', making the process a part of the supervision tree. %% %% `IntervalMs' specifies time, in milliseconds, between output handler invocations. %% %% Fields specifies the list of field names to report, and the order in which columns are printed. %% see {@link erlperf_monitor:monitor_sample()} for options. Passing `undefined' prints all columns %% known by this version of `erlperf'. %% @end -spec start_link(Handler :: handler(), IntervalMs :: pos_integer(), Fields :: [atom()] | undefined) -> {ok, Pid :: pid()} | {error, Reason :: term()}. start_link(Handler, Interval, Fields) -> gen_server:start_link(?MODULE, [Handler, Interval, Fields], []). %%%=================================================================== %%% gen_server callbacks %% System monitor state -record(state, { next :: integer(), %% absolute timer for the next tick interval :: pos_integer(), handler :: handler(), fields :: [atom()] | undefined, %% previously printed header %% if the new header is different from the previous one, it gets printed header = [] :: [atom() | [pid()]] }). %% @private init([Handler, Interval, Fields0]) -> Fields = if Fields0 =:= undefined -> ?KNOWN_FIELDS; true -> Fields0 end, %% use absolute timer to avoid skipping ticks Now = erlang:monotonic_time(millisecond), {ok, handle_tick(#state{next = Now, interval = Interval, handler = make_handler(Handler), fields = Fields})}. %% @private handle_call(_Request, _From, _State) -> erlang:error(notsup). %% @private handle_cast(_Request, _State) -> erlang:error(notsup). %% @private handle_info({timeout, _, tick}, State) -> {noreply, handle_tick(State)}. %%%=================================================================== %%% Internal functions handle_tick(#state{next = Now, interval = Interval, fields = Fields, handler = Handler, header = Header} = State) -> Next = Now + Interval, %% erlang:start_timer(Next, self(), tick, [{abs, true}]), %% last interval updates GetHistoryTo = Now + erlang:time_offset(millisecond), %% be careful not to overlap the timings (history:get is inclusive) Samples = erlperf_history:get(GetHistoryTo - Interval + 1, GetHistoryTo), %% now invoke the handler {NewHandler, NewHeader} = run_handler(Handler, Fields, Header, Samples), State#state{next = Next, handler = NewHandler, header = NewHeader}. make_handler({_M, _F, _A} = MFA) -> MFA; make_handler(IoDevice) when is_pid(IoDevice); is_atom(IoDevice) -> {fd, IoDevice}; make_handler({fd, IoDevice}) when is_pid(IoDevice); is_atom(IoDevice) -> {fd, IoDevice}; make_handler(Filename) when is_list(Filename); is_binary(Filename) -> {ok, Fd} = file:open(Filename, [raw, append]), {fd, Fd}. run_handler(Handler, _Fields, Header, []) -> {Handler, Header}; %% handler: MFA callback run_handler({M, F, A}, Fields, Header, Samples) -> Filtered = [{Time, maps:with(Fields, Sample)} || {Time, Sample} <- Samples], {{M, F, M:F(Filtered, A)}, Header}; %% built-in handler: file/console output run_handler({fd, IoDevice}, Fields, Header, Samples) -> %% the idea of the formatter below is to print lines like this: %% Dane Time node sched ets memory <123.456.1> <0.123.0> %% 2022-11-12 08:35:16 node1@host 33.5% 16 128111 12345 %% 2022-11-12 08:35:16 node1@host 33.5% 16 128111 9111 %% collect all jobs from all samples Jobs = lists:usort(lists:foldl( fun ({_Time, #{jobs := Jobs}}, Acc) -> {Pids, _} = lists:unzip(Jobs), Pids ++ Acc end, [], Samples)), %% replace atom 'jobs' with list of Jobs. This is effectively lists:keyreplace, but with no key NewHeader = [if F =:= jobs -> Jobs; true -> F end || F <- Fields], %% format specific fields of samples Formatted = [ [formatter(F, if is_list(F) -> maps:get(jobs, Sample); true -> maps:get(F, Sample) end) || F <- NewHeader] || {_Time, Sample} <- Samples], NewLine = io_lib:nl(), BinNl = list_to_binary(NewLine), %% check if header has changed and print if it has NewHeader =/= Header andalso begin FmtHdr = [header(S) || S <- NewHeader] ++ [BinNl], ok = file:write(IoDevice, FmtHdr) end, %% print the actual line Data = [F ++ NewLine || F <- Formatted], ok = file:write(IoDevice, Data), {{fd, IoDevice}, NewHeader}. header(time) -> <<" date time TZ ">>; header(sched_util) -> <<" %sched">>; header(dcpu) -> <<" %dcpu">>; header(dio) -> <<" %dio">>; header(processes) -> <<" procs">>; header(ports) -> <<" ports">>; header(ets) -> <<" ets">>; header(memory_total) -> <<" mem_total">>; header(memory_processes) -> <<" mem_proc">>; header(memory_binary) -> <<" mem_bin">>; header(memory_ets) -> <<" mem_ets">>; header(Jobs) when is_list(Jobs) -> iolist_to_binary([io_lib:format("~16s", [pid_to_list(Pid)]) || Pid <- Jobs]); header(node) -> <<"node ">>. formatter(time, Time) -> calendar:system_time_to_rfc3339(Time div 1000) ++ " "; formatter(Percent, Num) when Percent =:= sched_util; Percent =:= dcpu; Percent =:= dio -> io_lib:format("~7.2f", [Num]); formatter(Number, Num) when Number =:= processes; Number =:= ports -> io_lib:format("~8b", [Num]); formatter(ets, Num) -> io_lib:format("~6b", [Num]); formatter(Size, Num) when Size =:= memory_total; Size =:= memory_processes; Size =:= memory_binary; Size =:= memory_ets -> io_lib:format("~10s", [erlperf_file_log:format_size(Num)]); formatter(Jobs, JobsInSample) when is_list(Jobs) -> %% here, all Jobs must be formatter, potentially empty (if they are not in JobsInSample) [case lists:keyfind(Job, 1, JobsInSample) of {Job, Num} -> io_lib:format("~16s", [erlperf_file_log:format_number(Num)]); false -> " " end || Job <- Jobs]; formatter(node, Node) -> io_lib:format("~*s", [-22, Node]).