fix: prometheus metrics observation with qoe enabled

fixes #282

Author: id

src/emqtt_bench.erl

@@ -1670,38 +1670,37 @@ qoe_store_insert(Prometheus,
                   , handshaked := HSTs
                   , connected := ConnTs
                   } = QoE) ->
-   ElapsedHandshake = HSTs - StartTs,
-   ElapsedConn = ConnTs - StartTs,
-   ElapsedSub = case maps:get(subscribed, QoE, undefined) of
-                   undefined ->
-                      %% invalid
-                      ?invalid_elapsed;
-                   SubTs ->
-                      SubTs - StartTs
-                end,
-
-   ElapsedTCPHS = case maps:get(tcp_connected_at, QoE, undefined) of
-                   undefined ->
-                      %% invalid
-                      ?invalid_elapsed;
-                   TCPTs ->
-                      TCPTs - StartTs
-                end,
-   histogram_observe(Prometheus, mqtt_client_tcp_handshake_duration, ElapsedTCPHS),
-   histogram_observe(Prometheus, mqtt_client_handshake_duration, ElapsedHandshake),
-   histogram_observe(Prometheus, mqtt_client_connect_duration, ElapsedConn),
-   histogram_observe(Prometheus, mqtt_client_subscribe_duration, ElapsedSub),
-
-   Offset = erlang:time_offset(millisecond),
-   Term = #qoe_rec_v2{key = {ClientId, Offset + StartTs},
-                      tcp_lat = ElapsedTCPHS,
-                      handshake_lat = ElapsedHandshake,
-                      connect_lat = ElapsedConn,
-                      subscribe_lat = ElapsedSub
-                      },
-   true = ets:insert(?qoe_store, Term),
-   ok.
-
+    CalcElapsed = fun(EndTs) when is_integer(EndTs) ->
+                      EndTs - StartTs;
+                     (_) ->
+                      ?invalid_elapsed
+                  end,
+
+    ObserveElapsed = fun(Time, Metric) when is_integer(Time) ->
+                         histogram_observe(Prometheus, Metric, Time),
+                         Time;
+                        (_, _) ->
+                         ?invalid_elapsed
+                     end,
+
+    TcpConnectedAtTs = maps:get(tcp_connected_at, QoE, undefined),
+    ElapsedTCPHS = ObserveElapsed(CalcElapsed(TcpConnectedAtTs), mqtt_client_tcp_handshake_duration),
+    ElapsedHandshake = ObserveElapsed(CalcElapsed(HSTs), mqtt_client_handshake_duration),
+    ElapsedConn = ObserveElapsed(CalcElapsed(ConnTs), mqtt_client_connect_duration),
+    SubscribedTs = maps:get(subscribed, QoE, undefined),
+    ElapsedSub = ObserveElapsed(CalcElapsed(SubscribedTs), mqtt_client_subscribe_duration),
+
+    Offset = erlang:time_offset(millisecond),
+    Term = #qoe_rec_v2{
+        key = {ClientId, Offset + StartTs},
+        tcp_lat = ElapsedTCPHS,
+        handshake_lat = ElapsedHandshake,
+        connect_lat = ElapsedConn,
+        subscribe_lat = ElapsedSub
+    },
+
+    true = ets:insert(?qoe_store, Term),
+    ok.
 
 open_qoe_disklog(File) ->
    case disk_log:open([{name, ?QoELog}, {file, File}, {repair, true}, {type, halt},