% 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. -module(couch_stats_aggregator). -include("couch_stats.hrl"). -behaviour(gen_server). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -export([start/0, stop/0, get/1, get/2, get_json/1, get_json/2, all/0, time_passed/0, clear_aggregates/1]). -record(state, { aggregates = [], descriptions = [], timer_refs = [] }). -define(COLLECTOR, couch_stats_collector). -define(QUEUE_MAX_LENGTH, 900). % maximimum number of seconds % PUBLIC API start() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). stop() -> gen_server:call(?MODULE, stop). get(Key) -> gen_server:call(?MODULE, {get, Key}). get(Key, Time) -> gen_server:call(?MODULE, {get, Key, Time}). get_json(Key) -> gen_server:call(?MODULE, {get_json, Key}). get_json(Key, Time) -> gen_server:call(?MODULE, {get_json, Key, Time}). time_passed() -> gen_server:call(?MODULE, time_passed, infinity). clear_aggregates(Time) -> gen_server:call(?MODULE, {clear_aggregates, Time}, infinity). all() -> gen_server:call(?MODULE, all). % GEN_SERVER init(_) -> ets:new(?MODULE, [named_table, set, protected]), TimerRefs = init_timers(), init_descriptions(), {ok, #state{timer_refs = TimerRefs}}. handle_call({get, Key}, _, State) -> Value = get_aggregate(Key, State), {reply, Value, State}; handle_call({get, Key, Time}, _, State) -> Value = get_aggregate(Key, State, Time), {reply, Value, State}; handle_call({get_json, Key}, _, State) -> Value = aggregate_to_json_term(get_aggregate(Key, State)), {reply, Value, State}; handle_call({get_json, Key, Time}, _, State) -> Value = aggregate_to_json_term(get_aggregate(Key, State, Time)), {reply, Value, State}; handle_call(time_passed, _, OldState) -> NewTimerRefs = update_timer(time_passed, OldState#state.timer_refs), % the foldls below could probably be refactored into a less code-duping form % update aggregates on incremental counters NextState = lists:foldl(fun(Counter, State) -> {Key, Value} = Counter, update_aggregates_loop(Key, Value, State, incremental) end, OldState, ?COLLECTOR:all(incremental)), % update aggregates on absolute value counters NewState = lists:foldl(fun(Counter, State) -> {Key, Value} = Counter, % clear the counter, we've got the important bits in State ?COLLECTOR:clear(Key), update_aggregates_loop(Key, Value, State, absolute) end, NextState, ?COLLECTOR:all(absolute)), {reply, ok, NewState#state{timer_refs = NewTimerRefs}}; handle_call({clear_aggregates, Time}, _, State) -> {reply, ok, do_clear_aggregates(Time, State)}; handle_call(all, _ , State) -> Results = do_get_all(State), {reply, Results, State}; handle_call(stop, _, State) -> {stop, normal, stopped, State}. % PRIVATE API % Stats = [{Key, TimesProplist}] % TimesProplist = [{Time, Aggrgates}] % Aggregates = #aggregates{} % % [ % {Key, [ % {TimeA, #aggregates{}}, % {TimeB, #aggregates{}}, % {TimeC, #aggregates{}}, % {TimeD, #aggregates{}} % ] % }, % % ] %% clear the aggregats record for a specific Time = 60 | 300 | 900 do_clear_aggregates(Time, #state{aggregates=Stats, timer_refs=Refs}) -> NewTimerRefs = update_timer(Time, Refs), NewStats = lists:map(fun({Key, TimesProplist}) -> {Key, case proplists:lookup(Time, TimesProplist) of % do have stats for this key, if we don't, return Stat unmodified none -> TimesProplist; % there are stats, let's unset the Time one {_Time, _Stat} -> [{Time, #aggregates{}} | proplists:delete(Time, TimesProplist)] end} end, Stats), #state{aggregates=NewStats, timer_refs=NewTimerRefs}. get_aggregate(Key, State) -> %% default Time is 0, which is when CouchDB started get_aggregate(Key, State, '0'). get_aggregate(Key, #state{aggregates=StatsList}, Time) -> Description = get_description(Key), Aggregates = case proplists:lookup(Key, StatsList) of % if we don't have any data here, return an empty record none -> #aggregates{description=Description}; {Key, Stats} -> case proplists:lookup(Time, Stats) of none -> #aggregates{description=Description}; % empty record again {Time, Stat} -> Stat#aggregates{description=Description} end end, Aggregates. get_description(Key) -> case ets:lookup(?MODULE, Key) of [] -> <<"No description yet.">>; [{_Key, Description}] -> Description end. %% updates all aggregates for Key update_aggregates_loop(Key, Values, State, CounterType) -> #state{aggregates=AllStats} = State, % if we don't have any aggregates yet, put a list of empty atoms in % so we can loop over them in update_aggregates(). % [{{httpd,requests}, % [{'0',{aggregates,1,1,1,0,0,1,1}}, % {'60',{aggregates,1,1,1,0,0,1,1}}, % {'300',{aggregates,1,1,1,0,0,1,1}}, % {'900',{aggregates,1,1,1,0,0,1,1}}]}] [{_Key, StatsList}] = case proplists:lookup(Key, AllStats) of none -> [{Key, [ {'0', empty}, {'60', empty}, {'300', empty}, {'900', empty} ]}]; AllStatsMatch -> [AllStatsMatch] end, % if we get called with a single value, wrap in in a list ValuesList = case is_list(Values) of false -> [Values]; _True -> Values end, % loop over all Time's NewStats = lists:map(fun({Time, Stats}) -> % loop over all values for Key lists:foldl(fun(Value, Stat) -> {Time, update_aggregates(Value, Stat, CounterType)} end, Stats, ValuesList) end, StatsList), % put the newly calculated aggregates into State and delete the previous % entry #state{ aggregates=[{Key, NewStats} | proplists:delete(Key, AllStats)] }. % does the actual updating of the aggregate record update_aggregates(Value, Stat, CounterType) -> case Stat of % the first time this is called, we don't have to calculate anything % we just populate the record with Value empty -> #aggregates{ min=Value, max=Value, mean=Value, variance=0, stddev=0, count=1, current=Value }; % this sure could look nicer -- any ideas? StatsRecord -> #aggregates{ min=Min, max=Max, mean=Mean, variance=Variance, count=Count, current=Current } = StatsRecord, % incremental counters need to keep track of the last update's value NewValue = case CounterType of incremental -> Value - Current; absolute -> Value end, % Knuth, The Art of Computer Programming, vol. 2, p. 232. NewCount = Count + 1, NewMean = Mean + (NewValue - Mean) / NewCount, % NewCount is never 0. NewVariance = Variance + (NewValue - Mean) * (NewValue - NewMean), #aggregates{ min=lists:min([NewValue, Min]), max=lists:max([NewValue, Max]), mean=NewMean, variance=NewVariance, stddev=math:sqrt(NewVariance / NewCount), count=NewCount, current=Value } end. aggregate_to_json_term(#aggregates{min=Min,max=Max,mean=Mean,stddev=Stddev,count=Count,current=Current,description=Description}) -> {[ {current, Current}, {count, Count}, {mean, Mean}, {min, Min}, {max, Max}, {stddev, Stddev}, {description, Description} ]}. get_stats(Key, State) -> aggregate_to_json_term(get_aggregate(Key, State)). % convert ets2list() list into JSON-erlang-terms. % Thanks to Paul Davis do_get_all(#state{aggregates=Stats}=State) -> case Stats of [] -> {[]}; _ -> [{LastMod, LastVals} | LastRestMods] = lists:foldl(fun({{Module, Key}, _Count}, AccIn) -> case AccIn of [] -> [{Module, [{Key, get_stats({Module, Key}, State)}]}]; [{Module, PrevVals} | RestMods] -> [{Module, [{Key, get_stats({Module, Key}, State)} | PrevVals]} | RestMods]; [{OtherMod, ModVals} | RestMods] -> [{Module, [{Key, get_stats({Module, Key}, State)}]}, {OtherMod, {lists:reverse(ModVals)}} | RestMods] end end, [], lists:sort(Stats)), {[{LastMod, {lists:sort(LastVals)}} | LastRestMods]} end. init_descriptions() -> % ets is probably overkill here, but I didn't manage to keep the % descriptions in the gen_server state. Which means there is probably % a bug in one of the handle_call() functions most likely the one that % handles the time_passed message. But don't tell anyone, the math is % correct :) -- Jan % Style guide for descriptions: Start with a lowercase letter & do not add % a trailing full-stop / period. % please keep this in alphabetical order ets:insert(?MODULE, {{couchdb, database_writes}, <<"number of times a database was changed">>}), ets:insert(?MODULE, {{couchdb, database_reads}, <<"number of times a document was read from a database">>}), ets:insert(?MODULE, {{couchdb, open_databases}, <<"number of open databases">>}), ets:insert(?MODULE, {{couchdb, open_os_files}, <<"number of file descriptors CouchDB has open">>}), ets:insert(?MODULE, {{couchdb, request_time}, <<"length of a request inside CouchDB without MochiWeb">>}), ets:insert(?MODULE, {{httpd, bulk_requests}, <<"number of bulk requests">>}), ets:insert(?MODULE, {{httpd, requests}, <<"number of HTTP requests">>}), ets:insert(?MODULE, {{httpd, temporary_view_reads}, <<"number of temporary view reads">>}), ets:insert(?MODULE, {{httpd, view_reads}, <<"number of view reads">>}), ets:insert(?MODULE, {{httpd, clients_requesting_changes}, <<"Number of clients currently requesting continuous _changes">>}), ets:insert(?MODULE, {{httpd_request_methods, 'COPY'}, <<"number of HTTP COPY requests">>}), ets:insert(?MODULE, {{httpd_request_methods, 'DELETE'}, <<"number of HTTP DELETE requests">>}), ets:insert(?MODULE, {{httpd_request_methods, 'GET'}, <<"number of HTTP GET requests">>}), ets:insert(?MODULE, {{httpd_request_methods, 'HEAD'}, <<"number of HTTP HEAD requests">>}), ets:insert(?MODULE, {{httpd_request_methods, 'MOVE'}, <<"number of HTTP MOVE requests">>}), ets:insert(?MODULE, {{httpd_request_methods, 'POST'}, <<"number of HTTP POST requests">>}), ets:insert(?MODULE, {{httpd_request_methods, 'PUT'}, <<"number of HTTP PUT requests">>}), ets:insert(?MODULE, {{httpd_status_codes, '200'}, <<"number of HTTP 200 OK responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '201'}, <<"number of HTTP 201 Created responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '202'}, <<"number of HTTP 202 Accepted responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '301'}, <<"number of HTTP 301 Moved Permanently responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '304'}, <<"number of HTTP 304 Not Modified responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '400'}, <<"number of HTTP 400 Bad Request responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '401'}, <<"number of HTTP 401 Unauthorized responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '403'}, <<"number of HTTP 403 Forbidden responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '404'}, <<"number of HTTP 404 Not Found responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '405'}, <<"number of HTTP 405 Method Not Allowed responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '409'}, <<"number of HTTP 409 Conflict responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '412'}, <<"number of HTTP 412 Precondition Failed responses">>}), ets:insert(?MODULE, {{httpd_status_codes, '500'}, <<"number of HTTP 500 Internal Server Error responses">>}). % please keep this in alphabetical order % Timer init_timers() -> % OTP docs on timer: http://erlang.org/doc/man/timer.html % start() -> ok % Starts the timer server. Normally, the server does not need to be % started explicitly. It is started dynamically if it is needed. This is % useful during development, but in a target system the server should be % started explicitly. Use configuration parameters for kernel for this. % % TODO: Add timer_start to kernel start options. % start timers every second, minute, five minutes and fifteen minutes % in the rare event of a timer death, couch_stats_aggregator will die, % too and restarted by the supervision tree, all stats (for the last % fifteen minutes) are gone. {ok, A} = timer:apply_after(1000, ?MODULE, time_passed, []), {ok, B} = timer:apply_after(60000, ?MODULE, clear_aggregates, ['60']), {ok, C} = timer:apply_after(300000, ?MODULE, clear_aggregates, ['300']), {ok, D} = timer:apply_after(900000, ?MODULE, clear_aggregates, ['900']), [{time_passed,A}, {'60',B}, {'300',C}, {'900',D}]. update_timer(Type, Refs) -> timer:cancel(proplists:get_value(Type, Refs)), {ok,NewRef} = if Type == time_passed -> timer:apply_after(1000, ?MODULE, time_passed, []); true -> IntType = list_to_integer(atom_to_list(Type)), timer:apply_after(1000*IntType, ?MODULE, clear_aggregates, [Type]) end, [{Type,NewRef} | proplists:delete(Type, Refs)]. % Unused gen_server behaviour API functions that we need to declare. %% @doc Unused handle_cast(foo, State) -> {noreply, State}. handle_info(_Info, State) -> {noreply, State}. %% @doc Unused terminate(_Reason, _State) -> ok. %% @doc Unused code_change(_OldVersion, State, _Extra) -> {ok, State}. %% Tests -ifdef(TEST). % Internal API unit tests go here -endif.