Add more detailed traffic metrics (tx and rx, per-listener and per-dc)

src/mtp_down_conn.erl

@@ -104,8 +104,8 @@ handle_cast(shutdown, State) ->
     {stop, shutdown, State}.
 
 
-handle_info({tcp, Sock, Data}, #state{sock = Sock} = S) ->
-    mtp_metric:count_inc([?APP, received, bytes], byte_size(Data), #{labels => [downstream]}),
+handle_info({tcp, Sock, Data}, #state{sock = Sock, dc_id = DcId} = S) ->
+    mtp_metric:count_inc([?APP, received, downstream, bytes], byte_size(Data), #{labels => [DcId]}),
     mtp_metric:histogram_observe([?APP, tracker_packet_size, bytes], byte_size(Data), #{labels => [downstream]}),
     {ok, S1} = handle_downstream_data(Data, S),
     activate_if_no_overflow(S1),
@@ -239,7 +239,10 @@ down_send(Packet, #state{sock = Sock, codec = Codec, dc_id = DcId} = St) ->
     mtp_metric:rt(
       [?APP, downstream_send_duration, seconds],
       fun() ->
-              ok = gen_tcp:send(Sock, Encoded)
+              ok = gen_tcp:send(Sock, Encoded),
+              mtp_metric:count_inc(
+                [?APP, sent, downstream, bytes],
+                iolist_size(Encoded), #{labels => [DcId]})
       end, #{labels => [DcId]}),
     {ok, St#state{codec = Codec1}}.
 

src/mtp_handler.erl

@@ -137,11 +137,11 @@ handle_cast(Other, State) ->
     lager:warning("Unexpected msg ~p", [Other]),
     {noreply, State}.
 
-handle_info({tcp, Sock, Data}, #state{sock = Sock,
-                                      transport = Transport} = S) ->
+handle_info({tcp, Sock, Data}, #state{sock = Sock, transport = Transport,
+                                      listener = Listener} = S) ->
     %% client -> proxy
     Size = byte_size(Data),
-    mtp_metric:count_inc([?APP, received, bytes], Size, #{labels => [upstream]}),
+    mtp_metric:count_inc([?APP, received, upstream, bytes], Size, #{labels => [Listener]}),
     mtp_metric:histogram_observe([?APP, tracker_packet_size, bytes], Size, #{labels => [upstream]}),
     case handle_upstream_data(Data, S) of
         {ok, S1} ->
@@ -274,7 +274,11 @@ up_send(Packet, #state{stage = tunnel,
     mtp_metric:rt([?APP, upstream_send_duration, seconds],
               fun() ->
                       case Transport:send(Sock, Encoded) of
-                          ok -> ok;
+                          ok ->
+                              mtp_metric:count_inc(
+                                [?APP, sent, upstream, bytes],
+                                iolist_size(Encoded), #{labels => [Listener]}),
+                              ok;
                           {error, Reason} ->
                               is_atom(Reason) andalso
                                   mtp_metric:count_inc(

src/mtp_metric.erl

@@ -115,9 +115,19 @@ active_metrics() ->
       "Upstream self-healthcheck triggered some action",
       #{labels => [action]}},
 
-     {count, [?APP, received, bytes],
-      "Bytes transmitted from upstream/downstream socket",
-      #{labels => [direction]}},
+     {count, [?APP, received, downstream, bytes],
+      "Bytes transmitted from downstream socket",
+      #{labels => [dc_id]}},
+     {count, [?APP, received, upstream, bytes],
+      "Bytes transmitted from upstream socket",
+      #{labels => [listener]}},
+     {count, [?APP, sent, downstream, bytes],
+      "Bytes sent to downstream socket",
+      #{labels => [dc_id]}},
+     {count, [?APP, sent, upstream, bytes],
+      "Bytes sent to upstream socket",
+      #{labels => [listener]}},
+
      {histogram, [?APP, tracker_packet_size, bytes],
       "Received packet size",
       #{labels => [direction],