From 7c22e5ef20a24fa8a1b8cc1cbc6436a83b150af1 Mon Sep 17 00:00:00 2001 From: Lois Soto Lopez Date: Thu, 7 Mar 2024 09:41:59 +0100 Subject: [PATCH] Prevent not_found on list web mqtt connections Fixes #9302 --- ...Commands.ListWebMqttConnectionsCommand.erl | 86 +++++++++++++++++++ .../src/rabbit_web_mqtt_app.erl | 32 +++++-- .../src/rabbit_web_mqtt_handler.erl | 18 ++++ 3 files changed, 127 insertions(+), 9 deletions(-) create mode 100644 deps/rabbitmq_web_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand.erl diff --git a/deps/rabbitmq_web_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand.erl b/deps/rabbitmq_web_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand.erl new file mode 100644 index 000000000000..7ecae78cde4e --- /dev/null +++ b/deps/rabbitmq_web_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand.erl @@ -0,0 +1,86 @@ +%% This Source Code Form is subject to the terms of the Mozilla Public +%% License, v. 2.0. If a copy of the MPL was not distributed with this +%% file, You can obtain one at https://mozilla.org/MPL/2.0/. +%% +%% Copyright (c) 2007-2024 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved. + +-module('Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand'). + +-include_lib("rabbitmq_mqtt/include/rabbit_mqtt.hrl"). + +-behaviour('Elixir.RabbitMQ.CLI.CommandBehaviour'). + +-export([formatter/0, + scopes/0, + switches/0, + aliases/0, + usage/0, + usage_additional/0, + usage_doc_guides/0, + banner/2, + validate/2, + merge_defaults/2, + run/2, + output/2, + description/0, + help_section/0]). + +formatter() -> 'Elixir.RabbitMQ.CLI.Formatters.Table'. +scopes() -> [ctl, diagnostics]. +switches() -> [{verbose, boolean}]. +aliases() -> [{'V', verbose}]. + +description() -> <<"Lists all Web MQTT connections">>. + +help_section() -> + {plugin, web_mqtt}. + +validate(Args, _) -> + InfoItems = lists:map(fun atom_to_list/1, ?INFO_ITEMS), + case 'Elixir.RabbitMQ.CLI.Ctl.InfoKeys':validate_info_keys(Args, + InfoItems) of + {ok, _} -> ok; + Error -> Error + end. + +merge_defaults([], Opts) -> + merge_defaults([<<"client_id">>, <<"conn_name">>], Opts); +merge_defaults(Args, Opts) -> + {Args, maps:merge(#{verbose => false}, Opts)}. + +usage() -> + <<"list_web_mqtt_connections [ ...]">>. + +usage_additional() -> + Prefix = <<" must be one of ">>, + InfoItems = 'Elixir.Enum':join(lists:usort(?INFO_ITEMS), <<", ">>), + [ + {<<"">>, <>} + ]. + +usage_doc_guides() -> + [?MQTT_GUIDE_URL]. + +run(Args, #{node := NodeName, + timeout := Timeout, + verbose := Verbose}) -> + InfoKeys = case Verbose of + true -> ?INFO_ITEMS; + false -> 'Elixir.RabbitMQ.CLI.Ctl.InfoKeys':prepare_info_keys(Args) + end, + + Nodes = 'Elixir.RabbitMQ.CLI.Core.Helpers':nodes_in_cluster(NodeName), + + 'Elixir.RabbitMQ.CLI.Ctl.RpcStream':receive_list_items( + NodeName, + rabbit_web_mqtt_app, + emit_connection_info_all, + [Nodes, InfoKeys], + Timeout, + InfoKeys, + length(Nodes)). + +banner(_, _) -> <<"Listing Web MQTT connections ...">>. + +output(Result, _Opts) -> + 'Elixir.RabbitMQ.CLI.DefaultOutput':output(Result). diff --git a/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl b/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl index 5f2d1bb9e569..0cccdfb14a2a 100644 --- a/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl +++ b/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_app.erl @@ -12,7 +12,9 @@ start/2, prep_stop/1, stop/1, - list_connections/0 + list_connections/0, + emit_connection_info_all/4, + emit_connection_info_local/3 ]). %% Dummy supervisor - see Ulf Wiger's comment at @@ -52,6 +54,25 @@ list_connections() -> TLSPids = connection_pids_of_protocol(?TLS_PROTOCOL), PlainPids ++ TLSPids. +-spec emit_connection_info_all([node()], rabbit_types:info_keys(), reference(), pid()) -> term(). +emit_connection_info_all(Nodes, Items, Ref, AggregatorPid) -> + Pids = [spawn_link(Node, ?MODULE, emit_connection_info_local, + [Items, Ref, AggregatorPid]) + || Node <- Nodes], + + rabbit_control_misc:await_emitters_termination(Pids). + +-spec emit_connection_info_local(rabbit_types:info_keys(), reference(), pid()) -> ok. +emit_connection_info_local(Items, Ref, AggregatorPid) -> + LocalPids = list_connections(), + emit_connection_info(Items, Ref, AggregatorPid, LocalPids). + +emit_connection_info(Items, Ref, AggregatorPid, Pids) -> + rabbit_control_misc:emitting_map_with_exit_handler( + AggregatorPid, Ref, + fun(Pid) -> + rabbit_web_mqtt_handler:info(Pid, Items) + end, Pids). %% %% Implementation %% @@ -59,16 +80,9 @@ list_connections() -> connection_pids_of_protocol(Protocol) -> case rabbit_networking:ranch_ref_of_protocol(Protocol) of undefined -> []; - AcceptorRef -> - lists:map(fun cowboy_ws_connection_pid/1, ranch:procs(AcceptorRef, connections)) + AcceptorRef -> ranch:procs(AcceptorRef, connections) end. --spec cowboy_ws_connection_pid(pid()) -> pid(). -cowboy_ws_connection_pid(RanchConnPid) -> - Children = supervisor:which_children(RanchConnPid), - {cowboy_clear, Pid, _, _} = lists:keyfind(cowboy_clear, 1, Children), - Pid. - mqtt_init() -> CowboyOpts0 = maps:from_list(get_env(cowboy_opts, [])), CowboyWsOpts = maps:from_list(get_env(cowboy_ws_opts, [])), diff --git a/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl b/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl index fd3df1c9290e..d336fb6a3e29 100644 --- a/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl +++ b/deps/rabbitmq_web_mqtt/src/rabbit_web_mqtt_handler.erl @@ -23,6 +23,7 @@ ]). -export([conserve_resources/3]). +-export([info/2]). %% cowboy_sub_protocol -export([upgrade/4, @@ -94,6 +95,19 @@ init(Req, Opts) -> end end. +%% We cannot use a gen_server call, because the handler process is a +%% special cowboy_websocket process (not a gen_server) which assumes +%% all gen_server calls are supervisor calls, and does not pass on the +%% request to this callback module. (see cowboy_websocket:loop/3 and +%% cowboy_children:handle_supervisor_call/4) However using a generic +%% gen:call with a special label ?MODULE works fine. +-spec info(pid(), rabbit_types:info_keys()) -> + rabbit_types:infos(). +info(Pid, all) -> + info(Pid, ?INFO_ITEMS); +info(Pid, Items) -> + {ok, Res} = gen:call(Pid, ?MODULE, {info, Items}), + Res. -spec websocket_init(state()) -> {cowboy_websocket:commands(), state()} | {cowboy_websocket:commands(), state(), hibernate}. @@ -244,6 +258,10 @@ websocket_info(connection_created, State) -> rabbit_core_metrics:connection_created(self(), Infos), rabbit_event:notify(connection_created, Infos), {[], State, hibernate}; +websocket_info({?MODULE, From, {info, Items}}, State) -> + Infos = infos(Items, State), + gen:reply(From, Infos), + {[], State, hibernate}; websocket_info(Msg, State) -> ?LOG_WARNING("Web MQTT: unexpected message ~tp", [Msg]), {[], State, hibernate}.