%%% @doc Worker process to handle a hackney metric. %%% %%% A worker process has two jobs: %%% %%% (1) Calculate metric values %%% %%% Hackney does not keep the state of its metrics, but instead emits events to %%% the metrics engine, like "increase this counter by 1", "set this gauge to X", %%% "add Y to this histogram". The job of a metric worker is to process these %%% events and keep up-to-date state representing the value of the tracked metric. %%% State updates run in constant time (O(1)), important since a single %%% request generates about nine metric updates. %%% %%% (2) Send metric values to Telemetry %%% %%% If we send the metric value to Telemetry after every update, then %%% telemetry processing may not be able to keep up, and Telemetry will apply %%% backpressure. %%% %%% Since the worker maintains the most up-to-date value, we can send the current %%% value periodically. Gauge metrics may be less accurate, but it avoids overload. %%% %%% Start options: %%% %%% - metric: name of the metric as a list of atoms (required). %%% %%% - report_interval: how often the worker reports data to telemetry, in milliseconds. %%% %%% 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 Generate worker child spec for a supervisor. -spec child_spec([{metric, hackney_metric()}]) -> supervisor:child_spec(). child_spec(Args) -> Metric = proplists:get_value(metric, Args), #{id => {?MODULE, Metric}, start => {?MODULE, start_link, [Args]}}. %% @doc Return name of the worker process -spec worker_name(hackney_metric()) -> {global, {atom(), hackney_metric()}}. worker_name(Metric) -> {global, {node(), Metric}}. %% @doc Update metric -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 Start server -spec start_link(keywords()) -> gen_server:start_ret(). start_link(Args) -> Metric = proplists:get_value(metric, Args), gen_server:start_link(worker_name(Metric), ?MODULE, Args, []). %% @doc Initialize server state 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. handle_call(_Message, _From, State) -> {reply, ok, State}. %% @doc Handle update event 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 Handle report events handle_info(report, State) -> report(State), maybe_schedule_report(State), {noreply, State}. %% @doc gen_server terminate callback terminate(_Reason, _State) -> ok. %% @doc gen_server code_change callback code_change(_OldVersion, State, _Extra) -> {ok, State}. %% @doc Evaluate current process state and report value to telemetry. -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 Report events to telemetry. %% %% If the report interval is greater than zero, 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 Fetch interval on which workers report metrics to telemetry. -spec fetch_report_interval(keywords()) -> non_neg_integer(). 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.