feat: introduce new metric active_mqtt_connections (#81)

VivekPipaliya23 · web-flow · commit 11072fb066d5 · 2025-07-21T14:26:28.000+05:30
* feat: introduce new metric active_mqtt_connections

* fix: add new araity for listener
diff --git a/apps/vmq_server/src/vmq_health_http.erl b/apps/vmq_server/src/vmq_health_http.erl
@@ -71,7 +71,7 @@ cluster_status() ->
 -spec listeners_status() -> ok | {error, Reason :: string()}.
 listeners_status() ->
     NotRunningListeners = lists:filtermap(
-        fun({Type, _, _, Status, _, _, _}) ->
+        fun({Type, _, _, Status, _, _, _, _, _}) ->
             case Status of
                 running ->
                     false;
diff --git a/apps/vmq_server/src/vmq_listener_cli.erl b/apps/vmq_server/src/vmq_listener_cli.erl
@@ -312,7 +312,7 @@ vmq_listener_show_cmd() ->
         fun(_, [], []) ->
             Table =
                 lists:foldl(
-                    fun({Type, Ip, Port, Status, MP, MaxConns, PP}, Acc) ->
+                    fun({Type, Ip, Port, Status, MP, MaxConns, PP, ActiveConns, AllConns}, Acc) ->
                         [
                             [
                                 {type, Type},
@@ -321,7 +321,9 @@ vmq_listener_show_cmd() ->
                                 {port, Port},
                                 {mountpoint, MP},
                                 {max_conns, MaxConns},
-                                {proxy_protocol, PP}
+                                {proxy_protocol, PP},
+                                {active_conns, ActiveConns},
+                                {all_conns, AllConns}
                             ]
                             | Acc
                         ]
diff --git a/apps/vmq_server/src/vmq_metrics.erl b/apps/vmq_server/src/vmq_metrics.erl
@@ -2153,13 +2153,19 @@ fetch_external_metric(Mod, Fun, Default) ->
 misc_statistics() ->
     {NrOfSubs, SMemory} = fetch_external_metric(vmq_reg_trie, stats, {0, 0}),
     {NrOfRetain, RMemory} = fetch_external_metric(vmq_retain_srv, stats, {0, 0}),
+    {NrOfMQTTConnections, NrOfMQTTWSConnections} = fetch_external_metric(
+        vmq_ranch_sup, active_mqtt_connections, {0, 0}
+    ),
     [
         {router_subscriptions, NrOfSubs},
         {router_memory, SMemory},
         {retain_messages, NrOfRetain},
         {retain_memory, RMemory},
         {queue_processes, fetch_external_metric(vmq_queue_sup_sup, nr_of_queues, 0)},
-        {offline_messages, fetch_external_metric(vmq_message_store, nr_of_offline_messages, 0)}
+        {offline_messages, fetch_external_metric(vmq_message_store, nr_of_offline_messages, 0)},
+        {active_mqttws_connections, NrOfMQTTWSConnections},
+        {active_mqtt_connections, NrOfMQTTConnections},
+        {total_active_connections, NrOfMQTTWSConnections + NrOfMQTTConnections}
     ].
 
 -spec misc_stats_def() -> [metric_def()].
@@ -2194,7 +2200,28 @@ misc_stats_def() ->
             <<"The number of bytes used for storing retained messages.">>
         ),
         m(gauge, [], queue_processes, queue_processes, <<"The number of MQTT queue processes.">>),
-        m(gauge, [], offline_messages, offline_messages, <<"The number of offline messages">>)
+        m(gauge, [], offline_messages, offline_messages, <<"The number of offline messages">>),
+        m(
+            gauge,
+            [],
+            active_mqtt_connections,
+            active_mqtt_connections,
+            <<"The number of active MQTT(S) connections.">>
+        ),
+        m(
+            gauge,
+            [],
+            active_mqttws_connections,
+            active_mqttws_connections,
+            <<"The number of active MQTT WS(S) connections.">>
+        ),
+        m(
+            gauge,
+            [],
+            total_active_connections,
+            total_active_connections,
+            <<"The total number of active MQTT and MQTTWS connections.">>
+        )
     ].
 
 -spec system_statistics() -> [{metric_id(), any()}].
diff --git a/apps/vmq_server/src/vmq_ranch_config.erl b/apps/vmq_server/src/vmq_ranch_config.erl
@@ -59,10 +59,10 @@ start_link() ->
 stop_all_mqtt_listeners(KillSessions) ->
     lists:foreach(
         fun
-            ({mqtt, Addr, Port, _, _, _, _}) -> stop_listener(Addr, Port, KillSessions);
-            ({mqtts, Addr, Port, _, _, _, _}) -> stop_listener(Addr, Port, KillSessions);
-            ({mqttws, Addr, Port, _, _, _, _}) -> stop_listener(Addr, Port, KillSessions);
-            ({mqttwss, Addr, Port, _, _, _, _}) -> stop_listener(Addr, Port, KillSessions);
+            ({mqtt, Addr, Port, _, _, _, _, _, _}) -> stop_listener(Addr, Port, KillSessions);
+            ({mqtts, Addr, Port, _, _, _, _, _, _}) -> stop_listener(Addr, Port, KillSessions);
+            ({mqttws, Addr, Port, _, _, _, _, _, _}) -> stop_listener(Addr, Port, KillSessions);
+            ({mqttwss, Addr, Port, _, _, _, _, _, _}) -> stop_listener(Addr, Port, KillSessions);
             (_) -> ignore
         end,
         listeners()
@@ -135,51 +135,38 @@ start_listener(Type, Addr, Port, {TransportOpts, Opts}) ->
     end.
 
 listeners() ->
-    lists:foldl(
-        fun
-            ({ranch_server, _, _, _}, Acc) ->
-                Acc;
-            ({{ranch_listener_sup, {Ip, Port}}, Status, supervisor, _}, Acc) ->
-                {ok, {Type, Opts}} = get_listener_config(Ip, Port),
-                MountPoint = proplists:get_value(mountpoint, Opts, ""),
-                MaxConnections = proplists:get_value(
-                    max_connections,
-                    Opts,
-                    vmq_config:get_env(max_connections)
-                ),
-                Status1 =
-                    case Status of
-                        restarting ->
-                            restarting;
-                        undefined ->
-                            stopped;
-                        Pid when is_pid(Pid) ->
-                            case
-                                lists:keyfind(
-                                    ranch_acceptors_sup, 1, supervisor:which_children(Pid)
-                                )
-                            of
-                                false ->
-                                    not_found;
-                                {_, restarting, supervisor, _} ->
-                                    restarting;
-                                {_, undefined, supervisor, _} ->
-                                    stopped;
-                                {_, AcceptorPid, supervisor, _} when is_pid(AcceptorPid) ->
-                                    running
-                            end
-                    end,
-                StrIp = inet:ntoa(Ip),
-                StrPort = integer_to_list(Port),
-                ProxyProtocol = proplists:get_value(
-                    proxy_protocol,
-                    Opts,
-                    false
-                ),
-                [{Type, StrIp, StrPort, Status1, MountPoint, MaxConnections, ProxyProtocol} | Acc]
+    maps:fold(
+        fun({Ip, Port}, ConfigMap, Acc) ->
+            {ok, {Type, Opts}} = get_listener_config(Ip, Port),
+            MountPoint = proplists:get_value(mountpoint, Opts, ""),
+            MaxConnections = proplists:get_value(
+                max_connections,
+                Opts,
+                vmq_config:get_env(max_connections)
+            ),
+            ActiveConnections = maps:get(active_connections, ConfigMap),
+            % the highest number of connections seen
+            AllConnections = maps:get(all_connections, ConfigMap),
+            Status = maps:get(status, ConfigMap),
+            StrIp =
+                case Ip of
+                    {local, FS} -> {local, FS};
+                    _ -> inet:ntoa(Ip)
+                end,
+            StrPort = integer_to_list(Port),
+            ProxyProtocol = proplists:get_value(
+                proxy_protocol,
+                Opts,
+                false
+            ),
+            [
+                {Type, StrIp, StrPort, Status, MountPoint, MaxConnections, ProxyProtocol,
+                    ActiveConnections, AllConnections}
+                | Acc
+            ]
         end,
         [],
-        supervisor:which_children(ranch_sup)
+        ranch:info()
     ).
 
 get_listener_config(Addr, Port) ->
diff --git a/apps/vmq_server/src/vmq_ranch_sup.erl b/apps/vmq_server/src/vmq_ranch_sup.erl
@@ -16,7 +16,7 @@
 -behaviour(supervisor).
 
 %% API
--export([start_link/0]).
+-export([start_link/0, active_mqtt_connections/0]).
 
 %% Supervisor callbacks
 -export([init/1]).
@@ -27,6 +27,26 @@
 %%% API functions
 %%%===================================================================
 
+-spec active_mqtt_connections() -> {non_neg_integer(), non_neg_integer()}.
+active_mqtt_connections() ->
+    lists:foldl(
+        fun
+            % Unix sockets will also be counted under "mqtt"
+            ({mqtt, _, _, _, _, _, _, Active, _}, {MQTT, WS}) ->
+                {MQTT + Active, WS};
+            ({mqtts, _, _, _, _, _, _, Active, _}, {MQTT, WS}) ->
+                {MQTT + Active, WS};
+            ({mqttws, _, _, _, _, _, _, Active, _}, {MQTT, WS}) ->
+                {MQTT, WS + Active};
+            ({mqttwss, _, _, _, _, _, _, Active, _}, {MQTT, WS}) ->
+                {MQTT, WS + Active};
+            (_, Sum) ->
+                Sum
+        end,
+        {0, 0},
+        vmq_ranch_config:listeners()
+    ).
+
 %%--------------------------------------------------------------------
 %% @doc
 %% Starts the supervisor
diff --git a/apps/vmq_server/src/vmq_status_http.erl b/apps/vmq_server/src/vmq_status_http.erl
@@ -69,12 +69,12 @@ cluster_status() ->
 
 node_status() ->
     % Total Connections
-    SocketOpen = counter_val(?METRIC_SOCKET_OPEN),
-    SocketClose = counter_val(?METRIC_SOCKET_CLOSE),
-    TotalConnections = SocketOpen - SocketClose,
+    TotalActiveMqttConnections = lists:sum(
+        tuple_to_list(vmq_ranch_sup:active_mqtt_connections())
+    ),
     % Total Online Queues
     TotalQueues = vmq_queue_sup_sup:nr_of_queues(),
-    TotalOfflineQueues = TotalQueues - TotalConnections,
+    TotalOfflineQueues = TotalQueues - TotalActiveMqttConnections,
     TotalPublishIn =
         counter_val(?MQTT4_PUBLISH_RECEIVED) +
             counter_val(?MQTT5_PUBLISH_RECEIVED),
@@ -90,7 +90,7 @@ node_status() ->
     {NrOfSubs, _SMemory} = vmq_reg_trie:stats(),
     {NrOfRetain, _RMemory} = vmq_retain_srv:stats(),
     {ok, [
-        {<<"num_online">>, TotalConnections},
+        {<<"num_online">>, TotalActiveMqttConnections},
         {<<"num_offline">>, TotalOfflineQueues},
         {<<"msg_in">>, TotalPublishIn},
         {<<"msg_out">>, TotalPublishOut},
@@ -118,12 +118,17 @@ counter_val(C) ->
 
 listeners() ->
     lists:foldl(
-        fun({Type, Ip, Port, Status, MP, MaxConns, PP}, Acc) ->
+        fun({Type, Ip, Port, Status, MP, MaxConns, PP, _, _}, Acc) ->
+            Ip1 =
+                case Ip of
+                    {local, FS} -> list_to_binary(FS);
+                    Any -> list_to_binary(Any)
+                end,
             [
                 [
                     {type, Type},
                     {status, Status},
-                    {ip, list_to_binary(Ip)},
+                    {ip, Ip1},
                     {port, list_to_integer(Port)},
                     {mountpoint, MP},
                     {max_conns, MaxConns},

Original file line number	Diff line number	Diff line change
`@@ -312,7 +312,7 @@ vmq_listener_show_cmd() ->`
`312`	`312`	`fun(_, [], []) ->`
`313`	`313`	`Table =`
`314`	`314`	`lists:foldl(`
`315`		`- fun({Type, Ip, Port, Status, MP, MaxConns, PP}, Acc) ->`
	`315`	`+ fun({Type, Ip, Port, Status, MP, MaxConns, PP, ActiveConns, AllConns}, Acc) ->`
`316`	`316`	`[`
`317`	`317`	`[`
`318`	`318`	`{type, Type},`
`@@ -321,7 +321,9 @@ vmq_listener_show_cmd() ->`
`321`	`321`	`{port, Port},`
`322`	`322`	`{mountpoint, MP},`
`323`	`323`	`{max_conns, MaxConns},`
`324`		`- {proxy_protocol, PP}`
	`324`	`+ {proxy_protocol, PP},`
	`325`	`+ {active_conns, ActiveConns},`
	`326`	`+ {all_conns, AllConns}`
`325`	`327`	`]`
`326`	`328`	`\| Acc`
`327`	`329`	`]`