-module(rcl_test). -export([at_init_testsuite/0, pmap/2, init_nodes/4, start_node/4, add_nodes_to_cluster/2]). init_nodes(Suite, Config, CommonConfig, NodesConfig) -> ct:pal("[~p]", [Suite]), at_init_testsuite(), StartDCs = fun (Nodes) -> pmap(fun (N) -> NodeConfig = maps:get(N, NodesConfig), {_Status, Node} = start_node(N, Config, CommonConfig, NodeConfig), Node end, Nodes) end, [Nodes] = pmap(fun (N) -> StartDCs(N) end, [maps:keys(NodesConfig)]), [{clusters, [Nodes]}, {nodes, Nodes} | Config]. at_init_testsuite() -> {ok, Hostname} = inet:gethostname(), case net_kernel:start([list_to_atom("runner@" ++ Hostname), shortnames]) of {ok, _} -> ok; {error, {already_started, _}} -> ok; {error, {{already_started, _}, _}} -> ok end. add_nodes_to_cluster(NodeTarget, NodesToAdd) -> rcl_cluster_manager:add_nodes_to_cluster(NodeTarget, NodesToAdd). -spec pmap(fun(), list()) -> list(). pmap(F, L) -> Parent = self(), lists:foldl(fun (X, N) -> spawn_link(fun () -> Parent ! {pmap, N, F(X)} end), N + 1 end, 0, L), L2 = [receive {pmap, N, R} -> {N, R} end || _ <- L], {_, L3} = lists:unzip(lists:keysort(1, L2)), L3. start_node(Name, Config, CommonConfig, NodeConfig) -> #{app := RclApp} = CommonConfig, #{base_port := BasePort} = NodeConfig, %% code path for compiled dependencies (ebin folders) CodePath = lists:filter(fun filelib:is_dir/1, code:get_path()), ct:log("Starting node ~p", [Name]), {ok, Cwd} = file:get_cwd(), AppFolder = filename:dirname(filename:dirname(Cwd)), PrivDir = proplists:get_value(priv_dir, Config), NodeDir = filename:join([PrivDir, Name]) ++ "/", filelib:ensure_dir(NodeDir), AppDataDirName = to_ls(maps:get(data_dir_name, CommonConfig, "rcl-data")), BuildEnv = maps:get(build_env, CommonConfig, "dev"), DefaultBuildPath = io_lib:format("/_build/~s/rel/~s/lib/", [BuildEnv, RclApp]), BuildPath = to_ls(maps:get(build_path, CommonConfig, DefaultBuildPath)), DefaultSchemaDirs = io_lib:format("~s~s", [AppFolder, BuildPath]), SchemaDirs = to_ls(maps:get(schema_dirs, CommonConfig, DefaultSchemaDirs)), SetupNodeFn = maps:get(setup_node_fn, CommonConfig, fun (_) -> ok end), %% have the slave nodes monitor the runner node, so they can't outlive it TestNodeConfig = [%% have the slave nodes monitor the runner node, so they can't outlive it {monitor_master, true}, %% set code path for dependencies {startup_functions, [{code, set_path, [CodePath]}]}], case ct_slave:start(Name, TestNodeConfig) of {ok, Node} -> % load application to allow for configuring the environment before starting ok = rpc:call(Node, application, load, [riak_core]), ok = rpc:call(Node, application, load, [RclApp]), %% get remote working dir of node {ok, NodeWorkingDir} = rpc:call(Node, file, get_cwd, []), %% DATA DIRS ok = rpc:call(Node, application, set_env, [RclApp, data_dir, filename:join([NodeWorkingDir, Node, AppDataDirName])]), ok = rpc:call(Node, application, set_env, [riak_core, ring_state_dir, filename:join([NodeWorkingDir, Node, "data"])]), ok = rpc:call(Node, application, set_env, [riak_core, platform_data_dir, filename:join([NodeWorkingDir, Node, "data"])]), ok = rpc:call(Node, application, set_env, [riak_core, schema_dirs, [SchemaDirs]]), %% PORTS ok = rpc:call(Node, application, set_env, [riak_core, handoff_port, BasePort]), SetupNodeFn(#{node => Node, base_port => BasePort + 1, node_dir => NodeWorkingDir, node_config => TestNodeConfig, common_config => CommonConfig}), %% LOGGING Configuration %% add additional logging handlers to ensure easy access to remote node logs %% for each logging level LogRoot = filename:join([NodeWorkingDir, Node, "logs"]), %% set the logger configuration ok = rpc:call(Node, application, set_env, [RclApp, logger, log_config(LogRoot)]), %% set primary output level, no filter rpc:call(Node, logger, set_primary_config, [level, all]), %% load additional logger handlers at remote node rpc:call(Node, logger, add_handlers, [RclApp]), %% redirect slave logs to ct_master logs ok = rpc:call(Node, application, set_env, [RclApp, ct_master, node()]), %% reduce number of actual log files created to 4, reduces start-up time of node ok = rpc:call(Node, application, set_env, [riak_core, ring_creation_size, 4]), {ok, _} = rpc:call(Node, application, ensure_all_started, [RclApp]), ct:pal("Node ~p started with base port ~p", [Node, BasePort]), {connect, Node}; {error, already_started, Node} -> ct:log("Node ~p already started, reusing node", [Node]), {ready, Node}; {error, Reason, Node} -> ct:pal("Error starting node ~w, reason ~w, will retry", [Node, Reason]), ct_slave:stop(Name), time_utils:wait_until_offline(Node), start_node(Name, Config, CommonConfig, NodeConfig) end. %% logger configuration for each level %% see http://erlang.org/doc/man/logger.html log_config(LogDir) -> DebugConfig = #{level => debug, formatter => {logger_formatter, #{single_line => true, max_size => 2048}}, config => #{type => {file, filename:join(LogDir, "debug.log")}}}, InfoConfig = #{level => info, formatter => {logger_formatter, #{single_line => true, max_size => 2048}}, config => #{type => {file, filename:join(LogDir, "info.log")}}}, NoticeConfig = #{level => notice, formatter => {logger_formatter, #{single_line => true, max_size => 2048}}, config => #{type => {file, filename:join(LogDir, "notice.log")}}}, WarningConfig = #{level => warning, formatter => {logger_formatter, #{single_line => true, max_size => 2048}}, config => #{type => {file, filename:join(LogDir, "warning.log")}}}, ErrorConfig = #{level => error, formatter => {logger_formatter, #{single_line => true, max_size => 2048}}, config => #{type => {file, filename:join(LogDir, "error.log")}}}, [{handler, debug_rcl_test, logger_std_h, DebugConfig}, {handler, info_rcl_test, logger_std_h, InfoConfig}, {handler, notice_rcl_test, logger_std_h, NoticeConfig}, {handler, warning_rcl_test, logger_std_h, WarningConfig}, {handler, error_rcl_test, logger_std_h, ErrorConfig}]. to_ls(V) when is_binary(V) -> binary_to_list(V); to_ls(V) when is_list(V) -> lists:flatten(V).