%% ------------------------------------------------------------------- %% %% Copyright (c) 2014 Basho Technologies, Inc. All Rights Reserved. %% %% This Source Code Form is subject to the terms of the Mozilla Public %% License, v. 2.0. If a copy of the MPL was not distributed with this %% file, You can obtain one at http://mozilla.org/MPL/2.0/. %% %% ------------------------------------------------------------------- %% @doc %% %% A custom reporter plugin, executing in its own process, can receive %% updated metric values by having its module referenced in an %% `exometer_report:subscribe()' call. %% %% The reporter, once it is setup as a subscription destination, will %% receive periodic calls with updated metrics and data points to be %% reported. %% %% Each custom plugin implements the exometer_report behavior. %% %% The life cycle of a a custom reporter consists of the following steps. %% %% + Reporter creation
`exometer_init/1' is invoked by exometer when %% the reporter is configured in the reporter application %% environment. See {@section Configuring reporter plugins} for %% details. %% %% + Setup subscription
When `exometer_report:subscribe()' is called, targeting the %% custom report plugin, the gen_server's `exometer_subscribe()' function %% will be invoked to notify the plugin of the new metrics subscription. %% %% + Report Metrics
Updated metrics are sent by exometer to the %% `exometer_report/4'. All reported metrics will have been notified %% to the recipient through a previous `exometer_report()' function. %% %% + Tear down subscription
When `exometer_report:unsubscribe()' is called, addressing the %% custom report plugin, the recipient's `exometer_unsubscribe()' function %% will be invoked to notify the plugin of the deleted subscription. %% %% %% The following chapters details each of the callbacks to be implemented %% in the exometer_report behavior. %% %% === exometer_init/1 === %% %% The `exometer_init()' function is invoked as follows: %% %% 
%%      exometer_init(Options)
%% %% The custom reporter plugin should create the necessary state for the %% new plugin and return a state to be used in future plugin calls. %% %% + `Options'
Provides the prop list with attributes from the application environment %% for the cusom recipient. See {@section Configuring reporter plugins} for %% %% The `exomoeter_init()' function should return `{ok, State}' where %% State is a tuple that will be provided as a reference argument to %% future calls made into the plugin. Any other return formats will %% cancel the creation of the custom reporting plugin. %% %% %% === exometer_subscribe/4 === %% %% The `exometer_subscribe()' function is invoked as follows: %% %% 
%%      exometer_subscribe(Metric, DataPoint, Interval State)
%% %% The custom plugin can use this notification to modify and return its %% state in order to prepare for future calls to `exometer_report()' with %% the given meteric and data point. %% %% + `Metric'
Specifies the metric that is now subscribed to by the plugin %% as a list of atoms. %% %% + `DataPoint'
Specifies the data point within the subscribed-to metric %% as an atom, or a list of atoms. %% %% + `Interval'
Specifies the interval, in milliseconds, that the %% subscribed-to value will be reported at, or an atom, referring to a named %% interval configured in the reporter. %% %% + `State'
Contains the state returned by the last called plugin function. %% %% The `exomoeter_subscribe()' function should return `{ok, State}' where %% State is a tuple that will be provided as a reference argument to %% future calls made into the plugin. Any other return formats will %% generate an error log message by exometer. %% %% %% === exometer_report/4 === %% %% The `exometer_report()' function is invoked as follows: %% %% 
%%      exometer_report(Metric, DataPoint, State)
%% %% The custom plugin will receive this call when a periodic subscription %% triggers and wants to report its current value through the plugin. %% The plugin should export the value to the external system it interfaces and %% return its possibly modified state. %% %% + `Metric'
Specifies the metric that is to be reported. %% %% + `DataPoint'
Specifies the data point or data points within the metric %% to be reported. %% %% + `State'
Contains the state returned by the last called plugin function. %% %% The `exomoeter_report()' function should return `{ok, State}' where %% State is a tuple that will be provided as a reference argument to %% future calls made into the plugin. Any other return formats will %% generate an error log message by exometer. %% %% %% === exometer_unsubscribe/3 === %% %% The `exometer_unsubscribe()' function is invoked as follows: %% %% 
%%      exometer_unsubscribe(Metric, DataPoint, State)
%% %% The custom plugin can use this notification to modify and return its %% state in order to free resources used to maintain the now de-activated %% subscription. When this call returns, the given metric / data point %% will not be present in future calls to `exometer_report()'. %% %% + `Metric'
Specifies the metric that is now subscribed to by the plugin %% as a list of atoms. %% %% + `DataPoint'
Specifies the data point or data points within the %% subscribed-to metric as an atom or a list of atoms. %% %% + `State'
Contains the state returned by the last called plugin function. %% %% The `exometer_unsubscribe()' function should return `{ok, State}' where %% State is a tuple that will be provided as a reference argument to %% future calls made into the plugin. Any other return formats will %% generate an error log message by exometer. %% %% @end -module(exometer_report). -behaviour(gen_server). %% API -export( [ start_link/0, subscribe/4, subscribe/5, subscribe/6, unsubscribe/3, unsubscribe/4, unsubscribe_all/2, list_metrics/0, list_metrics/1, list_reporters/0, list_subscriptions/1, add_reporter/2, set_interval/3, delete_interval/2, restart_intervals/1, get_intervals/1, remove_reporter/1, remove_reporter/2, terminate_reporter/1, enable_reporter/1, disable_reporter/1, call_reporter/2, cast_reporter/2, setopts/3, new_entry/1 ]). %% Start phase function -export([start_reporters/0]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -export([disable_me/2]). -export_type([metric/0, datapoint/0, interval/0, extra/0]). -include("exometer.hrl"). -include("log.hrl"). -define(SERVER, ?MODULE). -type error() :: {error, any()}. -type metric() :: exometer:name() | {find, exometer:name()} | {select, ets:match_spec()}. -type datapoint() :: exometer:datapoint(). -type datapoints() :: datapoint() | [datapoint()]. -type options() :: [{atom(), any()}]. -type mod_state() :: any(). -type value() :: any(). -type interval() :: pos_integer() | atom(). -type time_ms() :: pos_integer(). -type delay() :: time_ms(). -type named_interval() :: {atom(), time_ms()} | {atom(), time_ms(), delay()}. -type callback_result() :: {ok, mod_state()} | any(). -type extra() :: any(). -type retry() :: boolean(). -type reporter_name() :: atom(). %% Restart specification -type maxR() :: pos_integer(). -type maxT() :: pos_integer(). -type action() :: {atom(), atom()}. -type restart() :: [{maxR(), maxT()} | action()]. %% Callback for function, not cast-based, reports that %% are invoked in-process. -callback exometer_init(options()) -> callback_result(). -callback exometer_report(metric(), datapoint(), value(), extra(), mod_state()) -> callback_result(). -callback exometer_subscribe(metric(), datapoint(), interval(), extra(), mod_state()) -> callback_result(). -callback exometer_unsubscribe(metric(), datapoint(), extra(), mod_state()) -> callback_result(). -callback exometer_info(any(),mod_state()) -> callback_result(). -callback exometer_call(any(), pid(), mod_state()) -> {reply, any(), mod_state()} | {noreply, mod_state()} | any(). -callback exometer_cast(any(), mod_state()) -> {noreply, mod_state()} | any(). -callback exometer_terminate(any(), mod_state()) -> any(). -callback exometer_setopts(exometer:entry(), options(), exometer:status(), mod_state()) -> callback_result(). -callback exometer_newentry(exometer:entry(), mod_state()) -> callback_result(). -record(key, { reporter :: module() | '_', metric :: metric() | '_', datapoint :: datapoints() | '_', retry_failed_metrics :: boolean() | '_', extra :: extra() | '_' }). -record(subscriber, { key :: #key{} | '_', interval :: interval() | '_', t_ref :: reference() | undefined | '_' }). -record(restart, { spec = default_restart() :: restart(), history = [] :: [pos_integer()], save_n = 10 :: pos_integer()} ). -record(interval, { name :: atom(), time = 0 :: non_neg_integer(), delay = 0 :: non_neg_integer(), t_ref :: reference() | undefined }). -record(reporter, { name :: atom() | '_', pid :: pid() | atom(), % in select() mref :: reference() | '_', module :: module() | '_', opts = [] :: [{atom(), any()}] | '_', intervals = [] :: [#interval{}] | '_', restart = #restart{} :: #restart{} | '_', status = enabled :: enabled | disabled | '_' }). -record(st, { subscribers = [] :: [#subscriber{}], reporters = [] :: [#reporter{}] }). %%%=================================================================== %%% API %%%=================================================================== %%-------------------------------------------------------------------- %% @doc Starts the server %%-------------------------------------------------------------------- -spec start_link() -> {ok, pid()} | ignore | {error, any()}. start_link() -> %% Launch the main server. gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). -spec subscribe(reporter_name(), metric(), datapoints(), interval()) -> ok | not_found | unknown_reporter | error. %% @equiv subscribe(Reporter, Metric, DataPoint, Interval, [], false) subscribe(Reporter, Metric, DataPoint, Interval) -> subscribe(Reporter, Metric, DataPoint, Interval, []). -spec subscribe(reporter_name(), metric(), datapoints(), interval(), extra()) -> ok | not_found | unknown_reporter | error. %% @equiv subscribe(Reporter, Metric, DataPoint, Interval, Extra, false) subscribe(Reporter, Metric, DataPoint, Interval, Extra) -> call({subscribe, #key{reporter = Reporter, metric = Metric, datapoint = DataPoint, retry_failed_metrics = false, extra = Extra}, Interval}). -spec subscribe(reporter_name(), metric(), datapoints(), interval(), extra(), retry()) -> ok | not_found | unknown_reporter | error. %% @doc Add a subscription to an existing reporter. %% %% The reporter must first be started using {@link add_reporter/2}, or through %% a static configuration. `Metric' is the name of an exometer entry. `DataPoint' %% is either a single data point (an atom) or a list of data points (a list). %% %% `Interval' is the sampling/reporting interval in milliseconds, or an atom, %% referring to a named interval configured in the reporter. The named %% interval need not be defined yet in the reporter (the subscription will %% not trigger until it is defined.) %% %% `Extra' can be anything that the chosen reporter understands (default: `[]'). %% If the reporter uses {@link exometer_util:report_type/3}, `Extra' should be %% a proplist, and the option `{report_type, T}' can control which type (e.g. %% for collectd or statsd) that the value corresponds to. %% %% `Retry': boolean(). If true, retry the subscription at the next interval, %% even if the metric cannot be read. %% @end subscribe(Reporter, Metric, DataPoint, Interval, Extra, Retry) when is_boolean(Retry) -> call({subscribe, #key{reporter = Reporter, metric = Metric, datapoint = DataPoint, retry_failed_metrics = Retry, extra = Extra}, Interval}). -spec unsubscribe(module(), metric(), datapoint()) -> ok | not_found. %% @equiv unsubscribe(Reporter, Metric, DataPoint, []) unsubscribe(Reporter, Metric, DataPoint) -> unsubscribe(Reporter, Metric, DataPoint, []). -spec unsubscribe(module(), metric(), datapoint() | [datapoint()], extra()) -> ok | not_found. %% @doc Removes a subscription. %% %% Note that the subscription is identified by the combination %% `{Reporter, Metric, DataPoint, Extra}'. The exact information can be %% extracted using {@link list_subscriptions/1}. %% @end unsubscribe(Reporter, Metric, DataPoint, Extra) -> call({unsubscribe, #key{reporter = Reporter, metric = Metric, datapoint = DataPoint, extra = Extra}}). -spec unsubscribe_all(module(), metric()) -> ok. %% @doc Removes all subscriptions related to Metric in Reporter. %% @end unsubscribe_all(Reporter, Metric) -> call({unsubscribe_all, Reporter, Metric}). -spec list_metrics() -> {ok, [{ exometer:name(), [datapoint()], [{reporter_name(), datapoint()}], exometer:status() }]} | {error, any()}. %% @equiv list_metrics([]) list_metrics() -> list_metrics([]). -spec list_metrics(Path :: metric()) -> {ok, [{ exometer:name(), [datapoint()], [{reporter_name(), datapoint()}], exometer:status() }]} | {error, any()}. %% @doc List all metrics matching `Path', together with subscription status. %% %% This function performs a metrics search using `exometer:find_entries/1', %% then matches the result against known subscriptions. It reports, for each %% metric, the available data points, as well as which reporters subscribe to %% which data points. %% @end list_metrics(Path) -> call({list_metrics, Path}). -spec list_reporters() -> [{reporter_name(), pid()}]. %% @doc List the name and pid of each known reporter. list_reporters() -> call(list_reporters). -spec list_subscriptions(reporter_name()) -> [{metric(), datapoint(), interval(), extra()}]. %% @doc List all subscriptions for `Reporter'. list_subscriptions(Reporter) -> call({list_subscriptions, Reporter}). -spec add_reporter(reporter_name(), options()) -> ok | {error, any()}. %% @doc Add a reporter. %% %% The reporter can be configured using the following options. Note that all %% options are also passed to the reporter callback module, which may support %% additional options. %% %% `{module, atom()}' - The name of the reporter callback module. If no module %% is given, the module name defaults to the given reporter name. %% %% `{status, enabled | disabled}' - The operational status of the reporter %% if enabled, the reporter will report values to its target. If disabled, the %% reporter process will be terminated and subscription timers canceled, but %% the subscriptions will remain, and it will also be possible to add new %% subscriptions to the reporter. %% %% `{intervals, [named_interval()]}' %% named_interval() :: {Name::atom(), Interval::pos_integer()} %% | {Name::atom(), Interval::time_ms(), delay()::time_ms()} %% Define named intervals. The name can be used by subscribers, so that all %% subsriptions for a given named interval will be reported when the interval %% triggers. An optional delay (in ms) can be given: this will cause the first %% interval to start in `Delay' milliseconds. When all intervals are named %% at the same time, the delay parameter can be used to achieve staggered %% reporting. %% @end add_reporter(Reporter, Options) -> call({add_reporter, Reporter, Options}). -spec remove_reporter(reporter_name()) -> ok | {error, any()}. %% @doc Remove reporter and all its subscriptions. remove_reporter(Reporter) -> call({remove_reporter, Reporter}). -spec set_interval(reporter_name(), atom(), time_ms() | {time_ms(), delay()}) -> ok |error(). %% @doc Specify a named interval. %% %% See {@link add_reporter/2} for a description of named intervals. %% The named interval is here specified as either `Time' (milliseconds) or %% `{Time, Delay}', where a delay in milliseconds is provided. %% %% If the named interval exists, it will be replaced with the new definition. %% Otherwise, it will be added. Use {@link restart_intervals/1} if you want %% all intervals to be restarted/resynched with corresponding relative delays. %% @end set_interval(Reporter, Name, Time) when is_atom(Name), is_integer(Time), Time >= 0 -> call({set_interval, Reporter, Name, Time}); set_interval(Reporter, Name, {Time, Delay}) when is_atom(Name), is_integer(Time), Time >= 0, is_integer(Delay), Delay >= 0 -> call({set_interval, Reporter, Name, {Time, Delay}}). -spec delete_interval(reporter_name(), atom()) -> ok | error(). %% @doc Delete a named interval. %% delete_interval(Reporter, Name) -> call({delete_interval, Reporter, Name}). -spec restart_intervals(reporter_name()) -> ok. %% @doc Restart all named intervals, respecting specified delays. %% %% This function can be used if named intervals are added incrementally, and %% it is important that all intervals trigger separated by the given delays. %% @end restart_intervals(Reporter) -> call({restart_intervals, Reporter}). -spec get_intervals(reporter_name()) -> [{atom(), [{time, pos_integer()} | {delay, pos_integer()} | {timer_ref, reference()}]}]. %% @doc List the named intervals for `Reporter'. get_intervals(Reporter) -> call({get_intervals, Reporter}). -spec enable_reporter(reporter_name()) -> ok | {error, any()}. %% @doc Enable `Reporter'. %% %% The reporter will be 'restarted' in the same way as if it had crashed %% and was restarted by the supervision logic, but without counting it as %% a restart. %% %% If the reporter was already enabled, nothing is changed. %% @end enable_reporter(Reporter) -> call({change_reporter_status, Reporter, enabled}). -spec disable_reporter(reporter_name()) -> ok | {error, any()}. %% @doc Disable `Reporter'. %% %% The reporter will be terminated, and all subscription timers will be %% canceled, but the subscriptions themselves and reporter metadata are kept. %% @end disable_reporter(Reporter) -> call({change_reporter_status, Reporter, disabled}). -spec disable_me(module(), any()) -> no_return(). %% @doc Used by a reporter to disable itself. %% %% This function can be called from a reporter instance if it wants to be %% disabled, e.g. after exhausting a configured number of connection attempts. %% The arguments passed are the name of the reporter callback module and the %% module state, and are used to call the `Mod:terminate/2' function. %% @end disable_me(Mod, St) -> cast({disable, self()}), receive {exometer_terminate, shutdown} -> Mod:exometer_terminate(shutdown, St), exit(shutdown) end. -spec call_reporter(reporter_name(), any()) -> any() | {error, any()}. %% @doc Send a custom (synchronous) call to `Reporter'. %% %% This function is used to make a client-server call to a given reporter %% instance. Note that the reporter type must recognize the request. %% @end call_reporter(Reporter, Msg) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{pid = Pid}] when is_pid(Pid) -> exometer_proc:call(Pid, Msg); [#reporter{status = disabled}] -> {error, disabled}; [] -> {error, {no_such_reporter, Reporter}} end. -spec cast_reporter(reporter_name(), any()) -> ok | {error, any()}. %% @doc Send a custom (asynchronous) cast to `Reporter'. %% %% This function is used to make an asynchronous cast to a given reporter %% instance. Note that the reporter type must recognize the message. %% @end cast_reporter(Reporter, Msg) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{pid = Pid}] when is_pid(Pid) -> exometer_proc:cast(Pid, Msg); [#reporter{status = disabled}] -> {error, disabled}; [] -> {error, {no_such_reporter, Reporter}} end. -spec remove_reporter(reporter_name(), _Reason::any()) -> ok | {error, any()}. %% @doc Remove `Reporter' (non-blocking call). %% %% This function can be used to order removal of a reporter with a custom %% reason. Note that the function is asynchronous, making it suitable e.g. %% for calling from within the reporter itself. %% @end remove_reporter(Reporter, Reason) -> cast({remove_reporter, Reporter, Reason}). -spec setopts(exometer:entry(), options(), exometer:status()) -> ok. %% @doc Called by exometer when options of a metric entry are changed. %% %% Reporters subscribing to the metric get a chance to process the options %% change in the function `Mod:exometer_setopts(Metric,Options,Status,St)'. %% @end setopts(Metric, Options, Status) -> call({setopts, Metric, Options, Status}). -spec new_entry(exometer:entry()) -> ok. %% @doc Called by exometer whenever a new entry is created. %% %% This function is called whenever a new metric is created, giving each %% reporter the chance to enable a subscription for it. Note that each %% reporter is free to call the subscription management functions, as there %% is no risk of deadlock. The callback function triggered by this call is %% `Mod:exometer_newentry(Entry, St)'. %% @end new_entry(Entry) -> cast({new_entry, Entry}). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %%-------------------------------------------------------------------- %% @private %% @doc %% Initializes the server %% %% @spec init(Args) -> {ok, State} | %% {ok, State, Timeout} | %% ignore | %% {stop, Reason} %% @end %%-------------------------------------------------------------------- init([]) -> process_flag(trap_exit, true), {ok, #st{}}. start_reporters() -> call(start_reporters). do_start_reporters(S) -> Opts = get_report_env(), ?info("Starting reporters with ~p~n", [ Opts ]), %% Dig out the mod opts. %% { reporters, [ {reporter1, [{opt1, val}, ...]}, {reporter2, [...]}]} %% Traverse list of reporter and launch reporter gen servers as dynamic %% supervisor children. case lists:keyfind(reporters, 1, Opts) of {reporters, ReporterList} -> ReporterRecs = make_reporter_recs(ReporterList), assert_no_duplicates(ReporterRecs), lists:foreach( fun(#reporter{name = Reporter, status = Status, opts = ROpts, intervals = Ints0} = R) -> Restart = get_restart(ROpts), {Pid, MRef, Ints} = if Status =:= enabled -> {P1,R1} = spawn_reporter(Reporter, ROpts), I1 = start_interval_timers(R), {P1,R1,I1}; true -> {undefined, undefined, Ints0} end, ets:insert(?EXOMETER_REPORTERS, R#reporter{pid = Pid, mref = MRef, intervals = Ints, restart = Restart}) end, ReporterRecs); false -> [] end, %% Dig out configured 'static' subscribers case lists:keyfind(subscribers, 1, Opts) of {subscribers, Subscribers} -> lists:foreach(fun init_subscriber/1, Subscribers); false -> [] end, S#st{}. make_reporter_recs([{R, Opts}|T]) when is_atom(R), is_list(Opts) -> [#reporter{name = R, module = get_module(R, Opts), status = proplists:get_value(status, Opts, enabled), opts = Opts, intervals = get_interval_opts(Opts)}|make_reporter_recs(T)]; make_reporter_recs([]) -> []. get_module(R, Opts) -> proplists:get_value(module, Opts, R). -spec get_interval_opts([named_interval() | any()]) -> [#interval{}]. get_interval_opts(Opts) -> Is1 = [singelton_interval(I) || {interval, I} <- Opts], Is = proplists:get_value(intervals, Opts, []), lists:map( fun({Name, Time}) when is_atom(Name), is_integer(Time), Time >= 0 -> #interval{name = Name, time = Time}; ({Name, Time, Delay}) when is_atom(Name), is_integer(Time), Time >= 0, is_integer(Delay), Delay >= 0 -> #interval{name = Name, time = Time, delay = Delay}; (Other) -> error({invalid_interval, Other}) end, Is ++ Is1). singelton_interval({N,T}=I) when is_atom(N), is_integer(T) -> I; singelton_interval({N,T,D}=I) when is_atom(N), is_integer(T), is_integer(D) -> I. start_interval_timers(#reporter{name = R, intervals = Ints}) -> lists:map(fun(I) -> start_interval_timer(I, R) end, Ints). start_interval_timer(#interval{name = Name, delay = Delay, t_ref = Ref} = I, R) -> cancel_timer(Ref), case Delay of 0 -> do_start_interval_timer(I, R); D -> TRef = erlang:send_after(D, self(), {start_interval, R, Name}), I#interval{t_ref = TRef} end. do_start_interval_timer(#interval{name = Name, time = Time} = I, R) -> TRef = erlang:send_after(Time, self(), batch_timer_msg(R, Name)), I#interval{t_ref = TRef}. batch_timer_msg(R, Name) -> {report_batch, R, Name}. get_report_env() -> Opts0 = exometer_util:get_env(report, []), {Rs1, Opts1} = split_env(reporters, Opts0), {Ss2, Opts2} = split_env(subscribers, Opts1), get_reporters(Rs1) ++ get_subscribers(Ss2) ++ Opts2. split_env(Tag, Opts) -> case lists:keytake(Tag, 1, Opts) of {value, {_, L}, Rest} -> {L, Rest}; false -> {[], Opts} end. get_reporters(L0) -> Rs = exometer_util:get_env(reporters, []), Ext = setup:find_env_vars(exometer_reporters), merge_env(reporters, Rs ++ L0, Ext). get_subscribers(L0) -> Ss = exometer_util:get_env(subscribers, []), Ext = setup:find_env_vars(exometer_subscribers), merge_env(subscribers, Ss ++ L0, Ext). merge_env(_, [], []) -> []; merge_env(Tag, L, E) -> [{Tag, L} || L =/= []] ++ [{Tag, X} || {_, X} <- E]. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling call messages %% %% @spec handle_call(Request, From, State) -> %% {reply, Reply, State} | %% {reply, Reply, State, Timeout} | %% {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, Reply, State} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_call(start_reporters, _From, S) -> {reply, ok, do_start_reporters(S)}; handle_call({subscribe, #key{ reporter = Reporter, metric = Metric, datapoint = DataPoint, retry_failed_metrics = RetryFailedMetrics, extra = Extra} , Interval }, _From, #st{} = St) -> %% Verify that the given metric/data point actually exist. case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{status = Status}] -> case is_valid_metric(Metric, DataPoint) of true -> if Status =:= enabled -> Reporter ! {exometer_subscribe, Metric, DataPoint, Interval, Extra}; true -> ignore end, subscribe_(Reporter, Metric, DataPoint, Interval, RetryFailedMetrics, Extra, Status), {reply, ok, St}; %% Nope - Not found. false -> case RetryFailedMetrics of true -> subscribe_(Reporter, Metric, DataPoint, Interval, RetryFailedMetrics, Extra, Status), {reply, ok, St}; false -> {reply, not_found, St} end; error -> {reply, error, St} end; [] -> {reply, unknown_reporter, St} end; handle_call({unsubscribe, #key{reporter = Reporter, metric = Metric, datapoint = DataPoint, extra = Extra}}, _, #st{} = St) -> Res = unsubscribe_(Reporter, Metric, DataPoint, Extra), {reply, Res, St}; handle_call({unsubscribe_all, Reporter, Metric}, _, #st{}=St) -> Subs = ets:select(?EXOMETER_SUBS, [{#subscriber{key = #key{reporter = Reporter, metric = Metric, _ = '_'}, _ = '_'}, [], ['$_']}]), lists:foreach(fun unsubscribe_/1, Subs), {reply, ok, St}; handle_call({list_metrics, Path}, _, St) -> if is_list(Path) -> DP = lists:foldr(fun(Metric, Acc) -> retrieve_metric(Metric, Acc) end, [], exometer:find_entries(Path)), {reply, {ok, DP}, St}; true -> {reply, {error, badarg}, St} end; handle_call({list_subscriptions, Reporter}, _, #st{} = St) -> Subs1 = lists:foldl( fun (#subscriber{key=#key{reporter=Rep}}=Sub, Acc) when Reporter == Rep -> #subscriber{ key=#key{ metric=Metric, datapoint=Dp, extra=Extra}, interval=Interval} = Sub, [{Metric, Dp, Interval, Extra} | Acc]; (_, Acc) -> Acc end, [], ets:select(?EXOMETER_SUBS, [{'_',[],['$_']}])), {reply, Subs1, St}; handle_call(list_reporters, _, #st{} = St) -> Info = ets:select(?EXOMETER_REPORTERS, [{#reporter{name = '$1', pid = '$2', _ = '_'}, [], [{{'$1', '$2'}}]}]), {reply, Info, St}; handle_call({add_reporter, Reporter, Opts}, _, #st{} = St) -> case ets:member(?EXOMETER_REPORTERS, Reporter) of true -> {reply, {error, already_running}, St}; false -> try [R] = make_reporter_recs([{Reporter, Opts}]), {Pid, MRef} = spawn_reporter(Reporter, Opts), Ints = start_interval_timers(R), R1 = R#reporter{intervals = Ints, pid = Pid, mref = MRef}, ets:insert(?EXOMETER_REPORTERS, R1), {reply, ok, St} catch error:Reason -> {reply, {error, Reason}, St} end end; handle_call({remove_reporter, Reporter}, _, St) -> case do_remove_reporter(Reporter) of ok -> {reply, ok, St}; E -> {reply, E, St} end; handle_call({change_reporter_status, Reporter, Status}, _, St) -> case change_reporter_status(Reporter, Status) of ok -> {reply, ok, St}; E -> {reply, E, St} end; handle_call({set_interval, Reporter, Name, Int}, _, #st{}=St) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{intervals = Ints}] -> try I0 = case lists:keyfind(Name, #interval.name, Ints) of false -> #interval{name = Name}; Interval -> Interval end, I1 = case Int of {Time, Delay} when is_integer(Time), Time >= 0, is_integer(Delay), Delay >= 0 -> I0#interval{time = Time, delay = Delay}; Time when is_integer(Time), Time >= 0 -> I0#interval{time = Time} end, ets:update_element(?EXOMETER_REPORTERS, Reporter, [{#reporter.intervals, lists:keystore( Name, #interval.name, Ints, start_interval_timer(I1, Reporter))}]), {reply, ok, St} catch error:Reason -> {reply, {error, Reason}, St} end; [] -> {reply, {error, not_found}, St} end; handle_call({delete_interval, Reporter, Name}, _, #st{} = St) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{intervals = Ints}] -> case lists:keyfind(Name, #interval.name, Ints) of #interval{t_ref = TRef} -> cancel_timer(TRef), ets:update_element(?EXOMETER_REPORTERS, Reporter, [{#reporter.intervals, lists:keydelete( Name, #interval.name, Ints)}]), {reply, ok, St}; false -> {reply, {error, not_found}, St} end; [] -> {reply, {error, not_found}, St} end; handle_call({restart_intervals, Reporter}, _, #st{} = St) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{} = R] -> Ints = start_interval_timers(R), ets:update_element(?EXOMETER_REPORTERS, Reporter, [{#reporter.intervals, Ints}]), {reply, ok, St}; [] -> {reply, {error, not_found}, St} end; handle_call({get_intervals, Reporter}, _, #st{} = St) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{intervals = Ints}] -> Info = [{Name, [{time, T}, {delay, D}, {timer_ref, TR}]} || #interval{name = Name, time = T, delay = D, t_ref = TR} <- Ints], {reply, Info, St}; [] -> {reply, {error, not_found}, St} end; handle_call({setopts, Metric, Options, Status}, _, #st{}=St) -> [erlang:send(Pid, {exometer_setopts, Metric, Options, Status}) || Pid <- reporter_pids()], {reply, ok, St}; handle_call(_Request, _From, State) -> {reply, {error, unknown_call}, State}. %%-------------------------------------------------------------------- %% @private %% @doc Handling cast messages. %% %% @spec handle_cast(Msg, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_cast({new_entry, Entry}, #st{} = St) -> [try erlang:send(Pid, {exometer_newentry, Entry}) catch error:_ -> ok end || Pid <- reporter_pids()], maybe_enable_subscriptions(Entry), {noreply, St}; handle_cast({remove_reporter, Reporter, Reason}, St) -> Terminate = case Reason of user -> true; _ -> false end, do_remove_reporter(Reporter, Terminate), {noreply, St}; handle_cast({disable, Pid}, #st{} = St) -> case reporter_by_pid(Pid) of [#reporter{} = Reporter] -> do_change_reporter_status(Reporter, disabled); [] -> ok end, {noreply, St}; handle_cast(_Msg, State) -> {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc Handling all non call/cast messages. %% %% @spec handle_info(Info, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_info({start_interval, Reporter, Name}, #st{} = St) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{intervals = Ints, status = enabled}] -> case lists:keyfind(Name, #interval.name, Ints) of #interval{} = I -> I1 = do_start_interval_timer(I, Reporter), ets:update_element(?EXOMETER_REPORTERS, Reporter, [{#reporter.intervals, lists:keyreplace( Name, #interval.name, Ints, I1)}]); false -> ok end; _ -> ok end, {noreply, St}; handle_info({report_batch, Reporter, Name}, #st{} = St) -> %% Find all entries where reporter is Reporter and interval is Name, %% and report them. report_batch(Reporter, Name), {noreply, St}; handle_info({report, #key{reporter = Reporter} = Key, Interval}, #st{} = St) -> case ets:member(?EXOMETER_SUBS, Key) andalso get_reporter_status(Reporter) == enabled of true -> case do_report(Key, Interval) of true -> restart_subscr_timer(Key, Interval); false -> ok end, {noreply, St}; false -> %% Possibly an unsubscribe removed the subscriber ?error("No such subscriber (Key=~p)~n", [Key]), {noreply, St} end; handle_info({'DOWN', Ref, process, _Pid, Reason}, #st{} = S) -> case reporter_by_mref(Ref) of [#reporter{module = Module, restart = Restart} = R] -> case add_restart(Restart) of {remove, How} -> case How of {M, F} when is_atom(M), is_atom(F) -> try M:F(Module, Reason) catch _:_ -> ok end; _ -> ok end, S; {restart, Restart1} -> restart_reporter(R#reporter{restart = Restart1}) end; _ -> S end, {noreply, S}; handle_info(_Info, State) -> ?warning("exometer_report:info(??): ~p~n", [ _Info ]), {noreply, State}. restart_reporter(#reporter{name = Name, opts = Opts, restart = Restart}) -> {Pid, MRef} = spawn_reporter(Name, Opts), [resubscribe(S) || S <- ets:select(?EXOMETER_SUBS, [{#subscriber{key = #key{reporter = Name, _ = '_'}, _ = '_'}, [], ['$_']}])], ets:update_element(?EXOMETER_REPORTERS, Name, [{#reporter.pid, Pid}, {#reporter.mref, MRef}, {#reporter.restart, Restart}, {#reporter.status, enabled}]), ok. %% If there are already subscriptions, enable them. maybe_enable_subscriptions(#exometer_entry{name = Metric}) -> lists:foreach( fun(#subscriber{key = #key{reporter = RName}} = S) -> case get_reporter_status(RName) of enabled -> resubscribe(S); _ -> ok end end, ets:select(?EXOMETER_SUBS, [{#subscriber{key = #key{metric = Metric, _ = '_'}, _ = '_'}, [], ['$_']}])). resubscribe(#subscriber{key = #key{reporter = RName, metric = Metric, datapoint = DataPoint, extra = Extra} = Key, t_ref = OldTRef, interval = Interval}) -> try_send(RName, {exometer_subscribe, Metric, DataPoint, Interval, Extra}), cancel_timer(OldTRef), TRef = erlang:send_after(Interval, self(), {report, Key, Interval}), ets:update_element(?EXOMETER_SUBS, Key, [{#subscriber.t_ref, TRef}]). do_report(#key{metric = Metric, datapoint = DataPoint, retry_failed_metrics = RetryFailedMetrics} = Key, Interval) -> case {RetryFailedMetrics, get_values(Metric, DataPoint)} of %% We found a value, or values. {_, [_|_] = Found} -> %% Distribute metric value to the correct process report_values(Found, Key), true; %% We did not find a value, but we should try again. {true, _ } -> if is_list(Metric) -> ?debug("Metric(~p) Datapoint(~p) not found." " Will try again in ~p msec~n", [Metric, DataPoint, Interval]), true; true -> false end; %% We did not find a value, and we should not retry. _ -> %% Entry removed while timer in progress. ?warning("Metric(~p) Datapoint(~p) not found. Will not try again~n", [Metric, DataPoint]), false end. report_batch(Reporter, Name) when is_atom(Name) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{status = disabled}] -> false; [R] -> Entries = ets:select(?EXOMETER_SUBS, [{#subscriber{key = #key{reporter = Reporter, _ = '_'}, interval = Name, _ = '_'}, [], ['$_']}]), lists:foreach( fun(#subscriber{key = Key}) -> do_report(Key, Name) end, Entries), restart_batch_timer(Name, R); [] -> false end. cancel_subscr_timers(Reporter) -> lists:foreach( fun(#subscriber{key = Key, t_ref = TRef}) -> cancel_timer(TRef), ets:update_element(?EXOMETER_SUBS, Key, [{#subscriber.t_ref, undefined}]) end, ets:select(?EXOMETER_SUBS, [{#subscriber{key = #key{reporter = Reporter, _ = '_'}, _ = '_'}, [], ['$_']}])). restart_subscr_timer(Key, Interval) when is_integer(Interval) -> TRef = erlang:send_after(Interval, self(), {report, Key, Interval}), ets:update_element(?EXOMETER_SUBS, Key, [{#subscriber.t_ref, TRef}]); restart_subscr_timer(_, _) -> true. restart_batch_timer(Name, #reporter{name = Reporter, intervals = Ints}) when is_list(Ints) -> case lists:keyfind(Name, #interval.name, Ints) of #interval{time = Time} = I -> TRef = erlang:send_after(Time, self(), batch_timer_msg(Reporter, Name)), ets:update_element(?EXOMETER_REPORTERS, Reporter, [{#reporter.intervals, lists:keyreplace(Name, #interval.name, Ints, I#interval{t_ref = TRef})}]); false -> false end. cancel_timer(undefined) -> false; cancel_timer(TRef) -> erlang:cancel_timer(TRef). %%-------------------------------------------------------------------- %% @private %% @doc %% This function is called by a gen_server when it is about to %% terminate. It should be the opposite of Reporter:init/1 and do any %% necessary cleaning up. When it returns, the gen_server terminates %% with Reason. The return value is ignored. %% %% @spec terminate(Reason, State) -> void() %% @end %%-------------------------------------------------------------------- terminate(_Reason, _) -> [terminate_reporter(R) || R <- ets:tab2list(?EXOMETER_REPORTERS)], ok. %%-------------------------------------------------------------------- %% @private %% @doc %% Convert process state when code is changed %% %% @spec code_change(OldVsn, State, Extra) -> {ok, NewState} %% @end %%-------------------------------------------------------------------- %% -record(reporter, { %% name :: atom(), %% pid :: pid(), %% mref :: reference(), %% module :: module(), %% opts = [] :: [{atom(), any()}], %% restart = #restart{} %% }). code_change(_OldVan, #st{reporters = Rs, subscribers = Ss} = S, _Extra) -> Rs1 = lists:map( fun({reporter,Pid,MRef,Module,Opts,Restart}) -> #reporter{name = Module, pid = Pid, mref = MRef, module = Module, opts = Opts, restart = Restart}; ({reporter,Name,Pid,MRef,Module,Opts,Restart}) -> #reporter{name = Name, pid = Pid, mref = MRef, module = Module, opts = Opts, restart = Restart}; ({reporter,Name,Pid,Mref,Module,Opts,Restart,Status}) -> #reporter{name = Name, pid = Pid, mref = Mref, module = Module, opts = Opts, restart = Restart, status = Status}; (#reporter{} = R) -> R end, Rs), [ets:insert(?EXOMETER_REPORTERS, R) || R <- Rs1], [ets:insert(?EXOMETER_SUBS, Sub) || Sub <- Ss], {ok, S#st{reporters = [], subscribers = []}}; code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== reporter_pids() -> ets:select(?EXOMETER_REPORTERS, [{#reporter{pid = '$1', _ = '_'}, [{is_pid,'$1'}], ['$1']}]). reporter_by_pid(Pid) -> ets:select(?EXOMETER_REPORTERS, [{#reporter{pid = Pid, _='_'}, [], ['$_']}]). reporter_by_mref(Ref) -> ets:select(?EXOMETER_REPORTERS, [{#reporter{mref = Ref, _='_'}, [], ['$_']}]). try_send(To, Msg) -> try To ! Msg catch error:_ -> Msg end. is_valid_metric({find, Name}, _DataPoint) when is_list(Name) -> true; is_valid_metric({select, Name}, _DataPoint) when is_list(Name) -> try ets:match_spec_compile(Name), true catch error:_ -> error end; is_valid_metric(Name, default) when is_list(Name) -> case exometer:info(Name, type) of undefined -> false; _ -> true end; is_valid_metric(Name, DataPoint) when is_list(Name) -> case dp_list(DataPoint) of [] -> error; [_|_] = DataPoints -> case exometer:info(Name, datapoints) of undefined -> false; DPs -> case DataPoints -- DPs of [] -> true; _ -> false end end end; is_valid_metric(_, _) -> false. dp_list(DP) when is_list(DP) -> DP; dp_list(DP) when is_atom(DP) -> [DP]; dp_list(50) -> [50]; dp_list(75) -> [75]; dp_list(90) -> [90]; dp_list(95) -> [95]; dp_list(99) -> [99]; dp_list(999) -> [999]. get_values(Name, DataPoint) when is_list(Name) -> case exometer:get_value(Name, DataPoint) of {ok, Values} when is_list(Values) -> [{Name, Values}]; _ -> [] end; get_values({How, Path}, DataPoint) -> Entries = case How of find -> exometer:find_entries(Path); select -> exometer:select(Path) end, lists:foldr( fun({Name, _, enabled}, Acc) -> case exometer:get_value(Name, DataPoint) of {ok, Values} when is_list(Values) -> [{Name, Values}|Acc]; _ -> Acc end; (_, Acc) -> Acc end, [], Entries). assert_no_duplicates([#reporter{name = R}|T]) -> case lists:keymember(R, #reporter.name, T) of true -> error({duplicate_reporter, R}); false -> assert_no_duplicates(T) end; assert_no_duplicates([]) -> ok. -spec spawn_reporter(reporter_name(), options()) -> {pid(), reference()}. spawn_reporter(Reporter, Opt) when is_atom(Reporter), is_list(Opt) -> Fun = fun() -> maybe_register(Reporter, Opt), {ok, Mod, St} = reporter_init(Reporter, Opt), reporter_loop(Mod, St) end, Pid = proc_lib:spawn(Fun), MRef = erlang:monitor(process, Pid), {Pid, MRef}. maybe_register(R, Opts) -> case lists:keyfind(registered_name, 1, Opts) of {_, none} -> ok; {_, Name} -> register(Name, self()); false -> register(R, self()) end. terminate_reporter(#reporter{pid = Pid, mref = MRef}) when is_pid(Pid) -> Pid ! {exometer_terminate, shutdown}, receive {'DOWN', MRef, _, _, _} -> ok after 1000 -> exit(Pid, kill), erlang:demonitor(MRef, [flush]) end; terminate_reporter(#reporter{pid = undefined}) -> ok. subscribe_(Reporter, Metric, DataPoint, Interval, RetryFailedMetrics, Extra, Status) -> Key = #key{reporter = Reporter, metric = Metric, datapoint = DataPoint, extra = Extra, retry_failed_metrics = RetryFailedMetrics }, ets:insert(?EXOMETER_SUBS, #subscriber{key = Key, interval = Interval, t_ref = maybe_send_after(Status, Key, Interval)}). maybe_send_after(enabled, Key, Interval) when is_integer(Interval) -> erlang:send_after(Interval, self(), {report, Key, Interval}); maybe_send_after(_, _, _) -> undefined. unsubscribe_(Reporter, Metric, DataPoint, Extra) -> ?info("unsubscribe_(~p, ~p, ~p, ~p)~n", [ Reporter, Metric, DataPoint, Extra]), case ets:lookup(?EXOMETER_SUBS, #key{reporter = Reporter, metric = Metric, datapoint = DataPoint, extra = Extra}) of [#subscriber{} = Sub] -> unsubscribe_(Sub); [] -> not_found end. unsubscribe_(#subscriber{key = #key{reporter = Reporter, metric = Metric, datapoint = DataPoint, extra = Extra} = Key, t_ref = TRef}) -> try_send( Reporter, {exometer_unsubscribe, Metric, DataPoint, Extra}), cancel_timer(TRef), ets:delete(?EXOMETER_SUBS, Key), ok. report_values(Found, #key{reporter = Reporter, extra = Extra} = Key) -> try [[report_value(Reporter, Name, DP, Extra, Val) || {DP, Val} <- Values] || {Name, Values} <- Found] catch error:Reason -> lager:error("ERROR ~p~nKey = ~p~nTrace: ~p", [Reason, Key, erlang:get_stacktrace()]) end. report_value(Reporter, Metric, DataPoint, Extra, Val) -> try Reporter ! {exometer_report, Metric, DataPoint, Extra, Val}, true catch error:_ -> false; exit:_ -> false end. retrieve_metric({Metric, Type, Enabled}, Acc) -> Cands = ets:select( ?EXOMETER_SUBS, [{#subscriber{key = #key{metric = Metric, _='_'}, _ = '_'}, [], ['$_']}]), [ { Metric, exometer:info(Metric, datapoints), get_subscribers(Metric, Type, Enabled, Cands), Enabled } | Acc ]. %% find_entries_in_list(find, Path, List) -> %% Pat = Path ++ '_', %% Spec = ets:match_spec_compile([{ {Pat, '_', '_'}, [], ['$_'] }]), %% ets:match_spec_run(List, Spec); %% find_entries_in_list(select, Pat, List) -> %% Spec = ets:match_spec_compile(Pat), %% ets:match_spec_run(List, Spec). get_subscribers(_Metric, _Type, _Status, []) -> []; %% This subscription matches Metric get_subscribers(Metric, Type, Status, [ #subscriber { key = #key { reporter = SReporter, metric = Metric, datapoint = SDataPoint }} | T ]) -> ?debug("get_subscribers(~p, ~p, ~p): match~n", [ Metric, SDataPoint, SReporter]), [ { SReporter, SDataPoint } | get_subscribers(Metric, Type, Status, T) ]; %% get_subscribers(Metric, Type, Status, %% [ #subscriber { %% key = #key { %% metric = {How, Path}, %% reporter = SReporter, %% datapoint = SDataPoint %% }} | T ]) -> %% case find_entries_in_list(How, Path, [{Metric, Type, Status}]) of %% [] -> %% get_subscribers(Metric, Type, Status, T); %% [_] -> %% [ { SReporter, SDataPoint } %% | get_subscribers(Metric, Type, Status, T) ] %% end; %% This subscription does not match Metric. get_subscribers(Metric, Type, Status, [ #subscriber { key = #key { reporter = SReporter, metric = SMetric, datapoint = SDataPoint }} | T]) -> ?debug("get_subscribers(~p, ~p, ~p) nomatch(~p) ~n", [ SMetric, SDataPoint, SReporter, Metric]), get_subscribers(Metric, Type, Status, T). %% Purge all subscriptions associated with a specific reporter %% (that just went down). purge_subscriptions(R) -> %% Go through all #subscriber elements in Subs and %% cancel the timer of those who match the provided reporter %% %% Return new #subscriber list with all original subscribers %% that do not reference reporter R. Subs = ets:select(?EXOMETER_SUBS, [{#subscriber{key = #key{reporter = R, _='_'}, _ = '_'}, [], ['$_']}]), lists:foreach(fun(#subscriber {key = Key, t_ref = TRef}) -> cancel_timer(TRef), ets:delete(?EXOMETER_SUBS, Key) end, Subs). %% Called by the spawn_monitor() call in init %% Loop and run reporters. %% Module is expected to implement exometer_report behavior reporter_init(Reporter, Opts) -> Module = proplists:get_value(module, Opts, Reporter), case Module:exometer_init(Opts) of {ok, St} -> {ok, Module, St}; {error, Reason} -> ?error("Failed to start reporter ~p: ~p~n", [Module, Reason]), exit(Reason) end. reporter_loop(Module, St) -> NSt = receive {exometer_report, Metric, DataPoint, Extra, Value } -> case Module:exometer_report(Metric, DataPoint, Extra, Value, St) of {ok, St1} -> {ok, St1}; _ -> {ok, St} end; {exometer_unsubscribe, Metric, DataPoint, Extra } -> case Module:exometer_unsubscribe(Metric, DataPoint, Extra, St) of {ok, St1} -> {ok, St1}; _ -> {ok, St} end; {exometer_subscribe, Metric, DataPoint, Extra, Interval } -> case Module:exometer_subscribe(Metric, DataPoint, Extra, Interval, St) of {ok, St1} -> {ok, St1}; _ -> {ok, St} end; {exometer_newentry, Entry} -> case Module:exometer_newentry(Entry, St) of {ok, St1} -> {ok, St1}; _ -> {ok, St} end; {exometer_setopts, Metric, Options, Status} -> case Module:exometer_setopts(Metric, Options, Status, St) of {ok, St1} -> {ok, St1}; _ -> {ok, St} end; {exometer_terminate, Reason} -> Module:exometer_terminate(Reason, St), terminate; {exometer_proc, {From, Ref}, Req} -> case Module:exometer_call(Req, From, St) of {reply, Reply, St1} -> From ! {Ref, Reply}, {ok, St1}; {noreply, St1} -> {ok, St1}; _ -> {ok, St} end; {exometer_proc, Req} -> case Module:exometer_cast(Req, St) of {noreply, St1} -> {ok, St1}; _ -> {ok, St} end; %% Allow reporters to generate their own callbacks. Other -> ?debug("Custom invocation: ~p(~p)~n", [ Module, Other]), case Module:exometer_info(Other, St) of {ok, St1} -> {ok, St1}; _ -> {ok, St} end end, case NSt of {ok, St2} -> reporter_loop(Module, St2); _ -> ok end. call(Req) -> gen_server:call(?MODULE, Req). cast(Req) -> gen_server:cast(?MODULE, Req). init_subscriber({Reporter, Metric, DataPoint, Interval, RetryFailedMetrics}) -> Status = get_reporter_status(Reporter), subscribe_(Reporter, Metric, DataPoint, Interval, RetryFailedMetrics, undefined, Status); init_subscriber({Reporter, Metric, DataPoint, Interval, RetryFailedMetrics, Extra}) -> Status = get_reporter_status(Reporter), subscribe_(Reporter, Metric, DataPoint, Interval, RetryFailedMetrics, Extra, Status); init_subscriber({Reporter, Metric, DataPoint, Interval}) -> Status = get_reporter_status(Reporter), subscribe_(Reporter, Metric, DataPoint, Interval, true, undefined, Status); init_subscriber({apply, {M, F, A}}) -> lists:foreach(fun(Sub) -> init_subscriber(Sub) end, apply(M, F, A)); init_subscriber({select, Expr}) when tuple_size(Expr)==3; tuple_size(Expr)==4; tuple_size(Expr)==5 -> {Pattern, Reporter, DataPoint, Interval, Retry, Extra} = case Expr of {P, R, D, I} -> {P, R, D, I, true, undefined}; {P, R, D, I, Rf} -> {P, R, D, I, Rf, undefined}; {P, R, D, I, Rf, X} -> {P, R, D, I, Rf, X} end, Status = get_reporter_status(Reporter), Entries = exometer:select(Pattern), lists:foreach( fun({Entry, _, _}) -> subscribe_(Reporter, Entry, DataPoint, Interval, Retry, Extra, Status) end, Entries); init_subscriber(Other) -> ?warning("Incorrect static subscriber spec ~p. " "Use { Reporter, Metric, DataPoint, Interval [, Extra ]}~n", [ Other ]). get_reporter_status(R) -> try ets:lookup_element(?EXOMETER_REPORTERS, R, #reporter.status) catch error:_ -> disabled end. add_restart(#restart{spec = Spec, history = H, save_n = N} = R) -> T = exometer_util:timestamp(), H1 = lists:sublist([T|H], 1, N), case match_frequency(H1, Spec) of {remove, Action} -> {remove, Action}; restart -> {restart, R#restart{history = H1}} end. match_frequency([H|T], Spec) -> match_frequency(T, 1, H, Spec). match_frequency([H|T], R, Since, Spec) -> R1 = R+1, %% Note that we traverse millisec timestamps backwards in time Span = (Since - H) div 1000, case find_match(Spec, R1, Span) of {true, Action} -> {remove, Action}; false -> match_frequency(T, R1, Since, Spec) end; match_frequency([], _, _, _) -> restart. find_match([{R1,T1}|Tail], R, T) when R1 =< R, T1 >= T -> {true, find_action(Tail)}; find_match([_|Tail], R, T) -> find_match(Tail, R, T); find_match([], _, _) -> false. find_action([{M,F} = H|_]) when is_atom(M), is_atom(F) -> H; find_action([_|T]) -> find_action(T); find_action([]) -> no_action. default_restart() -> [{3, 1}, {10, 30}, {?MODULE, remove_reporter}]. get_restart(Opts) -> case lists:keyfind(restart, 1, Opts) of {_, R} -> restart_rec(valid_restart(R)); false -> restart_rec(default_restart()) end. restart_rec(L) -> Save = lists:foldl( fun ({R,_}, Acc) when is_integer(R) -> erlang:max(R, Acc); (_, Acc) -> Acc end, 0, L), #restart{spec = L, save_n = Save}. valid_restart(L) when is_list(L) -> lists:foreach( fun({R,T}) when is_integer(R), is_integer(T), R > 0, T > 0 -> ok; ({M,F}) when is_atom(M), is_atom(F) -> ok; (_) -> erlang:error({invalid_restart_spec, L}) end, L), L. do_remove_reporter(Reporter) -> do_remove_reporter(Reporter, true). do_remove_reporter(Reporter, Terminate) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [#reporter{} = R] -> case Terminate of true -> terminate_reporter(R); false -> ok end, ets:delete(?EXOMETER_REPORTERS, Reporter), purge_subscriptions(Reporter), ok; [] -> {error, not_found} end. change_reporter_status(Reporter, New) -> case ets:lookup(?EXOMETER_REPORTERS, Reporter) of [R] -> do_change_reporter_status(R, New); [] -> {error, not_found} end. do_change_reporter_status(#reporter{name = Reporter, status = Old} = R, New) -> case {Old, New} of {disabled, enabled} -> restart_reporter(R); {enabled, disabled} -> cancel_subscr_timers(Reporter), terminate_reporter(R), ets:update_element(?EXOMETER_REPORTERS, Reporter, [{#reporter.status, disabled}]); {Old, Old} -> ok end, ok.