%%%============================================================================= %%% @doc Worker process to handle hackney metrics %%% %%% A worker process has two jobs: %%% %%% (1) Calculate the metric value. %%% %%% Hackney does not keep the state of its metrics, but instead emits events to %%% the metrics engine, like "+1 on this counter", "set X on this gauge", "add %%% Y on this histogram". The job of a metric worker is to process these events %%% and keep an up-to-date state that represent the real value of the tracked %%% metric. State updates runs on constant time - aka O(1) complexity, which is %%% important since each single request generates about 9 metric updates. %%% %%% (2) Send the metric value to Telemetry %%% %%% Telemetry will apply backpressure if we send too many events - which may %%% cause the process inbox to grow if we try to send the metric value after %%% every update. But because the state already has the most up-to-date value, %%% we can send this value on intervals - this does cause the metric to be less %%% accurate but allows us to use telemetry. %%% %%% Start options: %%% %%% - metric: the name of the metric, as a list of atoms. It's required. %%% - report_interval: the interval, in milliseconds, that a worker reports data %%% to telemetry. If set to 0, scheduled reports are disabled and the worker %%% will report after every update. Defaults to the value configured in the %%% hackney_telemetry/report_interval config. %%% %%% @end %%%============================================================================= -module(hackney_telemetry_worker). -behaviour(gen_server). % gen_server callbacks -export( [code_change/3, init/1, handle_call/3, handle_cast/2, handle_info/2, start_link/1, terminate/2] ). % public functions -export([child_spec/1, update/3, worker_name/1]). -include("hackney_telemetry.hrl"). -include("hackney_telemetry_worker.hrl"). -define(DEFAULT_REPORT_INTERVAL, 5000). %%------------------------------------------------------------------------------ %% @doc Generates a worker child spec based on a %% @end %%------------------------------------------------------------------------------ -spec child_spec(hackney_metric()) -> map(). child_spec(Args) -> Metric = proplists:get_value(metric, Args), #{id => {?MODULE, Metric}, start => {?MODULE, start_link, [Args]}}. %%------------------------------------------------------------------------------ %% @doc Returns the name of the worker process %% @end %%------------------------------------------------------------------------------ -spec worker_name(hackney_metric()) -> {global, {atom(), hackney_metric()}}. worker_name(Metric) -> {global, {node(), Metric}}. %%----------------------------------------------------------------------------- %% @doc Updates a metric %% @end %%----------------------------------------------------------------------------- -spec update(hackney_metric(), any(), transform_fun()) -> ok. update(Metric, EventValue, TransformFun) -> ProcessName = worker_name(Metric), gen_server:cast(ProcessName, {update_event, EventValue, TransformFun}). %%----------------------------------------------------------------------------- %% @doc Starts the server %% @end %%----------------------------------------------------------------------------- start_link(Args) -> Metric = proplists:get_value(metric, Args), gen_server:start_link(worker_name(Metric), ?MODULE, Args, []). %%----------------------------------------------------------------------------- %% @doc Initialize the state of the server %% @end %%----------------------------------------------------------------------------- init(Args) -> case telemetry_settings(Args) of {ok, TelemetrySettings} -> State = #worker_state{ value = 0, report_interval = fetch_report_interval(Args), telemetry_settings = TelemetrySettings }, maybe_schedule_report(State), {ok, State}; {error, Error} -> {stop, Error} end. telemetry_settings(Args) -> Metric = proplists:get_value(metric, Args), case Metric of [hackney, MeasurementKey] -> {ok, {[hackney], MeasurementKey, #{}}}; [hackney_pool, PoolName, MeasurementKey] -> {ok, {[hackney_pool], MeasurementKey, #{pool => PoolName}}}; _ -> {error, unsupported_metric} end. %%----------------------------------------------------------------------------- %% @doc gen_server handle_call implementation. %% @end %%----------------------------------------------------------------------------- handle_call(_Message, _From, State) -> {reply, ok, State}. %%----------------------------------------------------------------------------- %% @doc Handles update events %% @end %%----------------------------------------------------------------------------- handle_cast({update_event, EventValue, TransformFun}, State) -> NewValue = TransformFun(State#worker_state.value, EventValue), UpdatedState = State#worker_state{value = NewValue}, if UpdatedState#worker_state.report_interval == 0 -> report(UpdatedState); true -> ok end, {noreply, UpdatedState}. %%----------------------------------------------------------------------------- %% @doc Handles report events %% @end %%----------------------------------------------------------------------------- handle_info(report, State) -> report(State), maybe_schedule_report(State), {noreply, State}. %%------------------------------------------------------------------------------ %% @doc gen_server terminate calback %% @end %%------------------------------------------------------------------------------ terminate(_Reason, _State) -> ok. %%------------------------------------------------------------------------------ %% @doc gen_server code_change calback %% @end %%------------------------------------------------------------------------------ code_change(_OldVersion, State, _Extra) -> {ok, State}. %%------------------------------------------------------------------------------ %% @doc Evaluates the current state of the process and report the value to %% telemetry %% @end %%------------------------------------------------------------------------------ -spec report(#worker_state{}) -> ok. report(State) -> {Metric, MeasurementKey, Metadata} = State#worker_state.telemetry_settings, Measurement = #{MeasurementKey => State#worker_state.value}, telemetry:execute(Metric, Measurement, Metadata), ok. %%------------------------------------------------------------------------------ %% @doc Reports events to telemetry If the report interval is greater than zero, %% we schedule the report to happen. %% @end %%------------------------------------------------------------------------------ -spec maybe_schedule_report(#worker_state{}) -> ok. maybe_schedule_report(State) -> ReportInterval = State#worker_state.report_interval, if ReportInterval > 0 -> erlang:send_after(ReportInterval, self(), report), ok; true -> ok end. %%------------------------------------------------------------------------------ %% @doc Fetches the interval on which workers report metrics to telemetry. %% @end %%------------------------------------------------------------------------------ -spec fetch_report_interval(#worker_state{}) -> ok. fetch_report_interval(Args) -> ValueFromArgs = proplists:get_value(report_interval, Args), ValueFromConfig = application:get_env(hackney_telemetry, report_interval), if ValueFromArgs =/= undefined -> ValueFromArgs; ValueFromConfig =/= undefined -> {ok, ActualValueFromConfig} = ValueFromConfig, ActualValueFromConfig; true -> ?DEFAULT_REPORT_INTERVAL end.