amoc

%% @copyright 2023 Erlang Solutions Ltd.
%% @doc API module to run tests in a distributed environment
-module(amoc_dist).

-export([do/3,
         add/1,
         add/2,
         remove/2,
         remove/3,
         update_settings/1,
         update_settings/2,
         stop/0,
         get_state/0]).

-compile({no_auto_import, [ceil/1]}).

-type cluster_state() :: idle | running | stopped.

%% ------------------------------------------------------------------
%% API
%% ------------------------------------------------------------------

%% @doc Start a scenario with the given number of users and configuration
-spec do(amoc:scenario(), non_neg_integer(), amoc_config:settings()) ->
    {ok, any()} | {error, any()}.
do(Scenario, Count, Settings) ->
    case {prepare_cluster(Scenario, Settings), Count} of
        {{ok, _}, 0} -> {ok, no_users};
        {{ok, _}, Count} -> add(Count);
        {Error, _} -> Error
    end.

%% @doc Dynamically add more users to a currently running scenario, in all nodes
%% @see add/2
-spec add(pos_integer()) -> {ok, any()} | {error, any()}.
add(Count) ->
    add(Count, amoc_cluster:slave_nodes()).

%% @doc Dynamically add more users to a currently running scenario, in the specified nodes
-spec add(pos_integer(), [node()]) -> {ok, any()} | {error, any()}.
add(Count, Nodes) when is_integer(Count), Count > 0 ->
    case check_nodes(Nodes) of
        ok -> add_users(Count, Nodes);
        Error -> Error
    end.

%% @doc Dynamically remove more users from a currently running scenario,
%% optionally forcibly, in all nodes
%% @see remove/3
-spec remove(pos_integer(), boolean()) -> {ok, any()} | {error, any()}.
remove(Count, ForceRemove) ->
    remove(Count, ForceRemove, amoc_cluster:slave_nodes()).

%% @doc Dynamically remove more users from a currently running scenario,
%% optionally forcibly, in the specified nodes
%%
%% Forcing user removal means that all users will be signal to exit in parallel,
%% and will forcibly be killed after a short timeout (2 seconds),
%% whether they have exited already or not.
-spec remove(pos_integer(), boolean(), [node()]) -> {ok, any()} | {error, any()}.
remove(Count, ForceRemove, Nodes) when is_integer(Count), Count > 0 ->
    case check_nodes(Nodes) of
        ok -> remove_users(Count, ForceRemove, Nodes);
        Error -> Error
    end.

%% @doc Update the settings of a currently running scenario, in all nodes
%% @see update_settings/2
-spec update_settings(amoc_config:settings()) -> {ok, any()} | {error, any()}.
update_settings(Settings) ->
    Ret = update_settings(Settings, amoc_cluster:slave_nodes()),
    case Ret of
        {ok, _} -> set_param(settings, Settings);
        {error, _} -> ok
    end,
    Ret.

%% @doc Update the settings of a currently running scenario, in the specified nodes
-spec update_settings(amoc_config:settings(), [node()]) -> {ok, any()} | {error, any()}.
update_settings(Settings, Nodes) ->
    case check_nodes(Nodes) of
        ok -> update_settings_on_nodes(Settings, Nodes);
        Error -> Error
    end.

%% @doc Stop a running scenario in the whole cluster
-spec stop() -> {ok, any()} | {error, any()}.
stop() ->
    stop_cluster().

%% @doc Get the scenario running status in the cluster
-spec get_state() -> cluster_state().
get_state() ->
    case {amoc_cluster:master_node(), get_param(state)} of
        {undefined, undefined} -> idle;
        {_, {ok, State}} -> State;
        {Node, undefined} -> erpc:call(Node, ?MODULE, ?FUNCTION_NAME, [])
    end.

%% ------------------------------------------------------------------
%% Local functions
%% ------------------------------------------------------------------

-spec get_param(any()) -> {ok, any()} | undefined.
get_param(Key) ->
    try persistent_term:get({amoc_dist, Key}) of
        Value -> {ok, Value}
    catch
        error:badarg -> undefined
    end.

-spec set_param(any(), any()) -> ok.
set_param(Key, Value) ->
    persistent_term:put({amoc_dist, Key}, Value).

-spec set_state(cluster_state()) -> any().
set_state(State) ->
    set_param(state, State).

-spec check_nodes([node()]) -> ok | {error, any()}.
check_nodes(Nodes) ->
    SlaveNodes = amoc_cluster:slave_nodes(),
    MasterNode = amoc_cluster:master_node(),
    case {Nodes -- SlaveNodes, Nodes, node()} of
        {[], [_ | _], MasterNode} ->
            %% running on the master node with the proper non-empty Nodes list
            ok;
        {[], [], MasterNode} ->
            %% Nodes list is empty
            {error, empty_nodes_list};
        {BadNodes, _, MasterNode} ->
            %% non-slave nodes in the list
            {error, {bad_nodes, BadNodes}};
        {_, _, _} ->
            {error, not_a_master}
    end.

-spec prepare_cluster(amoc:scenario(), amoc_config:settings()) -> {ok, any()} | {error, any()}.
prepare_cluster(Scenario, Settings) ->
    case setup_master_node() of
        ok ->
            set_param(scenario, Scenario),
            set_param(settings, Settings),
            set_state(running),
            set_param(last_id, 0),
            setup_cluster();
        Error -> Error
    end.

-spec setup_master_node() -> ok | {error, any()}.
setup_master_node() ->
    case {amoc_cluster:set_master_node(node()), get_param(scenario)} of
        {ok, undefined} ->
            amoc_controller:disable();
        {ok, AnotherScenario} ->
            {error, {scenario_is_running, AnotherScenario}};
        {Error, _} -> Error
    end.

-spec setup_cluster() -> {ok, any()} | {error, any()}.
setup_cluster() ->
    case amoc_cluster:on_new_connection(fun setup_slave_node/1) of
        {ok, SlaveNodes} ->
            Result = [{Node, amoc_cluster:set_master_node(Node)} || Node <- SlaveNodes],
            maybe_error(Result);
        Error -> Error
    end.

-spec setup_slave_node(node()) -> ok | {error, any()} | {badrpc, any()}.
setup_slave_node(Node) ->
    case propagate_uploaded_modules(Node) of
        {ok, _} ->
            {ok, Scenario} = get_param(scenario),
            {ok, Settings} = get_param(settings),
            erpc:call(Node, amoc_controller, start_scenario, [Scenario, Settings]);
        Error -> Error
    end.

-spec propagate_uploaded_modules(node()) -> {ok, any()} | {error, any()}.
propagate_uploaded_modules(Node) ->
    Result = amoc_code_server:distribute_modules(Node),
    maybe_error(Result).

-spec add_users(pos_integer(), [node()]) -> {ok, any()} | {error, any()}.
add_users(Count, Nodes) ->
    {ok, LastId} = get_param(last_id),
    set_param(last_id, LastId + Count),
    Result = add_users([], LastId, Count, Nodes),
    maybe_error(Result).

-spec add_users([{node(), any()}], non_neg_integer(), non_neg_integer(), [node()]) ->
    [{node(), any()}].
add_users(Result, _, 0, []) -> Result;
add_users(Result, LastId, Count, [Node | T] = Nodes) ->
    case Count div length(Nodes) of
        0 ->
            add_users([{Node, {ok, node_skipped}} | Result], LastId, Count, T);
        N ->
            Ret = erpc:call(Node, amoc_controller, add_users, [LastId + 1, LastId + N]),
            add_users([{Node, Ret} | Result], LastId + N, Count - N, T)
    end.

-spec remove_users(pos_integer(), boolean(), [node()]) -> {ok, any()} | {error, any()}.
remove_users(Count, ForceRemove, Nodes) ->
    Result = remove_users([], Count, ForceRemove, Nodes),
    maybe_error(Result).

-spec remove_users([{node(), any()}], non_neg_integer(), boolean(), [node()]) ->
    [{node(), any()}].
remove_users(Result, 0, _, []) -> Result;
remove_users(Result, Count, ForceRemove, [Node | T] = Nodes) ->
    case Count div length(Nodes) of
        0 ->
            remove_users([{Node, {ok, node_skipped}} | Result], Count, ForceRemove, T);
        N ->
            Ret = erpc:call(Node, amoc_controller, remove_users, [N, ForceRemove]),
            remove_users([{Node, Ret} | Result], Count - N, ForceRemove, T)
    end.

-spec update_settings_on_nodes(amoc_config:settings(), [node()]) -> {ok, any()} | {error, any()}.
update_settings_on_nodes(Settings, Nodes) ->
    Result = [{Node, update_settings_on_node(Settings, Node)} || Node <- Nodes],
    maybe_error(Result).

-spec update_settings_on_node(amoc_config:settings(), node()) ->
          ok | {badrpc, any()} | {error, any()}.
update_settings_on_node(Settings, Node) ->
    erpc:call(Node, amoc_controller, update_settings, [Settings]).

-spec stop_cluster() -> {ok, any()} | {error, any()}.
stop_cluster() ->
    MasterNode = amoc_cluster:master_node(),
    case {node(), amoc_cluster:slave_nodes()} of
        {_, []} -> {error, no_slave_nodes};
        {MasterNode, Slaves} ->
            set_state(stopped),
            Result = [{Node, erpc:call(Node, amoc_controller, stop_scenario, [])} ||
                         Node <- Slaves],
            maybe_error(Result);
        {_, _} -> {error, not_a_master}
    end.

-spec maybe_error([{any(), any()}]) -> {ok, any()} | {error, any()}.
maybe_error(Result) ->
    ValuesAndErrors = lists:foldl(
        fun({Tag, {ok, Value}}, {OkValues, Errors}) ->
            {[{Tag, Value} | OkValues], Errors};
           ({Tag, ok}, {OkValues, Errors}) ->
               {[Tag | OkValues], Errors};
           ({Tag, {badrpc, Reason}}, {OkValues, Errors}) ->
               {OkValues, [{Tag, {badrpc, Reason}} | Errors]};
           ({Tag, {error, Reason}}, {OkValues, Errors}) ->
               {OkValues, [{Tag, Reason} | Errors]};
           ({Tag, InvalidRetValue}, {OkValues, Errors}) ->
               {OkValues, [{Tag, {invalid_ret_value, InvalidRetValue}} | Errors]}
        end, {[], []}, Result),
    case ValuesAndErrors of
        {OkValues, []} -> {ok, OkValues};
        {_, Errors} -> {error, Errors}
    end.