%%% @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]).
