Merge pull request #14976 from rabbitmq/su_aws/default_oper_policy_apply_to

Add apply_to to default oper policy, and fix boot steps

Author: michaelklishin

deps/rabbit/priv/schema/rabbit.schema

@@ -749,6 +749,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}}
 ]}.
@@ -815,7 +819,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

@@ -91,7 +91,7 @@
      {cleanup,  {rabbit_registry, unregister,
                  [queue, <<"classic">>]}},
      {requires, rabbit_registry},
-     {enables,     ?MODULE}]}).
+     {enables, [?MODULE, rabbit_policy]}]}).
 
 -rabbit_boot_step(
    {?MODULE,

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

@@ -112,7 +112,8 @@
                     [queue, <<"quorum">>, ?MODULE]}},
      {cleanup,  {rabbit_registry, unregister,
                  [queue, <<"quorum">>]}},
-     {requires, rabbit_registry}]}).
+     {requires, rabbit_registry},
+     {enables, rabbit_policy}]}).
 
 -type msg_id() :: non_neg_integer().
 -type qmsg() :: {rabbit_types:r('queue'), pid(), msg_id(), boolean(),

deps/rabbit/src/rabbit_stream_queue.erl

@@ -126,7 +126,8 @@
                     [queue, <<"stream">>, ?MODULE]}},
      {cleanup,  {rabbit_registry, unregister,
                  [queue, <<"stream">>]}},
-     {requires, rabbit_registry}
+     {requires, rabbit_registry},
+     {enables, rabbit_policy}
     ]}).
 
 -type client() :: #stream_client{}.

deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets

@@ -199,17 +199,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"}]}]}]}]}],