diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/ht.erl | 28 | ||||
| -rw-r--r-- | src/http_auth.erl | 5 | ||||
| -rw-r--r-- | src/plop_compat.erl | 27 | ||||
| -rw-r--r-- | src/plop_sup.erl | 1 | ||||
| -rw-r--r-- | src/statusreport.erl | 200 | ||||
| -rw-r--r-- | src/storage.erl | 14 |
6 files changed, 270 insertions, 5 deletions
@@ -29,7 +29,7 @@ -behaviour(gen_server). -export([reset_tree/1, load_tree/1, size/0, leaf_hash/1]). --export([add/1, root/0, root/1, path/2, consistency/2]). +-export([add/1, root/0, root/1, path/2, consistency/2, dump/1]). -export([start_link/0, start_link/1, stop/0]). -export([init/1, handle_call/3, terminate/2, handle_cast/2, handle_info/2, code_change/3]). @@ -96,6 +96,8 @@ print_tree() -> call(?MODULE, {print_tree, 4}). print_tree(HashOutputLen) -> call(?MODULE, {print_tree, HashOutputLen}). +dump(Filename) -> + call(?MODULE, {dump, Filename}). %% gen_server callbacks init(Args) -> @@ -141,7 +143,9 @@ handle_call({consistency, Version1, Version2}, _From, State) -> handle_call(testing_get_state, _From, State) -> {reply, State, State}; handle_call({print_tree, HashOutputLen}, _From, State) -> - {reply, print_tree(State, HashOutputLen), State}. + {reply, print_tree(State, HashOutputLen), State}; +handle_call({dump, Filename}, _From, State) -> + {reply, dump(State, Filename), State}. %%%%%%%%%%%%%%%%%%%% %% Private. @@ -444,6 +448,26 @@ print_layer(Tree, HashOutputLen, Layer, ILast) -> lists:seq(0, ILast)), io:format("~n"). +dump(Tree, Filename) -> + {ok, File} = file:open(Filename, [read, write, binary]), + ok = file:write(File, "# evaluated " ++ integer_to_list(Tree#tree.evaluated) ++ "\n"), + lists:foreach(fun (R) -> + write_layer_header(File, R), + lists:foreach(fun (I) -> + Entry = ts:retrieve(Tree#tree.store, {R, I}), + dumpone(File, Entry) + end, lists:seq(0, ts:count(Tree#tree.store, R) - 1)) + end, lists:seq(0, depth(Tree) - 1)), + file:close(File). + +write_layer_header(File, Layer) -> + EntryText = "# layer " ++ integer_to_list(Layer) ++ "\n", + ok = file:write(File, EntryText). + +dumpone(File, Entry) -> + EntryText = hex:bin_to_hexstr(Entry) ++ "\n", + ok = file:write(File, EntryText). + %%%%%%%%%%%%%%%%%%%% %% Testing ht. diff --git a/src/http_auth.erl b/src/http_auth.erl index 2cee51f..c8a8389 100644 --- a/src/http_auth.erl +++ b/src/http_auth.erl @@ -2,7 +2,7 @@ %%% See LICENSE for licensing information. -module(http_auth). --export([verify_auth/4, create_auth/3, init_key_table/0, sign_stored/1, verify_stored/3]). +-export([verify_auth/4, create_auth/3, init_key_table/0, sign_stored/1, verify_stored/3, own_name/0]). -define(KEY_TABLE, http_auth_keys). @@ -25,6 +25,9 @@ read_key_table() -> none end. +own_name() -> + {ok, {KeyName, _}} = application:get_env(plop, own_key), + KeyName. own_key() -> case application:get_env(plop, own_key, none) of diff --git a/src/plop_compat.erl b/src/plop_compat.erl index 9a98e3d..4d45590 100644 --- a/src/plop_compat.erl +++ b/src/plop_compat.erl @@ -2,14 +2,17 @@ %%% See LICENSE for licensing information. -module(plop_compat). --export([unpack_spki/1, timestamp/0]). +-export([unpack_spki/1, timestamp/0, monotonic_time/1, start_timer/4]). -include_lib("public_key/include/public_key.hrl"). unpack_spki(SPKI) -> unpack_spki(erlang:system_info(otp_release), SPKI). timestamp() -> timestamp(erlang:system_info(otp_release)). - +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). unpack_spki("R16" ++ _, SPKI) -> #'SubjectPublicKeyInfo'{subjectPublicKey = {_, Octets}, @@ -37,3 +40,23 @@ timestamp("18") -> erlang:timestamp(); timestamp("19") -> erlang:timestamp(). + +monotonic_time("R16" ++ _, millisecond) -> + {MeS, S, MiS} = timestamp(), + MeS * 1000000 + S * 1000 + MiS; +monotonic_time("17", millisecond) -> + {MeS, S, MiS} = timestamp(), + MeS * 1000000 + S * 1000 + MiS; +monotonic_time("18", Unit) -> + erlang:monotonic_time(Unit); +monotonic_time("19", Unit) -> + erlang:monotonic_time(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}]) -> + erlang:start_timer(max(0, Time - monotonic_time(millisecond)), Dest, Msg); +start_timer("18", Time, Dest, Msg, Options) -> + erlang:start_timer(Time, Dest, Msg, Options); +start_timer("19", Time, Dest, Msg, Options) -> + erlang:start_timer(Time, Dest, Msg, Options). diff --git a/src/plop_sup.erl b/src/plop_sup.erl index 27f7680..15f3fe8 100644 --- a/src/plop_sup.erl +++ b/src/plop_sup.erl @@ -50,6 +50,7 @@ init([]) -> Children = [permanent_worker(the_db, {db, start_link, []}, [db]), permanent_worker(the_storagedb, {storagedb, start_link, []}), permanent_worker(fsync, {fsyncport, start_link, []}), + permanent_worker(the_statusreport, {statusreport, start_link, []}), permanent_worker(plopcontrol, {plopcontrol, start_link, []})], OptionalChildren = lists:map(fun (ServiceName) -> case ServiceName of diff --git a/src/statusreport.erl b/src/statusreport.erl new file mode 100644 index 0000000..db85b84 --- /dev/null +++ b/src/statusreport.erl @@ -0,0 +1,200 @@ +%%% Copyright (c) 2017, NORDUnet A/S. +%%% See LICENSE for licensing information. + +-module(statusreport). +-behaviour(gen_server). + +-export([start_link/0]). +-export([init/1, handle_call/3, terminate/2, handle_cast/2, handle_info/2, + code_change/3]). +-export([report/4]). +-export([report_multi/4]). + +-record(state, { + timer :: none|reference(), + nodename :: string(), + statusreports :: dict:dict(), + lastsent :: integer() + }). + +start_link() -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +init([]) -> + process_flag(trap_exit, true), + ReportInterval = application:get_env(plop, status_report_interval, 1000), + lager:info("~p: starting", [?MODULE]), + HeartbeatInterval = application:get_env(plop, heartbeat_interval, 1000), + erlang:start_timer(HeartbeatInterval, self(), heartbeat), + {ok, #state{timer = none, + nodename = http_auth:own_name(), + statusreports = dict:new(), + lastsent = plop_compat:monotonic_time(millisecond) - ReportInterval}}. + +store_status(State, Service, Target, Variable, Status) -> + Statusreports = dict:store({Service, Target, Variable}, + {single, Status}, + State#state.statusreports), + State#state{statusreports = Statusreports}. + +dict_append_set(Key, Value, Dict) -> + AppendSet = sets:from_list([Value]), + dict:update(Key, fun ({multi, Old}) -> + {multi, sets:union(Old, AppendSet)} + end, + {multi, AppendSet}, Dict). + +store_multi_status(State, Service, Target, Variable, Status) -> + Statusreports = dict_append_set({Service, Target, Variable}, + Status, + State#state.statusreports), + State#state{statusreports = Statusreports}. + +store_set_status(State, Service, Target, Variable, Statuses) -> + Statusreports = dict:store({Service, Target, Variable}, + {multi, Statuses}, + State#state.statusreports), + State#state{statusreports = Statusreports}. + +heartbeat(State) -> + {ok, ConfigVersion} = plopconfig:get_env(version), + RunningApps = [atom_to_list(App) ++ " " ++ Vsn || {App, _Desc, Vsn} <- application:which_applications()], + + NewState1 = store_status(State, "heartbeat", "", "configversion", ConfigVersion), + NewState2 = store_set_status(NewState1, "heartbeat", "", "applications", sets:from_list(RunningApps)), + NewState2. + +handle_call(_, _From, State) -> + {noreply, State}. + +handle_cast({report, Service, Target, Variable, Status}, State) -> + NewState = store_status(State, Service, Target, Variable, Status), + {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_info({timeout, _Timer, force_send}, State) -> + lager:debug("statusreport timer timeout"), + {noreply, force_send(State)}; + +handle_info({timeout, _Timer, heartbeat}, State) -> + lager:debug("statusreport timer timeout"), + HeartbeatInterval = application:get_env(plop, heartbeat_interval, 1000), + erlang:start_timer(HeartbeatInterval, self(), heartbeat), + NewState = heartbeat(State), + {noreply, try_send(NewState)}. + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +cancel_timer(State) -> + case State#state.timer of + none -> + none; + _ -> + erlang:cancel_timer(State#state.timer) + end, + State#state{timer = none}. + +set_timer(State) -> + case State#state.timer of + none -> + ReportInterval = application:get_env(plop, status_report_interval, 1000), + Timer = plop_compat:start_timer(State#state.lastsent + ReportInterval, self(), force_send, [{abs, true}]), + State#state{timer = Timer}; + _ -> + State + end. + +terminate(Reason, State) -> + lager:info("~p terminating: ~p", [?MODULE, Reason]), + NewState = cancel_timer(State), + case Reason of + shutdown -> + force_send(NewState); + _ -> + none + end, + ok. + + + +group_by_service(Statusreports) -> + dict:to_list( + lists:foldl( + fun ({{Service, Target, Variable}, Status}, Acc) -> + dict:append(Service, {Target, Variable, Status}, Acc) + end, dict:new(), dict:to_list(Statusreports))). + +encode_one_status(Status) when is_number(Status) -> + Status; +encode_one_status(Status) when is_list(Status) -> + list_to_binary(Status); +encode_one_status(Status) when is_binary(Status) -> + Status; +encode_one_status(null) -> + null. + +encode_status({single, Status}) -> + encode_one_status(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, []), + 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 + ])), + case plop_httputil:request(DebugTag, URL, Headers, RequestBody) of + {error, Err} -> + lager:debug("request error ~p ~p", [DebugTag, Err]); + {failure, {none, StatusCode, none}, _RespHeaders, _Body} -> + lager:debug("request failure ~p ~p", [DebugTag, StatusCode]); + {success, {_, StatusCode, _}, _, Body} when StatusCode == 200 -> + case (catch mochijson2:decode(Body)) of + {error, Err} -> + lager:debug("error returned ~p ~p", [DebugTag, Err]); + {struct, _PropList} -> + none + end + end. + +force_send(State) -> + lists:foreach(fun ({Service, Statusreports}) -> + send(Service, Statusreports, State#state.nodename) + end, group_by_service(State#state.statusreports)), + NewState = cancel_timer(State), + NewState#state{statusreports = dict:new(), lastsent = plop_compat:monotonic_time(millisecond)}. + +try_send(State) -> + ReportInterval = application:get_env(plop, status_report_interval, 1000), + NextSend = State#state.lastsent + ReportInterval, + Now = plop_compat:monotonic_time(millisecond), + if + NextSend > Now -> + lager:debug("status report sent ~p ms ago, setting timer", [NextSend - Now]), + set_timer(State); + true -> + lager:debug("status report send long enough ago"), + force_send(State) + end. + +report(Service, Target, Variable, Status) when is_number(Status); is_list(Status); is_binary(Status); Status == null -> + lager:debug("reporting status ~p ~p ~p ~p", [Service, Target, Variable, Status]), + gen_server:cast(?MODULE, {report, Service, Target, Variable, 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}). diff --git a/src/storage.erl b/src/storage.erl index 7498635..6114a57 100644 --- a/src/storage.erl +++ b/src/storage.erl @@ -9,6 +9,18 @@ -define(APPURL_PLOP_STORAGE, "plop/v1/storage"). +reportnewentries() -> + NodeName = http_auth:own_name(), + {LastIndexOrZero, LastHash} = storagedb:lastverifiednewentry(), + VerifiedEntries = case LastHash of + none -> + 0; + _ -> + LastIndexOrZero + 1 + end, + NewEntries = index:indexsize(newentries_db) - VerifiedEntries, + statusreport:report("storage", NodeName, "newentries", NewEntries). + request(post, ?APPURL_PLOP_STORAGE, "sendentry", Input) -> case (catch mochijson2:decode(Input)) of {error, E} -> @@ -20,6 +32,7 @@ request(post, ?APPURL_PLOP_STORAGE, "sendentry", Input) -> ok = db:add_entry_sync(TreeLeafHash, LogEntry), ok = storagedb:add(TreeLeafHash), {KeyName, Sig} = http_auth:sign_stored(plop:spt_data_from_entry(LogEntry)), + reportnewentries(), success({[{result, <<"ok">>}, {"sig", KeyName ++ ":" ++ base64:encode_to_string(Sig)} ]}) @@ -55,6 +68,7 @@ request(get, ?APPURL_PLOP_STORAGE, "fetchnewentries", _Input) -> Entries = lists:map(fun(LeafHash) -> base64:encode(LeafHash) end, NewHashes), + reportnewentries(), success({[{result, <<"ok">>}, {entries, Entries}]}); request(get, ?APPURL_PLOP_STORAGE, "getentry", Query) -> |