Add apply_to to default oper policy, and fix boot steps (backport #14976) (backport #14988)

deps/rabbit/priv/schema/rabbit.schema

@@ -766,6 +766,10 @@ end}.
     {datatype, string}
 ]}.
 
+{mapping, "default_policies.operator.$id.apply_to", "rabbit.default_policies.operator", [
+    {datatype, string}
+]}.
+
 {mapping, "default_policies.operator.$id.expires", "rabbit.default_policies.operator", [
     {datatype, {duration, ms}}
 ]}.
@@ -832,7 +836,9 @@ end}.
                           end,
                     {["default_policies","operator",ID|T], NewV};
                    ({["default_policies","operator",ID, "queue_pattern"], V}) ->
-                    {["default_policies","operator",ID,"queue_pattern"], list_to_binary(V)};
+                    {["default_policies","operator",ID, "queue_pattern"], list_to_binary(V)};
+                   ({["default_policies","operator",ID, "apply_to"], V}) ->
+                    {["default_policies","operator",ID, "apply_to"], list_to_binary(V)};
                    (E) -> E
                 end),
     case Props of

deps/rabbit/src/rabbit_classic_queue.erl

@@ -81,6 +81,19 @@
 -export([validate_policy/1]).
 
 -rabbit_boot_step(
+<<<<<<< HEAD
+=======
+   {rabbit_classic_queue_type,
+    [{description, "Classic queue: queue type"},
+     {mfa,      {rabbit_registry, register,
+                    [queue, <<"classic">>, ?MODULE]}},
+     {cleanup,  {rabbit_registry, unregister,
+                 [queue, <<"classic">>]}},
+     {requires, rabbit_registry},
+     {enables, [?MODULE, rabbit_policy]}]}).
+
+-rabbit_boot_step(
+>>>>>>> accb38351 (Add apply_to to default oper policy, and fix boot steps)
    {?MODULE,
     [{description, "Deprecated queue-master-locator support."
       "Use queue-leader-locator instead."},

deps/rabbit/src/rabbit_db_vhost_defaults.erl

@@ -18,7 +18,8 @@
 -record(seeding_policy, {
     name                     :: binary(),
     queue_pattern = <<".*">> :: binary(),
-    definition    = []       :: definitions()
+    definition    = []       :: definitions(),
+    apply_to      = <<"all">>
 }).
 
 -type seeded_user_properties() :: #{
@@ -43,9 +44,9 @@ apply(VHost, ActingUser) ->
     end,
     lists:foreach(
         fun(P) ->
-            ok = rabbit_policy:set_op(VHost, P#seeding_policy.name, P#seeding_policy.queue_pattern, P#seeding_policy.definition,
-                                      undefined, undefined, ActingUser),
-            ?LOG_INFO("Applied default operator policy to vhost '~tp': ~tp", [VHost, P])
+                ok = rabbit_policy:set_op(VHost, P#seeding_policy.name, P#seeding_policy.queue_pattern, P#seeding_policy.definition,
+                                          undefined, P#seeding_policy.apply_to, ActingUser),
+                ?LOG_INFO("Applied default operator policy to vhost '~tp': ~tp", [VHost, P])
         end,
         list_operator_policies(VHost)
     ),
@@ -93,12 +94,15 @@ list_operator_policies(VHost) ->
             case re:run(VHost, RE, [{capture, none}]) of
                 match ->
                     QPattern = proplists:get_value(<<"queue_pattern">>, Ss, <<".*">>),
+                    ApplyTo = proplists:get_value(<<"apply_to">>, Ss, <<"all">>),
                     Ss1 = proplists:delete(<<"queue_pattern">>, Ss),
                     Ss2 = proplists:delete(<<"vhost_pattern">>, Ss1),
+                    Ss3 = proplists:delete(<<"apply_to">>, Ss2),
                     {true, #seeding_policy{
                         name = PolicyName,
                         queue_pattern = QPattern,
-                        definition = underscore_to_dash(Ss2)
+                        apply_to = ApplyTo,
+                        definition = underscore_to_dash(Ss3)
                     }};
                 _ ->
                     false

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -101,6 +101,19 @@
 -include("amqqueue.hrl").
 -include_lib("kernel/include/logger.hrl").
 
+<<<<<<< HEAD
+=======
+-rabbit_boot_step(
+   {rabbit_quorum_queue_type,
+    [{description, "Quorum queue: queue type"},
+     {mfa,      {rabbit_registry, register,
+                    [queue, <<"quorum">>, ?MODULE]}},
+     {cleanup,  {rabbit_registry, unregister,
+                 [queue, <<"quorum">>]}},
+     {requires, rabbit_registry},
+     {enables, rabbit_policy}]}).
+
+>>>>>>> accb38351 (Add apply_to to default oper policy, and fix boot steps)
 -type msg_id() :: non_neg_integer().
 -type qmsg() :: {rabbit_types:r('queue'), pid(), msg_id(), boolean(),
                  mc:state()}.

deps/rabbit/src/rabbit_stream_queue.erl

@@ -104,6 +104,20 @@
 -import(rabbit_queue_type_util, [args_policy_lookup/3]).
 -import(rabbit_misc, [queue_resource/2]).
 
+<<<<<<< HEAD
+=======
+-rabbit_boot_step(
+   {?MODULE,
+    [{description, "Stream queue: queue type"},
+     {mfa,      {rabbit_registry, register,
+                    [queue, <<"stream">>, ?MODULE]}},
+     {cleanup,  {rabbit_registry, unregister,
+                 [queue, <<"stream">>]}},
+     {requires, rabbit_registry},
+     {enables, rabbit_policy}
+    ]}).
+
+>>>>>>> accb38351 (Add apply_to to default oper policy, and fix boot steps)
 -type client() :: #stream_client{}.
 
 -spec is_enabled() -> boolean().

deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets

@@ -191,17 +191,13 @@ ssl_options.fail_if_no_peer_cert = true",
   default_policies.operator.a.expires = 1h
   default_policies.operator.a.queue_pattern = apple
   default_policies.operator.a.vhost_pattern = banana
-  default_policies.operator.a.classic_queues.ha_mode = exactly
-  default_policies.operator.a.classic_queues.ha_params = 2
-  default_policies.operator.a.classic_queues.ha_sync_mode = automatic
+  default_policies.operator.a.apply_to = quorum_queues
   default_policies.operator.a.classic_queues.queue_version = 2
 
  ",
   [{rabbit, [{default_policies, [{operator, [
-      {<<"a">>, [{<<"expires">>, 3600000},
-                 {<<"ha_mode">>, <<"exactly">>},
-                 {<<"ha_params">>, 2},
-                 {<<"ha_sync_mode">>, <<"automatic">>},
+      {<<"a">>, [{<<"apply_to">>,<<"quorum_queues">>},
+                 {<<"expires">>, 3600000},
                  {<<"queue_pattern">>, <<"apple">>},
                  {<<"queue_version">>, 2},
                  {<<"vhost_pattern">>, "banana"}]}]}]}]}],