Make 'queue.declare' aware of virtual host DQT

at validation time.

DQT = default queue type.

When a client provides no queue type, validation
should take the defaults (virtual host, global,
and the last resort fallback) into account
instead of considering the type to
be "undefined".

References #11457 ##11528

Author: michaelklishin

deps/rabbit/src/rabbit_amqqueue.erl

@@ -63,7 +63,7 @@
 -export([is_server_named_allowed/1]).
 
 -export([check_max_age/1]).
--export([get_queue_type/1, get_resource_vhost_name/1, get_resource_name/1]).
+-export([get_queue_type/1, get_queue_type/2, get_resource_vhost_name/1, get_resource_name/1]).
 
 -export([deactivate_limit_all/2]).
 
@@ -220,8 +220,10 @@ declare(QueueName, Durable, AutoDelete, Args, Owner, ActingUser) ->
     {protocol_error, Type :: atom(), Reason :: string(), Args :: term()}.
 declare(QueueName = #resource{virtual_host = VHost}, Durable, AutoDelete, Args,
         Owner, ActingUser, Node) ->
-    ok = check_declare_arguments(QueueName, Args),
-    Type = get_queue_type(Args),
+    %% note: this is a module name, not a shortcut such as <<"quorum">>
+    DQT = rabbit_vhost:default_queue_type(VHost, rabbit_queue_type:fallback()),
+    ok = check_declare_arguments(QueueName, Args, DQT),
+    Type = get_queue_type(Args, DQT),
     case rabbit_queue_type:is_enabled(Type) of
         true ->
             Q = amqqueue:new(QueueName,
@@ -248,10 +250,19 @@ declare(QueueName = #resource{virtual_host = VHost}, Durable, AutoDelete, Args,
              [rabbit_misc:rs(QueueName), Type, Node]}
     end.
 
+-spec get_queue_type(Args :: rabbit_framing:amqp_table()) -> rabbit_queue_type:queue_type().
+%% This version is not virtual host metadata-aware but will use
+%% the node-wide default type as well as 'rabbit_queue_type:fallback/0'.
 get_queue_type(Args) ->
+    get_queue_type(Args, rabbit_queue_type:default()).
+
+%% This version should be used together with 'rabbit_vhost:default_queue_type/{1,2}'
+get_queue_type(Args, DefaultQueueType) ->
     case rabbit_misc:table_lookup(Args, <<"x-queue-type">>) of
-        undefined ->
-            rabbit_queue_type:default();
+        {longstr, undefined} ->
+            rabbit_queue_type:discover(DefaultQueueType);
+        {longstr, <<"undefined">>} ->
+            rabbit_queue_type:discover(DefaultQueueType);
         {_, V} ->
             rabbit_queue_type:discover(V)
     end.
@@ -783,7 +794,31 @@ assert_args_equivalence(Q, NewArgs) ->
     QueueTypeArgs = rabbit_queue_type:arguments(queue_arguments, Type),
     rabbit_misc:assert_args_equivalence(ExistingArgs, NewArgs, QueueName, QueueTypeArgs).
 
-check_declare_arguments(QueueName, Args) ->
+-spec maybe_inject_default_queue_type_shortcut_into_args(
+    rabbit_framing:amqp_table(), rabbit_queue_type:queue_type()) -> rabbit_framing:amqp_table().
+maybe_inject_default_queue_type_shortcut_into_args(Args0, DefaultQueueType) ->
+    case rabbit_misc:table_lookup(Args0, <<"x-queue-type">>) of
+        {longstr, undefined} ->
+            %% Important: use a shortcut such as 'quorum' or 'stream' that for the given queue type module
+            inject_default_queue_type_shortcut_into_args(Args0, DefaultQueueType);
+        {longstr, <<"undefined">>} ->
+            %% Important: use a shortcut such as 'quorum' or 'stream' that for the given queue type module
+            inject_default_queue_type_shortcut_into_args(Args0, DefaultQueueType);
+        _ValueIsAlreadySet ->
+            Args0
+    end.
+
+-spec inject_default_queue_type_shortcut_into_args(
+    rabbit_framing:amqp_table(), rabbit_queue_type:queue_type()) -> rabbit_framing:amqp_table().
+inject_default_queue_type_shortcut_into_args(Args0, QueueType) ->
+    Shortcut = rabbit_queue_type:short_alias_of(QueueType),
+    NewVal = rabbit_data_coercion:to_binary(Shortcut),
+    rabbit_misc:set_table_value(Args0, <<"x-queue-type">>, longstr, NewVal).
+
+check_declare_arguments(QueueName, Args0, DefaultQueueType) ->
+    %% If the x-queue-type was not provided by the client, inject the
+    %% (virtual host, global or fallback) default before performing validation. MK.
+    Args = maybe_inject_default_queue_type_shortcut_into_args(Args0, DefaultQueueType),
     check_arguments_type_and_value(QueueName, Args, [{<<"x-queue-type">>, fun check_queue_type/2}]),
     Type = get_queue_type(Args),
     QueueTypeArgs = rabbit_queue_type:arguments(queue_arguments, Type),

deps/rabbit/src/rabbit_queue_type.erl

@@ -18,9 +18,11 @@
          init/0,
          close/1,
          discover/1,
+         short_alias_of/1,
          feature_flag_name/1,
          to_binary/1,
          default/0,
+         fallback/0,
          is_enabled/1,
          is_compatible/4,
          declare/2,
@@ -70,7 +72,7 @@
 %% sequence number typically
 -type correlation() :: term().
 -type arguments() :: queue_arguments | consumer_arguments.
--type queue_type() :: rabbit_classic_queue | rabbit_quorum_queue | rabbit_stream_queue.
+-type queue_type() :: rabbit_classic_queue | rabbit_quorum_queue | rabbit_stream_queue | module().
 %% see AMQP 1.0 §2.6.7
 -type delivery_count() :: sequence_no().
 -type credit() :: uint().
@@ -253,20 +255,49 @@
 -callback notify_decorators(amqqueue:amqqueue()) ->
     ok.
 
+-spec discover(binary() | atom()) -> queue_type().
+discover(<<"undefined">>) ->
+    fallback();
+discover(undefined) ->
+    fallback();
 %% TODO: should this use a registry that's populated on boot?
 discover(<<"quorum">>) ->
     rabbit_quorum_queue;
+discover(rabbit_quorum_queue) ->
+    rabbit_quorum_queue;
 discover(<<"classic">>) ->
     rabbit_classic_queue;
+discover(rabbit_classic_queue) ->
+    rabbit_classic_queue;
+discover(rabbit_stream_queue) ->
+    rabbit_stream_queue;
 discover(<<"stream">>) ->
     rabbit_stream_queue;
 discover(Other) when is_atom(Other) ->
     discover(rabbit_data_coercion:to_binary(Other));
 discover(Other) when is_binary(Other) ->
     T = rabbit_registry:binary_to_type(Other),
+    rabbit_log:debug("Queue type discovery: will look up a module for type '~tp'", [T]),
     {ok, Mod} = rabbit_registry:lookup_module(queue, T),
     Mod.
 
+-spec short_alias_of(queue_type()) -> binary().
+%% The opposite of discover/1: returns a short alias given a module name
+short_alias_of(<<"rabbit_quorum_queue">>) ->
+    <<"quorum">>;
+short_alias_of(rabbit_quorum_queue) ->
+    <<"quorum">>;
+short_alias_of(<<"rabbit_classic_queue">>) ->
+    <<"classic">>;
+short_alias_of(rabbit_classic_queue) ->
+    <<"classic">>;
+short_alias_of(<<"rabbit_stream_queue">>) ->
+    <<"stream">>;
+short_alias_of(rabbit_stream_queue) ->
+    <<"stream">>;
+short_alias_of(_Other) ->
+    undefined.
+
 feature_flag_name(<<"quorum">>) ->
     quorum_queue;
 feature_flag_name(<<"classic">>) ->
@@ -276,10 +307,19 @@ feature_flag_name(<<"stream">>) ->
 feature_flag_name(_) ->
     undefined.
 
+%% If the client does not specify the type, the virtual host does not have any
+%% metadata default, and rabbit.default_queue_type is not set in the application env,
+%% use this type as the last resort.
+-spec fallback() -> queue_type().
+fallback() ->
+    rabbit_classic_queue.
+
+-spec default() -> queue_type().
 default() ->
-    rabbit_misc:get_env(rabbit,
-                        default_queue_type,
-                        rabbit_classic_queue).
+    V = rabbit_misc:get_env(rabbit,
+                            default_queue_type,
+                            fallback()),
+    rabbit_data_coercion:to_atom(V).
 
 -spec to_binary(module()) -> binary().
 to_binary(rabbit_classic_queue) ->

deps/rabbit/src/rabbit_vhost.erl

@@ -22,6 +22,7 @@
 -export([delete_storage/1]).
 -export([vhost_down/1]).
 -export([put_vhost/6]).
+-export([default_queue_type/1, default_queue_type/2]).
 
 %%
 %% API
@@ -481,6 +482,22 @@ default_name() ->
         undefined   -> <<"/">>
     end.
 
+-spec default_queue_type(VirtualHost :: vhost:name()) -> rabbit_queue_type:queue_type().
+default_queue_type(VirtualHost) ->
+    default_queue_type(VirtualHost, rabbit_queue_type:fallback()).
+-spec default_queue_type(VirtualHost :: vhost:name(), Fallback :: rabbit_queue_type:queue_type()) -> rabbit_queue_type:queue_type().
+default_queue_type(VirtualHost, FallbackQueueType) ->
+    case exists(VirtualHost) of
+        false -> FallbackQueueType;
+        true ->
+            Record = lookup(VirtualHost),
+            case vhost:get_default_queue_type(Record) of
+                undefined       -> FallbackQueueType;
+                <<"undefined">> -> FallbackQueueType;
+                Type            -> Type
+            end
+end.
+
 -spec lookup(vhost:name()) -> vhost:vhost() | rabbit_types:ok_or_error(any()).
 lookup(VHostName) ->
     case rabbit_db_vhost:get(VHostName) of

deps/rabbitmq_management/test/stats_SUITE.erl

@@ -8,6 +8,7 @@
 -module(stats_SUITE).
 
 -include_lib("proper/include/proper.hrl").
+-include_lib("eunit/include/eunit.hrl").
 -include_lib("rabbitmq_management_agent/include/rabbit_mgmt_records.hrl").
 
 -compile(export_all).
@@ -175,4 +176,4 @@ format_range_constant(_Config) ->
                                          SamplesFun),
     5 = proplists:get_value(publish, Got),
     PD = proplists:get_value(publish_details, Got),
-    0.0 = proplists:get_value(rate, PD).
+    ?assertEqual(0.0, proplists:get_value(rate, PD)).