Merge pull request #8986 from lafirest/fix/slow_subs_unit

fix(slow_subs): fix timestamp unit error and update config not work

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
 

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

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]},

apps/emqx_slow_subs/src/emqx_slow_subs.erl

@@ -166,11 +166,11 @@ init([]) ->
         expire_timer => undefined
     },
 
-    Enable = emqx:get_config([slow_subs, enable]),
+    Cfg = emqx:get_config([slow_subs]),
-    {ok, check_enable(Enable, InitState)}.
+    {ok, check_enable(Cfg, InitState)}.
 
-handle_call({update_settings, #{enable := Enable}}, _From, State) ->
+handle_call({update_settings, Cfg}, _From, State) ->
-    State2 = check_enable(Enable, 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.

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]),
+    update_config(<<"expire_interval">>, <<"1500ms">>),
-    emqx_slow_subs:update_settings(Cfg#{expire_interval := 1500}),
     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
+    _ = [stats_with_type(Type) || Type <- [whole, internal, response]],
-    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],
     ok.
 
 t_expire(_) ->
@@ -135,16 +106,12 @@ t_expire(_) ->
     ?assertEqual(0, Size),
     ok.
 
-start_client(Subs) ->
+start_client(Type, Subs) ->
-    [spawn(fun() -> client(I, Subs) end) || I <- lists:seq(1, 10)].
+    [spawn(fun() -> client(I, Type, Subs) end) || I <- lists:seq(1, 10)].
 
-client(I, Subs) ->
+client(I, Type, Subs) ->
-    {ok, C} = emqtt:start_link([
+    ConnOptions = make_conn_options(Type, I),
-        {host, "localhost"},
+    {ok, C} = emqtt:start_link(ConnOptions),
-        {clientid, io_lib:format("slow_subs_~p", [I])},
-        {username, <<"plain">>},
-        {password, <<"plain">>}
-    ]),
     {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">>}
+    ].