Name changes and handling quorum queue leader election not being complete

SimonUnge · SimonUnge · commit e99c8fc37a43 · 2023-07-06T20:36:54.000-07:00
diff --git a/deps/rabbit/BUILD.bazel b/deps/rabbit/BUILD.bazel
@@ -5,8 +5,8 @@ load("//:rabbitmq_home.bzl", "rabbitmq_home")
 load("//:rabbitmq_run.bzl", "rabbitmq_run")
 load(
     "//:rabbitmq.bzl",
-    "RABBITMQ_DIALYZER_OPTS",
     "ENABLE_FEATURE_MAYBE_EXPR",
+    "RABBITMQ_DIALYZER_OPTS",
     "assert_suites",
     "rabbitmq_app",
     "rabbitmq_integration_suite",
@@ -332,7 +332,7 @@ rabbitmq_integration_suite(
 )
 
 rabbitmq_integration_suite(
-    name = "member_evaluation_SUITE",
+    name = "quorum_queue_member_reconciliation_SUITE",
     size = "medium",
     additional_beam = [
         ":test_quorum_queue_utils_beam",
diff --git a/deps/rabbit/app.bzl b/deps/rabbit/app.bzl
@@ -190,9 +190,9 @@ def all_beam_files(name = "all_beam_files"):
             "src/rabbit_queue_location_validator.erl",
             "src/rabbit_queue_master_location_misc.erl",
             "src/rabbit_queue_type_util.erl",
-            "src/rabbit_queue_member_eval.erl",
             "src/rabbit_quorum_memory_manager.erl",
             "src/rabbit_quorum_queue.erl",
+            "src/rabbit_quorum_queue_periodic_membership_reconciliation.erl",
             "src/rabbit_ra_registry.erl",
             "src/rabbit_ra_systems.erl",
             "src/rabbit_reader.erl",
@@ -435,9 +435,9 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/rabbit_queue_location_validator.erl",
             "src/rabbit_queue_master_location_misc.erl",
             "src/rabbit_queue_type_util.erl",
-            "src/rabbit_queue_member_eval.erl",
             "src/rabbit_quorum_memory_manager.erl",
             "src/rabbit_quorum_queue.erl",
+            "src/rabbit_quorum_queue_periodic_membership_reconciliation.erl",
             "src/rabbit_ra_registry.erl",
             "src/rabbit_ra_systems.erl",
             "src/rabbit_reader.erl",
@@ -696,11 +696,11 @@ def all_srcs(name = "all_srcs"):
             "src/rabbit_queue_location_validator.erl",
             "src/rabbit_queue_master_location_misc.erl",
             "src/rabbit_queue_master_locator.erl",
-            "src/rabbit_queue_member_eval.erl",
             "src/rabbit_queue_type.erl",
             "src/rabbit_queue_type_util.erl",
             "src/rabbit_quorum_memory_manager.erl",
             "src/rabbit_quorum_queue.erl",
+            "src/rabbit_quorum_queue_periodic_membership_reconciliation.erl",
             "src/rabbit_ra_registry.erl",
             "src/rabbit_ra_systems.erl",
             "src/rabbit_reader.erl",
@@ -805,16 +805,7 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app"],
     )
-    erlang_bytecode(
-        name = "member_evaluation_SUITE_beam_files",
-        testonly = True,
-        srcs = ["test/member_evaluation_SUITE.erl"],
-        outs = ["test/member_evaluation_SUITE.beam"],
-        hdrs = ["include/amqqueue.hrl", "include/amqqueue_v2.hrl"],
-        app_name = "rabbit",
-        erlc_opts = "//:test_erlc_opts",
-        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
-    )
+
     erlang_bytecode(
         name = "cluster_rename_SUITE_beam_files",
         testonly = True,
@@ -1985,3 +1976,12 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app"],
     )
+    erlang_bytecode(
+        name = "quorum_queue_member_reconciliation_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/quorum_queue_member_reconciliation_SUITE.erl"],
+        outs = ["test/quorum_queue_member_reconciliation_SUITE.beam"],
+        app_name = "rabbit",
+        erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
+    )
diff --git a/deps/rabbit/priv/schema/rabbit.schema b/deps/rabbit/priv/schema/rabbit.schema
@@ -2489,22 +2489,22 @@ end}.
 
 
 %%
-%% Queue membership evaluation
+%% Quorum Queue membership reconciliation
 %%
 
-{mapping, "quorum_queue_membership_evaluation.enabled", "rabbit.quorum_membership_changes_enabled", [
+{mapping, "quorum_queue.continuous_membership_reconciliation.enabled", "rabbit.quorum_membership_reconciliation_enabled", [
     {datatype, {enum, [true, false]}}]}.
 
 
-{mapping, "quorum_queue_membership_evaluation.default_timeout", "rabbit.quorum_membership_changes_default_timeout", [
+{mapping, "quorum_queue.continuous_membership_reconciliation.default_timeout", "rabbit.quorum_membership_reconciliation_default_timeout", [
     {datatype, integer}, {validators, ["non_negative_integer"]}
 ]}.
 
-{mapping, "quorum_queue_membership_evaluation.short_timeout", "rabbit.quorum_membership_changes_short_timeout", [
+{mapping, "quorum_queue.continuous_membership_reconciliation.short_timeout", "rabbit.quorum_membership_reconciliation_short_timeout", [
     {datatype, integer}, {validators, ["non_negative_integer"]}
 ]}.
 
-{mapping, "quorum_queue_membership_evaluation.target_group_size", "rabbit.quorum_membership_changes_target_group_size", [
+{mapping, "quorum_queue.continuous_membership_reconciliation.target_group_size", "rabbit.quorum_membership_reconciliation_target_group_size", [
     {datatype, integer}, {validators, ["non_negative_integer"]}
 ]}.
 
diff --git a/deps/rabbit/src/rabbit.erl b/deps/rabbit/src/rabbit.erl
@@ -175,10 +175,10 @@
                     {requires,    [rabbit_alarm, guid_generator]},
                     {enables,     core_initialized}]}).
 
--rabbit_boot_step({rabbit_queue_member_eval,
-                   [{description, "Queue membership evaluation"},
+-rabbit_boot_step({rabbit_quorum_queue_periodic_membership_reconciliation,
+                   [{description, "Quorums Queue membership reconciliation"},
                     {mfa,         {rabbit_sup, start_restartable_child,
-                                   [rabbit_queue_member_eval]}},
+                                   [rabbit_quorum_queue_periodic_membership_reconciliation]}},
                     {requires, [database]}]}).
 
 -rabbit_boot_step({rabbit_epmd_monitor,
diff --git a/deps/rabbit/src/rabbit_node_monitor.erl b/deps/rabbit/src/rabbit_node_monitor.erl
@@ -815,7 +815,7 @@ handle_dead_rabbit(Node, State = #state{partitions = Partitions,
     ok = rabbit_amqqueue:on_node_down(Node),
     ok = rabbit_alarm:on_node_down(Node),
     ok = rabbit_mnesia:on_node_down(Node),
-    ok = rabbit_queue_member_eval:on_node_down(Node),
+    ok = rabbit_quorum_queue_periodic_membership_reconciliation:on_node_down(Node),
     %% If we have been partitioned, and we are now in the only remaining
     %% partition, we no longer care about partitions - forget them. Note
     %% that we do not attempt to deal with individual (other) partitions
@@ -845,7 +845,7 @@ handle_live_rabbit(Node) ->
     ok = rabbit_amqqueue:on_node_up(Node),
     ok = rabbit_alarm:on_node_up(Node),
     ok = rabbit_mnesia:on_node_up(Node),
-    ok = rabbit_queue_member_eval:on_node_up(Node).
+    ok = rabbit_quorum_queue_periodic_membership_reconciliation:on_node_up(Node).
 
 maybe_autoheal(State = #state{partitions = []}) ->
     State;
diff --git a/deps/rabbit/src/rabbit_policy.erl b/deps/rabbit/src/rabbit_policy.erl
@@ -367,13 +367,13 @@ validate(_VHost, <<"operator_policy">>, Name, Term, _User) ->
 notify(VHost, <<"policy">>, Name, Term0, ActingUser) ->
     Term = rabbit_data_coercion:atomize_keys(Term0),
     update_matched_objects(VHost, Term, ActingUser),
-    rabbit_queue_member_eval:policy_set(),
+    rabbit_quorum_queue_periodic_membership_reconciliation:policy_set(),
     rabbit_event:notify(policy_set, [{name, Name}, {vhost, VHost},
                                      {user_who_performed_action, ActingUser} | Term]);
 notify(VHost, <<"operator_policy">>, Name, Term0, ActingUser) ->
     Term = rabbit_data_coercion:atomize_keys(Term0),
     update_matched_objects(VHost, Term, ActingUser),
-    rabbit_queue_member_eval:policy_set(),
+    rabbit_quorum_queue_periodic_membership_reconciliation:policy_set(),
     rabbit_event:notify(policy_set, [{name, Name}, {vhost, VHost},
                                      {user_who_performed_action, ActingUser} | Term]).
 
diff --git a/deps/rabbit/src/rabbit_quorum_queue.erl b/deps/rabbit/src/rabbit_quorum_queue.erl
@@ -246,7 +246,7 @@ start_cluster(Q) ->
                     ok = rabbit_fifo_client:update_machine_state(LeaderId,
                                                                  ra_machine_config(NewQ)),
                     notify_decorators(QName, startup),
-                    rabbit_queue_member_eval:queue_created(NewQ),
+                    rabbit_quorum_queue_periodic_membership_reconciliation:queue_created(NewQ),
                     rabbit_event:notify(queue_created,
                                         [{name, QName},
                                          {durable, Durable},
diff --git a/deps/rabbit/src/rabbit_quorum_queue_periodic_membership_reconciliation.erl b/deps/rabbit/src/rabbit_quorum_queue_periodic_membership_reconciliation.erl
@@ -5,7 +5,7 @@
 %% Copyright (c) 2007-2023 VMware, Inc. or its affiliates.  All rights reserved.
 %%
 
--module(rabbit_queue_member_eval).
+-module(rabbit_quorum_queue_periodic_membership_reconciliation).
 
 -feature(maybe_expr, enable).
 
@@ -24,7 +24,7 @@
 -define(SHORT_INTERVAL, 10_000).
 -define(QUEUE_COUNT_START_RANDOM_SELECTION, 1_000).
 
--define(EVAL_MSG, member_eval).
+-define(EVAL_MSG, membership_reconciliation).
 
 -record(state, {timer_ref :: reference() | undefined,
                 default_interval :: non_neg_integer(),
@@ -43,35 +43,31 @@ start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
 %% API
 %%----------------------------------------------------------------------------
 
-on_node_up(_Node) ->
-    rabbit_log:debug("SIMON: NODEUP",[]),
-    gen_server:cast(?SERVER, eval_trigger).
+on_node_up(Node) ->
+    gen_server:cast(?SERVER, {membership_reconciliation_trigger, {node_up, Node}}).
 
-on_node_down(_Node) ->
-    rabbit_log:debug("SIMON: NODEDOWN",[]),
-    gen_server:cast(?SERVER, eval_trigger).
+on_node_down(Node) ->
+    gen_server:cast(?SERVER, {membership_reconciliation_trigger, {node_down, Node}}).
 
-queue_created(_Node) ->
-    rabbit_log:debug("SIMON: QUEUE CREATED",[]),
-    gen_server:cast(?SERVER, eval_trigger).
+queue_created(Q) ->
+    gen_server:cast(?SERVER, {membership_reconciliation_trigger, {queue_created, Q}}).
 
 policy_set() ->
-    rabbit_log:debug("SIMON: POLICY SET",[]),
-    gen_server:cast(?SERVER, eval_trigger).
+    gen_server:cast(?SERVER, {membership_reconciliation_trigger, policy_set}).
 
 %%----------------------------------------------------------------------------
 %% gen_server callbacks
 %%----------------------------------------------------------------------------
 
 init([]) ->
-    DefaultInterval = rabbit_misc:get_env(rabbit, quorum_membership_changes_default_timeout,
+    Enabled = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_enabled,
+                                  false),
+    DefaultInterval = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_default_timeout,
                                           ?DEFAULT_INTERVAL),
-    ShortInterval = rabbit_misc:get_env(rabbit, quorum_membership_changes_short_timeout,
+    ShortInterval = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_short_timeout,
                                         ?SHORT_INTERVAL),
-    TargetGroupSize = rabbit_misc:get_env(rabbit, quorum_membership_changes_target_group_size,
+    TargetGroupSize = rabbit_misc:get_env(rabbit, quorum_membership_reconciliation_target_group_size,
                                           undefined),
-    Enabled = rabbit_misc:get_env(rabbit, quorum_membership_changes_enabled,
-                                  false),
     State = #state{default_interval = DefaultInterval,
                    short_interval = ShortInterval,
                    target_group_size = TargetGroupSize,
@@ -87,10 +83,12 @@ init([]) ->
 handle_call(_Request, _From, State) ->
     {reply, ok, State}.
 
-handle_cast(eval_trigger, #state{enabled = false} = State) ->
+handle_cast({membership_reconciliation_trigger, _Reason}, #state{enabled = false} = State) ->
     {noreply, State, hibernate};
-handle_cast(eval_trigger, #state{timer_ref = OldRef,
+handle_cast({membership_reconciliation_trigger, Reason}, #state{timer_ref = OldRef,
                                short_interval = Time} = State) ->
+    rabbit_log:debug("Quorum Queue membership reconciliation triggered: ~p",
+                     [Reason]),
     _ = erlang:cancel_timer(OldRef),
     Ref = erlang:send_after(Time, self(), ?EVAL_MSG),
     {noreply, State#state{timer_ref = Ref}};
@@ -100,7 +98,7 @@ handle_cast(_Msg, State) ->
 handle_info(?EVAL_MSG, #state{default_interval = DefaultInterval,
                               short_interval = ShortInterval,
                               target_group_size = TargetSize} = State) ->
-    Res = eval_quorum_queue_membership(TargetSize),
+    Res = reconclitiate_quorum_queue_membership(TargetSize),
     NewTimeout = case Res of
                      noop ->
                          DefaultInterval;
@@ -124,39 +122,41 @@ code_change(_OldVsn, State, _Extra) ->
 %% Internal functions
 %%----------------------------------------------------------------------------
 
-eval_quorum_queue_membership(TargetSize) ->
+reconclitiate_quorum_queue_membership(TargetSize) ->
     LocalLeaders = rabbit_amqqueue:list_local_leaders(),
     ExpectedNodes = rabbit_nodes:list_members(),
     Running = rabbit_nodes:list_running(),
-    eval_quorum_members(ExpectedNodes, Running, LocalLeaders, TargetSize, noop).
+    reconclitiate_quorum_members(ExpectedNodes, Running, LocalLeaders, TargetSize, noop).
 
-eval_quorum_members(_ExpectedNodes, _Running, [], _TargetSize, Result) ->
+reconclitiate_quorum_members(_ExpectedNodes, _Running, [], _TargetSize, Result) ->
     Result;
-eval_quorum_members(ExpectedNodes, Running, [Q | LocalLeaders], TargetSize, OldResult) ->
-    {ok, Members, {_, LeaderNode}} = ra:members(amqqueue:get_pid(Q)),
-
+reconclitiate_quorum_members(ExpectedNodes, Running, [Q | LocalLeaders], TargetSize, OldResult) ->
     Result =
-    maybe
-        %% Check if Leader is indeed this node
-        LeaderNode ?= node(),
-        %% 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
-            [] ->
-                add_member(Q, Running, MemberNodes, get_target_size(Q, TargetSize));
+        maybe
+            {ok, Members, {_, LeaderNode}} = ra:members(amqqueue:get_pid(Q)),
+            %% Check if Leader is indeed this node
+            LeaderNode ?= node(),
+            %% 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
+                [] ->
+                    maybe_add_member(Q, Running, MemberNodes, get_target_size(Q, TargetSize));
+                _ ->
+                    remove_members(Q, Remove)
+            end
+        else
+            {timeout, Reason} ->
+                rabbit_log:debug("Find leader timeout: ~p", [Reason]),
+                ok;
             _ ->
-                remove_members(Q, Remove)
-        end
-    else
-        _ ->
-            noop
-    end,
-    eval_quorum_members(ExpectedNodes, Running, LocalLeaders, TargetSize,
-                        update_result(OldResult, Result)).
+                noop
+        end,
+    reconclitiate_quorum_members(ExpectedNodes, Running, LocalLeaders, TargetSize,
+                                 update_result(OldResult, Result)).
 
-add_member(Q, Running, MemberNodes, TargetSize) ->
+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),
     case should_add_node(MemberNodes, New, TargetSize) of
@@ -191,13 +191,10 @@ should_add_node(MemberNodes, New, TargetSize) ->
     maybe
         true ?= NumberOfNewNodes > 0, %% There are new nodes to grow to
         true ?= CurrentSize < TargetSize, %% Target size not reached
-        true ?= is_even(CurrentSize) orelse NumberOfNewNodes > 1, %% Enough nodes to grow to odd member size
+        true ?= rabbit_misc:is_even(CurrentSize) orelse NumberOfNewNodes > 1, %% Enough nodes to grow to odd member size
         true ?= rabbit_nodes:is_running(lists:delete(node(), MemberNodes))
     end.
 
-is_even(N) ->
-    (N band 1) =:= 0.
-
 get_target_size(Q, undefined) ->
     get_target_size(Q);
 get_target_size(Q, N) when N > 0 ->
diff --git a/deps/rabbit/test/quorum_queue_member_reconciliation_SUITE.erl b/deps/rabbit/test/quorum_queue_member_reconciliation_SUITE.erl
@@ -4,7 +4,8 @@
 %%
 %% Copyright (c) 2018-2023 VMware, Inc. or its affiliates.  All rights reserved.
 
--module(member_evaluation_SUITE).
+
+-module(quorum_queue_member_reconciliation_SUITE).
 
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
@@ -35,10 +36,10 @@ init_per_suite(Config0) ->
     rabbit_ct_helpers:log_environment(),
     Config1 = rabbit_ct_helpers:merge_app_env(
                 Config0, {rabbit, [{quorum_tick_interval, 1000},
-                                   {quorum_membership_changes_enabled, true},
-                                   {quorum_membership_changes_default_timeout, 5000},
-                                   {quorum_membership_changes_short_timeout, 2000},
-                                   {quorum_membership_changes_target_group_size, 3}]}),
+                                   {quorum_membership_reconciliation_enabled, true},
+                                   {quorum_membership_reconciliation_default_timeout, 5000},
+                                   {quorum_membership_reconciliation_short_timeout, 2000},
+                                   {quorum_membership_reconciliation_target_group_size, 3}]}),
     rabbit_ct_helpers:run_setup_steps(Config1, []).
 
 end_per_suite(Config) ->
diff --git a/deps/rabbit_common/src/rabbit_misc.erl b/deps/rabbit_common/src/rabbit_misc.erl
@@ -80,6 +80,8 @@
 -export([find_child/2]).
 -export([is_regular_file/1]).
 -export([maps_any/2]).
+-export([is_even/1, is_odd/1]).
+
 
 %% Horrible macro to use in guards
 -define(IS_BENIGN_EXIT(R),
@@ -1470,3 +1472,10 @@ maps_any_1(Pred, {K, V, I}) ->
         false ->
             maps_any_1(Pred, maps:next(I))
     end.
+
+-spec is_even(integer()) -> boolean().
+is_even(N) ->
+    (N band 1) =:= 0.
+-spec is_odd(integer()) -> boolean().
+is_odd(N) ->
+    (N band 1) =:= 1.
diff --git a/moduleindex.yaml b/moduleindex.yaml
@@ -736,6 +736,7 @@ rabbit:
 - rabbit_queue_type_util
 - rabbit_quorum_memory_manager
 - rabbit_quorum_queue
+- rabbit_quorum_queue_periodic_membership_reconciliation
 - rabbit_ra_registry
 - rabbit_ra_systems
 - rabbit_reader
