%%% ---------------------------------------------------------------------------- %%% @author Jim Rosenblum %%% @copyright (C) 2015, Jim Rosenblum %%% @doc This gen_server provides update_statistcs/2 that is called by {@link %%% jc_store. jc_store} whenever a json_query is used. Frequency of use is %%% tracked, and if the {Map, Json Query} is used sufficiently often, an index %%% is created. %%% %%% @end %%% Created : 27 Aug 2015 by Jim Rosenblum %%% ---------------------------------------------------------------------------- -module(jc_analyzer). -behaviour(gen_server). %% Module API -export([dump/0, start_link/0, update_statistic/2]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). % Record and type definitions. -include("../include/records.hrl"). % Use QLC for some querying. -include_lib("stdlib/include/qlc.hrl"). -define(SERVER, ?MODULE). -define(INTERVAL, (1000 * 60 * 60)). % 1 hour interval to do clean-up job. % Start indexing if occurances of json query exceeds limit within window_sec. -record(jc_analyzer_state, {limit :: non_neg_integer(), window_sec :: non_neg_integer()}). %%% ============================================================================ %%% MODULE API %%% ============================================================================ %% ----------------------------------------------------------------------------- %% Create or update the statistics entry which tracks ths use of the json query %% on the given map. %% -spec update_statistic(map(), tuple()) -> ok. update_statistic(Map, Path) -> gen_server:cast({global, ?MODULE}, {update_stat, Map, Path}). %% ----------------------------------------------------------------------------- %% Return a list of #auto_index rows being tracked. %%. -spec dump() -> [#auto_index{}] | {error, timeout}. dump() -> gen_server:call({global, ?MODULE}, dump, 1000). %% ----------------------------------------------------------------------------- %% @doc Start the server and link the caller to it. %% -spec start_link() -> {ok, pid()} | ignore | {error, any()}. start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). %%% ============================================================================ %%% gen_server callbacks %%% ============================================================================ %% ----------------------------------------------------------------------------- %% @private Initialize the server by getting its default configuration and %% trying to become the master, and, if successful, schedule the first clean %% job. %% -spec init([]) -> {ok, #jc_analyzer_state{}}. init([]) -> {Cnt, Window} = application:get_env(jc, analyze_freq, {5, 120}), lager:info("~p: up with threshold of ~p times in ~p seconds.", [?MODULE, Cnt, Window]), grab_name(), {ok, #jc_analyzer_state{limit = Cnt, window_sec = Window}}. %% ----------------------------------------------------------------------------- %% @private Hande call messages. Dump the auto_index table. %% -spec handle_call(term(), {pid(), _}, #jc_analyzer_state{}) -> {reply, ok, #jc_analyzer_state{}}. handle_call(dump, _From, State) -> Reply = mnesia:dirty_match_object(#auto_index{_ = '_'}), {reply, Reply, State}; handle_call(Request, _From, State) -> lager:warning("~p: unexpected call request: ~p.",[?SERVER, Request]), {reply, ok, State}. %% ----------------------------------------------------------------------------- %% @private Handle cast messages: Update statistics about a Map and Path. %% -spec handle_cast(any(), #jc_analyzer_state{}) -> {noreply, #jc_analyzer_state{}}. handle_cast({update_stat, Map, Path}, State) -> update_stat(Map, Path, State), {noreply, State}; handle_cast(Msg, State) -> lager:warning("~p: unexpected cast message: ~p.",[?SERVER, Msg]), {noreply, State}. %% ----------------------------------------------------------------------------- %% @private Handle info messages: Master down, try to become master; %% Clean: clean the table of orphans rows - a sinlge use of a query search will %% be deleted via periodic process. %% -spec handle_info(any(), #jc_analyzer_state{}) -> {noreply, #jc_analyzer_state{}}. handle_info({'DOWN', _Mon, _Type, Obj, Info}, State) -> lager:debug("~p: master down at ~p with ~p.", [?SERVER, Obj, Info]), grab_name(), {noreply, State}; handle_info(clean, State) -> clean(State#jc_analyzer_state.window_sec), erlang:send_after(?INTERVAL, jc_analyzer, clean), {noreply, State}; handle_info(Info, State) -> lager:warning("~p: unexpected info message: ~p.",[?SERVER, Info]), {noreply, State}. %% ----------------------------------------------------------------------------- %% @private Terminate server. %% -spec terminate(any(), #jc_analyzer_state{}) -> any(). terminate(_Reason, _State) -> ok. %% ----------------------------------------------------------------------------- %% @private Convert process state when code is changed. %% -spec code_change(term(), #jc_analyzer_state{}, any()) -> {ok, #jc_analyzer_state{}}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%% ============================================================================ %%% internal functions %%% ============================================================================ %% ----------------------------------------------------------------------------- %% If the global registry doesn't have a PID registered for jc_analyzer, then %% claim it and schedule frst clean; otherwise, monitor the master. grab_name() -> case global:register_name(?MODULE, self()) of yes -> lager:info("~p: master is: ~p.", [?MODULE, self()]), erlang:send_after(?INTERVAL, jc_analyzer, clean); no -> Master = global:whereis_name(?MODULE), lager:info("~p: master is: ~p.", [?MODULE, Master]), _ = monitor(process, Master), ok end. %% ----------------------------------------------------------------------------- %% Create or update the row which is gathering statistics on the use of JSON %% querries. If one is used and there isn't an index for it, then a row should %% be created; otherwise updated. If it has been used sufficiently ask for an %% index. %% -spec update_stat(map(), tuple(), #jc_analyzer_state{}) -> ok. update_stat(Map, Path, #jc_analyzer_state{limit=Limit, window_sec=Window}) -> NowSecs = calendar:datetime_to_gregorian_seconds(calendar:universal_time()), case mnesia:dirty_read(auto_index, {Map, Path}) of [] -> % no records exists, create it. lager:debug("~p: initiating tracking of ~p.", [?SERVER, {Map, Path}]), mnesia:dirty_write(#auto_index{map_path = {Map, Path}, count = 1, first = NowSecs, last = NowSecs, times = [NowSecs], indexed = false, success = undefined, error_message = none}); % we have met the threshold and aren't indexing. [#auto_index{count=C, first=F, indexed=Indexed}=R] when (C + 1) >= Limit, Indexed == false -> case (NowSecs - F) =< Window of false -> % Occurances took longer than window to accumulate slide_window(R, NowSecs, Map, Path, Window); true -> initiate_index(R, NowSecs, Map, Path) end; % Already being indexed or haven't met the theshold in the window. [R] -> slide_window(R, NowSecs, Map, Path, Window) end. % Ask jc_store to create an index, record the response and mark the row indexed. initiate_index(#auto_index{count = C, times = T} = R, NowSecs, Map, Path) -> {Success, Msg} = case jc_store:create_index(Map, Path) of ok -> lager:debug("~p: initiating index on ~p.", [?SERVER, {Map, Path}]), {true, none}; {error, E} -> lager:warning("~p: not able to start index: ~p, ~p.", [?SERVER, {Map, Path}, E]), {false, E} end, mnesia:dirty_write(R#auto_index{count = C + 1, last = NowSecs, times = lists:sort([NowSecs|T]), indexed = true, success = Success, error_message = Msg}). % Remove all times, T, such that now - T > WindowSize. What's left is the new % count. % slide_window(#auto_index{times = T} = R, NowSecs, Map, Path, Window) -> lager:debug("~p: increment count and slidinf window on ~p.", [?SERVER, {Map, Path}]), NewTimes = cull(lists:sort([NowSecs|T]), NowSecs, Window), case NewTimes of [] -> mnesia:dirty_delete_object(R); NewTimes -> mnesia:dirty_write(R#auto_index{first = hd(NewTimes), last = NowSecs, times = NewTimes, count = length(NewTimes)}) end. % Remove all times, T, until the gap between T and now is less than the % window. % cull([], _NowSecs, _Window) -> []; cull([T|Ts], NowSecs, Window) when (NowSecs - T) >= Window -> cull(Ts, NowSecs, Window); cull(Ts, _NowSecs, _Window) -> Ts. % Delete anything from the table, not being indexed, whose (now - last) is > % then the time allowed (the Window). %% clean(Window) -> lager:debug("~p: cleaning auto_index table.", [?MODULE]), NowSecs = calendar:datetime_to_gregorian_seconds(calendar:universal_time()), QH = qlc:q([R || R <- mnesia:table(auto_index), R#auto_index.indexed == false, (NowSecs - R#auto_index.last) > Window]), Records = [mnesia:dirty_delete_object(R) || R <- mnesia:async_dirty(fun() -> qlc:e(QH) end)], lager:debug("~p: deleted ~p records from auto_index table.", [?MODULE, length(Records)]).