Current section

Files

Jump to
ra src ra.erl
Raw

src/ra.erl

%% This Source Code Form is subject to the terms of the Mozilla Public
%% License, v. 2.0. If a copy of the MPL was not distributed with this
%% file, You can obtain one at https://mozilla.org/MPL/2.0/.
%%
%% Copyright (c) 2017-2020 VMware, Inc. or its affiliates. All rights reserved.
%%
%% @doc The primary module for interacting with ra servers and clusters.
-module(ra).
-include("ra.hrl").
-export([
start/0,
start/1,
start_in/1,
%% command execution
process_command/2,
process_command/3,
pipeline_command/2,
pipeline_command/3,
pipeline_command/4,
%% queries
members/1,
members/2,
initial_members/1,
initial_members/2,
local_query/2,
local_query/3,
leader_query/2,
leader_query/3,
consistent_query/2,
consistent_query/3,
% cluster operations
start_cluster/1,
start_cluster/2,
start_cluster/3,
start_cluster/4,
start_or_restart_cluster/3,
start_or_restart_cluster/4,
delete_cluster/1,
delete_cluster/2,
% server management
start_server/1,
start_server/4,
restart_server/1,
restart_server/2,
stop_server/1,
force_delete_server/1,
trigger_election/1,
trigger_election/2,
%% membership changes
add_member/2,
add_member/3,
remove_member/2,
remove_member/3,
leave_and_terminate/1,
leave_and_terminate/2,
leave_and_terminate/3,
leave_and_delete_server/1,
leave_and_delete_server/2,
leave_and_delete_server/3,
%% troubleshooting
overview/0,
%% helpers
new_uid/1,
%% rebalancing
transfer_leadership/2,
aux_command/2,
cast_aux_command/2,
register_external_log_reader/1
]).
-define(START_TIMEOUT, ?DEFAULT_TIMEOUT).
-type ra_cmd_ret() :: ra_server_proc:ra_cmd_ret().
-type environment_param() ::
{data_dir, file:filename()} |
{wal_data_dir, file:filename()} |
{segment_max_entries, non_neg_integer()} |
{wal_max_size_bytes, non_neg_integer()} |
{wal_compute_checksums, boolean()} |
{wal_write_strategy, default | o_sync}.
-type query_fun() :: fun((term()) -> term()) |
{M :: module(), F :: atom(), A :: list()}.
%% export some internal types
-type index() :: ra_index().
-type server_id() :: ra_server_id().
-export_type([index/0,
server_id/0,
query_fun/0,
from/0]).
%% @doc Starts the ra application.
%% @end
-spec start() -> ok.
start() ->
{ok, _} = application:ensure_all_started(ra),
ok.
%% @doc Starts the ra application.
%% If the application is running it will be stopped and restarted.
%% @param DataDir: the data directory to run the application in.
%% @end
-spec start(Params :: [environment_param()]) ->
{ok, [Started]} | {error, term()} when Started :: term().
start(Params) when is_list(Params) ->
_ = application:stop(ra),
_ = application:load(ra),
[ok = application:set_env(ra, Param, Value)
|| {Param, Value} <- Params],
application:ensure_all_started(ra).
%% @doc Starts the ra application with a provided data directory.
%% The same as ra:start([{data_dir, dir}])
%% If the application is running it will be stopped and restarted.
%% @param DataDir: the data directory to run the application in.
%% @end
-spec start_in(DataDir :: file:filename()) ->
{ok, [Started]} | {error, term()}
when Started :: term().
start_in(DataDir) ->
start([{data_dir, DataDir}]).
%% @doc Restarts a previously successfully started ra server
%% @param ServerId the ra_server_id() of the server
%% @returns `{ok | error, Error}' where error can be
%% `not_found' or `name_not_registered' when the ra server has never before
%% been started on the Erlang node.
%% @end
-spec restart_server(ra_server_id()) -> ok | {error, term()}.
restart_server(ServerId) ->
% don't match on return value in case it is already running
case catch ra_server_sup_sup:restart_server(ServerId, #{}) of
{ok, _} -> ok;
{ok, _, _} -> ok;
{error, _} = Err -> Err;
{'EXIT', Err} -> {error, Err}
end.
%% @doc Restarts a previously successfully started ra server
%% @param ServerId the ra_server_id() of the server
%% @param AddConfig additional config parameters to be merged into the
%% original config.
%% @returns `{ok | error, Error}' where error can be
%% `not_found' or `name_not_registered' when the ra server has never before
%% been started on the Erlang node.
%% @end
-spec restart_server(ra_server_id(), ra_server:mutable_config()) ->
ok | {error, term()}.
restart_server(ServerId, AddConfig) ->
% don't match on return value in case it is already running
case catch ra_server_sup_sup:restart_server(ServerId, AddConfig) of
{ok, _} -> ok;
{ok, _, _} -> ok;
{error, _} = Err -> Err;
{'EXIT', Err} -> {error, Err}
end.
%% @doc Stops a ra server
%% @param ServerId the ra_server_id() of the server
%% @returns `{ok | error, nodedown}'
%% @end
-spec stop_server(ra_server_id()) -> ok | {error, nodedown}.
stop_server(ServerId) ->
try ra_server_sup_sup:stop_server(ServerId) of
ok -> ok;
{error, not_found} -> ok;
{error, {badrpc, nodedown}} ->
{error, nodedown}
catch
exit:noproc -> ok;
exit:{{nodedown, _}, _} ->
{error, nodedown}
end.
%% @doc Deletes a ra server
%% The server is forcefully deleted.
%% @param ServerId the ra_server_id() of the server
%% @returns `ok | {error, nodedown} | {badrpc, Reason}'
%% @end
-spec force_delete_server(ServerId :: ra_server_id()) -> ok | {error, term()} | {badrpc, term()}.
force_delete_server(ServerId) ->
ra_server_sup_sup:delete_server(ServerId).
%% @doc Starts or restarts a ra cluster.
%%
%%
%% @param ClusterName the name of the cluster.
%% @param Machine The {@link ra_machine:machine/0} configuration.
%% @param ServerIds The list of ra server ids.
%% @returns
%% `{ok, Started, NotStarted}' if a cluster could be successfully
%% started. A cluster can be successfully started if more than half of the
%% servers provided could be started. Servers that could not be started need to
%% be retried periodically using {@link start_server/1}
%%
%% `{error, cluster_not_formed}' if a cluster could not be started.
%%
%% If there was no existing cluster and a new cluster could not be formed
%% any servers that did manage to start are
%% forcefully deleted.
%% @end
-spec start_or_restart_cluster(ra_cluster_name(), ra_server:machine_conf(),
[ra_server_id()]) ->
{ok, [ra_server_id()], [ra_server_id()]} |
{error, cluster_not_formed}.
start_or_restart_cluster(ClusterName, Machine, ServerIds) ->
start_or_restart_cluster(ClusterName, Machine, ServerIds, ?START_TIMEOUT).
%% @doc Same as `start_or_restart_cluster/3' but accepts a custom timeout.
%% @param ClusterName the name of the cluster.
%% @param Machine The {@link ra_machine:machine/0} configuration.
%% @param ServerIds The list of ra server ids.
%% @param Timeout The time to wait for any server to restart or start
%% @returns
%% `{ok, Started, NotStarted}' if a cluster could be successfully
%% started. A cluster can be successfully started if more than half of the
%% servers provided could be started. Servers that could not be started need to
%% be retried periodically using {@link start_server/1}
%%
%% `{error, cluster_not_formed}' if a cluster could not be started.
%%
%% If there was no existing cluster and a new cluster could not be formed
%% any servers that did manage to start are
%% forcefully deleted.
%% @see start_or_restart_cluster/3
%% @end
-spec start_or_restart_cluster(ra_cluster_name(), ra_server:machine_conf(),
[ra_server_id()], non_neg_integer()) ->
{ok, [ra_server_id()], [ra_server_id()]} |
{error, cluster_not_formed}.
start_or_restart_cluster(ClusterName, Machine,
[FirstServer | RemServers] = ServerIds, Timeout) ->
case ra_server_sup_sup:restart_server(FirstServer, #{}) of
{ok, _} ->
%% restart the rest of the servers
_ = [{ok, _} = ra_server_sup_sup:restart_server(N, #{})
|| N <- RemServers],
{ok, ServerIds, []};
{error, Err} ->
?ERR("start_or_restart_cluster: got an error: ~p~n", [Err]),
start_cluster(ClusterName, Machine, ServerIds, Timeout)
end.
%% @doc Starts a new distributed ra cluster.
%%
%% @param ClusterName the name of the cluster.
%% @param Machine The {@link ra_machine:machine/0} configuration.
%% @param ServerIds The list of ra server ids.
%% @returns
%% `{ok, Started, NotStarted}' if a cluster could be successfully
%% started. A cluster can be successfully started if more than half of the
%% servers provided could be started. Servers that could not be started need to
%% be retried periodically using {@link start_server/1}
%%
%% `{error, cluster_not_formed}' if a cluster could not be started.
%%
%% If a cluster could not be formed any servers that did manage to start are
%% forcefully deleted.
%% @end
-spec start_cluster(ra_cluster_name(),
ra_server:machine_conf(),
[ra_server_id()]) ->
{ok, [ra_server_id()], [ra_server_id()]} |
{error, cluster_not_formed}.
start_cluster(ClusterName, Machine, ServerIds) ->
start_cluster(ClusterName, Machine, ServerIds, ?START_TIMEOUT).
%% @doc Starts a new distributed ra cluster.
%%
%% @param ClusterName the name of the cluster.
%% @param Machine The {@link ra_machine:machine/0} configuration.
%% @param ServerIds The list of ra server ids.
%% @param Timeout The time to wait for each server to start
%% @returns
%% `{ok, Started, NotStarted}' if a cluster could be successfully
%% started. A cluster can be successfully started if more than half of the
%% servers provided could be started. Servers that could not be started need to
%% be retried periodically using {@link start_server/1}
%%
%% `{error, cluster_not_formed}' if a cluster could not be started.
%%
%% If a cluster could not be formed any servers that did manage to start are
%% forcefully deleted.
%% @end
-spec start_cluster(ra_cluster_name(),
ra_server:machine_conf(),
[ra_server_id()],
non_neg_integer()) ->
{ok, [ra_server_id()], [ra_server_id()]} |
{error, cluster_not_formed}.
start_cluster(ClusterName, Machine, ServerIds, Timeout) ->
Configs = [begin
UId = new_uid(ra_lib:to_binary(ClusterName)),
#{id => Id,
uid => UId,
cluster_name => ClusterName,
log_init_args => #{uid => UId},
initial_members => ServerIds,
machine => Machine}
end || Id <- ServerIds],
start_cluster(Configs, Timeout).
%% @doc Starts a new distributed ra cluster.
%%
%% @param ServerConfigs a list of initial server configurations
%% @returns
%% `{ok, Started, NotStarted}' if a cluster could be successfully
%% started. A cluster can be successfully started if more than half of the
%% servers provided could be started. Servers that could not be started need to
%% be retried periodically using {@link start_server/1}
%%
%% `{error, cluster_not_formed}' if a cluster could not be started.
%%
%% If a cluster could not be formed any servers that did manage to start are
%% forcefully deleted.
%% @end
-spec start_cluster([ra_server:ra_server_config()]) ->
{ok, [ra_server_id()], [ra_server_id()]} |
{error, cluster_not_formed}.
start_cluster(ServerConfigs) ->
start_cluster(ServerConfigs, ?START_TIMEOUT).
%% @doc Same as `start_cluster/1' but accepts a custom timeout.
%% @param ServerConfigs a list of initial server configurations
%% @param Timeout the timeout to use
%% @end
-spec start_cluster([ra_server:ra_server_config()], non_neg_integer()) ->
{ok, [ra_server_id()], [ra_server_id()]} |
{error, cluster_not_formed}.
start_cluster([#{cluster_name := ClusterName} | _] =
ServerConfigs, Timeout) ->
{Started, NotStarted} =
ra_lib:partition_parallel(
fun (C) ->
case start_server(C) of
ok -> true;
Err ->
?ERR("ra: failed to start a server ~w, error: ~p~n",
[C, Err]),
false
end
end, ServerConfigs),
case Started of
[] ->
?ERR("ra: failed to form a new cluster ~w.~n "
"No servers were succesfully started.~n",
[ClusterName]),
{error, cluster_not_formed};
_ ->
StartedIds = [I || #{id := I} <- Started],
NotStartedIds = [I || #{id := I} <- NotStarted],
%% try triggering elections until one succeeds
_ = lists:any(fun (N) -> ok == trigger_election(N) end,
sort_by_local(StartedIds, [])),
%% TODO: handle case where no election was successfully triggered
case members(hd(StartedIds),
length(ServerConfigs) * Timeout) of
{ok, _, Leader} ->
?INFO("ra: started cluster ~s with ~b servers~n"
"~b servers failed to start: ~w~n"
"Leader: ~w", [ClusterName, length(ServerConfigs),
length(NotStarted), NotStartedIds,
Leader]),
% we have a functioning cluster
{ok, StartedIds, NotStartedIds};
Err ->
?WARN("ra: failed to form new cluster ~w.~n "
"Error: ~w~n", [ClusterName, Err]),
_ = [force_delete_server(N) || N <- StartedIds],
% we do not have a functioning cluster
{error, cluster_not_formed}
end
end.
%% @doc Starts a new distributed ra cluster.
%% @param ClusterName the name of the cluster.
%% @param ServerId the ra_server_id() of the server
%% @param Machine The {@link ra_machine:machine/0} configuration.
%% @param ServerConfigs a list of initial server configurations
%% @returns
%% `{ok, Started, NotStarted}' if a cluster could be successfully
%% started. A cluster can be successfully started if more than half of the
%% servers provided could be started. Servers that could not be started need to
%% be retried periodically using {@link start_server/1}
%%
%% `{error, cluster_not_formed}' if a cluster could not be started.
%%
%% If a cluster could not be formed any servers that did manage to start are
%% forcefully deleted.
%% @see start_server/1
%% @end
-spec start_server(ra_cluster_name(), ra_server_id(),
ra_server:machine_conf(), [ra_server_id()]) ->
ok | {error, term()}.
start_server(ClusterName, ServerId, Machine, ServerIds) ->
UId = new_uid(ra_lib:to_binary(ClusterName)),
Conf = #{cluster_name => ClusterName,
id => ServerId,
uid => UId,
initial_members => ServerIds,
log_init_args => #{uid => UId},
machine => Machine},
start_server(Conf).
%% @doc Starts a ra server
%% @param Conf a ra_server_config() configuration map.
%% @returns `{ok | error, Error}'
%% @end
-spec start_server(ra_server:ra_server_config()) -> ok | {error, term()}.
start_server(Conf) ->
%% validate UID is safe
case ra_lib:validate_base64uri(maps:get(uid, Conf)) of
true ->
% don't match on return value in case it is already running
case catch ra_server_sup_sup:start_server(Conf) of
{ok, _} -> ok;
{ok, _, _} -> ok;
{error, _} = Err -> Err;
{'EXIT', Err} -> {error, Err};
{badrpc, Reason} -> {error, Reason}
end;
false ->
{error, invalid_uid}
end.
%% @doc Deletes a ra cluster in an orderly fashion.
%% This function commits an end of life command which after each server applies
%% it will cause that server to shut down and delete all its data.
%% The leader will stay up until it has successfully replicated the end of life
%% command to all servers after which it too will shut down and delete all of its
%% data.
%% @param ServerIds the ra_server_ids of the cluster
%% @returns `{{ok, Leader} | error, nodedown}'
%% @end
-spec delete_cluster(ServerIds :: [ra_server_id()]) ->
{ok, ra_server_id()} | {error, term()}.
delete_cluster(ServerIds) ->
delete_cluster(ServerIds, ?DEFAULT_TIMEOUT).
%% @doc Same as `delete_cluster/1' but also accepts a timeout.
%% @see delete_cluster/1
%% @end
-spec delete_cluster(ServerIds :: [ra_server_id()], timeout()) ->
{ok, Leader :: ra_server_id()} | {error, term()}.
delete_cluster(ServerIds, Timeout) ->
DeleteCmd = {'$ra_cluster', delete, await_consensus},
case ra_server_proc:command(ServerIds, DeleteCmd, Timeout) of
{ok, _, Leader} ->
{ok, Leader};
{timeout, _} ->
{error, timeout};
Err ->
Err
end.
%% @doc Add a ra server id to a ra cluster's membership configuration.
%% This commits a join command to the leader log. After this has been replicated
%% the leader will start replicating entries to the new server.
%% This function returns after appending the command to the log.
%% New node does not have to be running at the time of addition but it is expected
%% to come online in a reasonable amount of time. A new member that's been
%% announced to its new cluster but taking a long time to start will
%% affect said cluster's availability characteristics (by increasing quorum node count).
%%
%% @param ServerLoc the ra server or servers to try to send the command to
%% @param ServerId the ra server id of the new server.
%% @end
-spec add_member(ra_server_id() | [ra_server_id()], ra_server_id()) ->
ra_cmd_ret() |
{error, already_member} |
{error, cluster_change_not_permitted}.
add_member(ServerLoc, ServerId) ->
add_member(ServerLoc, ServerId, ?DEFAULT_TIMEOUT).
%% @doc Same as `add_member/2' but also accepts a timeout.
%% @see add_member/2
%% @end
-spec add_member(ra_server_id() | [ra_server_id()],
ra_server_id(), timeout()) ->
ra_cmd_ret() |
{error, already_member} |
{error, cluster_change_not_permitted}.
add_member(ServerLoc, ServerId, Timeout) ->
ra_server_proc:command(ServerLoc,
{'$ra_join', ServerId, after_log_append},
Timeout).
%% @doc Removes a server from the cluster's membership configuration.
%% This function returns after appending a cluster membership change
%% command to the log.
%%
%% After a server is removed from its cluster and the membership change is replicated,
%% it would reject any commands it's asked to process.
%%
%% @param ServerRef the ra server to send the command to
%% @param ServerId the ra server id of the server to remove
%% @see leave_and_terminate/2
%% @see leave_and_delete_server/2
%% @see add_member/2
%% @see remove_member/3
%% @end
-spec remove_member(ra_server_id() | [ra_server_id()], ra_server_id()) ->
ra_cmd_ret() |
{error, not_member} |
{error, cluster_change_not_permitted}.
remove_member(ServerRef, ServerId) ->
remove_member(ServerRef, ServerId, ?DEFAULT_TIMEOUT).
%% @doc Same as `remove_member/2' but also accepts a timeout.
%% @see remove_member/2
%% @end
-spec remove_member(ra_server_id() | [ra_server_id()],
ra_server_id(), timeout()) ->
ra_cmd_ret() |
{error, not_member} |
{error, cluster_change_not_permitted}.
remove_member(ServerRef, ServerId, Timeout) ->
ra_server_proc:command(ServerRef,
{'$ra_leave', ServerId, after_log_append},
Timeout).
%% @doc Makes the server to enter a pre-vote state and attempt to become the leader.
%% It is necessary to call this function when starting a new cluster as a
%% brand new Ra server (node) will not automatically enter the pre-vote state.
%% This does not apply to recovering (previously started) servers: they will
%% enter the pre-vote state and proceed to participate in an election on boot.
%%
%% @param ServerId the ra server id of the server to trigger the election on.
%% @end
-spec trigger_election(ra_server_id()) -> ok.
trigger_election(ServerId) ->
trigger_election(ServerId, ?DEFAULT_TIMEOUT).
%% @doc Same as `trigger_election/2' but also accepts a timeout.
%% @end
-spec trigger_election(ra_server_id(), timeout()) -> ok.
trigger_election(ServerId, Timeout) ->
ra_server_proc:trigger_election(ServerId, Timeout).
%% @doc A safe way for a an active server to remove itself from its cluster.
%% The server is first removed, then asked to terminate.
%% Use this when a node knows that it is being permanently decommissioned.
%% @param ServerId the ra server to send the command to and to remove
%% @see remove_member/2
%% @end
leave_and_terminate(ServerId) ->
leave_and_terminate(ServerId, ServerId).
%% @doc A safe way to remove an active server from its cluster.
%% The command is added to the log by the `ServerRef' node.
%% Use this to decommission a node that's unable to start
%% or is permanently lost.
%% @param ServerRef the ra server to send the command to and to remove
%% @param ServerId the ra server to remove
%% @see leave_and_terminate/3
%% @end
-spec leave_and_terminate(ra_server_id() | [ra_server_id()], ra_server_id()) ->
ok | timeout | {error, noproc}.
leave_and_terminate(ServerRef, ServerId) ->
leave_and_terminate(ServerRef, ServerId, ?DEFAULT_TIMEOUT).
%% @doc Same as `leave_and_terminate/2' but also accepts a timeout.
%% @param ServerRef the ra server to send the command to and to remove
%% @param ServerId the ra server to remove
%% @param Timeout timeout to use
%% @see leave_and_terminate/2
%% @end
-spec leave_and_terminate(ra_server_id() | [ra_server_id()],
ra_server_id(), timeout()) ->
ok | timeout | {error, noproc}.
leave_and_terminate(ServerRef, ServerId, Timeout) ->
LeaveCmd = {'$ra_leave', ServerId, await_consensus},
case ra_server_proc:command(ServerRef, LeaveCmd, Timeout) of
{timeout, Who} ->
?ERR("Failed to leave the cluster: request to ~w timed out", [Who]),
timeout;
{error, noproc} = Err ->
Err;
{ok, _, _} ->
?INFO("We (Ra node ~w) has successfully left the cluster. Terminating.", [ServerId]),
stop_server(ServerId)
end.
%% @doc A safe way for a an active server to force remove itself from its cluster.
%% The server will be force removed after a membership transition command was
%% added to the log.
%% Use this when a node knows that it is being permanently decommissioned.
%% @param ServerRef the ra server to send the command to and to force remove
%% @see leave_and_terminate/2
%% @see leave_and_delete_server/3
%% @end
leave_and_delete_server(ServerId) ->
leave_and_delete_server(ServerId, ServerId).
%% @doc A safe way to remove an active server from its cluster.
%% The server will be force removed after a membership transition command was
%% added to the log.
%% The command is added to the log by the `ServerRef' node.
%% Use this to decommission a node that's unable to start
%% or is permanently lost.
%% @param ServerRef the ra server to send the command to and to remove
%% @param ServerId the ra server to force remove
%% @see leave_and_delete_server/3
%% @end
-spec leave_and_delete_server(ra_server_id() | [ra_server_id()],
ra_server_id()) ->
ok | timeout | {error, noproc}.
leave_and_delete_server(ServerRef, ServerId) ->
leave_and_delete_server(ServerRef, ServerId, ?DEFAULT_TIMEOUT).
%% @doc Same as `leave_and_delete_server/2' but also accepts a timeout.
%% @param ServerRef the ra server to send the command to and to remove
%% @param ServerId the ra server to force remove
%% @param Timeout timeout to use
%% @see leave_and_delete_server/2
%% @end
-spec leave_and_delete_server(ra_server_id() | [ra_server_id()],
ra_server_id(), timeout()) ->
ok | timeout | {error, noproc}.
leave_and_delete_server(ServerRef, ServerId, Timeout) ->
LeaveCmd = {'$ra_leave', ServerId, await_consensus},
case ra_server_proc:command(ServerRef, LeaveCmd, Timeout) of
{timeout, Who} ->
?ERR("Failed to leave the cluster: request to ~w timed out", [Who]),
timeout;
{error, _} = Err ->
Err;
{ok, _, _} ->
?INFO("Ra node ~w has successfully left the cluster.", [ServerId]),
force_delete_server(ServerId)
end.
%% @doc generates a random uid using the provided source material for the first
%% 6 characters.
%% @end
new_uid(Source) when is_binary(Source) ->
Prefix = ra_lib:derive_safe_string(ra_lib:to_binary(Source), 6),
ra_lib:make_uid(string:uppercase(Prefix)).
%% @doc Returns a map of overview data of the Ra system on the current Erlang
%% node.
%% @end
-spec overview() -> map().
overview() ->
#{node => node(),
servers => ra_directory:overview(),
counters => ra_counters:overview(),
wal => #{status => lists:nth(5, element(4, sys:get_status(ra_log_wal))),
open_mem_tables => ets:info(ra_log_open_mem_tables, size),
closed_mem_tables => ets:info(ra_log_closed_mem_tables, size)},
segment_writer => ra_log_segment_writer:overview()
}.
%% @doc Submits a command to a ra server. Returns after the command has
%% been applied to the Raft state machine. If the state machine returned a
%% response it is included in the second element of the response tuple.
%% If no response was returned the second element is the atom `noreply'.
%% If the server receiving the command isn't the current leader it will
%% redirect the call to the leader (if known) or hold on to the command until
%% a leader is known. The leader's server id is returned as the 3rd element
%% of the success reply tuple.
%%
%% If there is no majority of Ra servers online, this function will return
%% a timeout.
%%
%% @param ServerId the server id to send the command to
%% @param Command an arbitrary term that the state machine can handle
%% @param Timeout the time to wait before returning {timeout, ServerId}
%% @end
-spec process_command(ServerId :: ra_server_id() | [ra_server_id()],
Command :: term(),
Timeout :: timeout()) ->
{ok, Reply :: term(), Leader :: ra_server_id()} |
{error, term()} |
{timeout, ra_server_id()}.
process_command(ServerId, Cmd, Timeout) ->
ra_server_proc:command(ServerId, usr(Cmd, await_consensus), Timeout).
%% @doc Same as `process_command/3' with the default timeout of 5000 ms.
%% @param ServerId the server id to send the command to
%% @param Command an arbitrary term that the state machine can handle
%% @end
-spec process_command(ServerId :: ra_server_id() | [ra_server_id()],
Command :: term()) ->
{ok, Reply :: term(), Leader :: ra_server_id()} |
{error, term()} |
{timeout, ra_server_id()}.
process_command(ServerId, Command) ->
process_command(ServerId, Command, ?DEFAULT_TIMEOUT).
%% @doc Submits a command to the ra server using a gen_statem:cast, passing
%% an optional process-scoped term as correlation identifier.
%% A correlation id can be included
%% to implement reliable async interactions with the ra system. The calling
%% process can retain a map of commands that have not yet been applied to the
%% state machine successfully and resend them if a notification is not received
%% withing some time window.
%% When the submitted command(s) is applied to the state machine, the ra server will send
%% the calling process a ra_event of the following structure:
%%
%% `{ra_event, CurrentLeader, {applied, [{Correlation, Reply}]}}'
%%
%% Ra will batch notification and thus return a list of correlation
%% and result tuples.
%%
%% If the receiving ra server is not the cluster leader, a ra event of the following
%% structure will be returned informing the caller that it cannot process the
%% message. The message will include the current cluster leader, if one is known:
%%
%% `{ra_event, FromId, {rejected, {not_leader, Leader | undefined, Correlation}}}'
%%
%% The caller must then redirect the command for the correlation identifier to
%% the correct ra server: the leader.
%%
%% If instead the atom `no_correlation' is passed for the correlation argument,
%% the calling process will not receive any notification of command processing
%% success or otherwise.
%%
%% This is the least reliable way to interact with a ra system ("fire and forget")
%% and should only be used if the command is of little importance to the application.
%%
%% @param ServerId the ra server id to send the command to
%% @param Command an arbitrary term that the state machine can handle
%% @param Correlation a correlation identifier to be included to receive an
%% async notification after the command is applied to the state machine. If the
%% Correlation is set to `no_correlation' then no notifications will be sent.
%% @param Priority command priority. `low' priority commands will be held back
%% and appended to the Raft log in batches. NB: A `normal' priority command sent
%% from the same process can overtake a low priority command that was
%% sent before. There is no high priority.
%% Only use priority level of `low' with commands that
%% do not rely on total execution ordering.
%% @end
-spec pipeline_command(ServerId :: ra_server_id(), Command :: term(),
Correlation :: ra_server:command_correlation() |
no_correlation,
Priority :: normal | low) ->
ok.
pipeline_command(ServerId, Command, Correlation, Priority)
when Correlation /= no_correlation ->
Cmd = usr(Command, {notify, Correlation, self()}),
ra_server_proc:cast_command(ServerId, Priority, Cmd);
pipeline_command(ServerId, Command, no_correlation, Priority) ->
Cmd = usr(Command, noreply),
ra_server_proc:cast_command(ServerId, Priority, Cmd).
%% @doc Same as `pipeline_command/4' but uses a hardcoded priority of `low'.
%% @param ServerId the ra server id to send the command to
%% @param Command an arbitrary term that the state machine can handle
%% @param Correlation a correlation identifier to be included to receive an
%% async notification after the command is applied to the state machine.
%% @see pipeline_command/4
%% @end
-spec pipeline_command(ServerId :: ra_server_id(), Command :: term(),
Correlation :: ra_server:command_correlation() |
no_correlation) ->
ok.
pipeline_command(ServerId, Command, Correlation) ->
pipeline_command(ServerId, Command, Correlation, low).
%% @doc Sends a command to the ra server using a gen_statem:cast without
%% any correlation identifier.
%% Effectively the same as
%% `ra:pipeline_command(ServerId, Command, low, no_correlation)'
%% This is the least reliable way to interact with a ra system ("fire and forget")
%% and should only be used for commands that are of little importance
%% and/or where waiting for a response is prohibitively slow.
%% @param ServerId the ra server id to send the command to
%% @param Command an arbitrary term that the state machine can handle
%% @see pipeline_command/4
%% @end
-spec pipeline_command(ServerId :: ra_server_id(),
Command :: term()) -> ok.
pipeline_command(ServerId, Command) ->
pipeline_command(ServerId, Command, no_correlation, low).
%% @doc Query the machine state on any available server.
%% This allows you to run the QueryFun over the server machine state and
%% return the result. Any ra server can be addressed and will returns its local
%% state at the time of querying.
%% This can return stale results, including infinitely stale ones.
%% @param ServerId the ra server id to send the query to
%% @param QueryFun the query function to run
%% @end
-spec local_query(ServerId :: ra_server_id(),
QueryFun :: query_fun()) ->
ra_server_proc:ra_leader_call_ret({ra_idxterm(), Reply :: term()}).
local_query(ServerId, QueryFun) ->
local_query(ServerId, QueryFun, ?DEFAULT_TIMEOUT).
%% @doc Same as `local_query/2' but accepts a custom timeout.
%% @param ServerId the ra server id to send the query to
%% @param QueryFun the query function to run
%% @param Timeout the timeout to use
%% @see local_query/2
%% @end
-spec local_query(ServerId :: ra_server_id(),
QueryFun :: query_fun(),
Timeout :: timeout()) ->
ra_server_proc:ra_leader_call_ret({ra_idxterm(), Reply :: term()}) |
{ok, {ra_idxterm(), Reply :: term()}, not_known}.
local_query(ServerId, QueryFun, Timeout) ->
ra_server_proc:query(ServerId, QueryFun, local, Timeout).
%% @doc Query the machine state on the current leader node.
%% This function works like local_query, but redirects to the current
%% leader node.
%% The leader state may be more up-to-date compared to local state of some followers.
%% This function may still return stale results as it reads the current state
%% and does not wait for commands to be applied.
%% @param ServerId the ra server id(s) to send the query to
%% @param QueryFun the query function to run
%% @end
-spec leader_query(ServerId :: ra_server_id() | [ra_server_id()],
QueryFun :: query_fun()) ->
ra_server_proc:ra_leader_call_ret({ra_idxterm(), Reply :: term()}) |
{ok, {ra_idxterm(), Reply :: term()}, not_known}.
leader_query(ServerId, QueryFun) ->
leader_query(ServerId, QueryFun, ?DEFAULT_TIMEOUT).
%% @doc Same as `leader_query/2' but accepts a custom timeout.
%% @param ServerId the ra server id(s) to send the query to
%% @param QueryFun the query function to run
%% @param Timeout the timeout to use
%% @see leader_query/2
%% @end
-spec leader_query(ServerId :: ra_server_id() | [ra_server_id()],
QueryFun :: query_fun(),
Timeout :: timeout()) ->
ra_server_proc:ra_leader_call_ret({ra_idxterm(), Reply :: term()}) |
{ok, {ra_idxterm(), Reply :: term()}, not_known}.
leader_query(ServerId, QueryFun, Timeout) ->
ra_server_proc:query(ServerId, QueryFun, leader, Timeout).
%% @doc Query the state machine with a consistency guarantee.
%% This allows the caller to query the state machine on the leader node with
%% an additional heartbeat to check that the node is still the leader.
%% Consistency guarantee is that the query will return result containing
%% at least all changes, committed before this query is issued.
%% This may include changes which were committed while the query is running.
%% @param ServerId the ra server id(s) to send the query to
%% @param QueryFun the query function to run
%% @end
-spec consistent_query(ServerId :: ra_server_id() | [ra_server_id()],
QueryFun :: query_fun()) ->
ra_server_proc:ra_leader_call_ret(Reply :: term()).
consistent_query(ServerId, QueryFun) ->
consistent_query(ServerId, QueryFun, ?DEFAULT_TIMEOUT).
%% @doc Same as `consistent_query/2' but accepts a custom timeout.
%% @param ServerId the ra server id(s) to send the query to
%% @param QueryFun the query function to run
%% @param Timeout the timeout to use
%% @see consistent_query/2
%% @end
-spec consistent_query(ServerId :: ra_server_id() | [ra_server_id()],
QueryFun :: query_fun(),
Timeout :: timeout()) ->
ra_server_proc:ra_leader_call_ret(Reply :: term()).
consistent_query(ServerId, QueryFun, Timeout) ->
ra_server_proc:query(ServerId, QueryFun, consistent, Timeout).
%% @doc Returns a list of cluster members
%% @param ServerId the Ra server(s) to send the query to
%% @end
-spec members(ra_server_id() | [ra_server_id()]) ->
ra_server_proc:ra_leader_call_ret([ra_server_id()]).
members(ServerId) ->
members(ServerId, ?DEFAULT_TIMEOUT).
%% @doc Returns a list of cluster members
%% @param ServerId the Ra server(s) to send the query to
%% @param Timeout the timeout to use
%% @end
-spec members(ra_server_id() | [ra_server_id()], timeout()) ->
ra_server_proc:ra_leader_call_ret([ra_server_id()]).
members(ServerId, Timeout) ->
ra_server_proc:state_query(ServerId, members, Timeout).
-spec initial_members(ra_server_id() | [ra_server_id()]) ->
ra_server_proc:ra_leader_call_ret([ra_server_id()] | error).
initial_members(ServerId) ->
initial_members(ServerId, ?DEFAULT_TIMEOUT).
-spec initial_members(ra_server_id() | [ra_server_id()], timeout()) ->
ra_server_proc:ra_leader_call_ret([ra_server_id()] | error).
initial_members(ServerId, Timeout) ->
ra_server_proc:state_query(ServerId, initial_members, Timeout).
%% @doc Transfers leadership from the leader to a follower.
%% Returns `already_leader' if the transfer targer is already the leader.
%% @end
-spec transfer_leadership(ra_server_id(), ra_server_id()) ->
ok | already_leader | {error, term()} | {timeout, ra_server_id()}.
transfer_leadership(ServerId, TargetServerId) ->
ra_server_proc:transfer_leadership(ServerId, TargetServerId, ?DEFAULT_TIMEOUT).
-spec aux_command(ra_server_id(), term()) -> term().
aux_command(ServerRef, Cmd) ->
gen_statem:call(ServerRef, {aux_command, Cmd}).
-spec cast_aux_command(ra_server_id(), term()) -> ok.
cast_aux_command(ServerRef, Cmd) ->
gen_statem:cast(ServerRef, {aux_command, Cmd}).
%% @doc Registers an external log reader. ServerId needs to be local to the node.
%% Returns an initiated ra_log_reader:state() state.
%% @end
-spec register_external_log_reader(ra_server_id()) ->
ra_log_reader:state().
register_external_log_reader({_, Node} = ServerId)
when Node =:= node() ->
{ok, UId, Idx, SegRefs} = gen_statem:call(ServerId,
{register_external_log_reader, self()}),
ra_log_reader:init(UId, Idx, 1, SegRefs).
%% internal
usr(Data, Mode) ->
{'$usr', Data, Mode}.
sort_by_local([], Acc) ->
Acc;
sort_by_local([{_, N} = X | Rem], Acc) when N =:= node() ->
[X | Acc] ++ Rem;
sort_by_local([X | Rem], Acc) ->
sort_by_local(Rem, [X | Acc]).