Merge pull request #5410 from rabbitmq/rabbitmq-server-5305-follow-up

ctl add_vhost: check if relevant feature flags are enabled

Author: michaelklishin

deps/rabbit/src/rabbit_queue_type.erl

@@ -13,6 +13,7 @@
          init/0,
          close/1,
          discover/1,
+         feature_flag_name/1,
          default/0,
          is_enabled/1,
          is_compatible/4,
@@ -223,14 +224,23 @@
 -callback notify_decorators(amqqueue:amqqueue()) ->
     ok.
 
-%% TODO: this should be controlled by a registry that is populated on boot
+%% TODO: should this use a registry that's populated on boot?
 discover(<<"quorum">>) ->
     rabbit_quorum_queue;
 discover(<<"classic">>) ->
     rabbit_classic_queue;
 discover(<<"stream">>) ->
     rabbit_stream_queue.
 
+feature_flag_name(<<"quorum">>) ->
+    quorum_queue;
+feature_flag_name(<<"classic">>) ->
+    undefined;
+feature_flag_name(<<"stream">>) ->
+    stream_queue;
+feature_flag_name(_) ->
+    undefined.
+
 default() ->
     rabbit_classic_queue.
 

deps/rabbit/src/rabbit_vhost.erl

@@ -166,9 +166,17 @@ do_add(Name, Metadata, ActingUser) ->
     %% validate default_queue_type
     case Metadata of
         #{default_queue_type := DQT} ->
+            %% check that the queue type is known
             try rabbit_queue_type:discover(DQT) of
                 _ ->
-                    ok
+                    case rabbit_queue_type:feature_flag_name(DQT) of
+                        undefined -> ok;
+                        Flag when is_atom(Flag) ->
+                            case rabbit_feature_flags:is_enabled(Flag) of
+                                true  -> ok;
+                                false -> throw({error, queue_type_feature_flag_is_not_enabled})
+                            end
+                    end
             catch _:_ ->
                       throw({error, invalid_queue_type})
             end;

deps/rabbitmq_cli/lib/rabbitmq/cli/core/feature_flags.ex

@@ -5,11 +5,31 @@
 ## Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 
 defmodule RabbitMQ.CLI.Core.FeatureFlags do
+  alias RabbitMQ.CLI.Core.ExitCodes
 
   #
   # API
   #
 
+  def is_enabled_remotely(node_name, feature_flag) do
+    case :rabbit_misc.rpc_call(node_name, :rabbit_feature_flags, :is_enabled, [feature_flag]) do
+      true  -> true
+      false -> false
+      {:error, _} = error -> error
+    end
+  end
+
+  def assert_feature_flag_enabled(node_name, feature_flag, success_fun) do
+    case is_enabled_remotely(node_name, feature_flag) do
+      true  ->
+        success_fun.()
+      false ->
+        {:error, ExitCodes.exit_dataerr(), "The #{feature_flag} feature flag is not enabled on the target node"}
+      {:error, _} = error ->
+        error
+    end
+  end
+
   def feature_flag_lines(feature_flags) do
     feature_flags
     |> Enum.map(fn %{name: name, state: state} ->

deps/rabbitmq_cli/lib/rabbitmq/cli/ctl/commands/add_vhost_command.ex

@@ -5,7 +5,7 @@
 ## Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 
 defmodule RabbitMQ.CLI.Ctl.Commands.AddVhostCommand do
-  alias RabbitMQ.CLI.Core.{DocGuide, Helpers}
+  alias RabbitMQ.CLI.Core.{DocGuide, ExitCodes, FeatureFlags, Helpers}
 
   @behaviour RabbitMQ.CLI.CommandBehaviour
 
@@ -24,15 +24,29 @@ defmodule RabbitMQ.CLI.Ctl.Commands.AddVhostCommand do
     meta = %{description: desc,
              tags: parse_tags(tags),
              default_queue_type: default_qt}
-    :rabbit_misc.rpc_call(node_name, :rabbit_vhost, :add, [vhost, meta, Helpers.cli_acting_user()])
+    # check if the respective feature flag is enabled
+    case default_qt do
+      "quorum" ->
+        FeatureFlags.assert_feature_flag_enabled(node_name, :quorum_queue, fn () ->
+          :rabbit_misc.rpc_call(node_name, :rabbit_vhost, :add, [vhost, meta, Helpers.cli_acting_user()])
+        end)
+      "stream" ->
+        FeatureFlags.assert_feature_flag_enabled(node_name, :stream_queue, fn () ->
+          :rabbit_misc.rpc_call(node_name, :rabbit_vhost, :add, [vhost, meta, Helpers.cli_acting_user()])
+        end)
+      _ ->
+        :rabbit_misc.rpc_call(node_name, :rabbit_vhost, :add, [vhost, meta, Helpers.cli_acting_user()])
+    end
   end
   def run([vhost], %{node: node_name, description: desc, tags: tags}) do
     :rabbit_misc.rpc_call(node_name, :rabbit_vhost, :add, [vhost, description, tags, Helpers.cli_acting_user()])
   end
   def run([vhost], %{node: node_name}) do
     :rabbit_misc.rpc_call(node_name, :rabbit_vhost, :add, [vhost, Helpers.cli_acting_user()])
   end
-
+  def output({:error, :invalid_queue_type}, _opts) do
+    {:error, ExitCodes.exit_usage, "Unsupported default queue type"}
+  end
   use RabbitMQ.CLI.DefaultOutput
 
   def usage, do: "add_vhost <vhost> [--description <description> --tags \"<tag1>,<tag2>,<...>\" --default-queue-type <quorum|classic|stream>]"