yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor(metrics): rename speed to rate in emqx_plugin_libs_metrics

This commit is contained in:
Shawn 2021-12-03 10:51:39 +08:00
parent e07fce791f
commit 24bded99d5
3 changed files with 70 additions and 70 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
apps/emqx_bridge/src/emqx_bridge_api.erl
View File

@ -40,17 +40,17 @@
#{ matched => MATCH, #{ matched => MATCH,
success => SUCC, success => SUCC,
failed => FAILED, failed => FAILED,
speed => RATE, rate => RATE,
speed_last5m => RATE_5, rate_last5m => RATE_5,
speed_max => RATE_MAX rate_max => RATE_MAX
}). }).
-define(metrics(MATCH, SUCC, FAILED, RATE, RATE_5, RATE_MAX), -define(metrics(MATCH, SUCC, FAILED, RATE, RATE_5, RATE_MAX),
#{ matched := MATCH, #{ matched := MATCH,
success := SUCC, success := SUCC,
failed := FAILED, failed := FAILED,
speed := RATE, rate := RATE,
speed_last5m := RATE_5, rate_last5m := RATE_5,
speed_max := RATE_MAX rate_max := RATE_MAX
}). }).
req_schema() -> req_schema() ->
@ -76,9 +76,9 @@ metrics_schema() ->
matched => #{type => integer, example => "0"}, matched => #{type => integer, example => "0"},
success => #{type => integer, example => "0"}, success => #{type => integer, example => "0"},
failed => #{type => integer, example => "0"}, failed => #{type => integer, example => "0"},
speed => #{type => number, format => float, example => "0.0"}, rate => #{type => number, format => float, example => "0.0"},
speed_last5m => #{type => number, format => float, example => "0.0"}, rate_last5m => #{type => number, format => float, example => "0.0"},
speed_max => #{type => number, format => float, example => "0.0"} rate_max => #{type => number, format => float, example => "0.0"}
} }
}. }.

110
apps/emqx_plugin_libs/src/emqx_plugin_libs_metrics.erl
View File

@ -27,7 +27,7 @@
-export([ inc/3 -export([ inc/3
, inc/4 , inc/4
, get/3 , get/3
, get_speed/2 , get_rate/2
, create_metrics/2 , create_metrics/2
, clear_metrics/2 , clear_metrics/2
]). ]).
@ -54,7 +54,7 @@
-define(SECS_5M, 300). -define(SECS_5M, 300).
-define(SAMPLING, 10). -define(SAMPLING, 10).
-else. -else.
%% Use 5 secs average speed instead of 5 mins in case of testing %% Use 5 secs average rate instead of 5 mins in case of testing
-define(SECS_5M, 5). -define(SECS_5M, 5).
-define(SAMPLING, 1). -define(SAMPLING, 1).
-endif. -endif.
@ -65,9 +65,9 @@
matched => integer(), matched => integer(),
success => integer(), success => integer(),
failed => integer(), failed => integer(),
speed => float(), rate => float(),
speed_max => float(), rate_max => float(),
speed_last5m => float() rate_last5m => float()
}. }.
-type handler_name() :: atom(). -type handler_name() :: atom().
-type metric_id() :: binary(). -type metric_id() :: binary().
@ -75,22 +75,22 @@
-define(CntrRef(Name), {?MODULE, Name}). -define(CntrRef(Name), {?MODULE, Name}).
-define(SAMPCOUNT_5M, (?SECS_5M div ?SAMPLING)). -define(SAMPCOUNT_5M, (?SECS_5M div ?SAMPLING)).
%% the speed of 'matched' %% the rate of 'matched'
-record(speed, { -record(rate, {
max = 0 :: number(), max = 0 :: number(),
current = 0 :: number(), current = 0 :: number(),
last5m = 0 :: number(), last5m = 0 :: number(),
%% metadata for calculating the avg speed %% metadata for calculating the avg rate
tick = 1 :: number(), tick = 1 :: number(),
last_v = 0 :: number(), last_v = 0 :: number(),
%% metadata for calculating the 5min avg speed %% metadata for calculating the 5min avg rate
last5m_acc = 0 :: number(), last5m_acc = 0 :: number(),
last5m_smpl = [] :: list() last5m_smpl = [] :: list()
}). }).
-record(state, { -record(state, {
metric_ids = sets:new(), metric_ids = sets:new(),
speeds :: undefined | #{metric_id() => #speed{}} rates :: undefined | #{metric_id() => #rate{}}
}). }).
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
@ -122,19 +122,19 @@ get(Name, Id, Metric) ->
Ref -> counters:get(Ref, metrics_idx(Metric)) Ref -> counters:get(Ref, metrics_idx(Metric))
end. end.
-spec(get_speed(handler_name(), metric_id()) -> map()). -spec(get_rate(handler_name(), metric_id()) -> map()).
get_speed(Name, Id) -> get_rate(Name, Id) ->
gen_server:call(Name, {get_speed, Id}). gen_server:call(Name, {get_rate, Id}).
-spec(get_metrics(handler_name(), metric_id()) -> metrics()). -spec(get_metrics(handler_name(), metric_id()) -> metrics()).
get_metrics(Name, Id) -> get_metrics(Name, Id) ->
#{max := Max, current := Current, last5m := Last5M} = get_speed(Name, Id), #{max := Max, current := Current, last5m := Last5M} = get_rate(Name, Id),
#{matched => get_matched(Name, Id), #{matched => get_matched(Name, Id),
success => get_success(Name, Id), success => get_success(Name, Id),
failed => get_failed(Name, Id), failed => get_failed(Name, Id),
speed => Current, rate => Current,
speed_max => Max, rate_max => Max,
speed_last5m => Last5M rate_last5m => Last5M
}. }.
-spec inc(handler_name(), metric_id(), atom()) -> ok. -spec inc(handler_name(), metric_id(), atom()) -> ok.
@ -176,35 +176,35 @@ start_link(Name) ->
init(Name) -> init(Name) ->
erlang:process_flag(trap_exit, true), erlang:process_flag(trap_exit, true),
%% the speed metrics %% the rate metrics
erlang:send_after(timer:seconds(?SAMPLING), self(), ticking), erlang:send_after(timer:seconds(?SAMPLING), self(), ticking),
persistent_term:put(?CntrRef(Name), #{}), persistent_term:put(?CntrRef(Name), #{}),
{ok, #state{}}. {ok, #state{}}.
handle_call({get_speed, _Id}, _From, State = #state{speeds = undefined}) -> handle_call({get_rate, _Id}, _From, State = #state{rates = undefined}) ->
{reply, format_speed(#speed{}), State}; {reply, format_rate(#rate{}), State};
handle_call({get_speed, Id}, _From, State = #state{speeds = Speeds}) -> handle_call({get_rate, Id}, _From, State = #state{rates = Rates}) ->
{reply, case maps:get(Id, Speeds, undefined) of {reply, case maps:get(Id, Rates, undefined) of
undefined -> format_speed(#speed{}); undefined -> format_rate(#rate{});
Speed -> format_speed(Speed) Rate -> format_rate(Rate)
end, State}; end, State};
handle_call({create_metrics, Id}, _From, handle_call({create_metrics, Id}, _From,
State = #state{metric_ids = MIDs, speeds = Speeds}) -> State = #state{metric_ids = MIDs, rates = Rates}) ->
{reply, create_counters(get_self_name(), Id), {reply, create_counters(get_self_name(), Id),
State#state{metric_ids = sets:add_element(Id, MIDs), State#state{metric_ids = sets:add_element(Id, MIDs),
speeds = case Speeds of rates = case Rates of
undefined -> #{Id => #speed{}}; undefined -> #{Id => #rate{}};
_ -> Speeds#{Id => #speed{}} _ -> Rates#{Id => #rate{}}
end}}; end}};
handle_call({delete_metrics, Id}, _From, handle_call({delete_metrics, Id}, _From,
State = #state{metric_ids = MIDs, speeds = Speeds}) -> State = #state{metric_ids = MIDs, rates = Rates}) ->
{reply, delete_counters(get_self_name(), Id), {reply, delete_counters(get_self_name(), Id),
State#state{metric_ids = sets:del_element(Id, MIDs), State#state{metric_ids = sets:del_element(Id, MIDs),
speeds = case Speeds of rates = case Rates of
undefined -> undefined; undefined -> undefined;
_ -> maps:remove(Id, Speeds) _ -> maps:remove(Id, Rates)
end}}; end}};
handle_call(_Request, _From, State) -> handle_call(_Request, _From, State) ->
@ -213,17 +213,17 @@ handle_call(_Request, _From, State) ->
handle_cast(_Msg, State) -> handle_cast(_Msg, State) ->
{noreply, State}. {noreply, State}.
handle_info(ticking, State = #state{speeds = undefined}) -> handle_info(ticking, State = #state{rates = undefined}) ->
erlang:send_after(timer:seconds(?SAMPLING), self(), ticking), erlang:send_after(timer:seconds(?SAMPLING), self(), ticking),
{noreply, State}; {noreply, State};
handle_info(ticking, State = #state{speeds = Speeds0}) -> handle_info(ticking, State = #state{rates = Rates0}) ->
Speeds = maps:map( Rates = maps:map(
fun(Id, Speed) -> fun(Id, Rate) ->
calculate_speed(get_matched(get_self_name(), Id), Speed) calculate_rate(get_matched(get_self_name(), Id), Rate)
end, Speeds0), end, Rates0),
erlang:send_after(timer:seconds(?SAMPLING), self(), ticking), erlang:send_after(timer:seconds(?SAMPLING), self(), ticking),
{noreply, State#state{speeds = Speeds}}; {noreply, State#state{rates = Rates}};
handle_info(_Info, State) -> handle_info(_Info, State) ->
{noreply, State}. {noreply, State}.
@ -261,38 +261,38 @@ get_couters_ref(Name, Id) ->
get_all_counters(Name) -> get_all_counters(Name) ->
persistent_term:get(?CntrRef(Name), #{}). persistent_term:get(?CntrRef(Name), #{}).
calculate_speed(_CurrVal, undefined) -> calculate_rate(_CurrVal, undefined) ->
undefined; undefined;
calculate_speed(CurrVal, #speed{max = MaxSpeed0, last_v = LastVal, calculate_rate(CurrVal, #rate{max = MaxRate0, last_v = LastVal,
tick = Tick, last5m_acc = AccSpeed5Min0, tick = Tick, last5m_acc = AccRate5Min0,
last5m_smpl = Last5MinSamples0}) -> last5m_smpl = Last5MinSamples0}) ->
%% calculate the current speed based on the last value of the counter %% calculate the current rate based on the last value of the counter
CurrSpeed = (CurrVal - LastVal) / ?SAMPLING, CurrRate = (CurrVal - LastVal) / ?SAMPLING,
%% calculate the max speed since the emqx startup %% calculate the max rate since the emqx startup
MaxSpeed = MaxRate =
if MaxSpeed0 >= CurrSpeed -> MaxSpeed0; if MaxRate0 >= CurrRate -> MaxRate0;
true -> CurrSpeed true -> CurrRate
end, end,
%% calculate the average speed in last 5 mins %% calculate the average rate in last 5 mins
{Last5MinSamples, Acc5Min, Last5Min} = {Last5MinSamples, Acc5Min, Last5Min} =
if Tick =< ?SAMPCOUNT_5M -> if Tick =< ?SAMPCOUNT_5M ->
Acc = AccSpeed5Min0 + CurrSpeed, Acc = AccRate5Min0 + CurrRate,
{lists:reverse([CurrSpeed | lists:reverse(Last5MinSamples0)]), {lists:reverse([CurrRate | lists:reverse(Last5MinSamples0)]),
Acc, Acc / Tick}; Acc, Acc / Tick};
true -> true ->
[FirstSpeed | Speeds] = Last5MinSamples0, [FirstRate | Rates] = Last5MinSamples0,
Acc = AccSpeed5Min0 + CurrSpeed - FirstSpeed, Acc = AccRate5Min0 + CurrRate - FirstRate,
{lists:reverse([CurrSpeed | lists:reverse(Speeds)]), {lists:reverse([CurrRate | lists:reverse(Rates)]),
Acc, Acc / ?SAMPCOUNT_5M} Acc, Acc / ?SAMPCOUNT_5M}
end, end,
#speed{max = MaxSpeed, current = CurrSpeed, last5m = Last5Min, #rate{max = MaxRate, current = CurrRate, last5m = Last5Min,
last_v = CurrVal, last5m_acc = Acc5Min, last_v = CurrVal, last5m_acc = Acc5Min,
last5m_smpl = Last5MinSamples, tick = Tick + 1}. last5m_smpl = Last5MinSamples, tick = Tick + 1}.
format_speed(#speed{max = Max, current = Current, last5m = Last5Min}) -> format_rate(#rate{max = Max, current = Current, last5m = Last5Min}) ->
#{max => Max, current => precision(Current, 2), last5m => precision(Last5Min, 2)}. #{max => Max, current => precision(Current, 2), last5m => precision(Last5Min, 2)}.
precision(Float, N) -> precision(Float, N) ->

12
apps/emqx_rule_engine/src/emqx_rule_engine_api.erl
View File

@ -339,14 +339,14 @@ do_format_output(BridgeChannelId) when is_binary(BridgeChannelId) ->
get_rule_metrics(Id) -> get_rule_metrics(Id) ->
Format = fun (Node, #{matched := Matched, Format = fun (Node, #{matched := Matched,
speed := Current, rate := Current,
speed_max := Max, rate_max := Max,
speed_last5m := Last5M rate_last5m := Last5M
}) -> }) ->
#{ matched => Matched #{ matched => Matched
, speed => Current , rate => Current
, speed_max => Max , rate_max => Max
, speed_last5m => Last5M , rate_last5m => Last5M
, node => Node , node => Node
} }
end, end,