Renaming and optional auto-remove

Author: SimonUnge

deps/rabbit/priv/schema/rabbit.schema

@@ -2495,12 +2495,14 @@ end}.
 {mapping, "quorum_queue.continuous_membership_reconciliation.enabled", "rabbit.quorum_membership_reconciliation_enabled", [
     {datatype, {enum, [true, false]}}]}.
 
+{mapping, "quorum_queue.continuous_membership_reconciliation.auto_remove", "rabbit.quorum_membership_reconciliation_auto_remove", [
+    {datatype, {enum, [true, false]}}]}.
 
-{mapping, "quorum_queue.continuous_membership_reconciliation.default_timeout", "rabbit.quorum_membership_reconciliation_default_timeout", [
+{mapping, "quorum_queue.continuous_membership_reconciliation.interval", "rabbit.quorum_membership_reconciliation_interval", [
     {datatype, integer}, {validators, ["non_negative_integer"]}
 ]}.
 
-{mapping, "quorum_queue.continuous_membership_reconciliation.short_timeout", "rabbit.quorum_membership_reconciliation_short_timeout", [
+{mapping, "quorum_queue.continuous_membership_reconciliation.trigger_interval", "rabbit.quorum_membership_reconciliation_trigger_interval", [
     {datatype, integer}, {validators, ["non_negative_integer"]}
 ]}.
 

deps/rabbit/src/rabbit_quorum_queue_periodic_membership_reconciliation.erl

@@ -21,16 +21,17 @@
 
 -define(SERVER, ?MODULE).
 -define(DEFAULT_INTERVAL, 60_000*60).
--define(SHORT_INTERVAL, 10_000).
+-define(DEFAULT_TRIGGER_INTERVAL, 10_000).
 -define(QUEUE_COUNT_START_RANDOM_SELECTION, 1_000).
 
 -define(EVAL_MSG, membership_reconciliation).
 
 -record(state, {timer_ref :: reference() | undefined,
-                default_interval :: non_neg_integer(),
-                short_interval :: non_neg_integer(),
+                interval :: non_neg_integer(),
+                trigger_interval :: non_neg_integer(),
                 target_group_size :: non_neg_integer() | undefined,
-                enabled :: boolean()}).
+                enabled :: boolean(),
+                auto_remove :: boolean()}).
 
 %%----------------------------------------------------------------------------
 %% Start
@@ -62,19 +63,22 @@ policy_set() ->
 init([]) ->
     Enabled = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_enabled,
                                   false),
-    DefaultInterval = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_default_timeout,
+    AutoRemove = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_auto_remove,
+                                     false),
+    Interval = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_interval,
                                           ?DEFAULT_INTERVAL),
-    ShortInterval = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_short_timeout,
-                                        ?SHORT_INTERVAL),
+    TriggerInterval = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_trigger_interval,
+                                        ?DEFAULT_TRIGGER_INTERVAL),
     TargetGroupSize = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_target_group_size,
                                           undefined),
-    State = #state{default_interval = DefaultInterval,
-                   short_interval = ShortInterval,
+    State = #state{interval = Interval,
+                   trigger_interval = TriggerInterval,
                    target_group_size = TargetGroupSize,
-                   enabled = Enabled},
+                   enabled = Enabled,
+                   auto_remove = AutoRemove},
     case Enabled of
         true ->
-            Ref = erlang:send_after(DefaultInterval, self(), ?EVAL_MSG),
+            Ref = erlang:send_after(Interval, self(), ?EVAL_MSG),
             {ok, State#state{timer_ref = Ref}};
         false ->
             {ok, State, hibernate}
@@ -86,7 +90,7 @@ handle_call(_Request, _From, State) ->
 handle_cast({membership_reconciliation_trigger, _Reason}, #state{enabled = false} = State) ->
     {noreply, State, hibernate};
 handle_cast({membership_reconciliation_trigger, Reason}, #state{timer_ref = OldRef,
-                               short_interval = Time} = State) ->
+                               trigger_interval = Time} = State) ->
     rabbit_log:debug("Quorum Queue membership reconciliation triggered: ~p",
                      [Reason]),
     _ = erlang:cancel_timer(OldRef),
@@ -95,15 +99,14 @@ handle_cast({membership_reconciliation_trigger, Reason}, #state{timer_ref = OldR
 handle_cast(_Msg, State) ->
     {noreply, State}.
 
-handle_info(?EVAL_MSG, #state{default_interval = DefaultInterval,
-                              short_interval = ShortInterval,
-                              target_group_size = TargetSize} = State) ->
-    Res = reconclitiate_quorum_queue_membership(TargetSize),
+handle_info(?EVAL_MSG, #state{interval = Interval,
+                              trigger_interval = TriggerInterval} = State) ->
+    Res = reconclitiate_quorum_queue_membership(State),
     NewTimeout = case Res of
                      noop ->
-                         DefaultInterval;
+                         Interval;
                      _ ->
-                         ShortInterval
+                         TriggerInterval
                  end,
     Ref = erlang:send_after(NewTimeout, self(), ?EVAL_MSG),
     {noreply, State#state{timer_ref = Ref}};
@@ -122,15 +125,17 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal functions
 %%----------------------------------------------------------------------------
 
-reconclitiate_quorum_queue_membership(TargetSize) ->
+reconclitiate_quorum_queue_membership(State) ->
     LocalLeaders = rabbit_amqqueue:list_local_leaders(),
     ExpectedNodes = rabbit_nodes:list_members(),
     Running = rabbit_nodes:list_running(),
-    reconclitiate_quorum_members(ExpectedNodes, Running, LocalLeaders, TargetSize, noop).
+    reconclitiate_quorum_members(ExpectedNodes, Running, LocalLeaders, State, noop).
 
-reconclitiate_quorum_members(_ExpectedNodes, _Running, [], _TargetSize, Result) ->
+reconclitiate_quorum_members(_ExpectedNodes, _Running, [], _State, Result) ->
     Result;
-reconclitiate_quorum_members(ExpectedNodes, Running, [Q | LocalLeaders], TargetSize, OldResult) ->
+reconclitiate_quorum_members(ExpectedNodes, Running, [Q | LocalLeaders],
+                             #state{target_group_size = TargetSize} = State,
+                             OldResult) ->
     Result =
         maybe
             {ok, Members, {_, LeaderNode}} = ra:members(amqqueue:get_pid(Q), 500),
@@ -139,12 +144,12 @@ reconclitiate_quorum_members(ExpectedNodes, Running, [Q | LocalLeaders], TargetS
             %% And that this not is not in maintenance mode
             true ?= not rabbit_maintenance:is_being_drained_local_read(node()),
             MemberNodes = [Node || {_, Node} <- Members],
-            Remove = MemberNodes -- ExpectedNodes,
-            case Remove of
-                [] ->
+            DanglingNodes = MemberNodes -- ExpectedNodes,
+            case maybe_remove(DanglingNodes, State) of
+                false ->
                     maybe_add_member(Q, Running, MemberNodes, get_target_size(Q, TargetSize));
-                _ ->
-                    remove_members(Q, Remove)
+                true ->
+                    remove_members(Q, DanglingNodes)
             end
         else
             {timeout, Reason} ->
@@ -153,9 +158,16 @@ reconclitiate_quorum_members(ExpectedNodes, Running, [Q | LocalLeaders], TargetS
             _ ->
                 noop
         end,
-    reconclitiate_quorum_members(ExpectedNodes, Running, LocalLeaders, TargetSize,
+    reconclitiate_quorum_members(ExpectedNodes, Running, LocalLeaders, State,
                                  update_result(OldResult, Result)).
 
+maybe_remove(_, #state{auto_remove = false}) ->
+    false;
+maybe_remove([], #state{auto_remove = true}) ->
+    false;
+maybe_remove(_Nodes, #state{auto_remove = true}) ->
+    true.
+
 maybe_add_member(Q, Running, MemberNodes, TargetSize) ->
     %% Filter out any new nodes under maintenance
     New = rabbit_maintenance:filter_out_drained_nodes_local_read(Running -- MemberNodes),

deps/rabbit/test/quorum_queue_member_reconciliation_SUITE.erl

@@ -37,8 +37,9 @@ init_per_suite(Config0) ->
     Config1 = rabbit_ct_helpers:merge_app_env(
                 Config0, {rabbit, [{quorum_tick_interval, 1000},
                                    {quorum_membership_reconciliation_enabled, true},
-                                   {quorum_membership_reconciliation_default_timeout, 5000},
-                                   {quorum_membership_reconciliation_short_timeout, 2000},
+                                   {quorum_membership_reconciliation_auto_remove, true},
+                                   {quorum_membership_reconciliation_interval, 5000},
+                                   {quorum_membership_reconciliation_trigger_interval, 2000},
                                    {quorum_membership_reconciliation_target_group_size, 3}]}),
     rabbit_ct_helpers:run_setup_steps(Config1, []).