diff --git a/src/instrument/mongoose_prometheus_sliding_window.erl b/src/instrument/mongoose_prometheus_sliding_window.erl index 16505d90759..5fe7df4b3d0 100644 --- a/src/instrument/mongoose_prometheus_sliding_window.erl +++ b/src/instrument/mongoose_prometheus_sliding_window.erl @@ -3,28 +3,46 @@ %% Manages a sliding window using DDSketch instances for quantile summaries. %% Window parameters are configurable via window_size_ms and window_step_ms constants. +-behaviour(gen_server). + +%% gen_server callbacks +-export([init/1, + handle_call/3, + handle_cast/2, + handle_info/2, + terminate/2]). + +%% API -export([child_spec/0, - declare/1, + start_link/0]). + +-export([declare/1, observe/3, value/2, values/1, remove/2, - get_all_metric_names/0, + get_all_metrics/0, get_metric_spec/1]). --export([start_link/0, - init/1, - handle_call/3, - handle_cast/2, - handle_info/2, - terminate/2, - code_change/3]). +-define(METRICS_TABLE, mongoose_prometheus_sliding_window_metrics). +-define(DDSKERL_TABLE, mongoose_prometheus_sliding_window_ddskerl). --export([window_size_ms/0, window_step_ms/0, window_count/0]). % Exported for mocking in tests +-record(state, { + timer_ref :: timer:tref() +}). --ignore_xref([start_link/0, value/2, window_size_ms/0, window_step_ms/0, window_count/0]). +-record(metric, { + name :: name(), + states = #{} :: #{label_values() => metric_state()}, + spec = proplists:proplist() +}). --behaviour(gen_server). +-type name() :: string(). +-type label_values() :: [mongoose_instrument:label_value()]. +-type window_data() :: #{sketch := ddskerl_ets:ddsketch(), + name := term()}. +-type metric_state() :: #{windows => [window_data()], + current_index => non_neg_integer()}. %% Window parameters @@ -35,7 +53,7 @@ window_size_ms() -> 60000. % Total window size: 60 seconds window_step_ms() -> 3000. % Step size: 3 seconds per sub-window -spec window_count() -> pos_integer(). -window_count() -> ?MODULE:window_size_ms() div ?MODULE:window_step_ms(). +window_count() -> window_size_ms() div window_step_ms(). %% DDSketch parameters @@ -52,115 +70,31 @@ default_error() -> 0.01. -spec default_bound() -> non_neg_integer(). default_bound() -> 1260. --record(state, { - timer_ref :: {ok, timer:tref()} | {error, term()} | undefined, - ets_table :: ets:tab() | undefined, - metrics = #{} :: #{name() => #{{name(), label_values()} => metric_state()}}, - metric_specs = #{} :: #{name() => proplists:proplist()} -}). - --type name() :: string(). --type label_values() :: [mongoose_instrument:label_value()]. --type window_data() :: #{sketch := ddskerl_ets:ddsketch(), - ref := ets:tab(), - name := term()}. --type metric_state() :: #{windows => [window_data()], - current_index => non_neg_integer()}. - -%% Public API - --spec declare(proplists:proplist()) -> boolean(). -declare(MetricSpec) -> - Name = proplists:get_value(name, MetricSpec), - gen_server:call(?MODULE, {declare, Name, MetricSpec}). - --spec observe(name(), label_values(), number()) -> ok. -observe(Name, LabelValues, Value) -> - gen_server:cast(?MODULE, {observe, Name, LabelValues, Value}). - --spec value(name(), label_values()) -> undefined | {non_neg_integer(), number(), [{number(), number()}]}. -value(Name, LabelValues) -> - gen_server:call(?MODULE, {value, Name, LabelValues}). - --spec values(name()) -> [{label_values(), {non_neg_integer(), number(), [{number(), number()}]}}]. -values(Name) -> - [{LV, V} || LV <- get_label_values(Name), V <- [value(Name, LV)], V =/= undefined]. - --spec get_label_values(name()) -> [label_values()]. -get_label_values(Name) -> - gen_server:call(?MODULE, {get_label_values, Name}). - --spec remove(name(), label_values()) -> boolean(). -remove(Name, LabelValues) -> - gen_server:call(?MODULE, {remove, Name, LabelValues}). - --spec get_all_metric_names() -> [name()]. -get_all_metric_names() -> - gen_server:call(?MODULE, get_all_metric_names). - --spec get_metric_spec(name()) -> proplists:proplist() | undefined. -get_metric_spec(Name) -> - gen_server:call(?MODULE, {get_metric_spec, Name}). +%% gen_server -%% gen_server callbacks +init(_Args) -> + EtsOpts = [named_table, set, public, {read_concurrency, true}, {write_concurrency, auto}], + ets:new(?METRICS_TABLE, [{keypos, #metric.name} | EtsOpts]), + ets:new(?DDSKERL_TABLE, EtsOpts), + {ok, TimerRef} = timer:send_interval(window_step_ms(), rotate), + {ok, #state{timer_ref = TimerRef}}. -start_link() -> - gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). +handle_call(_Message, _From, State) -> + {noreply, State}. -init([]) -> - EtsTable = ets:new(?MODULE, [set, private, {read_concurrency, true}, {write_concurrency, true}]), - TimerRef = timer:send_interval(?MODULE:window_step_ms(), rotate), - {ok, #state{timer_ref = TimerRef, ets_table = EtsTable}}. - -handle_call({declare, Name, MetricSpec}, _From, State) -> - case maps:is_key(Name, State#state.metrics) of - true -> - {reply, false, State}; - false -> - NewMetrics = maps:put(Name, #{}, State#state.metrics), - NewSpecs = maps:put(Name, MetricSpec, State#state.metric_specs), - {reply, true, State#state{metrics = NewMetrics, metric_specs = NewSpecs}} - end; -handle_call({value, Name, LabelValues}, _From, State) -> - Result = get_value(Name, LabelValues, State), - {reply, Result, State}; -handle_call({get_label_values, Name}, _From, State) -> - Result = do_get_label_values(Name, State), - {reply, Result, State}; -handle_call({remove, Name, LabelValues}, _From, State) -> - NewState = do_remove(Name, LabelValues, State), - {reply, true, NewState}; -handle_call(get_all_metric_names, _From, State) -> - Names = maps:keys(State#state.metrics), - {reply, Names, State}; -handle_call({get_metric_spec, Name}, _From, State) -> - Spec = maps:get(Name, State#state.metric_specs, undefined), - {reply, Spec, State}; -handle_call(_Request, _From, State) -> - {reply, ok, State}. - -handle_cast({observe, Name, LabelValues, Value}, State) -> - NewState = do_observe(Name, LabelValues, Value, State), - {noreply, NewState}; -handle_cast(_Msg, State) -> +handle_cast(_Message, State) -> {noreply, State}. handle_info(rotate, State) -> - NewState = rotate_all_windows(State), - {noreply, NewState}; + rotate_all_windows(), + {noreply, State}; handle_info(_Info, State) -> {noreply, State}. -terminate(_Reason, #state{timer_ref = {ok, TRef}}) -> - timer:cancel(TRef), - ok; -terminate(_Reason, _State) -> - ok. - -code_change(_OldVsn, State, _Extra) -> - {ok, State}. +terminate(_Reason, #state{timer_ref = TimerRef}) -> + timer:cancel(TimerRef). -%% Child spec for supervision tree +%% API -spec child_spec() -> supervisor:child_spec(). child_spec() -> @@ -171,17 +105,23 @@ child_spec() -> type => worker, modules => [?MODULE]}. -%% Internal functions +-spec start_link() -> gen_server:start_ret(). +start_link() -> + gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). + +-spec declare(proplists:proplist()) -> boolean(). +declare(Spec) -> + Name = proplists:get_value(name, Spec), + ets:insert_new(?METRICS_TABLE, #metric{name = Name, spec = Spec}). --spec do_observe(name(), label_values(), number(), #state{}) -> #state{}. -do_observe(Name, LabelValues, Value, State) -> - case maps:find(Name, State#state.metrics) of - error -> - erlang:error({unknown_metric, Name}); - {ok, MetricState} -> - Key = {Name, LabelValues}, - {Windows, CurrentIndex} = - ensure_windows_initialized(Key, State), +-spec observe(name(), label_values(), number()) -> ok. +observe(Name, LabelValues, Value) -> + case ets:lookup(?METRICS_TABLE, Name) of + [] -> + % TODO: log error + ok; + [Metric] -> + {Windows, CurrentIndex} = ensure_windows_initialized(LabelValues, Metric), %% Get current window and add observation CurrentWindow = lists:nth(CurrentIndex + 1, Windows), @@ -191,161 +131,152 @@ do_observe(Name, LabelValues, Value, State) -> %% Update windows list and metric state UpdatedWindows = set_nth(CurrentIndex + 1, UpdatedWindow, Windows), - UpdatedMetricState = maps:put(Key, - #{windows => UpdatedWindows, - current_index => CurrentIndex}, - MetricState), + UpdatedMetricState = maps:put(LabelValues, + #{windows => UpdatedWindows, current_index => CurrentIndex}, + Metric#metric.states), - UpdatedMetrics = maps:put(Name, UpdatedMetricState, State#state.metrics), - State#state{metrics = UpdatedMetrics} + ets:insert(?METRICS_TABLE, Metric#metric{states = UpdatedMetricState}), + ok end. --spec ensure_windows_initialized({name(), label_values()}, #state{}) -> {[window_data()], non_neg_integer()}. -ensure_windows_initialized(Key, State) -> - WindowCount = ?MODULE:window_count(), - case get_metric_state(Key, State) of - undefined -> - Windows = [new_window(Key, Index, State) || Index <- lists:seq(0, WindowCount - 1)], - {Windows, 0}; - #{windows := Windows, current_index := CurrentIndex} -> - {Windows, CurrentIndex} +-spec value(name(), label_values()) -> undefined | {non_neg_integer(), number(), [{number(), number()}]}. +value(Name, LabelValues) -> + case get_metric_state(Name, LabelValues) of + undefined -> undefined; + State -> get_state_value(State) end. --spec get_value(name(), label_values(), #state{}) -> undefined | {non_neg_integer(), number(), [{number(), number()}]}. -get_value(Name, LabelValues, State) -> - Key = {Name, LabelValues}, - case get_metric_state(Key, State) of - undefined -> - undefined; - #{windows := Windows} -> - WindowTuples = [window_tuple(Window) || Window <- Windows], - Merged = lists:foldl(fun - (undefined, Acc) -> Acc; - (Val, undefined) -> Val; - (Val, Acc) -> ddskerl_ets:merge_tuples(Acc, Val) - end, undefined, WindowTuples), - case Merged of - undefined -> - undefined; - _ -> - Count = ddskerl_ets:total_tuple(Merged), - case Count of - 0 -> - undefined; - _ -> - Sum = ddskerl_ets:sum_tuple(Merged), - Quantiles = [{Q, ddskerl_ets:quantile_tuple(Merged, Q)} || Q <- default_quantiles()], - {Count, Sum, Quantiles} - end - end +-spec values(name()) -> [{label_values(), {non_neg_integer(), number(), [{number(), number()}]}}]. +values(Name) -> + case ets:lookup(?METRICS_TABLE, Name) of + [] -> + []; + [#metric{states = States}] -> + [{LabelValues, get_state_value(State)} || LabelValues := State <- States] end. --spec do_get_label_values(name(), #state{}) -> [label_values()]. -do_get_label_values(Name, State) -> - case State#state.metrics of - #{Name := MetricState} -> - [LabelValues || {_, LabelValues} <- maps:keys(MetricState)]; - #{} -> - [] - end. - --spec do_remove(name(), label_values(), #state{}) -> #state{}. -do_remove(Name, LabelValues, State) -> - Key = {Name, LabelValues}, - case maps:find(Name, State#state.metrics) of - error -> - State; - {ok, MetricState} -> - UpdatedMetricState = remove_metric_state(Key, MetricState), - UpdatedMetrics = maps:put(Name, UpdatedMetricState, State#state.metrics), - State#state{metrics = UpdatedMetrics} +-spec remove(name(), label_values()) -> boolean(). +remove(Name, LabelValues) -> + case ets:lookup(?METRICS_TABLE, Name) of + [] -> + true; + [#metric{states = States} = Metric] -> + States1 = remove_state(LabelValues, States), + ets:insert(?METRICS_TABLE, Metric#metric{states = States1}) end. + +%% -spec get_all_metrics() -> [{name(), #{label_values() => metric_state()}}]. +get_all_metrics() -> + ets:foldl(fun(#metric{name = Name, states = States}, Acc) -> + Values = [{LabelValues, get_state_value(State)} || LabelValues := State <- States], + [{Name, Values} | Acc] + end, + [], + ?METRICS_TABLE). --spec get_metric_state({name(), label_values()}, #state{}) -> undefined | metric_state(). -get_metric_state({Name, LabelValues}, State) -> - case State#state.metrics of - #{Name := MetricState} -> - maps:get({Name, LabelValues}, MetricState, undefined); - #{} -> - undefined - end. - --spec set_nth(pos_integer(), T, [T]) -> [T]. -set_nth(1, Value, [_ | Rest]) -> - [Value | Rest]; -set_nth(N, Value, [H | Rest]) when N > 1 -> - [H | set_nth(N - 1, Value, Rest)]. +-spec get_metric_spec(name()) -> proplists:proplist() | undefined. +get_metric_spec(Name) -> + case ets:lookup(?METRICS_TABLE, Name) of + [] -> undefined; + [#metric{spec = Spec}] -> Spec + end. -%% Window helpers +%% Private + +rotate_all_windows() -> + WindowCount = window_count(), + ets:foldl( + fun(#metric{states = States} = Metric, _Acc) -> + States1 = maps:map(rotate_windows(WindowCount), States), + ets:insert(?METRICS_TABLE, Metric#metric{states = States1}) + end, + ok, + ?METRICS_TABLE + ). + +rotate_windows(WindowCount) -> + fun(_, #{windows := Windows, current_index := CurrentIndex}) -> + NewIndex = (CurrentIndex + 1) rem WindowCount, + WindowToReset = lists:nth(NewIndex + 1, Windows), + ResetWindow = reset_window(WindowToReset), + UpdatedWindows = set_nth(NewIndex + 1, ResetWindow, Windows), + #{windows => UpdatedWindows, current_index => NewIndex} + end. --spec metric_options(name(), #state{}) -> {number(), non_neg_integer()}. -metric_options(Name, State) -> - MetricSpec = maps:get(Name, State#state.metric_specs, []), - Error = proplists:get_value(error, MetricSpec, default_error()), - Bound = proplists:get_value(bound, MetricSpec, default_bound()), - {Error, Bound}. +reset_window(#{sketch := Sketch} = Window) -> + ResetSketch = ddskerl_ets:reset(Sketch), + Window#{sketch := ResetSketch}. --spec window_name({name(), label_values()}, non_neg_integer()) -> term(). -window_name({Name, LabelValues}, Index) -> - {Name, LabelValues, Index}. +ensure_windows_initialized(LabelValues, #metric{states = States} = Metric) -> + case maps:get(LabelValues, States, undefined) of + undefined -> + Windows = [new_window(LabelValues, Metric, Index) || Index <- lists:seq(0, window_count() - 1)], + {Windows, 0}; + #{windows := Windows, current_index := CurrentIndex} -> + {Windows, CurrentIndex} + end. --spec new_window({name(), label_values()}, non_neg_integer(), #state{}) -> window_data(). -new_window(Key = {Name, _LabelValues}, Index, State) -> - {Error, Bound} = metric_options(Name, State), - WindowName = window_name(Key, Index), - Sketch = ddskerl_ets:new(#{ets_table => State#state.ets_table, +new_window(LabelValues, #metric{name = Name, spec = Spec}, Index) -> + WindowName = {Name, LabelValues, Index}, + {Error, Bound} = metric_options(Spec), + Sketch = ddskerl_ets:new(#{ets_table => ?DDSKERL_TABLE, name => WindowName, error => Error, bound => Bound}), #{sketch => Sketch, - ref => State#state.ets_table, name => WindowName}. --spec rotate_all_windows(#state{}) -> #state{}. -rotate_all_windows(State) -> - WindowCount = ?MODULE:window_count(), - UpdatedMetrics = maps:map( - fun(_Name, MetricState) -> - maps:map( - fun(_Key, #{windows := Windows, current_index := CurrentIndex}) -> - %% Move to next window and reset it - NewIndex = (CurrentIndex + 1) rem WindowCount, - WindowToReset = lists:nth(NewIndex + 1, Windows), - ResetWindow = reset_window(WindowToReset), - UpdatedWindows = set_nth(NewIndex + 1, ResetWindow, Windows), - #{windows => UpdatedWindows, current_index => NewIndex} - end, MetricState) - end, State#state.metrics), - State#state{metrics = UpdatedMetrics}. - --spec reset_window(window_data()) -> window_data(). -reset_window(WindowData) -> - Sketch = maps:get(sketch, WindowData), - ResetSketch = ddskerl_ets:reset(Sketch), - WindowData#{sketch := ResetSketch}. +metric_options(Spec) -> + Error = proplists:get_value(error, Spec, default_error()), + Bound = proplists:get_value(bound, Spec, default_bound()), + {Error, Bound}. + +set_nth(1, Value, [_ | Rest]) -> + [Value | Rest]; +set_nth(N, Value, [H | Rest]) when N > 1 -> + [H | set_nth(N - 1, Value, Rest)]. + +get_metric_state(Name, LabelValues) -> + case ets:lookup(?METRICS_TABLE, Name) of + [] -> + undefined; + [#metric{states = States}] -> + maps:get(LabelValues, States, undefined) + end. + +get_state_value(#{windows := Windows}) -> + WindowTuples = [window_tuple(Window) || Window <- Windows], + Merged = lists:foldl(fun(undefined, Acc) -> Acc; + (Val, undefined) -> Val; + (Val, Acc) -> ddskerl_ets:merge_tuples(Acc, Val) + end, undefined, WindowTuples), + case Merged of + undefined -> + undefined; + _ -> + Count = ddskerl_ets:total_tuple(Merged), + case Count of + 0 -> + undefined; + _ -> + Sum = ddskerl_ets:sum_tuple(Merged), + Quantiles = [{Q, ddskerl_ets:quantile_tuple(Merged, Q)} || Q <- default_quantiles()], + {Count, Sum, Quantiles} + end + end. -spec window_tuple(window_data()) -> ddskerl_ets:object() | undefined. -window_tuple(WindowData) -> - Ref = maps:get(ref, WindowData), - Name = maps:get(name, WindowData), - case ets:lookup(Ref, Name) of +window_tuple(#{name := Name}) -> + case ets:lookup(?DDSKERL_TABLE, Name) of [Val] -> Val; [] -> undefined end. --spec remove_metric_state({name(), label_values()}, #{}) -> #{}. -remove_metric_state(Key, MetricState) -> - case maps:find(Key, MetricState) of +remove_state(LabelValues, States) -> + case maps:take(LabelValues, States) of + {#{windows := Windows}, States1} -> + [ets:delete(Ref, Name) || #{ref := Ref, name := Name} <- Windows], + States1; error -> - MetricState; - {ok, #{windows := Windows}} -> - lists:foreach(fun(Window) -> delete_window(Window) end, Windows), - maps:remove(Key, MetricState) + States end. - --spec delete_window(window_data()) -> ok. -delete_window(WindowData) -> - Ref = maps:get(ref, WindowData), - Name = maps:get(name, WindowData), - ets:delete(Ref, Name), - ok. diff --git a/src/instrument/mongoose_prometheus_sliding_window_collector.erl b/src/instrument/mongoose_prometheus_sliding_window_collector.erl index 7dd46613338..6dfe70f7daf 100644 --- a/src/instrument/mongoose_prometheus_sliding_window_collector.erl +++ b/src/instrument/mongoose_prometheus_sliding_window_collector.erl @@ -11,40 +11,29 @@ deregister_cleanup(_) -> ok. collect_mf(_Registry, Callback) -> - %% Get all metric names from the sliding window manager - MetricNames = mongoose_prometheus_sliding_window:get_all_metric_names(), - lists:foreach(fun(Name) -> - collect_metric_family(Name, Callback) - end, MetricNames), + Metrics = mongoose_prometheus_sliding_window:get_all_metrics(), + lists:foreach(fun(Metric) -> + collect_metric_family(Metric, Callback) + end, Metrics), ok. %% Internal functions -collect_metric_family(Name, Callback) -> - Values = mongoose_prometheus_sliding_window:values(Name), - case Values of - [] -> - ok; - _ -> - SWName = sliding_window_metric_name(Name), - MetricSpec = mongoose_prometheus_sliding_window:get_metric_spec(Name), - Help = case MetricSpec of - undefined -> ""; - Spec -> proplists:get_value(help, Spec, "") - end, - LabelKeys = case MetricSpec of - undefined -> []; - Spec2 -> proplists:get_value(labels, Spec2, []) - end, - %% Convert to Prometheus format - Metrics = [create_summary_metric(LabelKeys, LabelValues, Count, Sum, Quantiles) - || {LabelValues, {Count, Sum, Quantiles}} <- Values], - %% Create and send the metric family - MF = prometheus_model_helpers:create_mf(SWName, Help, summary, Metrics), - Callback(MF) - end. - -sliding_window_metric_name(Name) -> Name. +collect_metric_family({_Name, []}, _Callback) -> + ok; +collect_metric_family({Name, Values}, Callback) -> + MetricSpec = mongoose_prometheus_sliding_window:get_metric_spec(Name), + Help = get_spec_value(MetricSpec, help, ""), + LabelKeys = get_spec_value(MetricSpec, labels, []), + %% Convert to Prometheus format + Metrics = [create_summary_metric(LabelKeys, LabelValues, Count, Sum, Quantiles) + || {LabelValues, {Count, Sum, Quantiles}} <- Values], + %% Create and send the metric family + MF = prometheus_model_helpers:create_mf(Name, Help, summary, Metrics), + Callback(MF). + +get_spec_value(undefined, _Key, Default) -> Default; +get_spec_value(Spec, Key, Default) -> proplists:get_value(Key, Spec, Default). create_summary_metric(LabelKeys, LabelValues, Count, Sum, Quantiles) -> %% Convert label keys and values to label pairs format