chore(slow-subs): change root name emqx_slow_subs > slow_subs
This commit is contained in:
parent
19dd5a7c97
commit
9a03869bd7
|
@ -2,7 +2,7 @@
|
||||||
## EMQ X Slow Subscribers Statistics
|
## EMQ X Slow Subscribers Statistics
|
||||||
##--------------------------------------------------------------------
|
##--------------------------------------------------------------------
|
||||||
|
|
||||||
emqx_slow_subs {
|
slow_subs {
|
||||||
enable = false
|
enable = false
|
||||||
|
|
||||||
threshold = 500ms
|
threshold = 500ms
|
||||||
|
|
|
@ -124,7 +124,7 @@ clear_history() ->
|
||||||
gen_server:call(?MODULE, ?FUNCTION_NAME, ?DEF_CALL_TIMEOUT).
|
gen_server:call(?MODULE, ?FUNCTION_NAME, ?DEF_CALL_TIMEOUT).
|
||||||
|
|
||||||
update_settings(Conf) ->
|
update_settings(Conf) ->
|
||||||
emqx_conf:update([emqx_slow_subs], Conf, #{override_to => cluster}).
|
emqx_conf:update([slow_subs], Conf, #{override_to => cluster}).
|
||||||
|
|
||||||
init_topk_tab() ->
|
init_topk_tab() ->
|
||||||
case ets:whereis(?TOPK_TAB) of
|
case ets:whereis(?TOPK_TAB) of
|
||||||
|
@ -146,7 +146,7 @@ post_config_update(_KeyPath, _UpdateReq, NewConf, _OldConf, _AppEnvs) ->
|
||||||
%%--------------------------------------------------------------------
|
%%--------------------------------------------------------------------
|
||||||
|
|
||||||
init([]) ->
|
init([]) ->
|
||||||
emqx_conf:add_handler([emqx_slow_subs], ?MODULE),
|
emqx_conf:add_handler([slow_subs], ?MODULE),
|
||||||
|
|
||||||
InitState = #{enable => false,
|
InitState = #{enable => false,
|
||||||
last_tick_at => 0,
|
last_tick_at => 0,
|
||||||
|
@ -154,11 +154,11 @@ init([]) ->
|
||||||
notice_timer => undefined
|
notice_timer => undefined
|
||||||
},
|
},
|
||||||
|
|
||||||
Enable = emqx:get_config([emqx_slow_subs, enable]),
|
Enable = emqx:get_config([slow_subs, enable]),
|
||||||
{ok, check_enable(Enable, InitState)}.
|
{ok, check_enable(Enable, InitState)}.
|
||||||
|
|
||||||
handle_call({update_settings, #{enable := Enable} = Conf}, _From, State) ->
|
handle_call({update_settings, #{enable := Enable} = Conf}, _From, State) ->
|
||||||
emqx_config:put([emqx_slow_subs], Conf),
|
emqx_config:put([slow_subs], Conf),
|
||||||
State2 = check_enable(Enable, State),
|
State2 = check_enable(Enable, State),
|
||||||
{reply, ok, State2};
|
{reply, ok, State2};
|
||||||
|
|
||||||
|
@ -204,7 +204,7 @@ expire_tick() ->
|
||||||
erlang:send_after(?EXPIRE_CHECK_INTERVAL, self(), ?FUNCTION_NAME).
|
erlang:send_after(?EXPIRE_CHECK_INTERVAL, self(), ?FUNCTION_NAME).
|
||||||
|
|
||||||
notice_tick() ->
|
notice_tick() ->
|
||||||
case emqx:get_config([emqx_slow_subs, notice_interval]) of
|
case emqx:get_config([slow_subs, notice_interval]) of
|
||||||
0 -> undefined;
|
0 -> undefined;
|
||||||
Interval ->
|
Interval ->
|
||||||
erlang:send_after(Interval, self(), ?FUNCTION_NAME)
|
erlang:send_after(Interval, self(), ?FUNCTION_NAME)
|
||||||
|
@ -225,7 +225,7 @@ do_publish([], _, _) ->
|
||||||
ok;
|
ok;
|
||||||
|
|
||||||
do_publish(Logs, Rank, TickTime) ->
|
do_publish(Logs, Rank, TickTime) ->
|
||||||
BatchSize = emqx:get_config([emqx_slow_subs, notice_batch_size]),
|
BatchSize = emqx:get_config([slow_subs, notice_batch_size]),
|
||||||
do_publish(Logs, BatchSize, Rank, TickTime, []).
|
do_publish(Logs, BatchSize, Rank, TickTime, []).
|
||||||
|
|
||||||
do_publish([Log | T], Size, Rank, TickTime, Cache) when Size > 0 ->
|
do_publish([Log | T], Size, Rank, TickTime, Cache) when Size > 0 ->
|
||||||
|
@ -254,7 +254,7 @@ publish(TickTime, Notices) ->
|
||||||
logs => lists:reverse(Notices)},
|
logs => lists:reverse(Notices)},
|
||||||
Payload = emqx_json:encode(WindowLog),
|
Payload = emqx_json:encode(WindowLog),
|
||||||
Msg = #message{ id = emqx_guid:gen()
|
Msg = #message{ id = emqx_guid:gen()
|
||||||
, qos = emqx:get_config([emqx_slow_subs, notice_qos])
|
, qos = emqx:get_config([slow_subs, notice_qos])
|
||||||
, from = ?MODULE
|
, from = ?MODULE
|
||||||
, topic = emqx_topic:systop(?NOTICE_TOPIC_NAME)
|
, topic = emqx_topic:systop(?NOTICE_TOPIC_NAME)
|
||||||
, payload = Payload
|
, payload = Payload
|
||||||
|
@ -264,7 +264,7 @@ publish(TickTime, Notices) ->
|
||||||
ok.
|
ok.
|
||||||
|
|
||||||
load(State) ->
|
load(State) ->
|
||||||
MaxSizeT = emqx:get_config([emqx_slow_subs, top_k_num]),
|
MaxSizeT = emqx:get_config([slow_subs, top_k_num]),
|
||||||
MaxSize = erlang:min(MaxSizeT, ?MAX_TAB_SIZE),
|
MaxSize = erlang:min(MaxSizeT, ?MAX_TAB_SIZE),
|
||||||
_ = emqx:hook('message.slow_subs_stats',
|
_ = emqx:hook('message.slow_subs_stats',
|
||||||
{?MODULE, on_stats_update, [#{max_size => MaxSize}]}
|
{?MODULE, on_stats_update, [#{max_size => MaxSize}]}
|
||||||
|
@ -283,7 +283,7 @@ unload(#{notice_timer := NoticeTimer, expire_timer := ExpireTimer} = State) ->
|
||||||
|
|
||||||
do_clear(Logs) ->
|
do_clear(Logs) ->
|
||||||
Now = ?NOW,
|
Now = ?NOW,
|
||||||
Interval = emqx:get_config([emqx_slow_subs, expire_interval]),
|
Interval = emqx:get_config([slow_subs, expire_interval]),
|
||||||
Each = fun(#top_k{index = Index, last_update_time = Ts}) ->
|
Each = fun(#top_k{index = Index, last_update_time = Ts}) ->
|
||||||
case Now - Ts >= Interval of
|
case Now - Ts >= Interval of
|
||||||
true ->
|
true ->
|
||||||
|
@ -330,7 +330,7 @@ check_enable(Enable, #{enable := IsEnable} = State) ->
|
||||||
end.
|
end.
|
||||||
|
|
||||||
update_threshold() ->
|
update_threshold() ->
|
||||||
Threshold = emqx:get_config([emqx_slow_subs, threshold]),
|
Threshold = emqx:get_config([slow_subs, threshold]),
|
||||||
emqx_message_latency_stats:update_threshold(Threshold),
|
emqx_message_latency_stats:update_threshold(Threshold),
|
||||||
ok.
|
ok.
|
||||||
|
|
||||||
|
|
|
@ -80,7 +80,7 @@ fields(record) ->
|
||||||
].
|
].
|
||||||
|
|
||||||
conf_schema() ->
|
conf_schema() ->
|
||||||
Ref = hoconsc:ref(emqx_slow_subs_schema, "emqx_slow_subs"),
|
Ref = hoconsc:ref(emqx_slow_subs_schema, "slow_subs"),
|
||||||
hoconsc:mk(Ref, #{}).
|
hoconsc:mk(Ref, #{}).
|
||||||
|
|
||||||
slow_subs(delete, _) ->
|
slow_subs(delete, _) ->
|
||||||
|
@ -104,8 +104,8 @@ encode_record(#top_k{index = ?INDEX(Latency, ClientId),
|
||||||
last_update_time => Ts}.
|
last_update_time => Ts}.
|
||||||
|
|
||||||
settings(get, _) ->
|
settings(get, _) ->
|
||||||
{200, emqx:get_raw_config([?APP_NAME], #{})};
|
{200, emqx:get_raw_config([slow_subs], #{})};
|
||||||
|
|
||||||
settings(put, #{body := Body}) ->
|
settings(put, #{body := Body}) ->
|
||||||
_ = emqx_slow_subs:update_settings(Body),
|
_ = emqx_slow_subs:update_settings(Body),
|
||||||
{200, emqx:get_raw_config([?APP_NAME], #{})}.
|
{200, emqx:get_raw_config([slow_subs], #{})}.
|
||||||
|
|
|
@ -4,9 +4,9 @@
|
||||||
|
|
||||||
-export([roots/0, fields/1]).
|
-export([roots/0, fields/1]).
|
||||||
|
|
||||||
roots() -> ["emqx_slow_subs"].
|
roots() -> ["slow_subs"].
|
||||||
|
|
||||||
fields("emqx_slow_subs") ->
|
fields("slow_subs") ->
|
||||||
[ {enable, sc(boolean(), false, "switch of this function")}
|
[ {enable, sc(boolean(), false, "switch of this function")}
|
||||||
, {threshold,
|
, {threshold,
|
||||||
sc(emqx_schema:duration_ms(),
|
sc(emqx_schema:duration_ms(),
|
||||||
|
|
|
@ -27,7 +27,7 @@
|
||||||
-define(NOW, erlang:system_time(millisecond)).
|
-define(NOW, erlang:system_time(millisecond)).
|
||||||
|
|
||||||
-define(BASE_CONF, <<"""
|
-define(BASE_CONF, <<"""
|
||||||
emqx_slow_subs {
|
slow_subs {
|
||||||
enable = true
|
enable = true
|
||||||
top_k_num = 5,
|
top_k_num = 5,
|
||||||
expire_interval = 3000
|
expire_interval = 3000
|
||||||
|
|
|
@ -35,7 +35,7 @@
|
||||||
-define(CLUSTER_RPC_SHARD, emqx_cluster_rpc_shard).
|
-define(CLUSTER_RPC_SHARD, emqx_cluster_rpc_shard).
|
||||||
|
|
||||||
-define(CONF_DEFAULT, <<"""
|
-define(CONF_DEFAULT, <<"""
|
||||||
emqx_slow_subs
|
slow_subs
|
||||||
{
|
{
|
||||||
enable = true
|
enable = true
|
||||||
top_k_num = 5,
|
top_k_num = 5,
|
||||||
|
@ -121,7 +121,7 @@ t_clear(_) ->
|
||||||
?assertEqual(0, ets:info(?TOPK_TAB, size)).
|
?assertEqual(0, ets:info(?TOPK_TAB, size)).
|
||||||
|
|
||||||
t_settting(_) ->
|
t_settting(_) ->
|
||||||
Conf = emqx:get_config([emqx_slow_subs]),
|
Conf = emqx:get_config([slow_subs]),
|
||||||
Conf2 = Conf#{threshold => 1000},
|
Conf2 = Conf#{threshold => 1000},
|
||||||
{ok, Data} = request_api(put,
|
{ok, Data} = request_api(put,
|
||||||
api_path(["slow_subscriptions", "settings"]),
|
api_path(["slow_subscriptions", "settings"]),
|
||||||
|
|
Loading…
Reference in New Issue