Merge pull request #9549 from rabbitmq/mergify/bp/v3.11.x/pr-9548

Add Classic Queue version to operator policies (backport #9547) (backport #9548)

Author: michaelklishin

README.md

@@ -51,10 +51,8 @@ RabbitMQ documentation is also [developed on GitHub](https://github.com/rabbitmq
 
 ## Getting Help from the Community
 
- * [Community Discord server](https://rabbitmq.com/discord/)
- * [Community Slack](https://rabbitmq.com/slack/)
  * [GitHub Discussions](https://github.com/rabbitmq/rabbitmq-server/discussions/)
- * [RabbitMQ mailing list](https://groups.google.com/forum/#!forum/rabbitmq-users)
+ * [Community Discord server](https://rabbitmq.com/discord/)
  * `#rabbitmq` on [Libera Chat](https://libera.chat/)
 
 

deps/rabbit/app.bzl

@@ -1104,7 +1104,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         outs = ["test/policy_SUITE.beam"],
         app_name = "rabbit",
         erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app"],
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
     erlang_bytecode(
         name = "priority_queue_SUITE_beam_files",

deps/rabbit/priv/schema/rabbit.schema

@@ -782,6 +782,12 @@ end}.
     {datatype, string}
 ]}.
 
+{mapping, "default_policies.operator.$id.classic_queues.queue_version", "rabbit.default_policies.operator",
+    [
+        {validators, ["non_zero_positive_integer"]},
+        {datatype, integer}
+    ]}.
+
 {translation, "rabbit.default_policies.operator", fun(Conf) ->
     Props = rabbit_cuttlefish:aggregate_props(
                 Conf,

deps/rabbit/src/rabbit_policies.erl

@@ -52,13 +52,15 @@ register() ->
                           {operator_policy_validator, <<"max-in-memory-length">>},
                           {operator_policy_validator, <<"max-in-memory-bytes">>},
                           {operator_policy_validator, <<"delivery-limit">>},
+                          {operator_policy_validator, <<"queue-version">>},
                           {policy_merge_strategy, <<"expires">>},
                           {policy_merge_strategy, <<"message-ttl">>},
                           {policy_merge_strategy, <<"max-length">>},
                           {policy_merge_strategy, <<"max-length-bytes">>},
                           {policy_merge_strategy, <<"max-in-memory-length">>},
                           {policy_merge_strategy, <<"max-in-memory-bytes">>},
-                          {policy_merge_strategy, <<"delivery-limit">>}]],
+                          {policy_merge_strategy, <<"delivery-limit">>},
+                          {policy_merge_strategy, <<"queue-version">>}]],
     ok.
 
 -spec validate_policy([{binary(), term()}]) -> rabbit_policy_validator:validate_results().
@@ -197,5 +199,6 @@ merge_policy_value(<<"max-in-memory-length">>, Val, OpVal) -> min(Val, OpVal);
 merge_policy_value(<<"max-in-memory-bytes">>, Val, OpVal) -> min(Val, OpVal);
 merge_policy_value(<<"expires">>, Val, OpVal)          -> min(Val, OpVal);
 merge_policy_value(<<"delivery-limit">>, Val, OpVal)   -> min(Val, OpVal);
+merge_policy_value(<<"queue-version">>, _Val, OpVal)   -> OpVal;
 %% use operator policy value for booleans
 merge_policy_value(_Key, Val, OpVal) when is_boolean(Val) andalso is_boolean(OpVal) -> OpVal.

deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets

@@ -151,6 +151,7 @@ ssl_options.fail_if_no_peer_cert = true",
   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.classic_queues.queue_version = 2
 
  ",
   [{rabbit, [{default_policies, [{operator, [
@@ -159,6 +160,7 @@ ssl_options.fail_if_no_peer_cert = true",
                  {<<"ha_params">>, 2},
                  {<<"ha_sync_mode">>, <<"automatic">>},
                  {<<"queue_pattern">>, <<"apple">>},
+                 {<<"queue_version">>, 2},
                  {<<"vhost_pattern">>, "banana"}]}]}]}]}],
   []},
 

deps/rabbit/test/policy_SUITE.erl

@@ -9,7 +9,8 @@
 
 -include_lib("common_test/include/ct.hrl").
 -include_lib("amqp_client/include/amqp_client.hrl").
-
+-include_lib("stdlib/include/assert.hrl").
+-include_lib("rabbitmq_ct_helpers/include/rabbit_assert.hrl").
 
 -compile(export_all).
 
@@ -25,7 +26,17 @@ groups() ->
                            policy_ttl,
                            operator_policy_ttl,
                            operator_retroactive_policy_ttl,
-                           operator_retroactive_policy_publish_ttl
+                           operator_retroactive_policy_publish_ttl,
+                           queue_type_specific_policies,
+                           queue_version_specific_policies,
+                           is_supported_operator_policy_expires,
+                           is_supported_operator_policy_message_ttl,
+                           is_supported_operator_policy_max_length,
+                           is_supported_operator_policy_max_length,
+                           is_supported_operator_policy_max_in_memory_length,
+                           is_supported_operator_policy_max_in_memory_bytes,
+                           is_supported_operator_policy_delivery_limit,
+                           is_supported_operator_policy_ha
                           ]}
     ].
 
@@ -208,14 +219,179 @@ target_count_policy(Config) ->
     rabbit_ct_client_helpers:close_connection(Conn),
     passed.
 
+queue_type_specific_policies(Config) ->
+    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    ClassicQ = <<"policy_ttl-classic_queue">>,
+    QuorumQ = <<"policy_ttl-quorum_queue">>,
+    StreamQ = <<"policy_ttl-stream_queue">>,
+
+    %% all policies match ".*" but different values should be applied based on queue type
+    rabbit_ct_broker_helpers:set_policy(Config, 0, <<"ttl-policy-classic">>,
+        <<".*">>, <<"classic_queues">>, [{<<"message-ttl">>, 20}]),
 
-%%----------------------------------------------------------------------------
+    rabbit_ct_broker_helpers:set_policy(Config, 0, <<"ttl-policy-quorum">>,
+        <<".*">>, <<"quorum_queues">>, [{<<"message-ttl">>, 40}]),
+
+    rabbit_ct_broker_helpers:set_policy(Config, 0, <<"ttl-policy-stream">>,
+        <<".*">>, <<"streams">>, [{<<"max-age">>, "1h"}]),
+
+    declare(Ch, ClassicQ, [{<<"x-queue-type">>, longstr, <<"classic">>}]),
+    declare(Ch, QuorumQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
+    declare(Ch, StreamQ, [{<<"x-queue-type">>, longstr, <<"stream">>}]),
+    timer:sleep(1),
+
+    ?assertMatch(20, check_policy_value(Server, ClassicQ, <<"message-ttl">>)),
+    ?assertMatch(40, check_policy_value(Server, QuorumQ, <<"message-ttl">>)),
+    ?assertMatch("1h", check_policy_value(Server, StreamQ, <<"max-age">>)),
+
+    delete(Ch, ClassicQ),
+    delete(Ch, QuorumQ),
+    delete(Ch, StreamQ),
+    rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"ttl-policy-classic">>),
+    rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"ttl-policy-quorum">>),
+    rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"ttl-policy-stream">>),
+
+    rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    passed.
+
+queue_version_specific_policies(Config) ->
+    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    QName = <<"policy_queue_version">>,
+    declare(Ch, QName),
+    QueueVersionOnePolicy = [{<<"queue-version">>, 1}],
+    QueueVersionTwoPolicy = [{<<"queue-version">>, 2}],
+
+    Opts = #{config => Config,
+             server => Server,
+             qname  => QName},
+
+    %% Queue version OperPolicy has precedence always
+    verify_policies(QueueVersionOnePolicy, QueueVersionTwoPolicy, QueueVersionTwoPolicy, Opts),
+    verify_policies(QueueVersionTwoPolicy, QueueVersionOnePolicy, QueueVersionOnePolicy, Opts),
+
+    delete(Ch, QName),
+    rabbit_ct_broker_helpers:clear_policy(Config, 0, <<"policy">>),
+    rabbit_ct_broker_helpers:clear_operator_policy(Config, 0, <<"op_policy">>),
+    rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    passed.
+
+%% See supported policies in https://www.rabbitmq.com/parameters.html#operator-policies
+%% This test applies all supported operator policies to all queue types,
+%% and later verifies the effective policy definitions.
+%% Just those supported by each queue type should be present.
+
+is_supported_operator_policy_expires(Config) ->
+    Value = 6000000,
+    effective_operator_policy_per_queue_type(
+      Config, <<"expires">>, Value, Value, Value, undefined).
+
+is_supported_operator_policy_message_ttl(Config) ->
+    Value = 1000,
+    effective_operator_policy_per_queue_type(
+      Config, <<"message-ttl">>, Value, Value, Value, undefined).
+
+is_supported_operator_policy_max_length(Config) ->
+    Value = 500,
+    effective_operator_policy_per_queue_type(
+      Config, <<"max-length">>, Value, Value, Value, undefined).
+
+is_supported_operator_policy_max_length_bytes(Config) ->
+    Value = 1500,
+    effective_operator_policy_per_queue_type(
+      Config, <<"max-length-bytes">>, Value, Value, Value, Value).
+
+is_supported_operator_policy_max_in_memory_length(Config) ->
+    Value = 30,
+    effective_operator_policy_per_queue_type(
+      Config, <<"max-in-memory-length">>, Value, undefined, Value, undefined).
+
+is_supported_operator_policy_max_in_memory_bytes(Config) ->
+    Value = 50000,
+    effective_operator_policy_per_queue_type(
+      Config, <<"max-in-memory-bytes">>, Value, undefined, Value, undefined).
+
+is_supported_operator_policy_delivery_limit(Config) ->
+    Value = 3,
+    effective_operator_policy_per_queue_type(
+      Config, <<"delivery-limit">>, Value, undefined, Value, undefined).
+
+is_supported_operator_policy_ha(Config) ->
+    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    ClassicQ = <<"classic_queue">>,
+    QuorumQ = <<"quorum_queue">>,
+    StreamQ = <<"stream_queue">>,
+
+    declare(Ch, ClassicQ, [{<<"x-queue-type">>, longstr, <<"classic">>}]),
+    declare(Ch, QuorumQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
+    declare(Ch, StreamQ, [{<<"x-queue-type">>, longstr, <<"stream">>}]),
+
+    rabbit_ct_broker_helpers:set_operator_policy(
+      Config, 0, <<"operator-policy">>, <<".*">>, <<"all">>,
+      [{<<"ha-mode">>, <<"exactly">>},
+       {<<"ha-params">>, 2},
+       {<<"ha-sync-mode">>, <<"automatic">>}]),
+
+    ?awaitMatch(<<"exactly">>, check_policy_value(Server, ClassicQ, <<"ha-mode">>), 30_000),
+    ?awaitMatch(2, check_policy_value(Server, ClassicQ, <<"ha-params">>), 30_000),
+    ?awaitMatch(<<"automatic">>, check_policy_value(Server, ClassicQ, <<"ha-sync-mode">>), 30_000),
+    ?awaitMatch(undefined, check_policy_value(Server, QuorumQ, <<"ha-mode">>), 30_000),
+    ?awaitMatch(undefined, check_policy_value(Server, StreamQ, <<"ha-mode">>), 30_000),
+
+    rabbit_ct_broker_helpers:clear_operator_policy(Config, 0, <<"operator-policy">>),
+
+    delete(Ch, ClassicQ),
+    delete(Ch, QuorumQ),
+    delete(Ch, StreamQ),
+
+    rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    passed.
+
+effective_operator_policy_per_queue_type(Config, Name, Value, ClassicValue, QuorumValue, StreamValue) ->
+    [Server | _] = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+    {Conn, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    ClassicQ = <<"classic_queue">>,
+    QuorumQ = <<"quorum_queue">>,
+    StreamQ = <<"stream_queue">>,
+
+    declare(Ch, ClassicQ, [{<<"x-queue-type">>, longstr, <<"classic">>}]),
+    declare(Ch, QuorumQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
+    declare(Ch, StreamQ, [{<<"x-queue-type">>, longstr, <<"stream">>}]),
+
+    rabbit_ct_broker_helpers:set_operator_policy(
+      Config, 0, <<"operator-policy">>, <<".*">>, <<"all">>,
+      [{Name, Value}]),
+
+    ?awaitMatch(ClassicValue, check_policy_value(Server, ClassicQ, Name), 30_000),
+    ?awaitMatch(QuorumValue, check_policy_value(Server, QuorumQ, Name), 30_000),
+    ?awaitMatch(StreamValue, check_policy_value(Server, StreamQ, Name), 30_000),
 
+    rabbit_ct_broker_helpers:clear_operator_policy(Config, 0, <<"operator-policy">>),
+
+    delete(Ch, ClassicQ),
+    delete(Ch, QuorumQ),
+    delete(Ch, StreamQ),
+
+    rabbit_ct_client_helpers:close_channel(Ch),
+    rabbit_ct_client_helpers:close_connection(Conn),
+    passed.
+
+%%----------------------------------------------------------------------------
 
 declare(Ch, Q) ->
     amqp_channel:call(Ch, #'queue.declare'{queue     = Q,
                                            durable   = true}).
 
+declare(Ch, Q, Args) ->
+    amqp_channel:call(Ch, #'queue.declare'{queue     = Q,
+                                           durable   = true,
+                                           arguments = Args}).
+
 delete(Ch, Q) ->
     amqp_channel:call(Ch, #'queue.delete'{queue = Q}).
 
@@ -269,10 +445,10 @@ verify_policies(Policy, OperPolicy, VerifyFuns, #{config := Config,
                                                   server := Server,
                                                   qname := QName}) ->
     rabbit_ct_broker_helpers:set_policy(Config, 0, <<"policy">>,
-                                        <<"policy_ha">>, <<"queues">>,
+                                        QName, <<"queues">>,
                                         Policy),
     rabbit_ct_broker_helpers:set_operator_policy(Config, 0, <<"op_policy">>,
-                                                 <<"policy_ha">>, <<"queues">>,
+                                                 QName, <<"queues">>,
                                                  OperPolicy),
     verify_policy(VerifyFuns, Server, QName).
 

deps/rabbitmq_management/priv/www/js/tmpl/policies.ejs

@@ -288,7 +288,8 @@
                   <span class="argument-link" field="definitionop" key="expires" type="number">Auto expire</span>
                   <span class="argument-link" field="definitionop" key="ha-mode" type="string">HA mode</span> <span class="help" id="policy-ha-mode"></span> |
                   <span class="argument-link" field="definitionop" key="ha-params" type="number">HA params</span> <span class="help" id="policy-ha-params"></span> |
-                  <span class="argument-link" field="definitionop" key="ha-sync-mode" type="string">HA sync mode</span> <span class="help" id="policy-ha-sync-mode"></span> </br>
+                  <span class="argument-link" field="definitionop" key="ha-sync-mode" type="string">HA sync mode</span> <span class="help" id="policy-ha-sync-mode"></span> </br> |
+                  <span class="argument-link" field="definitionop" key="queue-version" type="number">Version</span> <span class="help" id="queue-version"></span> </br>
                 </td>
               </tr>
               <tr>