Allow to disable Classic Mirrored Queues via configuration

PR #7390 introduces deprecated features and their lifecycle
management. One of the first applications should be kick-starting the
process of Classic Mirrored Queues deprecation. But that would be to
big of a change to be backported to any of the current releases, so
this commit introduces a simplified version of that deprecation.

To disable CMQs one need to add the following line to the config:
```
permit_deprecated_features.classic_mirrored_queues = false
```

What it does when CMQs are disabled via configuration:
- Doesn't allow to create user/operator policies that enable mirroring ("ha-mode")
- Prevent RabbitMQ startup if such a policy was previously configured

Differences from the final implementation that will be using deprecated features system:
- No warnings are issued when CMQs are not disabled, but being used
- It's not possible to set `permit_deprecated_features` option to the `true` value

This differences ensure that one only enables this feature when they
are absolutely sure what they are doing, but in a way that won't
interfere with a subsequent phased deprecation process.

deps/rabbit/priv/schema/rabbit.schema

@@ -2088,6 +2088,32 @@ end}.
     {datatype, integer}
 ]}.
 
+%%
+%% Feature flags and deprecated features
+%% =====================================
+%%
+
+%% NOTE: `true` is intentionally omitted - add it back when mirrored
+%% queue deprecation is converted to use deprecated features system.
+{mapping,
+ "permit_deprecated_features.$name", "rabbit.permit_deprecated_features",
+    [{datatype, {enum, [false]}}]
+}.
+
+%% This converts:
+%%     permit_deprecated_features.my_feature = true
+%% to:
+%%     {rabbit, [{permit_deprecated_features, #{my_feature => true}}]}.
+{translation, "rabbit.permit_deprecated_features",
+ fun(Conf) ->
+         Settings = cuttlefish_variable:filter_by_prefix(
+                      "permit_deprecated_features", Conf),
+         maps:from_list(
+           [{list_to_atom(FeatureName), State}
+            || {["permit_deprecated_features", FeatureName], State}
+               <- Settings])
+ end}.
+
 % ==========================
 % Kernel section
 % ==========================

deps/rabbit/src/rabbit.erl

@@ -193,6 +193,12 @@
                    [{description, "core initialized"},
                     {requires,    kernel_ready}]}).
 
+-rabbit_boot_step({deprecate_cmqs,
+                   [{description, "checks whether mirrored queues are disabled"},
+                    {mfa, {rabbit_mirror_queue_misc, prevent_startup_when_mirroring_is_disabled_but_configured, []}},
+                    {requires, [database]},
+                    {enables, [recovery]}]}).
+
 -rabbit_boot_step({recovery,
                    [{description, "exchange, queue and binding recovery"},
                     {mfa,         {rabbit, recover, []}},

deps/rabbit/src/rabbit_mirror_queue_misc.erl

@@ -26,6 +26,8 @@
 
 -export([get_replicas/1, transfer_leadership/2, migrate_leadership_to_existing_replica/2]).
 
+-export([prevent_startup_when_mirroring_is_disabled_but_configured/0]).
+
 %% for testing only
 -export([module/1]).
 
@@ -724,6 +726,63 @@ maybe_drop_master_after_sync(Q) when ?is_amqqueue(Q) ->
 
 %%----------------------------------------------------------------------------
 
+mirroring_policies() ->
+    Policies = rabbit_policy:list_as_maps(),
+    OpPolicies = rabbit_policy:list_op_as_maps(),
+    IsMirroringPolicy = fun
+        (#{definition := #{<<"ha-mode">> := _}}) ->
+            true;
+        (_) ->
+            false
+    end,
+    {lists:filter(IsMirroringPolicy, Policies), lists:filter(IsMirroringPolicy, OpPolicies)}.
+
+report_vhosts_using_mirroring(MirrorPolicies, PolicyType) ->
+    PerVhost = lists:foldr(
+                 fun (#{vhost := VHost, name := Name}, Acc) ->
+                         maps:update_with(
+                           VHost,
+                           fun (Message) -> <<Name/binary, ", ", Message/binary>> end,
+                           <<"Virtual host ", VHost/binary, " has ", PolicyType/binary,
+                             " policies that configure mirroring: ", Name/binary>>,
+                           Acc)
+                 end,
+                 #{}, MirrorPolicies),
+    lists:foreach(
+      fun (Msg) -> rabbit_log:error("~ts", [Msg]) end,
+      maps:values(PerVhost)).
+
+prevent_startup_when_mirroring_is_disabled_but_configured() ->
+    case are_cmqs_permitted() of
+        true ->
+            ok;
+        false ->
+            Error = {error, {failed_to_deny_deprecated_features, [classic_mirrored_queues]}},
+            case mirroring_policies() of
+                {[], []} ->
+                    ok;
+                {Pols, []} ->
+                    report_vhosts_using_mirroring(Pols, <<"user">>),
+                    exit(Error);
+                {[], OpPols} ->
+                    report_vhosts_using_mirroring(OpPols, <<"operator">>),
+                    exit(Error);
+                {Pols, OpPols} ->
+                    report_vhosts_using_mirroring(Pols, <<"user">>),
+                    report_vhosts_using_mirroring(OpPols, <<"operator">>),
+                    exit(Error)
+            end
+    end.
+
+are_cmqs_permitted() ->
+    %% FeatureName = classic_mirrored_queues,
+    %% rabbit_deprecated_features:is_permitted(FeatureName).
+    case application:get_env(rabbit, permit_deprecated_features) of
+        {ok, #{classic_mirrored_queues := false}} ->
+            false;
+        _ -> true
+    end.
+
 validate_policy(KeyList) ->
     Mode = proplists:get_value(<<"ha-mode">>, KeyList, none),
     Params = proplists:get_value(<<"ha-params">>, KeyList, none),
@@ -734,10 +793,12 @@ validate_policy(KeyList) ->
                           <<"ha-promote-on-shutdown">>, KeyList, none),
     PromoteOnFailure = proplists:get_value(
                           <<"ha-promote-on-failure">>, KeyList, none),
-    case {Mode, Params, SyncMode, SyncBatchSize, PromoteOnShutdown, PromoteOnFailure} of
-        {none, none, none, none, none, none} ->
+    case {are_cmqs_permitted(), Mode, Params, SyncMode, SyncBatchSize, PromoteOnShutdown, PromoteOnFailure} of
+        {_, none, none, none, none, none, none} ->
             ok;
-        {none, _, _, _, _, _} ->
+        {false, _, _, _, _, _, _} ->
+            {error, "Mirrored queues disabled via configuration", []};
+        {_, none, _, _, _, _, _} ->
             {error, "ha-mode must be specified to specify ha-params, "
              "ha-sync-mode or ha-promote-on-shutdown", []};
         _ ->

deps/rabbit/src/rabbit_policy.erl

@@ -41,7 +41,7 @@
 -export([parse_set_op/7, set_op/7, delete_op/3, lookup_op/2, list_op/0, list_op/1, list_op/2,
          list_formatted_op/1, list_formatted_op/3,
          match_all/2, match_as_map/1, match_op_as_map/1, definition_keys/1,
-         list_in/1, list_in/2, list_as_maps/0, list_as_maps/1, list_op_as_maps/1
+         list_in/1, list_in/2, list_as_maps/0, list_as_maps/1, list_op_as_maps/0, list_op_as_maps/1
         ]).
 -export([sort_by_priority/1]).
 
@@ -126,6 +126,9 @@ list_as_maps() ->
 list_as_maps(VHost) ->
     [maps:from_list(PL) || PL <- sort_by_priority(list0(VHost, fun maps:from_list/1))].
 
+list_op_as_maps() ->
+    list_op_as_maps('_').
+
 list_op_as_maps(VHost) ->
     [maps:from_list(PL) || PL <- sort_by_priority(list0_op(VHost, fun maps:from_list/1))].