Add batch commit for metrics

src/emqx_broker.erl

@@ -327,8 +327,6 @@ handle_call(Req, _From, State) ->
     emqx_logger:error("[Broker] unexpected call: ~p", [Req]),
     {reply, ignored, State}.
 
-
-
 handle_cast({From, #subscribe{topic = Topic, subpid = SubPid, subid = SubId, subopts = SubOpts}}, State) ->
     Subscriber = {SubPid, SubId},
     case ets:member(?SUBOPTION, {Topic, Subscriber}) of

src/emqx_connection.erl

@@ -170,7 +170,7 @@ send_fun(Transport, Socket) ->
         Data = emqx_frame:serialize(Packet, Options),
         try Transport:async_send(Socket, Data) of
             ok ->
-                emqx_metrics:inc('bytes/sent', iolist_size(Data)),
+                emqx_metrics:trans(inc, 'bytes/sent', iolist_size(Data)),
                 ok;
             Error -> Error
         catch
@@ -215,6 +215,7 @@ handle_info({timeout, Timer, emit_stats},
             State = #state{stats_timer = Timer,
                            proto_state = ProtoState
                           }) ->
+    emqx_metrics:commit(),
     emqx_cm:set_conn_stats(emqx_protocol:client_id(ProtoState), stats(State)),
     NewState = State#state{stats_timer = undefined},
     Limits = erlang:get(force_shutdown_policy),
@@ -248,7 +249,7 @@ handle_info(activate_sock, State) ->
 handle_info({inet_async, _Sock, _Ref, {ok, Data}}, State) ->
     ?LOG(debug, "RECV ~p", [Data]),
     Size = iolist_size(Data),
-    emqx_metrics:inc('bytes/received', Size),
+    emqx_metrics:trans(inc, 'bytes/received', Size),
     Incoming = #{bytes => Size, packets => 0},
     handle_packet(Data, State#state{await_recv = false, incoming = Incoming});
 

src/emqx_metrics.erl

@@ -19,6 +19,7 @@
 -export([start_link/0]).
 -export([new/1, all/0]).
 -export([val/1, inc/1, inc/2, inc/3, dec/2, dec/3, set/2]).
+-export([trans/2, trans/3, trans/4, commit/0]).
 %% Received/sent metrics
 -export([received/1, sent/1]).
 
@@ -133,10 +134,8 @@ inc(Metric, Val) when is_atom(Metric) ->
 
 %% @doc Increase metric value
 -spec(inc(counter | gauge, atom(), pos_integer()) -> pos_integer()).
-inc(gauge, Metric, Val) ->
-    update_counter(key(gauge, Metric), {2, Val});
-inc(counter, Metric, Val) ->
-    update_counter(key(counter, Metric), {2, Val}).
+inc(Type, Metric, Val) ->
+    update_counter(key(Type, Metric), {2, Val}).
 
 %% @doc Decrease metric value
 -spec(dec(gauge, atom()) -> integer()).
@@ -154,6 +153,41 @@ set(Metric, Val) when is_atom(Metric) ->
 set(gauge, Metric, Val) ->
     ets:insert(?TAB, {key(gauge, Metric), Val}).
 
+trans(inc, Metric) ->
+    trans(inc, {counter, Metric}, 1).
+
+trans(Opt, {gauge, Metric}, Val) ->
+    trans(Opt, gauge, Metric, Val);
+trans(inc, {counter, Metric}, Val) ->
+    trans(inc, counter, Metric, Val);
+trans(inc, Metric, Val) when is_atom(Metric) ->
+    trans(inc, counter, Metric, Val);
+trans(dec, gauge, Metric) ->
+    trans(dec, gauge, Metric, 1).
+
+trans(inc, Type, Metric, Val) ->
+    hold(Type, Metric, Val);
+trans(dec, gauge, Metric, Val) ->
+    hold(gauge, Metric, -Val).
+
+hold(Type, Metric, Val) when Type =:= counter orelse Type =:= gauge ->
+    put('$metrics', case get('$metrics') of
+                        undefined ->
+                            #{{Type, Metric} => Val};
+                        Metrics ->
+                            maps:update_with({Type, Metric}, fun(Cnt) -> Cnt + Val end, Val, Metrics)
+                    end).
+
+commit() ->
+    case get('$metrics') of
+        undefined -> ok;
+        Metrics ->
+            maps:fold(fun({Type, Metric}, Val, _Acc) -> 
+                          update_counter(key(Type, Metric), {2, Val})
+                      end, 0, Metrics),
+            erase('$metrics')
+    end.
+
 %% @doc Metric key
 key(gauge, Metric) ->
     {Metric, 0};

src/emqx_session.erl

@@ -421,7 +421,7 @@ handle_call({register_publish_packet_id, PacketId, Ts}, _From,
                           {ok, ensure_await_rel_timer(State1)}
                   end;
               true ->
-                  emqx_metrics:inc('messages/qos2/dropped'),
+                  emqx_metrics:trans(inc, 'messages/qos2/dropped'),
                   ?LOG(warning, "Dropped qos2 packet ~w for too many awaiting_rel", [PacketId], State),
                   {{error, ?RC_RECEIVE_MAXIMUM_EXCEEDED}, State}
           end);
@@ -432,7 +432,7 @@ handle_call({pubrec, PacketId, _ReasonCode}, _From, State = #state{inflight = In
               true ->
                   {ok, acked(pubrec, PacketId, State)};
               false ->
-                  emqx_metrics:inc('packets/pubrec/missed'),
+                  emqx_metrics:trans(inc, 'packets/pubrec/missed'),
                   ?LOG(warning, "The PUBREC PacketId ~w is not found.", [PacketId], State),
                   {{error, ?RC_PACKET_IDENTIFIER_NOT_FOUND}, State}
           end);
@@ -443,7 +443,7 @@ handle_call({pubrel, PacketId, _ReasonCode}, _From, State = #state{awaiting_rel
               {_Ts, AwaitingRel1} ->
                   {ok, State#state{awaiting_rel = AwaitingRel1}};
               error ->
-                  emqx_metrics:inc('packets/pubrel/missed'),
+                  emqx_metrics:trans(inc, 'packets/pubrel/missed'),
                   ?LOG(warning, "Cannot find PUBREL: ~w", [PacketId], State),
                   {{error, ?RC_PACKET_IDENTIFIER_NOT_FOUND}, State}
           end);
@@ -502,7 +502,7 @@ handle_cast({puback, PacketId, _ReasonCode}, State = #state{inflight = Inflight}
             noreply(dequeue(acked(puback, PacketId, State)));
         false ->
             ?LOG(warning, "The PUBACK PacketId ~w is not found", [PacketId], State),
-            emqx_metrics:inc('packets/puback/missed'),
+            emqx_metrics:trans(inc, 'packets/puback/missed'),
             {noreply, State}
     end;
 
@@ -513,7 +513,7 @@ handle_cast({pubcomp, PacketId, _ReasonCode}, State = #state{inflight = Inflight
             noreply(dequeue(acked(pubcomp, PacketId, State)));
         false ->
             ?LOG(warning, "The PUBCOMP PacketId ~w is not found", [PacketId], State),
-            emqx_metrics:inc('packets/pubcomp/missed'),
+            emqx_metrics:trans(inc, 'packets/pubcomp/missed'),
             {noreply, State}
     end;
 
@@ -603,6 +603,7 @@ handle_info({timeout, Timer, check_awaiting_rel}, State = #state{await_rel_timer
 handle_info({timeout, Timer, emit_stats},
             State = #state{client_id = ClientId,
                            stats_timer = Timer}) ->
+    emqx_metrics:commit(),
     _ = emqx_sm:set_session_stats(ClientId, stats(State)),
     NewState = State#state{stats_timer = undefined},
     Limits = erlang:get(force_shutdown_policy),
@@ -734,7 +735,7 @@ retry_delivery(Force, [{Type, Msg0, Ts} | Msgs], Now,
                             {publish, {PacketId, Msg}} ->
                                 case emqx_message:is_expired(Msg) of
                                     true ->
-                                        emqx_metrics:inc('messages/expired'),
+                                        emqx_metrics:trans(inc, 'messages/expired'),
                                         emqx_inflight:delete(PacketId, Inflight);
                                     false ->
                                         redeliver({PacketId, Msg}, State),
@@ -774,7 +775,7 @@ expire_awaiting_rel([{PacketId, Ts} | More], Now,
                     State = #state{awaiting_rel = AwaitingRel, await_rel_timeout = Timeout}) ->
     case (timer:now_diff(Now, Ts) div 1000) of
         Age when Age >= Timeout ->
-            emqx_metrics:inc('messages/qos2/expired'),
+            emqx_metrics:trans(inc, 'messages/qos2/expired'),
             ?LOG(warning, "Dropped qos2 packet ~s for await_rel_timeout", [PacketId], State),
             expire_awaiting_rel(More, Now, State#state{awaiting_rel = maps:remove(PacketId, AwaitingRel)});
         Age ->

src/emqx_ws_connection.erl

@@ -148,7 +148,7 @@ send_fun(WsPid) ->
     fun(Packet, Options) ->
         Data = emqx_frame:serialize(Packet, Options),
         BinSize = iolist_size(Data),
-        emqx_metrics:inc('bytes/sent', BinSize),
+        emqx_metrics:trans(inc, 'bytes/sent', BinSize),
         put(send_oct, get(send_oct) + BinSize),
         put(send_cnt, get(send_cnt) + 1),
         WsPid ! {binary, iolist_to_binary(Data)},
@@ -167,7 +167,7 @@ websocket_handle({binary, Data}, State = #state{parser_state = ParserState,
     BinSize = iolist_size(Data),
     put(recv_oct, get(recv_oct) + BinSize),
     ?LOG(debug, "RECV ~p", [Data]),
-    emqx_metrics:inc('bytes/received', BinSize),
+    emqx_metrics:trans(inc, 'bytes/received', BinSize),
     case catch emqx_frame:parse(iolist_to_binary(Data), ParserState) of
         {more, NewParserState} ->
             {ok, State#state{parser_state = NewParserState}};
@@ -223,6 +223,7 @@ websocket_info({deliver, PubOrAck}, State = #state{proto_state = ProtoState}) ->
 
 websocket_info({timeout, Timer, emit_stats},
                State = #state{stats_timer = Timer, proto_state = ProtoState}) ->
+    emqx_metrics:commit(),
     emqx_cm:set_conn_stats(emqx_protocol:client_id(ProtoState), stats(State)),
     {ok, State#state{stats_timer = undefined}, hibernate};
 

test/emqx_metrics_SUITE.erl

@@ -19,21 +19,40 @@
 
 -include("emqx_mqtt.hrl").
 
-all() -> [t_inc_dec_metrics].
+all() -> [t_inc_dec_metrics, t_trans].
 
 t_inc_dec_metrics(_) ->
     {ok, _} = emqx_metrics:start_link(),
     {0, 0} = {emqx_metrics:val('bytes/received'), emqx_metrics:val('messages/retained')},
     emqx_metrics:inc('bytes/received'),
     emqx_metrics:inc({counter, 'bytes/received'}, 2),
-    emqx_metrics:inc(counter, 'bytes/received', 2),
+    emqx_metrics:inc(counter, 'bytes/received', 1),
+    emqx_metrics:inc('bytes/received', 1),
     emqx_metrics:inc({gauge, 'messages/retained'}, 2),
     emqx_metrics:inc(gauge, 'messages/retained', 2),
     {5, 4} = {emqx_metrics:val('bytes/received'), emqx_metrics:val('messages/retained')},
     emqx_metrics:dec(gauge, 'messages/retained'),
     emqx_metrics:dec(gauge, 'messages/retained', 1),
     2 = emqx_metrics:val('messages/retained'),
+    emqx_metrics:set('messages/retained', 3),
+    3 = emqx_metrics:val('messages/retained'),
     emqx_metrics:received(#mqtt_packet{header = #mqtt_packet_header{type = ?CONNECT}}),
     {1, 1} = {emqx_metrics:val('packets/received'), emqx_metrics:val('packets/connect')},
     emqx_metrics:sent(#mqtt_packet{header = #mqtt_packet_header{type = ?CONNACK}}),
     {1, 1} = {emqx_metrics:val('packets/sent'), emqx_metrics:val('packets/connack')}.
+
+t_trans(_) ->
+    {ok, _} = emqx_metrics:start_link(),
+    emqx_metrics:trans(inc, 'bytes/received'),
+    emqx_metrics:trans(inc, {counter, 'bytes/received'}, 2),
+    emqx_metrics:trans(inc, counter, 'bytes/received', 2),
+    emqx_metrics:trans(inc, {gauge, 'messages/retained'}, 2),
+    emqx_metrics:trans(inc, gauge, 'messages/retained', 2),
+    {0, 0} = {emqx_metrics:val('bytes/received'), emqx_metrics:val('messages/retained')},
+    emqx_metrics:commit(),
+    {5, 4} = {emqx_metrics:val('bytes/received'), emqx_metrics:val('messages/retained')},
+    emqx_metrics:trans(dec, gauge, 'messages/retained'),
+    emqx_metrics:trans(dec, gauge, 'messages/retained', 1),
+    4 = emqx_metrics:val('messages/retained'),
+    emqx_metrics:commit(),
+    2 = emqx_metrics:val('messages/retained').

test/emqx_session_SUITE.erl

@@ -45,6 +45,7 @@ ignore_loop(_Config) ->
     application:set_env(emqx, mqtt_ignore_loop_deliver, false).
 
 t_session_all(_) ->
+    emqx_zone:set_env(internal, idle_timeout, 100),
     ClientId = <<"ClientId">>,
     {ok, ConnPid} = emqx_mock_client:start_link(ClientId),
     {ok, SPid} = emqx_mock_client:open_session(ConnPid, ClientId, internal),