Merge pull request #1834 from rabbitmq/qq-queue-length-drop-head

Queue length limit by byte size and number of messages (Drop head only)

Author: michaelklishin

src/rabbit_fifo.erl

@@ -4,12 +4,14 @@
          empty/0,
          fetch/2,
          append/3,
+         update_if_present/3,
          return/3,
          delete/2,
          size/1,
          smallest/1,
          next_key_after/2,
-         map/2
+         map/2,
+         to_map/1
         ]).
 
 -include_lib("ra/include/ra.hrl").
@@ -36,12 +38,22 @@ fetch(Key, #?MODULE{data = Data}) ->
 -spec append(integer(), term(), state()) -> state().
 append(Key, Value,
        #?MODULE{data = Data,
-              smallest = Smallest,
-              largest = Largest} = State)
+                smallest = Smallest,
+                largest = Largest} = State)
   when Key > Largest orelse Largest =:= undefined ->
     State#?MODULE{data = maps:put(Key, Value, Data),
-                smallest = ra_lib:default(Smallest, Key),
-                largest = Key}.
+                  smallest = ra_lib:default(Smallest, Key),
+                  largest = Key}.
+
+-spec update_if_present(integer(), term(), state()) -> state().
+update_if_present(Key, Value, #?MODULE{data = Data} = State) ->
+    case Data of
+        #{Key := _} ->
+            State#?MODULE{data = maps:put(Key, Value, Data)};
+        _ ->
+            State
+    end.
+
 
 -spec return(integer(), term(), state()) -> state().
 return(Key, Value, #?MODULE{data = Data, smallest = Smallest} = State)
@@ -76,6 +88,10 @@ delete(Key, #?MODULE{data = Data} = State) ->
 size(#?MODULE{data = Data}) ->
     maps:size(Data).
 
+-spec to_map(state()) -> #{integer() => term()}.
+to_map(#?MODULE{data = Data}) ->
+    Data.
+
 -spec smallest(state()) -> undefined | {integer(), term()}.
 smallest(#?MODULE{smallest = undefined}) ->
     undefined;

src/rabbit_fifo_index.erl

@@ -152,10 +152,15 @@ ra_machine(Q) ->
 
 ra_machine_config(Q = #amqqueue{name = QName,
                                 pid = {Name, _}}) ->
+    %% take the minimum value of the policy and the queue arg if present
+    MaxLength = args_policy_lookup(<<"max-length">>, fun min/2, Q),
+    MaxBytes = args_policy_lookup(<<"max-length-bytes">>, fun min/2, Q),
     #{name => Name,
       queue_resource => QName,
       dead_letter_handler => dlx_mfa(Q),
       become_leader_handler => {?MODULE, become_leader, [QName]},
+      max_length => MaxLength,
+      max_bytes => MaxBytes,
       single_active_consumer_on => single_active_consumer_on(Q)}.
 
 single_active_consumer_on(#amqqueue{arguments = QArguments}) ->
@@ -636,8 +641,10 @@ delete_member(#amqqueue{pid = {RaName, _}, name = QName}, Node) ->
 
 %%----------------------------------------------------------------------------
 dlx_mfa(Q) ->
-    DLX = init_dlx(args_policy_lookup(<<"dead-letter-exchange">>, fun res_arg/2, Q), Q),
-    DLXRKey = args_policy_lookup(<<"dead-letter-routing-key">>, fun res_arg/2, Q),
+    DLX = init_dlx(args_policy_lookup(<<"dead-letter-exchange">>,
+                                      fun res_arg/2, Q), Q),
+    DLXRKey = args_policy_lookup(<<"dead-letter-routing-key">>,
+                                 fun res_arg/2, Q),
     {?MODULE, dead_letter_publish, [DLX, DLXRKey, Q#amqqueue.name]}.
 
 init_dlx(undefined, _Q) ->
@@ -829,9 +836,8 @@ qnode({_, Node}) ->
     Node.
 
 check_invalid_arguments(QueueName, Args) ->
-    Keys = [<<"x-expires">>, <<"x-message-ttl">>, <<"x-max-length">>,
-            <<"x-max-length-bytes">>, <<"x-max-priority">>, <<"x-overflow">>,
-            <<"x-queue-mode">>],
+    Keys = [<<"x-expires">>, <<"x-message-ttl">>,
+            <<"x-max-priority">>, <<"x-queue-mode">>, <<"x-overflow">>],
     [case rabbit_misc:table_lookup(Args, Key) of
          undefined -> ok;
          _TypeVal   -> rabbit_misc:protocol_error(

src/rabbit_quorum_queue.erl

@@ -22,6 +22,7 @@
 
 -import(quorum_queue_utils, [wait_for_messages_ready/3,
                              wait_for_messages_pending_ack/3,
+                             wait_for_messages_total/3,
                              dirty_query/3,
                              ra_name/1]).
 
@@ -37,6 +38,7 @@ all() ->
 groups() ->
     [
      {single_node, [], all_tests()},
+     {single_node, [], memory_tests()},
      {unclustered, [], [
                         {cluster_size_2, [], [add_member]}
                        ]},
@@ -51,6 +53,7 @@ groups() ->
                                             delete_member_not_found,
                                             delete_member]
                        ++ all_tests()},
+                      {cluster_size_2, [], memory_tests()},
                       {cluster_size_3, [], [
                                             declare_during_node_down,
                                             simple_confirm_availability_on_leader_change,
@@ -61,7 +64,8 @@ groups() ->
                                             delete_declare,
                                             metrics_cleanup_on_leadership_takeover,
                                             metrics_cleanup_on_leader_crash,
-                                            consume_in_minority]},
+                                            consume_in_minority
+                                            ]},
                       {cluster_size_5, [], [start_queue,
                                             start_queue_concurrent,
                                             quorum_cluster_size_3,
@@ -126,6 +130,11 @@ all_tests() ->
      consume_redelivery_count,
      subscribe_redelivery_count,
      message_bytes_metrics,
+     queue_length_limit_drop_head
+    ].
+
+memory_tests() ->
+    [
      memory_alarm_rolls_wal
     ].
 
@@ -240,7 +249,9 @@ declare_args(Config) ->
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
     LQ = ?config(queue_name, Config),
-    declare(Ch, LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>}]),
+    declare(Ch, LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
+                     {<<"x-max-length">>, long, 2000},
+                     {<<"x-max-length-bytes">>, long, 2000}]),
     assert_queue_type(Server, LQ, quorum),
 
     DQ = <<"classic-declare-args-q">>,
@@ -293,16 +304,6 @@ declare_invalid_args(Config) ->
        declare(rabbit_ct_client_helpers:open_channel(Config, Server),
                LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
                     {<<"x-message-ttl">>, long, 2000}])),
-    ?assertExit(
-       {{shutdown, {server_initiated_close, 406, _}}, _},
-       declare(rabbit_ct_client_helpers:open_channel(Config, Server),
-               LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
-                    {<<"x-max-length">>, long, 2000}])),
-    ?assertExit(
-       {{shutdown, {server_initiated_close, 406, _}}, _},
-       declare(rabbit_ct_client_helpers:open_channel(Config, Server),
-               LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
-                    {<<"x-max-length-bytes">>, long, 2000}])),
 
     ?assertExit(
        {{shutdown, {server_initiated_close, 406, _}}, _},
@@ -314,7 +315,7 @@ declare_invalid_args(Config) ->
        {{shutdown, {server_initiated_close, 406, _}}, _},
        declare(rabbit_ct_client_helpers:open_channel(Config, Server),
                LQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
-                    {<<"x-overflow">>, longstr, <<"drop-head">>}])),
+                    {<<"x-overflow">>, longstr, <<"reject-publish">>}])),
 
     ?assertExit(
        {{shutdown, {server_initiated_close, 406, _}}, _},
@@ -1422,7 +1423,7 @@ metrics_cleanup_on_leadership_takeover(Config) ->
                   _ -> false
               end
       end),
-    force_leader_change(Leader, Servers, QQ),
+    force_leader_change(Servers, QQ),
     wait_until(fun () ->
                        [] =:= rpc:call(Leader, ets, lookup, [queue_coarse_metrics, QRes]) andalso
                        [] =:= rpc:call(Leader, ets, lookup, [queue_metrics, QRes])
@@ -2151,6 +2152,32 @@ memory_alarm_rolls_wal(Config) ->
     timer:sleep(1000),
     ok.
 
+queue_length_limit_drop_head(Config) ->
+    [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
+
+    Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
+    QQ = ?config(queue_name, Config),
+    ?assertEqual({'queue.declare_ok', QQ, 0, 0},
+                 declare(Ch, QQ, [{<<"x-queue-type">>, longstr, <<"quorum">>},
+                                  {<<"x-max-length">>, long, 1}])),
+
+    RaName = ra_name(QQ),
+    ok = amqp_channel:cast(Ch,
+                           #'basic.publish'{routing_key = QQ},
+                           #amqp_msg{props   = #'P_basic'{delivery_mode = 2},
+                                     payload = <<"msg1">>}),
+    ok = amqp_channel:cast(Ch,
+                           #'basic.publish'{routing_key = QQ},
+                           #amqp_msg{props   = #'P_basic'{delivery_mode = 2},
+                                     payload = <<"msg2">>}),
+    wait_for_consensus(QQ, Config),
+    wait_for_messages_ready(Servers, RaName, 1),
+    wait_for_messages_pending_ack(Servers, RaName, 0),
+    wait_for_messages_total(Servers, RaName, 1),
+    ?assertMatch({#'basic.get_ok'{}, #amqp_msg{payload = <<"msg2">>}},
+                 amqp_channel:call(Ch, #'basic.get'{queue = QQ,
+                                                    no_ack = true})).
+
 %%----------------------------------------------------------------------------
 
 declare(Ch, Q) ->
@@ -2201,6 +2228,9 @@ filter_queues(Expected, Got) ->
                          lists:member(K, Keys)
                  end, Got).
 
+publish_many(Ch, Queue, Count) ->
+    [publish(Ch, Queue) || _ <- lists:seq(1, Count)].
+
 publish(Ch, Queue) ->
     ok = amqp_channel:cast(Ch,
                            #'basic.publish'{routing_key = Queue},
@@ -2268,14 +2298,16 @@ wait_until(Condition, N) ->
             wait_until(Condition, N - 1)
     end.
 
-force_leader_change(Leader, Servers, Q) ->
+
+force_leader_change([Server | _] = Servers, Q) ->
     RaName = ra_name(Q),
+    {ok, _, {_, Leader}} = ra:members({RaName, Server}),
     [F1, _] = Servers -- [Leader],
     ok = rpc:call(F1, ra, trigger_election, [{RaName, F1}]),
     case ra:members({RaName, Leader}) of
         {ok, _, {_, Leader}} ->
             %% Leader has been re-elected
-            force_leader_change(Leader, Servers, Q);
+            force_leader_change(Servers, Q);
         {ok, _, _} ->
             %% Leader has changed
             ok
@@ -2297,3 +2329,8 @@ get_message_bytes(Leader, QRes) ->
         _ ->
             []
     end.
+
+wait_for_consensus(Name, Config) ->
+    Server = rabbit_ct_broker_helpers:get_node_config(Config, 0, nodename),
+    RaName = ra_name(Name),
+    {ok, _, _} = ra:members({RaName, Server}).

test/quorum_queue_SUITE.erl

@@ -5,6 +5,7 @@
 -export([
          wait_for_messages_ready/3,
          wait_for_messages_pending_ack/3,
+         wait_for_messages_total/3,
          dirty_query/3,
          ra_name/1
         ]).
@@ -17,6 +18,10 @@ wait_for_messages_pending_ack(Servers, QName, Ready) ->
     wait_for_messages(Servers, QName, Ready,
                       fun rabbit_fifo:query_messages_checked_out/1, 60).
 
+wait_for_messages_total(Servers, QName, Total) ->
+    wait_for_messages(Servers, QName, Total,
+                      fun rabbit_fifo:query_messages_total/1, 60).
+
 wait_for_messages(Servers, QName, Number, Fun, 0) ->
     Msgs = dirty_query(Servers, QName, Fun),
     Totals = lists:map(fun(M) when is_map(M) ->
@@ -28,8 +33,8 @@ wait_for_messages(Servers, QName, Number, Fun, 0) ->
 wait_for_messages(Servers, QName, Number, Fun, N) ->
     Msgs = dirty_query(Servers, QName, Fun),
     ct:pal("Got messages ~p", [Msgs]),
-    case lists:all(fun(M) when is_map(M) ->
-                           maps:size(M) == Number;
+    case lists:all(fun(C) when is_integer(C) ->
+                           C == Number;
                       (_) ->
                            false
                    end, Msgs) of

test/quorum_queue_utils.erl

@@ -124,7 +124,6 @@ basics(Config) ->
         {ra_event, Frm, E} ->
             case rabbit_fifo_client:handle_ra_event(Frm, E, FState6b) of
                 {internal, _, _, _FState7} ->
-                    ct:pal("unexpected event ~p~n", [E]),
                     exit({unexpected_internal_event, E});
                 {{delivery, Ctag, [{Mid, {_, two}}]}, FState7} ->
                     {ok, _S} = rabbit_fifo_client:return(Ctag, [Mid], FState7),
@@ -218,9 +217,7 @@ usage(Config) ->
     % force tick and usage stats emission
     ServerId ! tick_timeout,
     timer:sleep(50),
-    % ct:pal("ets ~w ~w ~w", [ets:tab2list(rabbit_fifo_usage), ServerId, UId]),
     Use = rabbit_fifo:usage(element(1, ServerId)),
-    ct:pal("Use ~w~n", [Use]),
     ra:stop_server(ServerId),
     ?assert(Use > 0.0),
     ok.
@@ -300,6 +297,7 @@ returns_after_down(Config) ->
                          Self ! checkout_done
                  end),
     receive checkout_done -> ok after 1000 -> exit(checkout_done_timeout) end,
+    timer:sleep(1000),
     % message should be available for dequeue
     {ok, {_, {_, msg1}}, _} = rabbit_fifo_client:dequeue(<<"tag">>, settled, F2),
     ra:stop_server(ServerId),
@@ -380,7 +378,6 @@ discard(Config) ->
     {ok, empty, _F4} = rabbit_fifo_client:dequeue(<<"tag1">>, settled, F3),
     receive
         {dead_letter, Letters} ->
-            ct:pal("dead letters ~p~n", [Letters]),
             [{_, msg1}] = Letters,
             ok
     after 500 ->
@@ -481,16 +478,13 @@ test_queries(Config) ->
     F0 = rabbit_fifo_client:init(ClusterName, [ServerId], 4),
     {ok, _} = rabbit_fifo_client:checkout(<<"tag">>, 1, undefined, F0),
     {ok, {_, Ready}, _} = ra:local_query(ServerId,
-                                             fun rabbit_fifo:query_messages_ready/1),
-    ?assertEqual(1, maps:size(Ready)),
-    ct:pal("Ready ~w~n", [Ready]),
+                                         fun rabbit_fifo:query_messages_ready/1),
+    ?assertEqual(1, Ready),
     {ok, {_, Checked}, _} = ra:local_query(ServerId,
-                                               fun rabbit_fifo:query_messages_checked_out/1),
-    ?assertEqual(1, maps:size(Checked)),
-    ct:pal("Checked ~w~n", [Checked]),
+                                           fun rabbit_fifo:query_messages_checked_out/1),
+    ?assertEqual(1, Checked),
     {ok, {_, Processes}, _} = ra:local_query(ServerId,
-                                                 fun rabbit_fifo:query_processes/1),
-    ct:pal("Processes ~w~n", [Processes]),
+                                             fun rabbit_fifo:query_processes/1),
     ?assertEqual(2, length(Processes)),
     P !  stop,
     ra:stop_server(ServerId),
@@ -565,7 +559,6 @@ process_ra_events(State, Acc, Wait) ->
 process_ra_events0(State0, Acc, Actions0, Wait, DeliveryFun) ->
     receive
         {ra_event, From, Evt} ->
-            % ct:pal("ra event ~w~n", [Evt]),
             case rabbit_fifo_client:handle_ra_event(From, Evt, State0) of
                 {internal, _, Actions,  State} ->
                     process_ra_events0(State, Acc, Actions0 ++ Actions,
@@ -588,7 +581,6 @@ discard_next_delivery(State0, Wait) ->
                     discard_next_delivery(State, Wait);
                 {{delivery, Tag, Msgs}, State1} ->
                     MsgIds = [element(1, M) || M <- Msgs],
-                    ct:pal("discarding ~p", [Msgs]),
                     {ok, State} = rabbit_fifo_client:discard(Tag, MsgIds,
                                                          State1),
                     State
@@ -605,7 +597,6 @@ return_next_delivery(State0, Wait) ->
                     return_next_delivery(State, Wait);
                 {{delivery, Tag, Msgs}, State1} ->
                     MsgIds = [element(1, M) || M <- Msgs],
-                    ct:pal("returning ~p", [Msgs]),
                     {ok, State} = rabbit_fifo_client:return(Tag, MsgIds,
                                                         State1),
                     State