WIP; more testing with MFv2 + testcase improvementsadd-feature-flags-controller

2 files changed, 177 insertions, 53 deletions

diff --git a/deps/rabbit/src/rabbit_ff_controller.erl b/deps/rabbit/src/rabbit_ff_controller.erl
index 4d150f18cc..2b5dcc9f77 100644
--- a/deps/rabbit/src/rabbit_ff_controller.erl
+++ b/deps/rabbit/src/rabbit_ff_controller.erl
@@ -765,8 +765,10 @@ rpc_call(Node, Module, Function, Args, Timeout) ->
             {error, pre_feature_flags_rabbitmq};
         {badrpc, Reason} = Error ->
             ?LOG_ERROR(
-               "Feature flags: error while running ~s:~s~p "
-               "on node `~s`: ~p",
+               "Feature flags: error while running:~n"
+               "Feature flags:   ~s:~s~p~n"
+               "Feature flags: on node `~s`:~n"
+               "Feature flags:   ~p",
                [?MODULE, Function, Args, Node, Reason],
                #{domain => ?RMQLOG_DOMAIN_FEAT_FLAGS}),
             {error, Error};
diff --git a/deps/rabbit/test/feature_flags_v2_SUITE.erl b/deps/rabbit/test/feature_flags_v2_SUITE.erl
index e3f0474d88..af6d2dd9b5 100644
--- a/deps/rabbit/test/feature_flags_v2_SUITE.erl
+++ b/deps/rabbit/test/feature_flags_v2_SUITE.erl
@@ -35,10 +35,11 @@
          enable_supported_feature_flag_in_a_3node_cluster/1,
          enable_partially_supported_feature_flag_in_a_3node_cluster/1,
          enable_unsupported_feature_flag_in_a_3node_cluster/1,
-         enable_feature_flag_in_a_cluster_and_add_member_after/1,
-         enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv1/1,
-         enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2/1,
-         enable_feature_flag_in_a_cluster_and_remove_member_concurrently/1
+         enable_feature_flag_in_cluster_and_add_member_after/1,
+         enable_feature_flag_in_cluster_and_add_member_concurrently_mfv1/1,
+         enable_feature_flag_in_cluster_and_add_member_concurrently_mfv2/1,
+         enable_feature_flag_in_cluster_and_remove_member_concurrently_mfv1/1,
+         enable_feature_flag_in_cluster_and_remove_member_concurrently_mfv2/1
         ]).
 
 suite() ->
@@ -64,10 +65,11 @@ groups() ->
        enable_supported_feature_flag_in_a_3node_cluster,
        enable_partially_supported_feature_flag_in_a_3node_cluster,
        enable_unsupported_feature_flag_in_a_3node_cluster,
-       enable_feature_flag_in_a_cluster_and_add_member_after,
-       enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv1,
-       enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2,
-       enable_feature_flag_in_a_cluster_and_remove_member_concurrently
+       enable_feature_flag_in_cluster_and_add_member_after,
+       enable_feature_flag_in_cluster_and_add_member_concurrently_mfv1,
+       enable_feature_flag_in_cluster_and_add_member_concurrently_mfv2,
+       enable_feature_flag_in_cluster_and_remove_member_concurrently_mfv1,
+       enable_feature_flag_in_cluster_and_remove_member_concurrently_mfv2
       ]}
     ],
     [
@@ -279,7 +281,7 @@ mf_count_runs(_FeatureName, _FeatureProps, enable) ->
 
 mf_wait_and_count_runs(FeatureName, FeatureProps, enable = Arg) ->
     Peer = get_peer_proc(),
-    Peer ! {self(), waiting},
+    Peer ! {node(), self(), waiting},
     ?LOG_NOTICE("Migration function: waiting for signal from ~p...", [Peer]),
     receive proceed -> ok end,
     ?LOG_NOTICE("Migration function: unblocked!", []),
@@ -290,7 +292,7 @@ mf_wait_and_count_runs_v2(
              props = FeatureProps,
              command = enable = Arg}) ->
     Peer = get_peer_proc(),
-    Peer ! {self(), waiting},
+    Peer ! {node(), self(), waiting},
     ?LOG_NOTICE("Migration function: waiting for signal from ~p...", [Peer]),
     receive proceed -> ok end,
     ?LOG_NOTICE("Migration function: unblocked!", []),
@@ -590,7 +592,7 @@ enable_unsupported_feature_flag_in_a_3node_cluster(Config) ->
          || Node <- Nodes],
     ok.
 
-enable_feature_flag_in_a_cluster_and_add_member_after(Config) ->
+enable_feature_flag_in_cluster_and_add_member_after(Config) ->
     AllNodes = [NewNode | [FirstNode | _] = Nodes] = ?config(nodes, Config),
     connect_nodes(Nodes),
     override_running_nodes([NewNode]),
@@ -700,7 +702,7 @@ enable_feature_flag_in_a_cluster_and_add_member_after(Config) ->
          || Node <- AllNodes],
     ok.
 
-enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv1(Config) ->
+enable_feature_flag_in_cluster_and_add_member_concurrently_mfv1(Config) ->
     AllNodes = [NewNode | [FirstNode | _] = Nodes] = ?config(nodes, Config),
     connect_nodes(Nodes),
     override_running_nodes([NewNode]),
@@ -755,7 +757,9 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv1(Config) ->
     %% blocked and waits for a message from this process. Therefore, we are
     %% sure the feature flag is in the `state_changing' state and we can try
     %% to add a new node and sync its feature flags.
-    FirstNodeMigFunPid = receive {MigFunPid1, waiting} -> MigFunPid1 end,
+    FirstNodeMigFunPid = receive
+                             {_Node, MigFunPid1, waiting} -> MigFunPid1
+                         end,
 
     %% Check compatibility between NewNodes and Nodes. This doesn't block.
     ok = run_on_node(
@@ -862,10 +866,10 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv1(Config) ->
     %% which ran.
     ct:pal("Unblocking other nodes, including the joining one"),
     OtherMigratedNodes = [receive
-                              {MigFunPid2, waiting} ->
+                              {Node, MigFunPid2, waiting} ->
                                   MigFunPid2 ! proceed,
-                                  node(MigFunPid2)
-                          end || _ <- ExpectedNodes -- [FirstMigratedNode]],
+                                  Node 
+                          end || Node <- ExpectedNodes -- [FirstMigratedNode]],
     MigratedNodes = [FirstMigratedNode | OtherMigratedNodes],
     ?assertEqual(lists:sort(ExpectedNodes), lists:sort(MigratedNodes)),
 
@@ -894,7 +898,7 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv1(Config) ->
          || Node <- AllNodes],
     ok.
 
-enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2(Config) ->
+enable_feature_flag_in_cluster_and_add_member_concurrently_mfv2(Config) ->
     AllNodes = [NewNode | [FirstNode | _] = Nodes] = ?config(nodes, Config),
     connect_nodes(Nodes),
     override_running_nodes([NewNode]),
@@ -924,18 +928,25 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2(Config) ->
     ct:pal(
       "Enabling the feature flag in the cluster (in a separate process)"),
     Peer = self(),
+    _ = [ok =
+         run_on_node(
+           Node,
+           fun() ->
+                   %% The migration function uses the `Peer' PID (the process
+                   %% executing the testcase) to notify its own PID and wait
+                   %% for a signal from `Peer' to proceed and finish the
+                   %% migration.
+                   record_peer_proc(Peer),
+                   ok
+           end,
+           [])
+         || Node <- AllNodes],
     Enabler = spawn_link(
                 fun() ->
                         ok =
                         run_on_node(
                           FirstNode,
                           fun() ->
-                                  %% The migration function uses the `Peer'
-                                  %% PID (the process executing the testcase)
-                                  %% to notify its own PID and wait for a
-                                  %% signal from `Peer' to proceed and finish
-                                  %% the migration.
-                                  record_peer_proc(Peer),
                                   ?assertEqual(
                                      ok,
                                      rabbit_feature_flags:enable(
@@ -944,25 +955,15 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2(Config) ->
                           end,
                           [])
                 end),
-    _ = [ok =
-         run_on_node(
-           Node,
-           fun() ->
-                   %% We just need to record `Peer' on other nodes in the
-                   %% cluster because the migration will run on them with
-                   %% `feature_flags_v2'.
-                   record_peer_proc(Peer),
-                   ok
-           end,
-           [])
-         || Node <- tl(Nodes)],
 
     %% By waiting for the message from one of the migration function
     %% instances, we make sure the feature flags controller on `FirstNode' is
     %% blocked and waits for a message from this process. Therefore, we are
     %% sure the feature flag is in the `state_changing' state and we can try
     %% to add a new node and sync its feature flags.
-    FirstNodeMigFunPid = receive {MigFunPid1, waiting} -> MigFunPid1 end,
+    FirstNodeMigFunPid = receive
+                             {_Node, MigFunPid1, waiting} -> MigFunPid1
+                         end,
 
     %% Check compatibility between NewNodes and Nodes. This doesn't block.
     ok = run_on_node(
@@ -994,7 +995,6 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2(Config) ->
                        run_on_node(
                          NewNode,
                          fun() ->
-                                 record_peer_proc(Peer),
                                  ?assertEqual(
                                     ok,
                                     rabbit_feature_flags:
@@ -1033,11 +1033,12 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2(Config) ->
     SyncerMRef = erlang:monitor(process, Syncer),
     unlink(Enabler),
     unlink(Syncer),
+
     %% The migration function runs on all clustered nodes with v2, including
     %% the one joining the cluster, thanks to the synchronization.
     %%
     %% When this testcase runs with feature flags v1, the feature flag we want
-    %% to enable uses the migration function API v2: this implicitely enables
+    %% to enable uses the migration function API v2: this implicitly enables
     %% `feature_flags_v2'. As part of the synchronization, the node still on
     %% feature flags v1 will try to sync `feature_flags_v2' specificaly first.
     %% After that, the controller-based sync proceeds.
@@ -1055,10 +1056,10 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2(Config) ->
     %% which ran.
     ct:pal("Unblocking other nodes, including the joining one"),
     OtherMigratedNodes = [receive
-                              {MigFunPid2, waiting} ->
+                              {Node, MigFunPid2, waiting} ->
                                   MigFunPid2 ! proceed,
-                                  node(MigFunPid2)
-                          end || _ <- ExpectedNodes -- [FirstMigratedNode]],
+                                  Node
+                          end || Node <- ExpectedNodes -- [FirstMigratedNode]],
     MigratedNodes = [FirstMigratedNode | OtherMigratedNodes],
     ?assertEqual(lists:sort(ExpectedNodes), lists:sort(MigratedNodes)),
 
@@ -1097,7 +1098,7 @@ enable_feature_flag_in_a_cluster_and_add_member_concurrently_mfv2(Config) ->
          || Node <- AllNodes],
     ok.
 
-enable_feature_flag_in_a_cluster_and_remove_member_concurrently(Config) ->
+enable_feature_flag_in_cluster_and_remove_member_concurrently_mfv1(Config) ->
     AllNodes = [LeavingNode | [FirstNode | _] = Nodes] = ?config(
                                                             nodes, Config),
     connect_nodes(AllNodes),
@@ -1110,8 +1111,6 @@ enable_feature_flag_in_a_cluster_and_remove_member_concurrently(Config) ->
                        migration_fun => {?MODULE, mf_wait_and_count_runs}}},
     inject_on_nodes(AllNodes, FeatureFlags),
 
-    UsingFFv1 = not ?config(enable_feature_flags_v2, Config),
-
     ct:pal(
       "Checking the feature flag is supported but disabled on all nodes"),
     _ = [ok =
@@ -1128,10 +1127,6 @@ enable_feature_flag_in_a_cluster_and_remove_member_concurrently(Config) ->
     ct:pal(
       "Enabling the feature flag in the cluster (in a separate process)"),
     Peer = self(),
-    ExpectedRet = case UsingFFv1 of
-                      true -> ok;
-                      false -> {error, {badrpc, nodedown}}
-                  end,
     Enabler = spawn_link(
                 fun() ->
                         ok =
@@ -1145,7 +1140,7 @@ enable_feature_flag_in_a_cluster_and_remove_member_concurrently(Config) ->
                                   %% the migration.
                                   record_peer_proc(Peer),
                                   ?assertEqual(
-                                     ExpectedRet,
+                                     ok,
                                      rabbit_feature_flags:enable(
                                        FeatureName)),
                                   ok
@@ -1158,7 +1153,9 @@ enable_feature_flag_in_a_cluster_and_remove_member_concurrently(Config) ->
     %% blocked and waits for a message from this process. Therefore, we are
     %% sure the feature flag is in the `state_changing' state and we can try
     %% to add a new node and sync its feature flags.
-    FirstNodeMigFunPid = receive {MigFunPid1, waiting} -> MigFunPid1 end,
+    FirstNodeMigFunPid = receive
+                             {_Node, MigFunPid1, waiting} -> MigFunPid1
+                         end,
 
     %% Remove node from cluster.
     stop_slave_node(LeavingNode),
@@ -1189,9 +1186,134 @@ enable_feature_flag_in_a_cluster_and_remove_member_concurrently(Config) ->
          run_on_node(
            Node,
            fun() ->
+                   ?assert(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ok
+           end,
+           [])
+         || Node <- Nodes],
+    ok.
+
+enable_feature_flag_in_cluster_and_remove_member_concurrently_mfv2(Config) ->
+    AllNodes = [LeavingNode | [FirstNode | _] = Nodes] = ?config(
+                                                            nodes, Config),
+    connect_nodes(AllNodes),
+    override_running_nodes(AllNodes),
+
+    FeatureName = ?FUNCTION_NAME,
+    FeatureFlags = #{FeatureName =>
+                     #{provided_by => ?MODULE,
+                       stability => stable,
+                       migration_fun => {?MODULE,
+                                         mf_wait_and_count_runs_v2}}},
+    inject_on_nodes(AllNodes, FeatureFlags),
+
+    UsingFFv1 = not ?config(enable_feature_flags_v2, Config),
+
+    ct:pal(
+      "Checking the feature flag is supported but disabled on all nodes"),
+    _ = [ok =
+         run_on_node(
+           Node,
+           fun() ->
+                   ?assert(rabbit_feature_flags:is_supported(FeatureName)),
+                   ?assertNot(rabbit_feature_flags:is_enabled(FeatureName)),
+                   ok
+           end,
+           [])
+         || Node <- AllNodes],
+
+    ct:pal(
+      "Enabling the feature flag in the cluster (in a separate process)"),
+    Peer = self(),
+    _ = [ok =
+         run_on_node(
+           Node,
+           fun() ->
+                   %% The migration function uses the `Peer' PID (the process
+                   %% executing the testcase) to notify its own PID and wait
+                   %% for a signal from `Peer' to proceed and finish the
+                   %% migration.
+                   record_peer_proc(Peer),
+                   ok
+           end,
+           [])
+         || Node <- AllNodes],
+    Enabler = spawn_link(
+                fun() ->
+                        ok =
+                        run_on_node(
+                          FirstNode,
+                          fun() ->
+                                  ?assertEqual(
+                                     {error, {badrpc, nodedown}},
+                                     rabbit_feature_flags:enable(
+                                       FeatureName)),
+                                  ok
+                          end,
+                          [])
+                end),
+
+    %% By waiting for the message from one of the migration function
+    %% instances, we make sure the feature flags controller on `FirstNode' is
+    %% blocked and waits for a message from this process. Therefore, we are
+    %% sure the feature flag is in the `state_changing' state and we can try
+    %% to add a new node and sync its feature flags.
+    FirstNodeMigFunPid = receive
+                             {Node, MigFunPid1, waiting} -> MigFunPid1
+                         end,
+
+    %% Remove node from cluster.
+    stop_slave_node(LeavingNode),
+    override_running_nodes(Nodes),
+
+    %% Unblock the migration functions on `Nodes'.
+    EnablerMRef = erlang:monitor(process, Enabler),
+    unlink(Enabler),
+
+    %% The migration function runs on all clustered nodes with v2.
+    %%
+    %% When this testcase runs with feature flags v1, the feature flag we want
+    %% to enable uses the migration function API v2: this implicitly enables
+    %% `feature_flags_v2'. As part of the synchronization, the node still on
+    %% feature flags v1 will try to sync `feature_flags_v2' specificaly first.
+    %% After that, the controller-based sync proceeds.
+    ExpectedNodes = Nodes,
+
+    %% Unblock the migration function for which we already consumed the
+    %% `waiting' notification.
+    FirstMigratedNode = node(FirstNodeMigFunPid),
+    ct:pal(
+      "Unblocking first node (~p @ ~s)",
+      [FirstNodeMigFunPid, FirstMigratedNode]),
+    FirstNodeMigFunPid ! proceed,
+
+    %% Unblock the rest and collect the node names of all migration functions
+    %% which ran.
+    ct:pal("Unblocking other nodes"),
+    OtherMigratedNodes = [receive
+                              {Node, MigFunPid2, waiting} ->
+                                  MigFunPid2 ! proceed,
+                                  Node
+                          end || Node <- ExpectedNodes -- [FirstMigratedNode]],
+    MigratedNodes = [FirstMigratedNode | OtherMigratedNodes],
+    ?assertEqual(lists:sort(ExpectedNodes), lists:sort(MigratedNodes)),
+
+    ct:pal("Waiting for spawned processes to terminate"),
+    receive
+        {'DOWN', EnablerMRef, process, Enabler, EnablerReason} ->
+            ?assertEqual(normal, EnablerReason)
+    end,
+
+    ct:pal(
+      "Checking the feature flag is enabled (v1) or disabled (v2) in the "
+      "cluster"),
+    _ = [ok =
+         run_on_node(
+           Node,
+           fun() ->
                    case UsingFFv1 of
                        true ->
-                           ?assert(
+                           ?assertNot(
                               rabbit_feature_flags:is_enabled(FeatureName));
                        false ->
                            ?assertNot(