make: Initial work on using ct_master to run tests

Because `ct_master` is yet another Erlang node, and it is used
to run multiple CT nodes, meaning it is in a cluster of CT
nodes, the tests that change the net_ticktime could not
work properly anymore. This is because net_ticktime must
be the same value across the cluster.

The same value had to be set for all tests in order to solve
this. This is why it was changed to 5s across the board. The
lower net_ticktime was used in most places to speed up tests
that must deal with cluster failures, so that value is good
enough for these cases.

One test in amqp_client was using the net_ticktime to test
the behavior of the direct connection timeout with varying
net_ticktime configurations. The test now mocks the
`net_kernel:get_net_ticktime()` function to achieve the
same result.

Author: lhoguin

deps/amqp_client/test/system_SUITE.erl

@@ -335,14 +335,16 @@ safe_call_timeouts_test(Params = #amqp_params_network{}) ->
     meck:unload(amqp_network_connection);
 
 safe_call_timeouts_test(Params = #amqp_params_direct{}) ->
+    %% We must mock net_kernel:get_net_ticktime/0 as changing
+    %% the tick time directly could lead to nodes disconnecting.
+    meck:new(net_kernel, [unstick, passthrough]),
+
     TestCallTimeout = 30000,
-    NetTicktime0 = net_kernel:get_net_ticktime(),
     amqp_util:update_call_timeout(TestCallTimeout),
 
     %% 1. NetTicktime >= DIRECT_OPERATION_TIMEOUT (120s)
     NetTicktime1 = 140,
-    net_kernel:set_net_ticktime(NetTicktime1, 1),
-    wait_until_net_ticktime(NetTicktime1),
+    meck:expect(net_kernel, get_net_ticktime, fun() -> NetTicktime1 end),
 
     {ok, Connection1} = amqp_connection:start(Params),
     ?assertEqual((NetTicktime1 * 1000) + ?CALL_TIMEOUT_DEVIATION,
@@ -356,15 +358,12 @@ safe_call_timeouts_test(Params = #amqp_params_direct{}) ->
 
     %% 2. Transitioning NetTicktime >= DIRECT_OPERATION_TIMEOUT (120s)
     NetTicktime2 = 120,
-    net_kernel:set_net_ticktime(NetTicktime2, 1),
-    ?assertEqual({ongoing_change_to, NetTicktime2}, net_kernel:get_net_ticktime()),
+    meck:expect(net_kernel, get_net_ticktime, fun() -> {ongoing_change_to, NetTicktime2} end),
 
     {ok, Connection2} = amqp_connection:start(Params),
     ?assertEqual((NetTicktime2 * 1000) + ?CALL_TIMEOUT_DEVIATION,
         amqp_util:call_timeout()),
 
-    wait_until_net_ticktime(NetTicktime2),
-
     ?assertEqual(ok, amqp_connection:close(Connection2)),
     wait_for_death(Connection2),
 
@@ -373,15 +372,14 @@ safe_call_timeouts_test(Params = #amqp_params_direct{}) ->
 
     %% 3. NetTicktime < DIRECT_OPERATION_TIMEOUT (120s)
     NetTicktime3 = 60,
-    net_kernel:set_net_ticktime(NetTicktime3, 1),
-    wait_until_net_ticktime(NetTicktime3),
+    meck:expect(net_kernel, get_net_ticktime, fun() -> NetTicktime3 end),
 
     {ok, Connection3} = amqp_connection:start(Params),
     ?assertEqual((?DIRECT_OPERATION_TIMEOUT + ?CALL_TIMEOUT_DEVIATION),
         amqp_util:call_timeout()),
 
-    net_kernel:set_net_ticktime(NetTicktime0, 1),
-    wait_until_net_ticktime(NetTicktime0),
+    meck:unload(net_kernel),
+
     ?assertEqual(ok, amqp_connection:close(Connection3)),
     wait_for_death(Connection3),
 
@@ -1578,16 +1576,6 @@ assert_down_with_error(MonitorRef, CodeAtom) ->
         exit(did_not_die)
     end.
 
-wait_until_net_ticktime(NetTicktime) ->
-    case net_kernel:get_net_ticktime() of
-        NetTicktime -> ok;
-        {ongoing_change_to, NetTicktime} ->
-            timer:sleep(1000),
-            wait_until_net_ticktime(NetTicktime);
-        _ ->
-            throw({error, {net_ticktime_not_set, NetTicktime}})
-    end.
-
 set_resource_alarm(Resource, Config)
   when Resource =:= memory orelse Resource =:= disk ->
     SrcDir = ?config(amqp_client_srcdir, Config),

deps/rabbit/Makefile

@@ -226,6 +226,36 @@ ct-slow:
 
 CT_OPTS += -ct_hooks rabbit_ct_hook []
 
+define ct_master.erl
+	StartOpts = #{
+		host => "localhost",
+		connection => standard_io,
+		args => ["-hidden"]
+	},
+	{ok, Pid1, _} = peer:start(StartOpts#{name => "rabbit_shard1"}),
+	{ok, Pid2, _} = peer:start(StartOpts#{name => "rabbit_shard2"}),
+	{ok, Pid3, _} = peer:start(StartOpts#{name => "rabbit_shard3"}),
+	{ok, Pid4, _} = peer:start(StartOpts#{name => "rabbit_shard4"}),
+	peer:call(Pid1, net_kernel, set_net_ticktime, [5]),
+	peer:call(Pid2, net_kernel, set_net_ticktime, [5]),
+	peer:call(Pid3, net_kernel, set_net_ticktime, [5]),
+	peer:call(Pid4, net_kernel, set_net_ticktime, [5]),
+	peer:call(Pid1, persistent_term, put, [rabbit_ct_tcp_port_base, 23000]),
+	peer:call(Pid2, persistent_term, put, [rabbit_ct_tcp_port_base, 25000]),
+	peer:call(Pid3, persistent_term, put, [rabbit_ct_tcp_port_base, 27000]),
+	peer:call(Pid4, persistent_term, put, [rabbit_ct_tcp_port_base, 29000]),
+	ct_master:run("ct.test.spec"),
+	peer:stop(Pid4),
+	peer:stop(Pid3),
+	peer:stop(Pid2),
+	peer:stop(Pid1),
+	halt()
+endef
+
+ct-master: test-build
+	$(verbose) mkdir -p $(CT_LOGS_DIR)
+	$(call erlang,$(ct_master.erl),-sname rabbit_master@localhost -hidden -kernel net_ticktime 5)
+
 # --------------------------------------------------------------------
 # Compilation.
 # --------------------------------------------------------------------

deps/rabbit/ct.test.spec

@@ -0,0 +1,186 @@
+{logdir, "logs/"}.
+{logdir, master, "logs/"}.
+{create_priv_dir, all_nodes, auto_per_run}.
+
+{node, shard1, 'rabbit_shard1@localhost'}.
+{node, shard2, 'rabbit_shard2@localhost'}.
+{node, shard3, 'rabbit_shard3@localhost'}.
+{node, shard4, 'rabbit_shard4@localhost'}.
+
+%%
+%% Sets of test suites that take around the same time to complete.
+%%
+
+{define, 'Set1', [
+    amqp_address_SUITE
+,   amqp_auth_SUITE
+,   amqp_client_SUITE
+,   amqp_credit_api_v2_SUITE
+,   amqp_proxy_protocol_SUITE
+,   amqp_system_SUITE
+,   amqpl_consumer_ack_SUITE
+,   amqpl_direct_reply_to_SUITE
+,   amqqueue_backward_compatibility_SUITE
+,   backing_queue_SUITE
+,   bindings_SUITE
+,   channel_interceptor_SUITE
+,   channel_operation_timeout_SUITE
+,   classic_queue_SUITE
+,   classic_queue_prop_SUITE
+]}.
+
+{define, 'Set2', [
+    cluster_SUITE
+,   config_schema_SUITE
+,   confirms_rejects_SUITE
+,   consumer_timeout_SUITE
+,   crashing_queues_SUITE
+,   deprecated_features_SUITE
+,   direct_exchange_routing_v2_SUITE
+,   disconnect_detected_during_alarm_SUITE
+,   disk_monitor_SUITE
+,   dynamic_qq_SUITE
+,   exchanges_SUITE
+,   rabbit_stream_queue_SUITE
+]}.
+
+{define, 'Set3', [
+    cli_forget_cluster_node_SUITE
+,   feature_flags_SUITE
+,   feature_flags_v2_SUITE
+,   feature_flags_with_unpriveleged_user_SUITE
+,   list_consumers_sanity_check_SUITE
+,   list_queues_online_and_offline_SUITE
+,   logging_SUITE
+,   lqueue_SUITE
+,   maintenance_mode_SUITE
+,   mc_unit_SUITE
+,   message_containers_deaths_v2_SUITE
+,   message_size_limit_SUITE
+,   metadata_store_migration_SUITE
+,   metadata_store_phase1_SUITE
+,   metrics_SUITE
+,   mirrored_supervisor_SUITE
+,   msg_store_SUITE
+,   peer_discovery_classic_config_SUITE
+]}.
+
+{define, 'Set4', [
+    peer_discovery_dns_SUITE
+,   peer_discovery_tmp_hidden_node_SUITE
+,   per_node_limit_SUITE
+,   per_user_connection_channel_limit_SUITE
+,   per_user_connection_channel_tracking_SUITE
+,   per_user_connection_tracking_SUITE
+,   per_vhost_connection_limit_SUITE
+,   per_vhost_msg_store_SUITE
+,   per_vhost_queue_limit_SUITE
+,   policy_SUITE
+,   priority_queue_SUITE
+,   priority_queue_recovery_SUITE
+,   product_info_SUITE
+,   proxy_protocol_SUITE
+,   publisher_confirms_parallel_SUITE
+]}.
+
+{define, 'Set5', [
+    clustering_recovery_SUITE
+,   metadata_store_clustering_SUITE
+,   queue_length_limits_SUITE
+,   queue_parallel_SUITE
+,   quorum_queue_SUITE
+,   rabbit_access_control_SUITE
+,   rabbit_confirms_SUITE
+,   rabbit_core_metrics_gc_SUITE
+,   rabbit_cuttlefish_SUITE
+,   rabbit_db_binding_SUITE
+,   rabbit_db_exchange_SUITE
+,   rabbit_db_maintenance_SUITE
+,   rabbit_db_msup_SUITE
+,   rabbit_db_policy_SUITE
+,   rabbit_db_queue_SUITE
+,   rabbit_db_topic_exchange_SUITE
+,   rabbit_direct_reply_to_prop_SUITE
+]}.
+
+{define, 'Set6', [
+    queue_type_SUITE
+,   quorum_queue_member_reconciliation_SUITE
+,   rabbit_fifo_SUITE
+,   rabbit_fifo_dlx_SUITE
+,   rabbit_fifo_dlx_integration_SUITE
+,   rabbit_fifo_int_SUITE
+,   rabbit_fifo_prop_SUITE
+,   rabbit_fifo_v0_SUITE
+,   rabbit_local_random_exchange_SUITE
+,   rabbit_message_interceptor_SUITE
+,   rabbit_stream_coordinator_SUITE
+,   rabbit_stream_sac_coordinator_SUITE
+,   rabbitmq_4_0_deprecations_SUITE
+,   rabbitmq_queues_cli_integration_SUITE
+,   rabbitmqctl_integration_SUITE
+,   rabbitmqctl_shutdown_SUITE
+,   routing_SUITE
+,   runtime_parameters_SUITE
+]}.
+
+{define, 'Set7', [
+    cluster_limit_SUITE
+,   cluster_minority_SUITE
+,   clustering_management_SUITE
+,   signal_handling_SUITE
+,   single_active_consumer_SUITE
+,   term_to_binary_compat_prop_SUITE
+,   topic_permission_SUITE
+,   transactions_SUITE
+,   unicode_SUITE
+,   unit_access_control_SUITE
+,   unit_access_control_authn_authz_context_propagation_SUITE
+,   unit_access_control_credential_validation_SUITE
+,   unit_amqp091_content_framing_SUITE
+,   unit_amqp091_server_properties_SUITE
+,   unit_app_management_SUITE
+,   unit_cluster_formation_locking_mocks_SUITE
+,   unit_cluster_formation_sort_nodes_SUITE
+,   unit_collections_SUITE
+,   unit_config_value_encryption_SUITE
+,   unit_connection_tracking_SUITE
+]}.
+
+{define, 'Set8', [
+    dead_lettering_SUITE
+,   definition_import_SUITE
+,   per_user_connection_channel_limit_partitions_SUITE
+,   per_vhost_connection_limit_partitions_SUITE
+,   unit_credit_flow_SUITE
+,   unit_disk_monitor_SUITE
+,   unit_file_handle_cache_SUITE
+,   unit_gen_server2_SUITE
+,   unit_log_management_SUITE
+,   unit_operator_policy_SUITE
+,   unit_pg_local_SUITE
+,   unit_plugin_directories_SUITE
+,   unit_plugin_versioning_SUITE
+,   unit_policy_validators_SUITE
+,   unit_priority_queue_SUITE
+,   unit_queue_consumers_SUITE
+,   unit_queue_location_SUITE
+,   unit_quorum_queue_SUITE
+,   unit_stats_and_metrics_SUITE
+,   unit_supervisor2_SUITE
+,   unit_vm_memory_monitor_SUITE
+,   upgrade_preparation_SUITE
+,   vhost_SUITE
+]}.
+
+{suites, shard1, "test/", 'Set1'}.
+{suites, shard1, "test/", 'Set2'}.
+
+{suites, shard2, "test/", 'Set3'}.
+{suites, shard2, "test/", 'Set4'}.
+
+{suites, shard3, "test/", 'Set5'}.
+{suites, shard3, "test/", 'Set6'}.
+
+{suites, shard4, "test/", 'Set7'}.
+{suites, shard4, "test/", 'Set8'}.

deps/rabbit/test/cluster_limit_SUITE.erl

@@ -54,8 +54,7 @@ init_per_group(Group, Config) ->
                                                    [{rmq_nodes_count, ClusterSize},
                                                     {rmq_nodename_suffix, Group},
                                                     {tcp_ports_base}]),
-            Config1b = rabbit_ct_helpers:set_config(Config1, [{net_ticktime, 10}]),
-            rabbit_ct_helpers:run_steps(Config1b,
+            rabbit_ct_helpers:run_steps(Config1,
                                         [fun merge_app_env/1 ] ++
                                             rabbit_ct_broker_helpers:setup_steps())
     end.

deps/rabbit/test/feature_flags_v2_SUITE.erl

@@ -167,7 +167,15 @@ start_slave_node(Parent, Config, Testcase, N) ->
     Name = list_to_atom(
              rabbit_misc:format("~ts-~b", [Testcase, N])),
     ct:pal("- Starting slave node `~ts@...`", [Name]),
-    {ok, Node} = slave:start(net_adm:localhost(), Name),
+    {ok, NodePid, Node} = peer:start(#{
+        name => Name,
+        connection => standard_io,
+        shutdown => close
+    }),
+    peer:call(NodePid, net_kernel, set_net_ticktime, [5]),
+
+    persistent_term:put({?MODULE, Node}, NodePid),
+
     ct:pal("- Slave node `~ts` started", [Node]),
 
     TestCodePath = filename:dirname(code:which(?MODULE)),
@@ -183,8 +191,16 @@ stop_slave_nodes(Config) ->
     rabbit_ct_helpers:delete_config(Config, nodes).
 
 stop_slave_node(Node) ->
-    ct:pal("- Stopping slave node `~ts`...", [Node]),
-    ok = slave:stop(Node).
+    case persistent_term:get({?MODULE, Node}, undefined) of
+        undefined ->
+            %% Node was already stopped (e.g. by the test case).
+            ok;
+        NodePid ->
+            persistent_term:erase({?MODULE, Node}),
+
+            ct:pal("- Stopping slave node `~ts`...", [Node]),
+            ok = peer:stop(NodePid)
+    end.
 
 connect_nodes([FirstNode | OtherNodes] = Nodes) ->
     lists:foreach(

deps/rabbit/test/metadata_store_clustering_SUITE.erl

@@ -73,8 +73,7 @@ end_per_suite(Config) ->
 init_per_group(unclustered, Config) ->
     rabbit_ct_helpers:set_config(Config, [{metadata_store, mnesia},
                                           {rmq_nodes_clustered, false},
-                                          {tcp_ports_base},
-                                          {net_ticktime, 10}]);
+                                          {tcp_ports_base}]);
 init_per_group(cluster_size_2, Config) ->
     rabbit_ct_helpers:set_config(Config, [{rmq_nodes_count, 2}]);
 init_per_group(cluster_size_3, Config) ->

deps/rabbit/test/per_vhost_connection_limit_SUITE.erl

@@ -105,9 +105,8 @@ init_per_multinode_group(_Group, Config, NodeCount) ->
     rabbit_ct_broker_helpers:setup_steps() ++
     rabbit_ct_client_helpers:setup_steps()).
 
-end_per_group(Group, Config) when Group == tests;
-                                  Group == khepri_migration ->
-    % The broker is managed by {init,end}_per_testcase().
+end_per_group(Group, Config) when Group == tests ->
+    % The broker is managed by sub-groups.
     Config;
 end_per_group(_Group, Config) ->
     rabbit_ct_helpers:run_steps(Config,

deps/rabbit/test/queue_parallel_SUITE.erl

@@ -646,7 +646,11 @@ delete_immediately_by_resource(Config) ->
     ok.
 
 cc_header_non_array_should_close_channel(Config) ->
-    {C, Ch} = rabbit_ct_client_helpers:open_connection_and_channel(Config, 0),
+    %% We use an unmanaged connection to avoid issues with
+    %% tests running in parallel: in this test we expect the
+    %% channel to close, but that channel is reused in other tests.
+    C = rabbit_ct_client_helpers:open_unmanaged_connection(Config, 0),
+    {ok, Ch} = amqp_connection:open_channel(C),
     Name0 = ?FUNCTION_NAME,
     Name = atom_to_binary(Name0),
     QName = <<"queue_cc_header_non_array", Name/binary>>,

deps/rabbit/test/queue_type_SUITE.erl

@@ -56,8 +56,7 @@ init_per_group(Group, Config) ->
                                             {tcp_ports_base, {skip_n_nodes, ClusterSize}}
                                             ]),
     Config1b = rabbit_ct_helpers:set_config(Config1,
-                                            [{queue_type, atom_to_binary(Group, utf8)},
-                                             {net_ticktime, 5}
+                                            [{queue_type, atom_to_binary(Group, utf8)}
                                             ]),
     Config2 = rabbit_ct_helpers:run_steps(Config1b,
                                           [fun merge_app_env/1 ] ++

deps/rabbit/test/quorum_queue_member_reconciliation_SUITE.erl

@@ -51,8 +51,7 @@ init_per_group(Group, Config) ->
                                            [{rmq_nodes_count, ClusterSize},
                                             {rmq_nodename_suffix, Group},
                                             {tcp_ports_base}]),
-    Config1b = rabbit_ct_helpers:set_config(Config1, [{net_ticktime, 10}]),
-    rabbit_ct_helpers:run_steps(Config1b,
+    rabbit_ct_helpers:run_steps(Config1,
                                 [fun merge_app_env/1 ] ++
                                     rabbit_ct_broker_helpers:setup_steps()).