%% ------------------------------------------------------------------- %% %% 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 Interface library for managing probes.
%% %% This library contains the main API for accessing all probes %% executing in exometer. %% %% All exported functions in the `exomter_probe' module are invoked %% by the `exometer' module; a developer will not have to call %% `exometer_probe' functions directly. %% %% A probe is an implementation of the `exometer_probe' behavior %% which runs in its own process in order to collect data to be %% handled and reported by exometer. The implementation will be %% invoked through the `exomoeter_probe' module, which, as stated %% above, in its turn is invoked by the `exometer' module. %% %% A custom exometer probe is invoked by mapping a type to the module %% name of the custom exometer probe module. All metrics created with the %% given type will trigger the invocation of the new probe module. See %% {@section Configuring type - entry maps} for details on how to setup %% such maps. %% %% If the data can be collected at a high speed, and without %% blocking, an `exometer_entry' implementation can be used instead %% to do the gathering in-process. %% %% A probe is created throgh the `exomter_probe:new/3' call, which in %% its turn is called by `exometer:new/3'. During probe creation, a %% new process is spawned to handle the probe and call its %% implementation. While the created process is not a gen_server, it %% behaves similarly and provides a state to all implementation %% calls. %% %% Once running, the probe collects data from a subsystem, such as %% `/proc', `sysfs', and `netlink', through timer-based calls to %% `probe_sample/1' or explicit calls to `probe_update/2'. %% %% A probe implementation can support any number of data points, %% where each data point is a specifric sample from the probe. For %% example, a probe that measures network traffic would have %% `rx_packets', `tx_packets', `errors', `dropped', and other data %% points reported by `ifconfig(8)' and `ip(8)'. %% %% Values are retrieved from the probe through the `probe_get_value/2' %% call, which specifies the data points to be returned. The probe is %% expected to gather the given data points and return them to the %% caller. %% %% == The probe callback interface == %% %% The following functions are to be implemented and exported by a probe %% implementation. %% %% === behaviour/0 === %% %% The `behaviour/0' function for an entry implementation should return %% the atom `probe'. This function will be involved by the %% exometer system in order to determine if a callback is %% an entry or a probe. %% %% === probe_init/3 === %% The `probe_init/3' function is invoked as follows: %% %%
%%      probe_init(Name, Type, Options)
%% %% The implementation shall initiate the probe, create the %% necessary state, and return it for furure access %% through `probe_update/2', `probe_sample/1' and `get_value/2' calls. %% %% The arguments are as follows: %% %% + `Name' %% Specifies the name of the metric to be created as a list of atoms. %% %% + `Type' %% Specifies the type provided to the `exometer:new/3' call (before it %% was translated by the type - exometer probe map). It can be used if several %% different types are mapped to the same probe module. %% %% + `Options' %% Specifies an option list that contains additional setup directives to %% the probe. The actual options to support are a combination of the %% standard options, described below, and probe specific options %% processed by `probe_init/3'. %% %% Standard options are processed directly by `new/3', before %% `probe_init/3' is calledm and are as follows: %% %% + `{priority, P}' %% Will be forwarded by the probe's process to `erlang:process_flag/2'. %% %% + `{min_heap_size, S}' %% Will be forwarded by the probe's process to `erlang:process_flag/2'. %% %% + `{min_bin_vheap_size, S}' %% Will be forwarded by the probe's process to `erlang:process_flag/2'. %% %% + `{sensitive, true | false}' %% Will be forwarded by the probe's process to `erlang:process_flag/2'. %% %% + `{sample_interval, t}' %% Specifies the interval, in milliseconds, that `exometer_probe:sample/1'. %% should be invoked at. %% %% The `probe_init/3' implementation is invoked by `exometer:new/3', %% which calls `exometer_probe:new/3', which invokes the probe %% implementation.. %% %% The `probe_init/3' function shall return `{ok, State}' where State %% is a tuple that will be provided as a the `State' argument to all %% future probe implementation calls for the metric. %% %% If the `sample_interval' option has been specified in `Opts', %% probe_sample/2' will be invoked immediately after `probe_init/2' %% returns to retrieve a first sample. After that, `probe_sample/2' %% will repeatedly will be called by the probe process at the %% millisecond-specified interval. %% %% Should `probe_init/3' return antyhing else but `{ok, State}', %% invoking `new/3' call will fail. %% %% %% === probe_terminate/1 === %% The `probe_terminate/1' function is invoked as follows: %% %%
%%      probe_terminate(State)
%% %% The custom probe shall release any resources associated with the %% given state and return `ok'. %% %% The arguments are as follows: %% %% + `State' %% The probe state, originally returned by `probe_init/3' and subsequentially %% modified by other probe implementation calls. %% %% %% The `probe_terminate/1' implementation is invoked by `exometer:delete/1', which %% calls `exometer_probe:delete/3', which invokes the probe %% implementation. %% %% %% === probe_setopts/3 === %% The `probe_setopts/2' function is invoked as follows: %% %%
%%      probe_setopts(Entry, Opts, State)
%% %% The `probe_setopts/4' implementation is invoked by %% `exometer:setopts/3', which calls `exometer_probe:setopts/3', %% which invokes the probe implementation. %% %% The implementation of this function shall modify the options of a %% probe. The `setopts/3' function, which will process standard %% options before invoking `probe_setopts/4' with the remaining %% options. See the documentation for `probe_init/3' for details. %% %% The arguments are as follows: %% %% + `Entry' %% The (opaque) exometer entry record. See {@link exometer_info} for %% information on how to inspect the data structure. %% %% + `Opts' %% The probe-specific options to be processed. %% %% + `Status' %% The new status of the entry. %% %% + `State' %% The probe state, originally returned by `probe_init/3' and subsequently %% modified by other probe implementation calls. %% %% This function shall return `{ok, NewState}' where `NewState' is %% the modified probe state that incorporates the new options. %% %% %% === probe_update/2 === %% The `probe_update/2' function is invoked as follows: %% %%
%%      probe_update(Value, State)
%% %% Incorporate a new value into the metric maintained by the metric. %% %% The arguments are as follows: %% %% + `Value' %% The value to integrate. %% %% + `State' %% The probe state, originally returned by `probe_init/3' and subsequentially %% modified by other probe implementation calls. %% %% This function can be called outside the periodic `probe_sample/1/' %% call to have the probe process a value given in `Value'. %% %% The `probe_update/2' implementation is invoked by `exometer:update/2', which %% calls `exometer_probe:update/4', which invokes the probe %% implementation. %% %% Once processed, `probe_update/2' shall return `{ok, NewState}', %% where `NewState' contains the new probe state with the processed %% value. %% %% %% === probe_get_value/2 === %% The `probe_get_value/2' function is invoked as follows: %% %%
%%      probe_get_value(DataPoints, State)
%% %% The `probe_get_value/2' implementation shall retrieve the value of %% one or more data points from the probe. %% %% The arguments are as follows: %% %% + `DataPoints' %% List of data point atoms to retrieve values for. %% %% + `State' %% The probe state, originally returned by `probe_init/3' and subsequentially %% modified by other probe implementation calls. %% %% The `probe_get_value/2' implementation is invoked by %% `exometer:get_value/2', which calls `exometer_probe:get_value/4', %% which invokes the probe implementation. %% %% If `exometer:get_value/2' is invoked with `default' as a single %% data point, the probe's `probe_get_datapoints/1' function will be %% called to retrieve all data points supported by the probe %% implementation. `probe_get_value/2' will then be called with the %% returned set of data points provided as an argument. %% %% This function shall return the value of all data points provided in %% `DataPoints', given that they are supported. %% %% The list in the returned tuple shall have the format: %% %%
%%      [{ DP, Val}, ...]
%% %% Where `DP' one of the data points in the `DataPoints' argument, and %% `Val' is the value of that data point. %% %% If one of the argument-provided data points are not supported by the probe, %% the tuple returned for that data point shall be `{ DP, {error, undefined}'. %% %% For example, if the provided `DataPoint' argument is set to `[ min, %% max, xyzzy ]', and only `min' and `max' are supported %% by the probe, the returned list shall look like below: %% %%
%%      [{ min, 0.1265 }, { max, 3338.21 }, { xyzzy, { error, unsupported } ]
%% %% The `probe_get_value/2' implementation shall return `{ok, List}', %% where `List' is the list of data points and their values described %% above. No new state is returned by this function. %% %% %% === probe_get_datapoints/1 === %% The `probe_get_datapoints/1' function is invoked as follows: %% %%
%%      probe_get_datapoints(State)
%% %% The `probe_get_datapoints/1' shall return a list with all data points %% supported by the probe %% %% The arguments are as follows: %% %% + `State' %% The probe state, originally returned by `probe_init/3' and subsequentially %% modified by other probe implementation calls. %% %% The `probe_get_datapoints/1' implementation is invoked by %% `exometer:info/2', which calls `exometer_probe:get_datapoints/3', %% which invokes the probe implementation. %% %% In cases where `exometer:get_value/2' is called with `default' as a %% single data point, `probe_get_datapoints/1' is also called to %% retrieve a list of all supported data points, which is then %% forwarded to `probe_get_value/2'. %% %% The implementation of `probe_get_datapoints/1' shall return `{ok, DpList}', %% where `DpList' is a list of data point atoms supported by the probe. %% %% %% === probe_reset/1 === %% The `probe_reset/1' function is invoked as follows: %% %%
%%      probe_reset(State)
%% %% The `probe_reset/1' shall reset the state of the probe to its initial state. %% %% The arguments are as follows: %% %% + `State' %% The probe state, originally returned by `probe_init/3' and subsequentially %% modified by other probe implementation calls. %% %% %% The `probe_reset/1' implementation is invoked by %% `exometer:reset/1', which calls `exometer_probe:reset/3', which %% invokes the probe implementation. %% %% The implementation of `probe_reset/1' shall return `{ok, %% NewState}', where `NewState' contains the reset state of the probe. %% %% %% === probe_sample/1 === %% The `probe_sample/1' function is invoked as follows: %% %%
%%      probe_sample(State)
%% %% The `probe_sample/1' implementation shall sample data from the %% subsystem the probe is integrated with. %% %% The arguments are as follows: %% %% + `State' %% The probe state, originally returned by `probe_init/3' and subsequentially %% modified by other probe implementation calls. %% %% When invoked, `probe_sample/1' is expected to interface the %% sub-system (/proc, /sysfs, etc) monitored by the probe, extract the %% relevant data from it, and return an updated probe state that %% incorporates the extracted data. %% %% The `probe_sample/1' function is invoked by the probe thread at %% intervals specified by the `{sample_interval, Intv}' option %% provided to `exometer_probe:new/3'. If this option is missing, or %% set to infinity, `probe_sample/1' will never be called. %% %% The implementation of `probe_sample/1' shall return `{ok, %% NewState}', where `NewState' contains the new state of the probe %% with the sampled data integrated into it. %% %% %% === probe_handle_msg/2 === %% The `probe_handle_msg/2' function is invoked as follows: %% %%
%%      probe_handle_msg(Msg, State)
%% %% The `probe_handle_msg/1' is invoked to process messages received %% by the probe process. %% %% The arguments are as follows: %% %% + `State' %% The probe state, originally returned by `probe_init/3' and subsequentially %% modified by other probe implementation calls. %% %% + `Msg' %% The probe state, originally returned by `probe_init/3' and subsequentially %% modified by other probe implementation calls. %% %% The implementation of this function will be called by the probe's %% process when it receives a message that is not recognized by the %% internal receive loop. %% %% The implementation of `probe_handle_msg/2' shall return `{ok, %% NewState}', where `NewState' contains the new state of the probe %% that reflects the processed message. %% -module(exometer_probe). -behaviour(exometer_entry). % exometer_entry callb -export( [ behaviour/0, new/3, delete/3, get_datapoints/3, get_value/3, get_value/4, update/4, reset/3, sample/3, setopts/3 ]). -include("exometer.hrl"). -record(st, { name, type, module = undefined, mod_state, sample_timer, sample_interval = infinity, %% msec. infinity = disable probe_sample() peridoc calls. opts = [] }). -type name() :: exometer:name(). -type options() :: exometer:options(). -type type() :: exometer:type(). -type mod_state() :: any(). -type data_points() :: [atom()]. -type probe_reply() :: ok | {ok, mod_state()} | {ok, any(), mod_state()} | {noreply, mod_state()} | {error, any()}. -type probe_noreply() :: ok | {ok, mod_state()} | {error, any()}. -callback behaviour() -> exometer:behaviour(). -callback probe_init(name(), type(), options()) -> probe_noreply(). -callback probe_terminate(mod_state()) -> probe_noreply(). -callback probe_setopts(exometer:entry(), options(), mod_state()) -> probe_reply(). -callback probe_update(any(), mod_state()) -> probe_noreply(). -callback probe_get_value(data_points(), mod_state()) -> probe_reply(). -callback probe_get_datapoints(mod_state()) -> {ok, data_points()}. -callback probe_reset(mod_state()) -> probe_noreply(). -callback probe_sample(mod_state()) -> probe_noreply(). -callback probe_handle_msg(any(), mod_state()) -> probe_noreply(). %% FIXME: Invoke this. -callback probe_code_change(any(), mod_state(), any()) -> {ok, mod_state()}. new(Name, Type, [{arg, Module}|Opts]) -> { ok, exometer_proc:spawn_process( Name, fun() -> init(Name, Type, Module, Opts) end) }; new(Name, Type, Options) -> %% Extract the module to use. {value, { module, Module }, Opts1 } = lists:keytake(module, 1, Options), new(Name, Type, [{arg, Module} | Opts1]). %% Should never be called directly for exometer_probe. behaviour() -> entry. delete(_Name, _Type, Pid) when is_pid(Pid) -> exometer_proc:cast(Pid, delete). get_value(_Name, _Type, Pid) when is_pid(Pid) -> exometer_proc:call(Pid, {get_value, default}). get_value(_Name, _Type, Pid, DataPoints) when is_pid(Pid) -> exometer_proc:call(Pid, {get_value, DataPoints}). get_datapoints(_Name, _Type, Pid) when is_pid(Pid) -> exometer_proc:call(Pid, get_datapoints). setopts(#exometer_entry{ref = Pid} = E, Opts, Status) when is_pid(Pid) -> exometer_proc:cast(Pid, {setopts, E, Opts, Status}). update(_Name, Value, _Type, Pid) when is_pid(Pid) -> exometer_proc:cast(Pid, {update, Value}). reset(_Name, _Type, Pid) when is_pid(Pid) -> exometer_proc:cast(Pid, reset). sample(_Name, _Type, Pid) when is_pid(Pid) -> exometer_proc:cast(Pid, sample). init(Name, Type, Mod, Opts) -> process_flag(min_heap_size, 40000), {St0, Opts1} = process_opts(Opts, #st{name = Name, type = Type, module = Mod}), St = St0#st{opts = Opts1}, %% Create a new state for the module case {Mod:probe_init(Name, Type, St#st.opts), St#st.sample_interval} of { ok, infinity} -> %% No sample timer to start. Return with undefined mod state loop(St#st{ mod_state = undefined }); {{ok, ModSt}, infinity} -> %% No sample timer to start. Return with the mod state returned by probe_init. loop(St#st{ mod_state = ModSt }); {ok, _} -> %% Fire up the timer, with undefined mod loop(sample(St#st{ mod_state = undefined })); {{ok, ModSt}, _ } -> %% Fire up the timer. loop(sample(St#st{ mod_state = ModSt })); {{error, Reason}, _} -> %% FIXME: Proper shutdown. {error, Reason} end. loop(St) -> receive Msg -> loop(handle_msg(Msg, St)) end. handle_msg(Msg, St) -> Module = St#st.module, case Msg of {system, From, Req} -> exometer_proc:handle_system_msg( Req, From, St, fun(St1) -> loop(St1) end); {exometer_proc, {From, Ref}, {get_value, default} } -> {ok, DataPoints} = Module:probe_get_datapoints(St#st.mod_state), {Reply, NSt} = process_probe_reply(St, Module:probe_get_value(DataPoints, St#st.mod_state)), From ! {Ref, Reply }, NSt; {exometer_proc, {From, Ref}, {get_value, DataPoints} } -> {Reply, NSt} = process_probe_reply(St, Module:probe_get_value(DataPoints, St#st.mod_state)), From ! {Ref, Reply }, NSt; {exometer_proc, {From, Ref}, get_datapoints } -> {ok, DataPoints} =Module:probe_get_datapoints(St#st.mod_state), From ! {Ref, DataPoints }, St; {exometer_proc, { update, Value } } -> Res = process_probe_noreply(St, Module:probe_update(Value, St#st.mod_state)), Res; {exometer_proc, reset } -> process_probe_noreply(St, Module:probe_reset(St#st.mod_state)); {exometer_proc, sample } -> process_probe_noreply(St, Module:probe_sample(St#st.mod_state)); {exometer_proc, {From, Ref}, {setopts, Entry, Options}} -> %% Extract probe-level options (sample_interval) {NSt, Opts1} = process_opts(Options, St), {Reply, NSt1} = %% Call module setopts for remainder of opts process_probe_reply( NSt, Module:probe_setopts(Entry, Opts1, NSt#st.mod_state)), From ! {Ref, Reply }, %% Return state with options and any non-duplicate original opts. NSt1#st { opts = Opts1 ++ [{K,V} || {K,V} <- St#st.opts, not lists:keymember(K,1,Opts1) ] }; {timeout, _TRef, {exometer_proc, sample_timer}} -> sample(St); {exometer_proc, delete} -> Module:probe_terminate(St#st.mod_state), exometer_proc:stop(); {exometer_proc, code_change} -> Module:probe_terminate(St#st.mod_state), exometer_proc:stop(); Other -> process_probe_noreply(St, Module:probe_handle_msg(Other, St#st.mod_state)) end. process_probe_reply(St, ok) -> {ok, St}; process_probe_reply(St, {ok, Reply}) -> {Reply, St}; process_probe_reply(St, {ok, Reply, ModSt}) -> {Reply, St#st { mod_state = ModSt }} ; process_probe_reply(St, {noreply, ModSt}) -> {noreply, St#st { mod_state = ModSt }}; process_probe_reply(St, {error, Reason}) -> {{error, Reason}, St}; process_probe_reply(St, Err) -> {{error, { unsupported, Err}}, St}. process_probe_noreply(St, {ok, ModSt}) -> St#st{mod_state=ModSt}; process_probe_noreply(St, _) -> St. %% =================================================================== sample(#st{module = Mod, mod_state = ModSt} = St) -> St1 = restart_timer(sample, St), process_probe_noreply(St1, Mod:probe_sample(ModSt)). restart_timer(sample, #st{sample_interval = Int} = St) -> St#st{sample_timer = start_timer(Int, {exometer_proc, sample_timer})}. start_timer(infinity, _) -> undefined; start_timer(T, Msg) when is_integer(T), T >= 0, T =< 16#FFffFFff -> erlang:start_timer(T, self(), Msg). process_opts(Options, #st{} = St) -> exometer_proc:process_options(Options), process_opts(Options, St, []). process_opts([{sample_interval, Val}|T], #st{} = St, Acc) -> process_opts(T, St#st{ sample_interval = Val }, Acc); process_opts([Opt|T], St, Acc) -> process_opts(T, St, [Opt | Acc]); process_opts([], St, Acc) -> {St, lists:reverse(Acc)}.