Merge pull request #13009 from lafirest/fix/rate_log

feat(limiter): lift the log level of rate limiter to `warning`

apps/emqx/src/emqx_connection.erl

@@ -1013,17 +1013,23 @@ check_limiter(
     Data,
     WhenOk,
     Msgs,
-    #state{limiter_timer = undefined, limiter = Limiter} = State
+    #state{channel = Channel, limiter_timer = undefined, limiter = Limiter} = State
 ) ->
     case emqx_limiter_container:check_list(Needs, Limiter) of
         {ok, Limiter2} ->
             WhenOk(Data, Msgs, State#state{limiter = Limiter2});
         {pause, Time, Limiter2} ->
-            ?SLOG(debug, #{
+            ?SLOG_THROTTLE(
-                msg => "pause_time_due_to_rate_limit",
+                warning,
-                needs => Needs,
+                #{
-                time_in_ms => Time
+                    msg => socket_receive_paused_by_rate_limit,
-            }),
+                    paused_ms => Time
+                },
+                #{
+                    tag => "RATE",
+                    clientid => emqx_channel:info(clientid, Channel)
+                }
+            ),
 
             Retry = #retry{
                 types = [Type || {_, Type} <- Needs],
@@ -1057,7 +1063,7 @@ check_limiter(
 
 %% try to perform a retry
 -spec retry_limiter(state()) -> _.
-retry_limiter(#state{limiter = Limiter} = State) ->
+retry_limiter(#state{channel = Channel, limiter = Limiter} = State) ->
     #retry{types = Types, data = Data, next = Next} =
         emqx_limiter_container:get_retry_context(Limiter),
     case emqx_limiter_container:retry_list(Types, Limiter) of
@@ -1071,11 +1077,17 @@ retry_limiter(#state{limiter = Limiter} = State) ->
                 }
             );
         {pause, Time, Limiter2} ->
-            ?SLOG(debug, #{
+            ?SLOG_THROTTLE(
-                msg => "pause_time_due_to_rate_limit",
+                warning,
-                types => Types,
+                #{
-                time_in_ms => Time
+                    msg => socket_receive_paused_by_rate_limit,
-            }),
+                    paused_ms => Time
+                },
+                #{
+                    tag => "RATE",
+                    clientid => emqx_channel:info(clientid, Channel)
+                }
+            ),
 
             TRef = start_timer(Time, limit_timeout),
 

apps/emqx/src/emqx_ws_connection.erl

@@ -600,17 +600,23 @@ check_limiter(
     Data,
     WhenOk,
     Msgs,
-    #state{limiter_timer = undefined, limiter = Limiter} = State
+    #state{channel = Channel, limiter_timer = undefined, limiter = Limiter} = State
 ) ->
     case emqx_limiter_container:check_list(Needs, Limiter) of
         {ok, Limiter2} ->
             WhenOk(Data, Msgs, State#state{limiter = Limiter2});
         {pause, Time, Limiter2} ->
-            ?SLOG(debug, #{
+            ?SLOG_THROTTLE(
-                msg => "pause_time_due_to_rate_limit",
+                warning,
-                needs => Needs,
+                #{
-                time_in_ms => Time
+                    msg => socket_receive_paused_by_rate_limit,
-            }),
+                    paused_ms => Time
+                },
+                #{
+                    tag => "RATE",
+                    clientid => emqx_channel:info(clientid, Channel)
+                }
+            ),
 
             Retry = #retry{
                 types = [Type || {_, Type} <- Needs],
@@ -644,7 +650,7 @@ check_limiter(
     State#state{limiter_buffer = queue:in(New, Buffer)}.
 
 -spec retry_limiter(state()) -> state().
-retry_limiter(#state{limiter = Limiter} = State) ->
+retry_limiter(#state{channel = Channel, limiter = Limiter} = State) ->
     #retry{types = Types, data = Data, next = Next} = emqx_limiter_container:get_retry_context(
         Limiter
     ),
@@ -659,11 +665,17 @@ retry_limiter(#state{limiter = Limiter} = State) ->
                 }
             );
         {pause, Time, Limiter2} ->
-            ?SLOG(debug, #{
+            ?SLOG_THROTTLE(
-                msg => "pause_time_due_to_rate_limit",
+                warning,
-                types => Types,
+                #{
-                time_in_ms => Time
+                    msg => socket_receive_paused_by_rate_limit,
-            }),
+                    paused_ms => Time
+                },
+                #{
+                    tag => "RATE",
+                    clientid => emqx_channel:info(clientid, Channel)
+                }
+            ),
 
             TRef = start_timer(Time, limit_timeout),
 

apps/emqx_conf/src/emqx_conf_schema.erl

@@ -84,7 +84,8 @@
     cannot_publish_to_topic_due_to_not_authorized,
     cannot_publish_to_topic_due_to_quota_exceeded,
     connection_rejected_due_to_license_limit_reached,
-    dropped_msg_due_to_mqueue_is_full
+    dropped_msg_due_to_mqueue_is_full,
+    socket_receive_paused_by_rate_limit
 ]).
 
 %% Callback to upgrade config after loaded from config file but before validation.

changes/ce/feat-13009.en.md

@@ -0,0 +1 @@
+Changed the log level of message receive pause from `debug` to `warning`. The log message `socket_receive_paused_by_rate_limit` is throttled to avoid excessive logging.