%% ========================================================================================================== %% Ram - An ephemeral distributed KV store for Erlang and Elixir. %% %% The MIT License (MIT) %% %% Copyright (c) 2021-2022 Roberto Ostinelli . %% %% Permission is hereby granted, free of charge, to any person obtaining a copy %% of this software and associated documentation files (the "Software"), to deal %% in the Software without restriction, including without limitation the rights %% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell %% copies of the Software, and to permit persons to whom the Software is %% furnished to do so, subject to the following conditions: %% %% The above copyright notice and this permission notice shall be included in %% all copies or substantial portions of the Software. %% %% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR %% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, %% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE %% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER %% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, %% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN %% THE SOFTWARE. %% ========================================================================================================== %% @private -module(ram_backbone). %% API -export([start_cluster/1, stop_cluster/1]). -export([add_node/2, remove_node/2, nodes/0]). -export([get_server_loc/0]). %% tests -ifdef(TEST). -export([get_node_from_server_id/1]). -endif. %% macros -define(SYSTEM, default). -define(CLUSTER_NAME, ram). -define(RA_MACHINE, {module, ram_kv, #{}}). %% =================================================================== %% API %% =================================================================== -spec start_cluster([node()]) -> ok | {error, Reason :: term()}. start_cluster(Nodes) -> %% init ServerIds = [make_server_id(Node) || Node <- Nodes], %% start ra lists:foreach(fun(Node) -> ok = rpc:call(Node, ra, start, []) end, Nodes), %% start cluster case ra:start_cluster(?SYSTEM, ?CLUSTER_NAME, ?RA_MACHINE, ServerIds) of {ok, StartedIds, _NotStartedIds} when length(StartedIds) =:= length(Nodes) -> error_logger:info_msg("RAM[~s] Cluster started on ~p", [node(), get_nodes_from_server_ids(ServerIds)]), ok; {ok, StartedIds, NotStartedIds} -> error_logger:warning_msg("RAM[~s] Cluster started on ~p but but not on ~p", [node(), get_nodes_from_server_ids(StartedIds), get_nodes_from_server_ids(NotStartedIds)]), ok; {error, Reason} -> error_logger:error_msg("RAM[~s] Could not start cluster on ~p: ~p", [node(), get_nodes_from_server_ids(ServerIds), Reason]), {error, Reason} end. %% @doc Stops the Ram cluster. -spec stop_cluster([node()]) -> ok | {error, Reason :: term()}. stop_cluster(Nodes) -> %% init ServerIds = [make_server_id(Node) || Node <- Nodes], case ra:delete_cluster(ServerIds) of {ok, _ServerLoc} -> error_logger:info_msg("RAM[~s] Cluster stopped on ~p", [node(), get_nodes_from_server_ids(ServerIds)]), ok; {error, Reason} -> error_logger:error_msg("RAM[~s] Could not stop cluster on ~p: ~p", [node(), get_nodes_from_server_ids(ServerIds), Reason]), {error, Reason} end. -spec add_node(Node :: node(), RefNode :: node()) -> ok | {error, Reason :: term()}. add_node(Node, RefNode) -> %% init ServerLoc = make_server_id(RefNode), ServerId = make_server_id(Node), %% start ra ok = rpc:call(Node, ra, start, []), %% add case ra:add_member(ServerLoc, ServerId) of {ok, _, _NewServerLoc} -> case ra:start_server(?SYSTEM, ?CLUSTER_NAME, ServerId, ?RA_MACHINE, [ServerLoc]) of ok -> error_logger:info_msg("RAM[~s] Node ~s added to cluster", [node(), Node]), ok; {error, Reason} -> error_logger:error_msg("RAM[~s] Error addding node ~s to cluster: ~p", [node(), Node, Reason]), {error, Reason} end; {error, Reason} -> error_logger:error_msg("RAM[~s] Error addding node ~s to cluster: ~p", [node(), Node, Reason]), {error, Reason} end. -spec remove_node(Node :: node(), RefNode :: node()) -> ok | {error, Reason :: term()}. remove_node(Node, RefNode) -> %% init ServerLoc = make_server_id(RefNode), ServerId = make_server_id(Node), %% remove case ra:leave_and_delete_server(?SYSTEM, ServerLoc, ServerId) of ok -> error_logger:info_msg("RAM[~s] Node ~s removed from cluster", [node(), Node]), ok; timeout -> error_logger:error_msg("RAM[~s] Timeout removing node ~s from cluster", [node(), Node]), {error, timeout}; Err -> error_logger:error_msg("RAM[~s] Error removing node ~s from cluster: ~p", [node(), Node, Err]), {error, Err} end. -spec nodes() -> [node()]. nodes() -> ServerId = make_server_id(node()), case ra:members(ServerId) of {ok, ServerIds, _} -> get_nodes_from_server_ids(ServerIds); {error, Reason} -> error_logger:error_msg("RAM[~s] Failed to retrieve members: ~p", [node(), Reason]), {error, Reason} end. -spec get_server_loc() -> ra:server_id(). get_server_loc() -> make_server_id(node()). %% =================================================================== %% Internals %% =================================================================== -spec make_server_id(node()) -> ra:server_id(). make_server_id(Node) -> {ram, Node}. -spec get_node_from_server_id(ra:server_id()) -> node(). get_node_from_server_id({ram, Node}) -> Node. -spec get_nodes_from_server_ids([ra:server_id()]) -> [node()]. get_nodes_from_server_ids(ServerIds) -> [get_node_from_server_id(ServerId) || ServerId <- ServerIds].