From 245fa0ece2c00d949ac1baea5a316adb8d751024 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 --- deps/rabbit/src/rabbit_networking.erl | 10 ++- deps/rabbitmq_mqtt/include/rabbit_mqtt.hrl | 3 + ...tl.Commands.ListMqttConnectionsCommand.erl | 1 + deps/rabbitmq_mqtt/src/rabbit_mqtt.erl | 9 +- deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl | 15 ++-- ...Commands.ListWebMqttConnectionsCommand.erl | 86 +++++++++++++++++++ .../src/rabbit_web_mqtt_app.erl | 40 +++++---- .../src/rabbit_web_mqtt_handler.erl | 18 ++++ 8 files changed, 155 insertions(+), 27 deletions(-) create mode 100644 deps/rabbitmq_web_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListWebMqttConnectionsCommand.erl diff --git a/deps/rabbit/src/rabbit_networking.erl b/deps/rabbit/src/rabbit_networking.erl index 9ae8561a378c..4563da64d1b4 100644 --- a/deps/rabbit/src/rabbit_networking.erl +++ b/deps/rabbit/src/rabbit_networking.erl @@ -34,7 +34,8 @@ force_connection_event_refresh/1, force_non_amqp_connection_event_refresh/1, handshake/2, tcp_host/1, ranch_ref/1, ranch_ref/2, ranch_ref_of_protocol/1, - listener_of_protocol/1, stop_ranch_listener_of_protocol/1]). + listener_of_protocol/1, stop_ranch_listener_of_protocol/1, + list_local_connections_of_protocol/1]). %% Used by TCP-based transports, e.g. STOMP adapter -export([tcp_listener_addresses/1, @@ -252,6 +253,13 @@ stop_ranch_listener_of_protocol(Protocol) -> ranch:stop_listener(Ref) end. +-spec list_local_connections_of_protocol(atom()) -> [pid()]. +list_local_connections_of_protocol(Protocol) -> + case ranch_ref_of_protocol(Protocol) of + undefined -> []; + AcceptorRef -> ranch:procs(AcceptorRef, connections) + end. + -spec start_tcp_listener( listener_config(), integer()) -> 'ok' | {'error', term()}. diff --git a/deps/rabbitmq_mqtt/include/rabbit_mqtt.hrl b/deps/rabbitmq_mqtt/include/rabbit_mqtt.hrl index a7376897bc3d..4d1b3656fa60 100644 --- a/deps/rabbitmq_mqtt/include/rabbit_mqtt.hrl +++ b/deps/rabbitmq_mqtt/include/rabbit_mqtt.hrl @@ -12,7 +12,10 @@ -define(PERSISTENT_TERM_EXCHANGE, mqtt_exchange). -define(DEFAULT_MQTT_EXCHANGE, <<"amq.topic">>). -define(MQTT_GUIDE_URL, <<"https://rabbitmq.com/docs/mqtt/">>). +-define(WEB_MQTT_GUIDE_URL, <<"https://rabbitmq.com/docs/web-mqtt/">>). +-define(MQTT_TCP_PROTOCOL, 'mqtt'). +-define(MQTT_TLS_PROTOCOL, 'mqtt/ssl'). -define(MQTT_PROTO_V3, mqtt310). -define(MQTT_PROTO_V4, mqtt311). -define(MQTT_PROTO_V5, mqtt50). diff --git a/deps/rabbitmq_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListMqttConnectionsCommand.erl b/deps/rabbitmq_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListMqttConnectionsCommand.erl index 98fe6968122d..6b3204051912 100644 --- a/deps/rabbitmq_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListMqttConnectionsCommand.erl +++ b/deps/rabbitmq_mqtt/src/Elixir.RabbitMQ.CLI.Ctl.Commands.ListMqttConnectionsCommand.erl @@ -80,6 +80,7 @@ run(Args, #{node := NodeName, InfoKeys, length(Nodes)). + banner(_, _) -> <<"Listing MQTT connections ...">>. output(Result, _Opts) -> diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt.erl index fb60b6610a02..86cf09d149b4 100644 --- a/deps/rabbitmq_mqtt/src/rabbit_mqtt.erl +++ b/deps/rabbitmq_mqtt/src/rabbit_mqtt.erl @@ -62,7 +62,7 @@ emit_connection_info_all(Nodes, Items, Ref, AggregatorPid) -> -spec emit_connection_info_local(rabbit_types:info_keys(), reference(), pid()) -> ok. emit_connection_info_local(Items, Ref, AggregatorPid) -> - LocalPids = local_connection_pids(), + LocalPids = list_local_mqtt_connections(), emit_connection_info(Items, Ref, AggregatorPid, LocalPids). emit_connection_info(Items, Ref, AggregatorPid, Pids) -> @@ -93,6 +93,13 @@ local_connection_pids() -> end, pg:which_groups(PgScope)) end. +%% This function excludes Web MQTT connections +list_local_mqtt_connections() -> + PlainPids = rabbit_networking:list_local_connections_of_protocol(?MQTT_TCP_PROTOCOL), + TLSPids = rabbit_networking:list_local_connections_of_protocol(?MQTT_TLS_PROTOCOL), + PlainPids ++ TLSPids. + + init_global_counters() -> lists:foreach(fun init_global_counters/1, [?MQTT_PROTO_V3, ?MQTT_PROTO_V4, diff --git a/deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl b/deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl index 79185a12348f..2bdacebb58e2 100644 --- a/deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl +++ b/deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl @@ -13,9 +13,6 @@ -export([start_link/2, init/1, stop_listeners/0]). --define(TCP_PROTOCOL, 'mqtt'). --define(TLS_PROTOCOL, 'mqtt/ssl'). - start_link(Listeners, []) -> supervisor:start_link({local, ?MODULE}, ?MODULE, [Listeners]). @@ -66,8 +63,8 @@ init([{Listeners, SslListeners0}]) -> -spec stop_listeners() -> ok. stop_listeners() -> - _ = rabbit_networking:stop_ranch_listener_of_protocol(?TCP_PROTOCOL), - _ = rabbit_networking:stop_ranch_listener_of_protocol(?TLS_PROTOCOL), + _ = rabbit_networking:stop_ranch_listener_of_protocol(?MQTT_TCP_PROTOCOL), + _ = rabbit_networking:stop_ranch_listener_of_protocol(?MQTT_TLS_PROTOCOL), ok. %% @@ -86,7 +83,7 @@ tcp_listener_spec([Address, SocketOpts, NumAcceptors, ConcurrentConnsSups]) -> rabbit_mqtt_listener_sup, Address, SocketOpts, - transport(?TCP_PROTOCOL), + transport(?MQTT_TCP_PROTOCOL), rabbit_mqtt_reader, [], mqtt, @@ -101,7 +98,7 @@ ssl_listener_spec([Address, SocketOpts, SslOpts, NumAcceptors, ConcurrentConnsSu rabbit_mqtt_listener_sup, Address, SocketOpts ++ SslOpts, - transport(?TLS_PROTOCOL), + transport(?MQTT_TLS_PROTOCOL), rabbit_mqtt_reader, [], 'mqtt/ssl', @@ -111,7 +108,7 @@ ssl_listener_spec([Address, SocketOpts, SslOpts, NumAcceptors, ConcurrentConnsSu "MQTT TLS listener" ). -transport(?TCP_PROTOCOL) -> +transport(?MQTT_TCP_PROTOCOL) -> ranch_tcp; -transport(?TLS_PROTOCOL) -> +transport(?MQTT_TLS_PROTOCOL) -> ranch_ssl. 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..5a89ab3ca36a --- /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() -> + [?WEB_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..fc6424ffae4f 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 @@ -48,27 +50,33 @@ init([]) -> {ok, {{one_for_one, 1, 5}, []}}. -spec list_connections() -> [pid()]. list_connections() -> - PlainPids = connection_pids_of_protocol(?TCP_PROTOCOL), - TLSPids = connection_pids_of_protocol(?TLS_PROTOCOL), + PlainPids = rabbit_networking:list_local_connections_of_protocol(?TCP_PROTOCOL), + TLSPids = rabbit_networking:list_local_connections_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 %% -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)) - 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}.