diff options
Diffstat (limited to 'src')
-rw-r--r-- | src/bench.erl | 18 | ||||
-rw-r--r-- | src/plop_compat.erl | 30 | ||||
-rw-r--r-- | src/statusreport.erl | 68 |
3 files changed, 96 insertions, 20 deletions
diff --git a/src/bench.erl b/src/bench.erl new file mode 100644 index 0000000..06e4777 --- /dev/null +++ b/src/bench.erl @@ -0,0 +1,18 @@ +%%% Copyright (c) 2017, NORDUnet A/S. +%%% See LICENSE for licensing information. + +-module(bench). + +-export([timingpoint/3]). + +timingpoint(Service, Target, Tag) -> + Thispoint = plop_compat:monotonic_time(millisecond), + Seq = plop_compat:unique_integer([monotonic]), + case get(bench_lastpoint) of + undefined -> + statusreport:bench(Service, Target, Tag, Seq, Thispoint + plop_compat:time_offset(millisecond), null); + Lastpoint -> + statusreport:bench(Service, Target, Tag, Seq, Lastpoint + plop_compat:time_offset(millisecond), Thispoint - Lastpoint) + end, + put(bench_lastpoint, Thispoint), + ok. diff --git a/src/plop_compat.erl b/src/plop_compat.erl index 4d45590..5c1fa17 100644 --- a/src/plop_compat.erl +++ b/src/plop_compat.erl @@ -2,7 +2,7 @@ %%% See LICENSE for licensing information. -module(plop_compat). --export([unpack_spki/1, timestamp/0, monotonic_time/1, start_timer/4]). +-export([unpack_spki/1, timestamp/0, monotonic_time/1, start_timer/4, unique_integer/1, time_offset/1]). -include_lib("public_key/include/public_key.hrl"). unpack_spki(SPKI) -> @@ -13,6 +13,10 @@ monotonic_time(Unit) -> monotonic_time(erlang:system_info(otp_release), Unit). start_timer(Time, Dest, Msg, Options) -> start_timer(erlang:system_info(otp_release), Time, Dest, Msg, Options). +unique_integer(Modifiers) -> + unique_integer(erlang:system_info(otp_release), Modifiers). +time_offset(Unit) -> + time_offset(erlang:system_info(otp_release), Unit). unpack_spki("R16" ++ _, SPKI) -> #'SubjectPublicKeyInfo'{subjectPublicKey = {_, Octets}, @@ -43,15 +47,35 @@ timestamp("19") -> monotonic_time("R16" ++ _, millisecond) -> {MeS, S, MiS} = timestamp(), - MeS * 1000000 + S * 1000 + MiS; + (MeS * 1000000 + S) * 1000 + MiS div 1000; monotonic_time("17", millisecond) -> {MeS, S, MiS} = timestamp(), - MeS * 1000000 + S * 1000 + MiS; + (MeS * 1000000 + S) * 1000 + MiS div 1000; monotonic_time("18", Unit) -> erlang:monotonic_time(Unit); monotonic_time("19", Unit) -> erlang:monotonic_time(Unit). +unique_integer("R16", _Modifiers) -> + {MeS, S, MiS} = erlang:now(), + (MeS * 1000000 + S) * 1000000 + MiS; +unique_integer("17", _Modifiers) -> + {MeS, S, MiS} = erlang:now(), + (MeS * 1000000 + S) * 1000000 + MiS; +unique_integer("18", Modifiers) -> + erlang:unique_integer(Modifiers); +unique_integer("19", Modifiers) -> + erlang:unique_integer(Modifiers). + +time_offset("R16", _Unit) -> + 0; +time_offset("17", _Unit) -> + 0; +time_offset("18", Unit) -> + erlang:time_offset(Unit); +time_offset("19", Unit) -> + erlang:time_offset(Unit). + start_timer("R16" ++ _, Time, Dest, Msg, [{abs, true}]) -> erlang:start_timer(max(0, Time - monotonic_time(millisecond)), Dest, Msg); start_timer("17", Time, Dest, Msg, [{abs, true}]) -> diff --git a/src/statusreport.erl b/src/statusreport.erl index db85b84..8099f86 100644 --- a/src/statusreport.erl +++ b/src/statusreport.erl @@ -8,7 +8,7 @@ -export([init/1, handle_call/3, terminate/2, handle_cast/2, handle_info/2, code_change/3]). -export([report/4]). --export([report_multi/4]). +-export([report_multi/4, bench/6]). -record(state, { timer :: none|reference(), @@ -32,7 +32,7 @@ init([]) -> lastsent = plop_compat:monotonic_time(millisecond) - ReportInterval}}. store_status(State, Service, Target, Variable, Status) -> - Statusreports = dict:store({Service, Target, Variable}, + Statusreports = dict:store({{status, Service}, {Target, Variable}}, {single, Status}, State#state.statusreports), State#state{statusreports = Statusreports}. @@ -45,13 +45,19 @@ dict_append_set(Key, Value, Dict) -> {multi, AppendSet}, Dict). store_multi_status(State, Service, Target, Variable, Status) -> - Statusreports = dict_append_set({Service, Target, Variable}, + Statusreports = dict_append_set({{status, Service}, {Target, Variable}}, Status, State#state.statusreports), State#state{statusreports = Statusreports}. +store_bench(State, Service, Target, Tag, Seq, Starttime, Elapsed) -> + Statusreports = dict:store({{bench, Service}, {Target, Tag, Seq}}, + {Starttime, Elapsed}, + State#state.statusreports), + State#state{statusreports = Statusreports}. + store_set_status(State, Service, Target, Variable, Statuses) -> - Statusreports = dict:store({Service, Target, Variable}, + Statusreports = dict:store({{status, Service}, {Target, Variable}}, {multi, Statuses}, State#state.statusreports), State#state{statusreports = Statusreports}. @@ -72,6 +78,9 @@ handle_cast({report, Service, Target, Variable, Status}, State) -> {noreply, try_send(NewState)}; handle_cast({report_multi, Service, Target, Variable, Status}, State) -> NewState = store_multi_status(State, Service, Target, Variable, Status), + {noreply, try_send(NewState)}; +handle_cast({bench, Service, Target, Tag, Seq, Starttime, Elapsed}, State) -> + NewState = store_bench(State, Service, Target, Tag, Seq, Starttime, Elapsed), {noreply, try_send(NewState)}. handle_info({timeout, _Timer, force_send}, State) -> @@ -123,8 +132,8 @@ terminate(Reason, State) -> group_by_service(Statusreports) -> dict:to_list( lists:foldl( - fun ({{Service, Target, Variable}, Status}, Acc) -> - dict:append(Service, {Target, Variable, Status}, Acc) + fun ({{Service, Group}, Status}, Acc) -> + dict:append(Service, {Group, Status}, Acc) end, dict:new(), dict:to_list(Statusreports))). encode_one_status(Status) when is_number(Status) -> @@ -141,22 +150,43 @@ encode_status({single, Status}) -> encode_status({multi, Statuses}) -> lists:map(fun encode_one_status/1, sets:to_list(Statuses)). -send(Service, Statusreports, Nodename) -> - lager:debug("reporting status to ~p: ~p", [Service, Statusreports]), - [NodeAddress] = plopconfig:get_env(statusservers, []), +encode_report(status, Nodename, {{Target, Variable}, Status}) -> + {struct, + [{"target", list_to_binary(Target)}, + {"source", list_to_binary(Nodename)}, + {"key", list_to_binary(Variable)}, + {"value", encode_status(Status)}]}; + +encode_report(bench, Nodename, {{Target, Tag, Seq}, {Starttime, Elapsed}}) -> + {struct, + [{"target", list_to_binary(Target)}, + {"source", list_to_binary(Nodename)}, + {"tag", list_to_binary(Tag)}, + {"seq", Seq}, + {"starttime", Starttime}, + {"elapsed", Elapsed}]}. + +addresses_for_servicetype(status) -> + plopconfig:get_env(statusservers, []); +addresses_for_servicetype(bench) -> + plopconfig:get_env(benchservers, []). + +send({ServiceType, Service}, Statusreports, Nodename) -> + lager:debug("reporting status to ~p ~p: ~p", [ServiceType, Service, Statusreports]), + NodeAddresses = addresses_for_servicetype(ServiceType), DebugTag = "statusreport", - URL = NodeAddress ++ Service, Headers = [{"Content-Type", "text/json"}], RequestBody = list_to_binary( mochijson2:encode( [ - {struct, - [{"target", list_to_binary(Target)}, - {"source", list_to_binary(Nodename)}, - {"key", list_to_binary(Variable)}, - {"value", encode_status(Status)}]} - || {Target, Variable, Status} <- Statusreports + encode_report(ServiceType, Nodename, Statusreport) + || Statusreport <- Statusreports ])), + lists:foreach(fun (NodeAddress) -> + send_one_server(DebugTag, NodeAddress ++ Service, Headers, RequestBody) + end, NodeAddresses). + +send_one_server(DebugTag, URL, Headers, RequestBody) -> case plop_httputil:request(DebugTag, URL, Headers, RequestBody) of {error, Err} -> lager:debug("request error ~p ~p", [DebugTag, Err]); @@ -187,7 +217,7 @@ try_send(State) -> lager:debug("status report sent ~p ms ago, setting timer", [NextSend - Now]), set_timer(State); true -> - lager:debug("status report send long enough ago"), + lager:debug("status report sent long enough ago"), force_send(State) end. @@ -198,3 +228,7 @@ report(Service, Target, Variable, Status) when is_number(Status); is_list(Status report_multi(Service, Target, Variable, Status) when is_number(Status); is_list(Status); is_binary(Status) -> lager:debug("reporting multi status ~p ~p ~p ~p", [Service, Target, Variable, Status]), gen_server:cast(?MODULE, {report_multi, Service, Target, Variable, Status}). + +bench(Service, Target, Tag, Seq, Starttime, Elapsed) -> + lager:debug("reporting bench ~p ~p ~p ~p ~p ~p", [Service, Target, Tag, Seq, Starttime, Elapsed]), + gen_server:cast(?MODULE, {bench, Service, Target, Tag, Seq, Starttime, Elapsed}). |