Fix cluster links statistic

Use the `sys_dist` ets table to get distribution port information.

Fixes #4981

Get cluster links stats for TLS dist

Use code from prometheus.erl to get dist links info

deps/rabbit_common/src/rabbit_net.erl

@@ -9,13 +9,15 @@
 -include("rabbit.hrl").
 
 -include_lib("kernel/include/inet.hrl").
+-include_lib("kernel/include/net_address.hrl").
 
 -export([is_ssl/1, ssl_info/1, controlling_process/2, getstat/2,
     recv/1, sync_recv/2, async_recv/3, port_command/2, getopts/2,
     setopts/2, send/2, close/1, fast_close/1, sockname/1, peername/1,
     peercert/1, connection_string/2, socket_ends/2, is_loopback/1,
     tcp_host/1, unwrap_socket/1, maybe_get_proxy_socket/1,
-    hostname/0, getifaddrs/0, proxy_ssl_info/2]).
+    hostname/0, getifaddrs/0, proxy_ssl_info/2,
+    dist_info/0]).
 
 %%---------------------------------------------------------------------------
 
@@ -329,3 +331,39 @@ maybe_get_proxy_socket(Sock={rabbit_proxy_socket, _, _}) ->
     Sock;
 maybe_get_proxy_socket(_Sock) ->
     undefined.
+
+%% deps/prometheus/src/collectors/vm/prometheus_vm_dist_collector.erl
+%% https://github.com/deadtrickster/prometheus.erl/blob/v4.8.2/src/collectors/vm/prometheus_vm_dist_collector.erl#L386-L450
+dist_info() ->
+    {ok, NodesInfo} = net_kernel:nodes_info(),
+    TcpInetPorts = [P || P <- erlang:ports(),
+                         erlang:port_info(P, name) =:= {name, "tcp_inet"}],
+    [dist_info(NodeInfo, TcpInetPorts) || NodeInfo <- NodesInfo].
+
+dist_info({Node, Info}, TcpInetPorts) ->
+    DistPid = proplists:get_value(owner, Info),
+    case proplists:get_value(address, Info, #net_address{}) of
+        #net_address{address=undefined} ->
+            %% No stats available
+            {Node, DistPid, []};
+        #net_address{address=PeerAddr} ->
+            dist_info(Node, TcpInetPorts, DistPid, PeerAddr)
+    end.
+
+dist_info(Node, TcpInetPorts, DistPid, PeerAddrArg) ->
+    [DistPort] = [P || P <- TcpInetPorts,
+                       inet:peername(P) =:= {ok, PeerAddrArg}],
+    S = case {socket_ends(DistPort, inbound), getstat(DistPort, [recv_oct, send_oct])} of
+            {{ok, {PeerAddr, PeerPort, SockAddr, SockPort}}, {ok, Stats}} ->
+                PeerAddrBin = rabbit_data_coercion:to_binary(maybe_ntoab(PeerAddr)),
+                SockAddrBin = rabbit_data_coercion:to_binary(maybe_ntoab(SockAddr)),
+                [{peer_addr, PeerAddrBin},
+                 {peer_port, PeerPort},
+                 {sock_addr, SockAddrBin},
+                 {sock_port, SockPort},
+                 {recv_bytes, rabbit_misc:pget(recv_oct, Stats)},
+                 {send_bytes, rabbit_misc:pget(send_oct, Stats)}];
+            _ ->
+                []
+        end,
+    {Node, DistPid, S}.

deps/rabbitmq_management_agent/src/rabbit_mgmt_external_stats.erl

@@ -15,8 +15,6 @@
 
 -export([list_registry_plugins/1]).
 
--import(rabbit_misc, [pget/2]).
-
 -include_lib("rabbit_common/include/rabbit.hrl").
 
 -define(METRICS_KEYS, [fd_used, sockets_used, mem_used, disk_free, proc_used, gc_num,
@@ -350,40 +348,7 @@ flatten_key({A, B}) ->
     list_to_atom(atom_to_list(A) ++ "_" ++ atom_to_list(B)).
 
 cluster_links() ->
-    {ok, Items} = net_kernel:nodes_info(),
-    [Link || Item <- Items,
-             Link <- [format_nodes_info(Item)], Link =/= undefined].
-
-format_nodes_info({Node, Info}) ->
-    Owner = proplists:get_value(owner, Info),
-    case catch process_info(Owner, links) of
-        {links, Links} ->
-            case [Link || Link <- Links, is_port(Link)] of
-                [Port] ->
-                    {Node, Owner, format_nodes_info1(Port)};
-                _ ->
-                    undefined
-            end;
-        _ ->
-            undefined
-    end.
-
-format_nodes_info1(Port) ->
-    case {rabbit_net:socket_ends(Port, inbound),
-          rabbit_net:getstat(Port, [recv_oct, send_oct])} of
-        {{ok, {PeerAddr, PeerPort, SockAddr, SockPort}}, {ok, Stats}} ->
-            [{peer_addr, maybe_ntoab(PeerAddr)},
-             {peer_port, PeerPort},
-             {sock_addr, maybe_ntoab(SockAddr)},
-             {sock_port, SockPort},
-             {recv_bytes, pget(recv_oct, Stats)},
-             {send_bytes, pget(send_oct, Stats)}];
-        _ ->
-            []
-    end.
-
-maybe_ntoab(A) when is_tuple(A) -> list_to_binary(rabbit_misc:ntoab(A));
-maybe_ntoab(H)                  -> H.
+    rabbit_net:dist_info().
 
 %%--------------------------------------------------------------------