Add node_queue_size metric to dist collector (#94)

Add node_queue_size_bytes metric to dist collector.

erlang/otp#2270 is available since Erlang/OTP 22.1, released 17th of September 2019, time to ship this feature 🚢

Thanks @essen!

Author: essen

.travis.yml

@@ -11,6 +11,6 @@ cache:
   directories:
   - $HOME/.cache/rebar3/
 otp_release:
-- 20.1
-- 20.0
+- 21.3
+- 22.3
 script: "./bin/checks.sh && ./rebar3 as test coveralls send"

doc/prometheus_vm_dist_collector.md

@@ -128,6 +128,10 @@ The current state of the distribution link.<br />
 The state is represented as a numerical value where `pending=1`,
 `up_pending=2` and `up=3`.
 
+* `erlang_vm_dist_node_queue_size_bytes`<br/>
+Type: gauge.<br/>
+The number of bytes in the output distribution queue.<br/>
+This queue sits between the Erlang code and the port driver.
 
 
 ### <a name="Configuration">Configuration</a> ###
@@ -185,5 +189,6 @@ Available options:
 
 * `node_state` for `erlang_vm_dist_node_state`.
 
+* `node_queue_size_bytes` for `erlang_vm_dist_node_queue_size_bytes`.
 
 By default all metrics are enabled.

src/collectors/vm/prometheus_vm_dist_collector.erl

@@ -142,6 +142,12 @@
 %%     The state is represented as a numerical value where `pending=1',
 %%     `up_pending=2' and `up=3'.
 %%   </li>
+%%   <li>
+%%     `erlang_vm_dist_node_queue_size_bytes'<br/>
+%%     Type: gauge.<br/>
+%%     The number of bytes in the output distribution queue.<br/>
+%%     This queue sits between the Erlang code and the port driver.
+%%   </li>
 %% </ul>
 %%
 %% ==Configuration==
@@ -223,6 +229,9 @@
 %%   <li>
 %%     `node_state' for `erlang_vm_dist_node_state'.
 %%   </li>
+%%   <li>
+%%     `node_queue_size_bytes' for `erlang_vm_dist_node_queue_size_bytes'.
+%%   </li>
 %% </ul>
 %%
 %% By default all metrics are enabled.
@@ -239,6 +248,8 @@
 
 -behaviour(prometheus_collector).
 
+-dialyzer({nowarn_function, node_queue_size/1}).
+
 %%====================================================================
 %% Macros
 %%====================================================================
@@ -360,7 +371,11 @@ metrics1() ->
     "The current state of the distribution link. "
     "The state is represented as a numerical value where `pending=1', "
     "`up_pending=2' and `up=3'.",
-    metric_node_state(Data)}].
+    metric_node_state(Data)},
+   {node_queue_size_bytes, gauge,
+    "The number of bytes in the output distribution queue. "
+    "This queue sits between the Erlang code and the port driver.",
+    metric_node_queue_size(Data)}].
 
 enabled_metrics() ->
   application:get_env(prometheus, vm_dist_collector_metrics, all).
@@ -482,3 +497,17 @@ metric_node_state(Data) ->
 node_state(pending) -> 1;
 node_state(up_pending) -> 2;
 node_state(up) -> 3.
+
+metric_node_queue_size(Data) ->
+    [
+        {[{peer, Node}], node_queue_size(Node)}
+    || {Node, _} <- Data].
+
+node_queue_size(Node) ->
+    ConnId = ets:lookup_element(sys_dist, Node, 3),
+    {ok, _, _, QueueSize} = erlang:dist_get_stat(ConnId),
+    case QueueSize of
+        false -> 0;
+        true -> -1;
+        _ -> QueueSize
+    end.

test/eunit/collectors/vm/prometheus_vm_dist_collector_tests.erl

@@ -35,7 +35,8 @@ test_no_distribution(_) ->
    ?_assertMatch(nomatch, re:run(Metrics, "erlang_vm_dist_proc_message_queue_len")),
    ?_assertMatch(nomatch, re:run(Metrics, "erlang_vm_dist_proc_reductions")),
    ?_assertMatch(nomatch, re:run(Metrics, "erlang_vm_dist_proc_status")),
-   ?_assertMatch(nomatch, re:run(Metrics, "erlang_vm_dist_node_state"))
+   ?_assertMatch(nomatch, re:run(Metrics, "erlang_vm_dist_node_state")),
+   ?_assertMatch(nomatch, re:run(Metrics, "erlang_vm_dist_node_queue_size_bytes"))
   ].
 
 
@@ -81,7 +82,8 @@ test_default_metrics(_) ->
    ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_proc_message_queue_len{peer")),
    ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_proc_reductions{peer")),
    ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_proc_status{peer")),
-   ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_node_state{peer"))
+   ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_node_state{peer")),
+   ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_node_queue_size_bytes{peer"))
   ].
 
 
@@ -112,7 +114,8 @@ test_all_metrics(_) ->
                          proc_message_queue_len,
                          proc_reductions,
                          proc_status,
-                         node_state
+                         node_state,
+                         node_queue_size_bytes
                         ]),
     prometheus_registry:register_collector(prometheus_vm_dist_collector),
     Metrics = prometheus_text_format:format(),
@@ -140,7 +143,8 @@ test_all_metrics(_) ->
      ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_proc_message_queue_len")),
      ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_proc_reductions")),
      ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_proc_status")),
-     ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_node_state"))
+     ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_node_state")),
+     ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_node_queue_size_bytes"))
     ]
 
   after
@@ -188,7 +192,8 @@ test_custom_metrics(_) ->
      ?_assertMatch(nomatch, re:run(Metrics, "erlang_vm_dist_proc_message_queue_len")),
      ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_proc_reductions")),
      ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_proc_status")),
-     ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_node_state"))
+     ?_assertMatch({match, _}, re:run(Metrics, "erlang_vm_dist_node_state")),
+     ?_assertMatch(nomatch, re:run(Metrics, "erlang_vm_dist_node_queue_size_bytes"))
     ]
 
   after