<commit_before>%  @copyright 2011-2014 Zuse Institute Berlin

%   Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
%
%   Unless required by applicable law or agreed to in writing, software
%   distributed under the License is distributed on an "AS IS" BASIS,
%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%   See the License for the specific language governing permissions and
%   limitations under the License.

%% @author <NAME> <<EMAIL>>
%% @doc    Periodically executes a small benchmark to monitor the overall
%%         performance of Scalaris.
%% @end
%% @version $Id$
-module(monitor_perf).
-author('<EMAIL>').
-vsn('$Id$').

-behaviour(gen_component).

-include("record_helpers.hrl").
-include("scalaris.hrl").

% monitor process functions
-export([start_link/1, init/1, on/2, check_config/0]).

-record(state,
        {id        = ?required(state, id)        :: uid:global_uid(),
         perf_rr   = ?required(state, perf_rr)   :: rrd:rrd(),
         perf_lh   = ?required(state, perf_lh)   :: rrd:rrd(),
         perf_tx   = ?required(state, perf_tx)   :: rrd:rrd()
        }).

-type state() :: {AllNodes::#state{}, CollectingAtLeader::#state{}, BenchPid::pid() | ok, IgnoreBenchTimeout::boolean()}.
-type message() ::
    {bench} |
    {bench_result, Time::erlang_timestamp(), TimeInMs::non_neg_integer()} |
    {bench_timeout, Time::erlang_timestamp(), BenchPid::pid()} |
    {collect_system_stats} |
    {propagate} |
    {get_node_details_response, node_details:node_details()} |
    {bulkowner, deliver, Id::uid:global_uid(), Range::intervals:interval(), {gather_stats, SourcePid::comm:mypid(), Id::uid:global_uid()}, Parents::[comm:mypid(),...]} |
    {{get_rrds_response, [{Process::atom(), Key::monitor:key(), DB::rrd:rrd() | undefined}]}, {SourcePid::comm:mypid(), Id::uid:global_uid()}} |
    {{get_rrds_response, [{Process::atom(), Key::monitor:key(), DB::rrd:rrd() | undefined}]}, {SourcePid::comm:mypid(), Id::uid:global_uid(), MyMonData::[{Process::atom(), Key::monitor:key(), Data::rrd:timing_type()}]}} |
    {bulkowner, gather, Id::uid:global_uid(), Target::comm:mypid(), Msgs::[comm:message(),...], Parents::[comm:mypid()]} |
    {bulkowner, reply, Id::uid:global_uid(), {gather_stats_response, Id::uid:global_uid(), [{Process::atom(), Key::monitor:key(), Data::rrd:timing_type()}]}} |
    {bulkowner, deliver, Id::uid:global_uid(), Range::intervals:interval(), {report_value, StatsOneRound::#state{}}, Parents::[comm:mypid(),...]} |
    {get_rrds, [{Process::atom(), Key::monitor:key()},...], SourcePid::comm:mypid()}.

%-define(TRACE(X,Y), log:pal(X,Y)).
-define(TRACE(X,Y), ok).
-define(TRACE1(Msg, State),
        ?TRACE("[ ~.0p ]~n  Msg: ~.0p~n  State: ~.0p)~n", [self(), Msg, State])).

%% @doc Creates a monitoring value for benchmarks with a 1m monitoring interval
%%      and a timing histogram, only keeping the newest value.
-spec init_bench() -> ok.
init_bench() ->
    monitor:proc_set_value(
      ?MODULE, 'read_read', rrd:create(60 * 1000000, 1, {timing_with_hist, ms})).

-spec bench_service(Owner::pid()) -> ok.
bench_service(Owner) ->
    erlang:monitor(process, Owner),
    bench_service_loop(Owner).

-spec bench_service_loop(Owner::comm:erl_local_pid()) -> ok.
bench_service_loop(Owner) ->
    trace_mpath:thread_yield(),
    receive
        ?SCALARIS_RECV({bench}, %% ->
            begin
                Key1 = randoms:getRandomString(),
                Key2 = randoms:getRandomString(),
                ReqList = [{read, Key1}, {read, Key2}, {commit}],
                Time = os:timestamp(),
                {TimeInUs, _Result} = util:tc(fun api_tx:req_list/1, [ReqList]),
                comm:send_local(Owner, {bench_result, Time, TimeInUs / 1000}),
                bench_service_loop(Owner)
            end);
        ?SCALARIS_RECV({tx_tm_rtm_commit_reply, _, _}, %% ->
            % left-over commit information from bench, more specifically api_tx:req_list/1
            bench_service_loop(Owner));
        {'DOWN', _MonitorRef, process, Owner, _Info1} -> ok
    end.

-spec init_system_stats() -> ok.
init_system_stats() ->
    % system stats in 10s intervals:
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_total', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_processes', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_system', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_atom', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_binary', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_ets', rrd:create(15 * 1000000, 1, gauge)),
    
    monitor:client_monitor_set_value(
      ?MODULE, 'rcv_count', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'rcv_bytes', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'send_count', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'send_bytes', rrd:create(15 * 1000000, 1, gauge)).

-spec collect_system_stats() -> ok.
collect_system_stats() ->
    [{total, MemTotal}, {processes, MemProcs}, {system, MemSys},
     {atom, MemAtom}, {binary, MemBin}, {ets, MemEts}] =
        erlang:memory([total, processes, system, atom, binary, ets]),
    
    monitor:client_monitor_set_value(?MODULE, 'mem_total',
                           fun(Old) -> rrd:add_now(MemTotal, Old) end),
    monitor:client_monitor_set_value(?MODULE, 'mem_processes',
                           fun(Old) -> rrd:add_now(MemProcs, Old) end),<commit_msg>monitor_perf: also set system stats at init<commit_after>%  @copyright 2011-2014 Zuse Institute Berlin

%   Licensed under the Apache License, Version 2.0 (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.apache.org/licenses/LICENSE-2.0
%
%   Unless required by applicable law or agreed to in writing, software
%   distributed under the License is distributed on an "AS IS" BASIS,
%   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%   See the License for the specific language governing permissions and
%   limitations under the License.

%% @author <NAME> <<EMAIL>>
%% @doc    Periodically executes a small benchmark to monitor the overall
%%         performance of Scalaris.
%% @end
%% @version $Id$
-module(monitor_perf).
-author('<EMAIL>').
-vsn('$Id$').

-behaviour(gen_component).

-include("record_helpers.hrl").
-include("scalaris.hrl").

% monitor process functions
-export([start_link/1, init/1, on/2, check_config/0]).

-record(state,
        {id        = ?required(state, id)        :: uid:global_uid(),
         perf_rr   = ?required(state, perf_rr)   :: rrd:rrd(),
         perf_lh   = ?required(state, perf_lh)   :: rrd:rrd(),
         perf_tx   = ?required(state, perf_tx)   :: rrd:rrd()
        }).

-type state() :: {AllNodes::#state{}, CollectingAtLeader::#state{}, BenchPid::pid() | ok, IgnoreBenchTimeout::boolean()}.
-type message() ::
    {bench} |
    {bench_result, Time::erlang_timestamp(), TimeInMs::non_neg_integer()} |
    {bench_timeout, Time::erlang_timestamp(), BenchPid::pid()} |
    {collect_system_stats} |
    {propagate} |
    {get_node_details_response, node_details:node_details()} |
    {bulkowner, deliver, Id::uid:global_uid(), Range::intervals:interval(), {gather_stats, SourcePid::comm:mypid(), Id::uid:global_uid()}, Parents::[comm:mypid(),...]} |
    {{get_rrds_response, [{Process::atom(), Key::monitor:key(), DB::rrd:rrd() | undefined}]}, {SourcePid::comm:mypid(), Id::uid:global_uid()}} |
    {{get_rrds_response, [{Process::atom(), Key::monitor:key(), DB::rrd:rrd() | undefined}]}, {SourcePid::comm:mypid(), Id::uid:global_uid(), MyMonData::[{Process::atom(), Key::monitor:key(), Data::rrd:timing_type()}]}} |
    {bulkowner, gather, Id::uid:global_uid(), Target::comm:mypid(), Msgs::[comm:message(),...], Parents::[comm:mypid()]} |
    {bulkowner, reply, Id::uid:global_uid(), {gather_stats_response, Id::uid:global_uid(), [{Process::atom(), Key::monitor:key(), Data::rrd:timing_type()}]}} |
    {bulkowner, deliver, Id::uid:global_uid(), Range::intervals:interval(), {report_value, StatsOneRound::#state{}}, Parents::[comm:mypid(),...]} |
    {get_rrds, [{Process::atom(), Key::monitor:key()},...], SourcePid::comm:mypid()}.

%-define(TRACE(X,Y), log:pal(X,Y)).
-define(TRACE(X,Y), ok).
-define(TRACE1(Msg, State),
        ?TRACE("[ ~.0p ]~n  Msg: ~.0p~n  State: ~.0p)~n", [self(), Msg, State])).

%% @doc Creates a monitoring value for benchmarks with a 1m monitoring interval
%%      and a timing histogram, only keeping the newest value.
-spec init_bench() -> ok.
init_bench() ->
    monitor:proc_set_value(
      ?MODULE, 'read_read', rrd:create(60 * 1000000, 1, {timing_with_hist, ms})).

-spec bench_service(Owner::pid()) -> ok.
bench_service(Owner) ->
    erlang:monitor(process, Owner),
    bench_service_loop(Owner).

-spec bench_service_loop(Owner::comm:erl_local_pid()) -> ok.
bench_service_loop(Owner) ->
    trace_mpath:thread_yield(),
    receive
        ?SCALARIS_RECV({bench}, %% ->
            begin
                Key1 = randoms:getRandomString(),
                Key2 = randoms:getRandomString(),
                ReqList = [{read, Key1}, {read, Key2}, {commit}],
                Time = os:timestamp(),
                {TimeInUs, _Result} = util:tc(fun api_tx:req_list/1, [ReqList]),
                comm:send_local(Owner, {bench_result, Time, TimeInUs / 1000}),
                bench_service_loop(Owner)
            end);
        ?SCALARIS_RECV({tx_tm_rtm_commit_reply, _, _}, %% ->
            % left-over commit information from bench, more specifically api_tx:req_list/1
            bench_service_loop(Owner));
        {'DOWN', _MonitorRef, process, Owner, _Info1} -> ok
    end.

-spec init_system_stats() -> ok.
init_system_stats() ->
    % system stats in 10s intervals:
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_total', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_processes', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_system', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_atom', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_binary', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'mem_ets', rrd:create(15 * 1000000, 1, gauge)),
    
    monitor:client_monitor_set_value(
      ?MODULE, 'rcv_count', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'rcv_bytes', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'send_count', rrd:create(15 * 1000000, 1, gauge)),
    monitor:client_monitor_set_value(
      ?MODULE, 'send_bytes', rrd:create(15 * 1000000, 1, gauge)),

    collect_system_stats().

-spec collect_system_stats() -> ok.
collect_system_stats() ->
    [{total, MemTotal}, {processes, MemProcs}, {system, MemSys},
     {atom, MemAtom}, {binary, MemBin}, {ets, MemEts}] =
        erlang:memory([total, processes, system, atom, binary, ets]),
    
    monitor:client_monitor_set_value(?MODULE, 'mem_total',
                           fun(Old) -> rrd:add_now(MemTotal, Old) end),
    monitor:client_monitor_set_value(?MODULE, 'mem_processes',
                           fun(Old) -> rrd:add_now(MemProcs, Old) end),
{'fraction_non_alphanumeric': 0.168980897079862, 'fraction_numerical': 0.024236303963645543, 'mean_word_length': 3.970305311585111, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '471553', 'n_tokens_mistral': 4207, 'n_tokens_neox': 3919, 'n_words': 848}