Make filter size configurable

as a queue arg and policy

Author: kjnilsson

deps/rabbit/src/rabbit_amqqueue.erl

@@ -1,4 +1,4 @@
-%% This Source Code Form is subject to the terms of the Mozilla Public
+% This Source Code Form is subject to the terms of the Mozilla Public
 %% License, v. 2.0. If a copy of the MPL was not distributed with this
 %% file, You can obtain one at https://mozilla.org/MPL/2.0/.
 %%
@@ -834,26 +834,27 @@ check_arguments_key(QueueName, QueueType, Args, InvalidArgs) ->
                   end, Args).
 
 declare_args() ->
-    [{<<"x-expires">>,                 fun check_expires_arg/2},
-     {<<"x-message-ttl">>,             fun check_message_ttl_arg/2},
-     {<<"x-dead-letter-exchange">>,    fun check_dlxname_arg/2},
+    [{<<"x-expires">>, fun check_expires_arg/2},
+     {<<"x-message-ttl">>, fun check_message_ttl_arg/2},
+     {<<"x-dead-letter-exchange">>, fun check_dlxname_arg/2},
      {<<"x-dead-letter-routing-key">>, fun check_dlxrk_arg/2},
-     {<<"x-dead-letter-strategy">>,    fun check_dlxstrategy_arg/2},
-     {<<"x-max-length">>,              fun check_non_neg_int_arg/2},
-     {<<"x-max-length-bytes">>,        fun check_non_neg_int_arg/2},
-     {<<"x-max-in-memory-length">>,    fun check_non_neg_int_arg/2},
-     {<<"x-max-in-memory-bytes">>,     fun check_non_neg_int_arg/2},
-     {<<"x-max-priority">>,            fun check_max_priority_arg/2},
-     {<<"x-overflow">>,                fun check_overflow/2},
-     {<<"x-queue-mode">>,              fun check_queue_mode/2},
-     {<<"x-queue-version">>,           fun check_queue_version/2},
-     {<<"x-single-active-consumer">>,  fun check_single_active_consumer_arg/2},
-     {<<"x-queue-type">>,              fun check_queue_type/2},
-     {<<"x-quorum-initial-group-size">>,     fun check_initial_cluster_size_arg/2},
-     {<<"x-max-age">>,                 fun check_max_age_arg/2},
-     {<<"x-stream-max-segment-size-bytes">>,        fun check_non_neg_int_arg/2},
-     {<<"x-initial-cluster-size">>,    fun check_initial_cluster_size_arg/2},
-     {<<"x-queue-leader-locator">>,    fun check_queue_leader_locator_arg/2}].
+     {<<"x-dead-letter-strategy">>, fun check_dlxstrategy_arg/2},
+     {<<"x-max-length">>, fun check_non_neg_int_arg/2},
+     {<<"x-max-length-bytes">>, fun check_non_neg_int_arg/2},
+     {<<"x-max-in-memory-length">>, fun check_non_neg_int_arg/2},
+     {<<"x-max-in-memory-bytes">>, fun check_non_neg_int_arg/2},
+     {<<"x-max-priority">>, fun check_max_priority_arg/2},
+     {<<"x-overflow">>, fun check_overflow/2},
+     {<<"x-queue-mode">>, fun check_queue_mode/2},
+     {<<"x-queue-version">>, fun check_queue_version/2},
+     {<<"x-single-active-consumer">>, fun check_single_active_consumer_arg/2},
+     {<<"x-queue-type">>, fun check_queue_type/2},
+     {<<"x-quorum-initial-group-size">>, fun check_initial_cluster_size_arg/2},
+     {<<"x-max-age">>, fun check_max_age_arg/2},
+     {<<"x-stream-max-segment-size-bytes">>, fun check_non_neg_int_arg/2},
+     {<<"x-stream-filter-size-bytes">>, fun check_non_neg_int_arg/2},
+     {<<"x-initial-cluster-size">>, fun check_initial_cluster_size_arg/2},
+     {<<"x-queue-leader-locator">>, fun check_queue_leader_locator_arg/2}].
 
 consume_args() -> [{<<"x-priority">>,              fun check_int_arg/2},
                    {<<"x-cancel-on-ha-failover">>, fun check_bool_arg/2},

deps/rabbit/src/rabbit_policies.erl

@@ -44,6 +44,7 @@ register() ->
                           {policy_validator, <<"delivery-limit">>},
                           {policy_validator, <<"max-age">>},
                           {policy_validator, <<"stream-max-segment-size-bytes">>},
+                          {policy_validator, <<"stream-filter-size-bytes">>},
                           {policy_validator, <<"queue-leader-locator">>},
                           {policy_validator, <<"initial-cluster-size">>},
                           {operator_policy_validator, <<"expires">>},
@@ -195,7 +196,13 @@ validate_policy0(<<"stream-max-segment-size-bytes">>, Value)
   when is_integer(Value), Value >= 0, Value =< ?MAX_STREAM_MAX_SEGMENT_SIZE ->
     ok;
 validate_policy0(<<"stream-max-segment-size-bytes">>, Value) ->
-    {error, "~tp is not a valid segment size", [Value]}.
+    {error, "~tp is not a valid segment size", [Value]};
+
+validate_policy0(<<"stream-filter-size-bytes">>, Value)
+  when is_integer(Value), Value >= 16, Value =< 255 ->
+    ok;
+validate_policy0(<<"stream-filter-size-bytes">>, Value) ->
+    {error, "~tp is not a valid filter size. Valid range is 16-255", [Value]}.
 
 merge_policy_value(<<"message-ttl">>, Val, OpVal)      -> min(Val, OpVal);
 merge_policy_value(<<"max-length">>, Val, OpVal)       -> min(Val, OpVal);

deps/rabbit/src/rabbit_stream_queue.erl

@@ -110,7 +110,9 @@ declare(Q0, _Node) when ?amqqueue_is_stream(Q0) ->
            [fun rabbit_queue_type_util:check_auto_delete/1,
             fun rabbit_queue_type_util:check_exclusive/1,
             fun rabbit_queue_type_util:check_non_durable/1,
-            fun rabbit_stream_queue:check_max_segment_size_bytes/1],
+            fun check_max_segment_size_bytes/1,
+            fun check_filter_size/1
+           ],
            Q0) of
         ok ->
             create_stream(Q0);
@@ -130,6 +132,18 @@ check_max_segment_size_bytes(Q) ->
             ok
     end.
 
+check_filter_size(Q) ->
+    Args = amqqueue:get_arguments(Q),
+    case rabbit_misc:table_lookup(Args, <<"x-stream-filter-size-bytes">>) of
+        undefined ->
+            ok;
+        {_Type, Val} when Val > 255 orelse Val < 16 ->
+            {protocol_error, precondition_failed,
+             "Invalid value for  x-stream-filter-size-bytes", []};
+        _ ->
+            ok
+    end.
+
 create_stream(Q0) ->
     Arguments = amqqueue:get_arguments(Q0),
     QName = amqqueue:get_name(Q0),
@@ -865,34 +879,29 @@ delete_replica(VHost, Name, Node) ->
 make_stream_conf(Q) ->
     QName = amqqueue:get_name(Q),
     Name = stream_name(QName),
-    %% MaxLength = args_policy_lookup(<<"max-length">>, policy_precedence/2, Q),
-    MaxBytes = args_policy_lookup(<<"max-length-bytes">>, fun policy_precedence/2, Q),
-    MaxAge = max_age(args_policy_lookup(<<"max-age">>, fun policy_precedence/2, Q)),
-    MaxSegmentSizeBytes = args_policy_lookup(<<"stream-max-segment-size-bytes">>, fun policy_precedence/2, Q),
     Formatter = {?MODULE, format_osiris_event, [QName]},
-    Retention = lists:filter(fun({_, R}) ->
-                                     R =/= undefined
-                             end, [{max_bytes, MaxBytes},
-                                   {max_age, MaxAge}]),
-    add_if_defined(max_segment_size_bytes, MaxSegmentSizeBytes,
-                   #{reference => QName,
-                     name => Name,
-                     retention => Retention,
-                     event_formatter => Formatter,
-                     epoch => 1}).
+    update_stream_conf(Q, #{reference => QName,
+                            name => Name,
+                            event_formatter => Formatter,
+                            epoch => 1}).
 
 update_stream_conf(undefined, #{} = Conf) ->
     Conf;
 update_stream_conf(Q, #{} = Conf) when ?is_amqqueue(Q) ->
     MaxBytes = args_policy_lookup(<<"max-length-bytes">>, fun policy_precedence/2, Q),
     MaxAge = max_age(args_policy_lookup(<<"max-age">>, fun policy_precedence/2, Q)),
-    MaxSegmentSizeBytes = args_policy_lookup(<<"stream-max-segment-size-bytes">>, fun policy_precedence/2, Q),
+    MaxSegmentSizeBytes = args_policy_lookup(<<"stream-max-segment-size-bytes">>,
+                                             fun policy_precedence/2, Q),
+    FilterSizeBytes = args_policy_lookup(<<"stream-filter-size-bytes">>,
+                                         fun policy_precedence/2, Q),
     Retention = lists:filter(fun({_, R}) ->
                                      R =/= undefined
                              end, [{max_bytes, MaxBytes},
                                    {max_age, MaxAge}]),
-    add_if_defined(max_segment_size_bytes, MaxSegmentSizeBytes,
-                   Conf#{retention => Retention}).
+    add_if_defined(
+      filter_size, FilterSizeBytes,
+      add_if_defined(max_segment_size_bytes, MaxSegmentSizeBytes,
+                     Conf#{retention => Retention})).
 
 add_if_defined(_, undefined, Map) ->
     Map;

deps/rabbit/test/rabbit_stream_queue_SUITE.erl

@@ -88,6 +88,7 @@ all_tests() ->
      declare_invalid_properties,
      declare_server_named,
      declare_invalid_arg,
+     declare_invalid_filter_size,
      consume_invalid_arg,
      declare_queue,
      delete_queue,
@@ -261,7 +262,11 @@ declare_args(Config) ->
     Q = ?config(queue_name, Config),
     ?assertEqual({'queue.declare_ok', Q, 0, 0},
                  declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>},
-                                 {<<"x-max-length-bytes">>, long, 2_000_000}])),
+                                 {<<"x-max-length-bytes">>, long, 2_000_000},
+                                 {<<"x-max-age">>, longstr, <<"10D">>},
+                                 {<<"x-stream-max-segment-size-bytes">>, long, 5_000_000},
+                                 {<<"x-stream-filter-size-bytes">>, long, 32}
+                                ])),
     assert_queue_type(Server, Q, rabbit_stream_queue),
     rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
 
@@ -331,6 +336,17 @@ declare_invalid_arg(Config) ->
        declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>},
                        {<<"x-overflow">>, longstr, <<"reject-publish">>}])).
 
+declare_invalid_filter_size(Config) ->
+    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    Q = ?config(queue_name, Config),
+
+    ExpectedError = <<"PRECONDITION_FAILED - Invalid value for  x-stream-filter-size-bytes">>,
+    ?assertExit(
+       {{shutdown, {server_initiated_close, 406, ExpectedError}}, _},
+       declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>},
+                       {<<"x-stream-filter-size-bytes">>, long, 256}])).
+
 consume_invalid_arg(Config) ->
     Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
@@ -2232,7 +2248,8 @@ update_retention_policy(Config) ->
     Q = ?config(queue_name, Config),
     ?assertEqual({'queue.declare_ok', Q, 0, 0},
                  declare(Ch, Q, [{<<"x-queue-type">>, longstr, <<"stream">>},
-                                 {<<"x-stream-max-segment-size-bytes">>, long, 200}
+                                 {<<"x-stream-max-segment-size-bytes">>, long, 200},
+                                 {<<"x-stream-filter-size-bytes">>, long, 32}
                                 ])),
     check_leader_and_replicas(Config, Servers),
 

deps/rabbitmq_management/priv/www/js/global.js

@@ -209,6 +209,9 @@ var HELP = {
     'queue-stream-max-segment-size-bytes':
       'Total segment size for stream segments on disk.<br/>(Sets the x-stream-max-segment-size-bytes argument.)',
 
+    'queue-stream-max-segment-size-bytes':
+      'Size of the filter data attached to each stream chunk.<br/>(Sets the x-stream-filter-size-bytes argument.)',
+
     'queue-auto-delete':
       'If yes, the queue will delete itself after at least one consumer has connected, and then all consumers have disconnected.',
 

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

@@ -141,6 +141,8 @@
                   <span class="help" id="queue-max-age"></span> |
                   <span class="argument-link" field="definition" key="stream-max-segment-size-bytes" type="number">Max segment size in bytes</span>
                   <span class="help" id="queue-stream-max-segment-size-bytes"></span> |
+                  <span class="argument-link" field="definition" key="stream-filter-size-bytes" type="number">Filter size in bytes. Valid range: 16-255</span>
+                  <span class="help" id="queue-stream-filter-size-bytes"></span> |
                   <span class="argument-link" field="definition" key="queue-leader-locator" type="string">Leader locator</span>
                   <span class="help" id="queue-leader-locator"></span>
                 </td>

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

@@ -341,6 +341,7 @@
                 <% if (queue_type == "stream") { %>
                   <span class="argument-link" field="arguments" key="x-max-age" type="string">Max time retention</span><span class="help" id="queue-max-age"></span>
                   | <span class="argument-link" field="arguments" key="x-stream-max-segment-size-bytes" type="number">Max segment size in bytes</span><span class="help" id="queue-stream-max-segment-size-bytes"></span>
+                  | <span class="argument-link" field="arguments" key="x-stream-filter-size-bytes" type="number">Filter size (per chunk) in bytes</span><span class="help" id="queue-stream-filter-size-bytes"></span>
                   | <span class="argument-link" field="arguments" key="x-initial-cluster-size" type="number">Initial cluster size</span><span class="help" id="queue-initial-cluster-size"></span>
                 <% } %>
                 <% if (queue_type != "classic") { %>