diff --git a/CHANGES-5.0.md b/CHANGES-5.0.md index fe586f1fb..a5c2ffa56 100644 --- a/CHANGES-5.0.md +++ b/CHANGES-5.0.md @@ -9,6 +9,7 @@ * Check ACLs for last will testament topic before publishing the message. [#8930](https://github.com/emqx/emqx/pull/8930) * Fix GET /listeners API crash When some nodes still in initial configuration. [#9002](https://github.com/emqx/emqx/pull/9002) * Fix empty variable interpolation in authentication and authorization. Placeholders for undefined variables are rendered now as empty strings and do not cause errors anymore. [#8963](https://github.com/emqx/emqx/pull/8963) +* Fix the latency statistics error of the slow subscription module when `stats_type` is `internal` or `response`. [#8986](https://github.com/emqx/emqx/pull/8986) # 5.0.8 diff --git a/apps/emqx/src/emqx_session.erl b/apps/emqx/src/emqx_session.erl index 71ffeb24e..8ce8a1802 100644 --- a/apps/emqx/src/emqx_session.erl +++ b/apps/emqx/src/emqx_session.erl @@ -892,7 +892,7 @@ on_delivery_completed( ). mark_begin_deliver(Msg) -> - emqx_message:set_header(deliver_begin_at, erlang:system_time(second), Msg). + emqx_message:set_header(deliver_begin_at, erlang:system_time(millisecond), Msg). %%-------------------------------------------------------------------- %% Helper functions diff --git a/apps/emqx_slow_subs/src/emqx_slow_subs.app.src b/apps/emqx_slow_subs/src/emqx_slow_subs.app.src index e87d293d3..866655b61 100644 --- a/apps/emqx_slow_subs/src/emqx_slow_subs.app.src +++ b/apps/emqx_slow_subs/src/emqx_slow_subs.app.src @@ -1,7 +1,7 @@ {application, emqx_slow_subs, [ {description, "EMQX Slow Subscribers Statistics"}, % strict semver, bump manually! - {vsn, "1.0.1"}, + {vsn, "1.0.2"}, {modules, []}, {registered, [emqx_slow_subs_sup]}, {applications, [kernel, stdlib, emqx]}, diff --git a/apps/emqx_slow_subs/src/emqx_slow_subs.erl b/apps/emqx_slow_subs/src/emqx_slow_subs.erl index 2a07518b3..9a82abeb9 100644 --- a/apps/emqx_slow_subs/src/emqx_slow_subs.erl +++ b/apps/emqx_slow_subs/src/emqx_slow_subs.erl @@ -166,11 +166,11 @@ init([]) -> expire_timer => undefined }, - Enable = emqx:get_config([slow_subs, enable]), - {ok, check_enable(Enable, InitState)}. + Cfg = emqx:get_config([slow_subs]), + {ok, check_enable(Cfg, InitState)}. -handle_call({update_settings, #{enable := Enable}}, _From, State) -> - State2 = check_enable(Enable, State), +handle_call({update_settings, Cfg}, _From, State) -> + State2 = check_enable(Cfg, State), {reply, ok, State2}; handle_call(clear_history, _, State) -> do_clear_history(), @@ -206,12 +206,14 @@ code_change(_OldVsn, State, _Extra) -> expire_tick() -> erlang:send_after(?EXPIRE_CHECK_INTERVAL, self(), ?FUNCTION_NAME). -load(State) -> +load( #{ top_k_num := MaxSizeT, stats_type := StatsType, threshold := Threshold - } = emqx:get_config([slow_subs]), + }, + State +) -> MaxSize = erlang:min(MaxSizeT, ?MAX_SIZE), ok = emqx_hooks:put( 'delivery.completed', @@ -334,15 +336,15 @@ do_clear_history() -> ets:delete_all_objects(?INDEX_TAB), ets:delete_all_objects(?TOPK_TAB). -check_enable(Enable, #{enable := IsEnable} = State) -> +check_enable(#{enable := Enable} = Cfg, #{enable := IsEnable} = State) -> case {IsEnable, Enable} of {false, true} -> - load(State); + load(Cfg, State); {true, false} -> unload(State); {true, true} -> S1 = unload(State), - load(S1); + load(Cfg, S1); _ -> State end. diff --git a/apps/emqx_slow_subs/test/emqx_slow_subs_SUITE.erl b/apps/emqx_slow_subs/test/emqx_slow_subs_SUITE.erl index 0547eb1f8..f7ef78ed8 100644 --- a/apps/emqx_slow_subs/test/emqx_slow_subs_SUITE.erl +++ b/apps/emqx_slow_subs/test/emqx_slow_subs_SUITE.erl @@ -26,13 +26,15 @@ -define(NOW, erlang:system_time(millisecond)). -define(CLUSTER_RPC_SHARD, emqx_cluster_rpc_shard). +-define(LANTENCY, 101). -define(BASE_CONF, << "" "\n" "slow_subs {\n" " enable = true\n" - " top_k_num = 5,\n" + " top_k_num = 5\n" + " threshold = 100ms\n" " expire_interval = 5m\n" " stats_type = whole\n" " }" @@ -64,10 +66,10 @@ end_per_suite(_Config) -> init_per_testcase(t_expire, Config) -> {ok, _} = emqx_cluster_rpc:start_link(), - Cfg = emqx_config:get([slow_subs]), - emqx_slow_subs:update_settings(Cfg#{expire_interval := 1500}), + update_config(<<"expire_interval">>, <<"1500ms">>), Config; init_per_testcase(_, Config) -> + {ok, _} = emqx_cluster_rpc:start_link(), Config. end_per_testcase(_, _) -> @@ -84,38 +86,7 @@ end_per_testcase(_, _) -> %% Test Cases %%-------------------------------------------------------------------- t_pub(_) -> - %% Sub topic first - Subs = [{<<"/test1/+">>, ?QOS_1}, {<<"/test2/+">>, ?QOS_2}], - Clients = start_client(Subs), - timer:sleep(1000), - Now = ?NOW, - %% publish - - lists:foreach( - fun(I) -> - Topic = list_to_binary(io_lib:format("/test1/~p", [I])), - Msg = emqx_message:make(undefined, ?QOS_1, Topic, <<"Hello">>), - emqx:publish(Msg#message{timestamp = Now - 500}), - timer:sleep(100) - end, - lists:seq(1, 10) - ), - - lists:foreach( - fun(I) -> - Topic = list_to_binary(io_lib:format("/test2/~p", [I])), - Msg = emqx_message:make(undefined, ?QOS_2, Topic, <<"Hello">>), - emqx:publish(Msg#message{timestamp = Now - 500}), - timer:sleep(100) - end, - lists:seq(1, 10) - ), - - timer:sleep(1000), - Size = ets:info(?TOPK_TAB, size), - ?assert(Size =< 10 andalso Size >= 3, io_lib:format("the size is :~p~n", [Size])), - - [Client ! stop || Client <- Clients], + _ = [stats_with_type(Type) || Type <- [whole, internal, response]], ok. t_expire(_) -> @@ -135,16 +106,12 @@ t_expire(_) -> ?assertEqual(0, Size), ok. -start_client(Subs) -> - [spawn(fun() -> client(I, Subs) end) || I <- lists:seq(1, 10)]. +start_client(Type, Subs) -> + [spawn(fun() -> client(I, Type, Subs) end) || I <- lists:seq(1, 10)]. -client(I, Subs) -> - {ok, C} = emqtt:start_link([ - {host, "localhost"}, - {clientid, io_lib:format("slow_subs_~p", [I])}, - {username, <<"plain">>}, - {password, <<"plain">>} - ]), +client(I, Type, Subs) -> + ConnOptions = make_conn_options(Type, I), + {ok, C} = emqtt:start_link(ConnOptions), {ok, _} = emqtt:connect(C), Len = erlang:length(Subs), @@ -155,3 +122,81 @@ client(I, Subs) -> stop -> ok end. + +stats_with_type(Type) -> + emqx_slow_subs:clear_history(), + update_stats_type(Type), + %% Sub topic first + Subs = [{<<"/test1/+">>, ?QOS_1}, {<<"/test2/+">>, ?QOS_2}], + Clients = start_client(Type, Subs), + timer:sleep(1000), + Now = ?NOW, + %% publish + + lists:foreach( + fun(I) -> + Topic = list_to_binary(io_lib:format("/test1/~p", [I])), + Msg = emqx_message:make(undefined, ?QOS_1, Topic, <<"Hello">>), + emqx:publish(Msg#message{timestamp = Now - ?LANTENCY}), + timer:sleep(100) + end, + lists:seq(1, 10) + ), + + lists:foreach( + fun(I) -> + Topic = list_to_binary(io_lib:format("/test2/~p", [I])), + Msg = emqx_message:make(undefined, ?QOS_2, Topic, <<"Hello">>), + emqx:publish(Msg#message{timestamp = Now - ?LANTENCY}), + timer:sleep(100) + end, + lists:seq(1, 10) + ), + + timer:sleep(1000), + Size = ets:info(?TOPK_TAB, size), + ?assert( + Size =< 10 andalso Size >= 3, + lists:flatten(io_lib:format("with_type:~p, the size is :~p~n", [Type, Size])) + ), + + ?assert( + lists:all( + fun(#{timespan := Ts}) -> + Ts >= 101 andalso Ts < ?NOW - Now + end, + emqx_slow_subs_api:get_history() + ) + ), + + [Client ! stop || Client <- Clients], + ok. + +update_stats_type(Type) -> + update_config(<<"stats_type">>, erlang:atom_to_binary(Type)). + +update_config(Key, Value) -> + Raw = #{ + <<"enable">> => true, + <<"expire_interval">> => <<"5m">>, + <<"stats_type">> => <<"whole">>, + <<"threshold">> => <<"100ms">>, + <<"top_k_num">> => 5 + }, + emqx_slow_subs:update_settings(Raw#{Key => Value}). + +make_conn_options(response, I) -> + [ + {msg_handler, #{ + publish => fun(_) -> timer:sleep(?LANTENCY) end, + disconnected => fun(_) -> ok end + }} + | make_conn_options(whole, I) + ]; +make_conn_options(_, I) -> + [ + {host, "localhost"}, + {clientid, io_lib:format("slow_subs_~p", [I])}, + {username, <<"plain">>}, + {password, <<"plain">>} + ].