%%%============================================================================= %%% %%% | o __ _| _ __ |_ _ _ _ (TM) %%% |_ | | | (_| (/_ | | |_) (_| |_| | | | %%% %%% @copyright (C) 2014, Lindenbaum GmbH %%% %%% Permission to use, copy, modify, and/or distribute this software for any %%% purpose with or without fee is hereby granted, provided that the above %%% copyright notice and this permission notice appear in all copies. %%% %%% THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES %%% WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF %%% MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR %%% ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES %%% WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN %%% ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF %%% OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE. %%% %%% @doc %%% A registered server that manages dynamic expansion and reduction of the %%% Mnesia cluster. The server also subscribes for Mnesia system events. In %%% case a DB inconsistency is detected (split brain) the server tries to %%% resolve the conflict using the vector clocks of the `mnkv' tables as well %%% as user-provided resolve callbacks. If conflict resolution fails, one of the %%% conflicting nodes will be restarted! %%% @end %%%============================================================================= -module(mnkv_mon). -behaviour(gen_server). %% Internal API -export([start_link/0, info/0]). %% gen_server callbacks -export([init/1, handle_cast/2, handle_call/3, handle_info/2, code_change/3, terminate/2]). -include("mnkv.hrl"). -define(DUMP, "mnesia_core.dump"). -define(ERR(Fmt, Args), error_logger:error_msg(Fmt, Args)). -define(INFO(Fmt, Args), error_logger:info_msg(Fmt, Args)). %%%============================================================================= %%% Internal API %%%============================================================================= %%------------------------------------------------------------------------------ %% @private %% Simply start the server (registered). %%------------------------------------------------------------------------------ -spec start_link() -> {ok, pid()} | {error, term()}. start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). %%------------------------------------------------------------------------------ %% @private %% Print the all `mnkv` tables visible to this node. %%------------------------------------------------------------------------------ -spec info() -> ok. info() -> Tables = relevant_tables(node()), io:format("~w mnkv tables visible to ~s~n", [length(Tables), node()]), [io:format(" * ~s~n", [Table]) || Table <- Tables], io:format("~n"). %%%============================================================================= %%% gen_server callbacks %%%============================================================================= -record(state, {}). %%------------------------------------------------------------------------------ %% @private %%------------------------------------------------------------------------------ init([]) -> {ok, _} = mnesia:subscribe(system), [self() ! {nodeup, Node} || Node <- nodes()], ok = net_kernel:monitor_nodes(true), {ok, #state{}}. %%------------------------------------------------------------------------------ %% @private %%------------------------------------------------------------------------------ handle_call(_Request, _From, State) -> {reply, undef, State}. %%------------------------------------------------------------------------------ %% @private %%------------------------------------------------------------------------------ handle_cast(_Request, State) -> {noreply, State}. %%------------------------------------------------------------------------------ %% @private %%------------------------------------------------------------------------------ handle_info({mnesia_system_event, Event}, State) -> handle_mnesia_event(Event, State), {noreply, State}; handle_info({nodeup, Node}, State) -> ok = global:trans({?MODULE, self()}, fun() -> expand(Node) end), {noreply, State}; handle_info({nodedown, Node}, State) -> ok = global:trans({?MODULE, self()}, fun() -> reduce(Node) end), {noreply, State}; handle_info(_Info, State) -> {noreply, State}. %%------------------------------------------------------------------------------ %% @private %%------------------------------------------------------------------------------ code_change(_OldVsn, State, _Extra) -> {ok, State}. %%------------------------------------------------------------------------------ %% @private %%------------------------------------------------------------------------------ terminate(_Reason, _State) -> ok. %%%============================================================================= %%% Internal functions %%%============================================================================= %%------------------------------------------------------------------------------ %% @private %% Handle Mnesia system events. Fatal conditions will be resolved with node %% restart. %%------------------------------------------------------------------------------ handle_mnesia_event({mnesia_fatal, Format, Args, BinaryCore}, _State) -> ?ERR("Fatal condition: " ++ Format, Args), file:write_file(?DUMP, BinaryCore), handle_unresolved_conflict(); handle_mnesia_event({mnesia_info, Format, Args}, _State) -> ?INFO(Format, Args); handle_mnesia_event(_Event, _State) -> ok. %%------------------------------------------------------------------------------ %% @private %% Expand the mnesia cluster to `Node'. This will add table copies of local %% tables to `Node' as well as table copies remote tables on the local node. %% This is the central function handling dynamic expansion of mnesia on new %% nodes as well as recovery from partitioned networks. %% %% A word about table replication: %% When using RAM copies, Mnesia is able to `merge' schema tables of different %% nodes, as long as one of the schema's to merge is clean (no tables with the %% same name on both nodes). By default tables are assigned a `cookie' value %% that differs even for tables with the same name and makes them incompatible %% by default. `mnkv' avoids this by assigning each of its tables a custom %% cookie (which is always the same). However, even using this trick, special %% magic is needed to merge these tables. Each of these tables must be %% configured as RAM copy on the remote node __before__ merging the schemas. %% %% Thanks again to Ulf Wiger (`unsplit') for pointing us in the right direction. %%------------------------------------------------------------------------------ expand(Node) -> expand(is_running(Node), Node). expand(true, Node) -> LocalTables = relevant_tables(node()), RemoteTables = relevant_tables(Node), ?MNKV_DBG("Relevant tables on ~s: ~w~n", [node(), LocalTables]), ?MNKV_DBG("Relevant tables on ~s: ~w~n", [Node, RemoteTables]), LocalOnlyTables = LocalTables -- RemoteTables, RemoteOnlyTables = RemoteTables -- LocalTables, ?MNKV_DBG("Local-only tables on ~s: ~w~n", [node(), LocalOnlyTables]), ?MNKV_DBG("Local-only tables on ~s: ~w~n", [Node, RemoteOnlyTables]), LocalSchema = get_cookie(node(), schema), RemoteSchema = get_cookie(Node, schema), IsRemoteRunning = lists:member(Node, get_running_nodes(node())), case {LocalSchema =:= RemoteSchema, IsRemoteRunning} of {true, true} -> %% Schemas are already merged and we see the node as running, %% someone else did the work. However, this also means that the %% respective remote node must have this node in its %% `running_db_nodes'. true = lists:member(node(), get_running_nodes(Node)), %% Now distribute tables only found on one of the nodes. add_table_copies(node(), Node, LocalOnlyTables), add_table_copies(Node, node(), RemoteOnlyTables), ?INFO("Already connected to ~s~n", [Node]); {IsMerged, IsRemoteRunning} -> case {IsMerged, IsRemoteRunning} of {false, false} -> %% The newly connected node has a different schema. That %% means the schemas must be merged now. ?INFO("Expanding to ~s...~n", [Node]); {true, false} -> %% Both nodes have the same schema, but none of the nodes %% has its respective counterpart in its `running_db_nodes', %% this must be a reconnect after a netsplit. ?INFO("Recovering from network partition (~s)...~n", [Node]) end, connect_nodes(Node, LocalTables, LocalOnlyTables, RemoteOnlyTables) end; expand(false, _) -> ok. %% Do not expand to the remote node if mnesia is not running. %%------------------------------------------------------------------------------ %% @private %%------------------------------------------------------------------------------ connect_nodes(Node, LocalTables, LocalOnlyTables, RemoteOnlyTables) -> %% All duplicate tables must be interconnected (with mutual RAM copies) %% __before__ connecting the mnesia instances. LocalAndRemoteTables = LocalTables -- LocalOnlyTables, add_table_copies(node(), Node, LocalAndRemoteTables), add_table_copies(Node, node(), LocalAndRemoteTables), %% Connect both nodes and merge values of duplicate tables. Fun = connect_nodes_user_fun(LocalAndRemoteTables, Node), case mnesia_controller:connect_nodes([Node], Fun) of {ok, NewNodes} -> %% The merge is only successful if Node is now part of the %% mnesia cluster true = lists:member(Node, NewNodes), %% Now distribute tables only found on one of the nodes. add_table_copies(node(), Node, LocalOnlyTables), add_table_copies(Node, node(), RemoteOnlyTables), ?INFO("Successfully connected to ~s~n", [Node]); Error -> ?ERR("Failed to connect to ~s: ~w~n", [Node, Error]), %% last resort handle_unresolved_conflict() end. %%------------------------------------------------------------------------------ %% @private %% A custom user function that merges the schemas of two node islands as well as %% the respective values in conflicting tables. %%------------------------------------------------------------------------------ connect_nodes_user_fun(TablesToMerge, Node) -> fun(SchemaMergeFun) -> case SchemaMergeFun(TablesToMerge) of Result = {merged, OldFriends, NewFriends} -> ?MNKV_DBG("Schemas successfully merged~n", []), ?MNKV_DBG("NewFriends: ~w~n", [NewFriends]), %% Sorry, but we must be part of `db_nodes' ourselves or I %% loose my mind (see mnesia_schema:do_merge_schema/1). true = lists:member(node(), OldFriends), %% What does this mean? %% Unfortunately, this is also not really clear to me. My %% assumption: %% `NewFriends' seems to be the list of nodes the local node %% successfully merged its schema with (without the local %% node itself). The local node is usually part of the %% `OldFriends' list. This seems to be equivalent to the %% current `db_nodes'. In case of asymmetric netsplits there %% may be a non-empty intersection between the two lists. %% The merge function may be called multiple times in case %% additional nodes become visible by merged schemas or %% through merge retries. %% %% What do we do here? %% We try to make sure to always merge with a sane candidate %% from another (new) island mnesia connects us to. case lists:member(Node, OldFriends) of false -> %% If we did not already merge with `Node', it %% __must__ be a member of `NewFriends'. true = lists:member(Node, NewFriends), MergeWith = [Node]; true -> %% Otherwise `Node' is already part of `db_nodes' %% and thus a member of `OldFriends'. In this case %% we simple choose a representative of %% `NewFriends' that is not already an old friend. MergeWith = NewFriends -- OldFriends end, case mnkv_merge:tables(TablesToMerge, MergeWith) of ok -> Result; {error, Reason} -> mnesia:abort(Reason) end; Result -> Result end end. %%------------------------------------------------------------------------------ %% @private %% Remove `Node' from the distributed mnesia cluster. %%------------------------------------------------------------------------------ reduce(Node) -> case lists:member(Node, get_db_nodes(node())) of true -> %% Remove the disconnected node from the global schema. This will %% remove all ram copies copied onto this node (for all tables). case mnesia:del_table_copy(schema, Node) of {atomic, ok} -> ?INFO("Successfully disconnected from ~s~n", [Node]); Error = {aborted, _} -> ?INFO("Failed to remove schema from ~s: ~w~n", [Node, Error]) end; false -> %% The disconnected node is not part of the seen `db_nodes' anymore, %% someone else did the work or the node did never participate. ok end. %%------------------------------------------------------------------------------ %% @private %% Returns whether mnesia is running on `Node'. %%------------------------------------------------------------------------------ is_running(Node) -> rpc_mnesia(Node, system_info, [is_running]) =:= yes. %%------------------------------------------------------------------------------ %% @private %% Add RAM copies of all `Tables' at `FromNode' to `ToNode'. Crashes, if a table %% cannot be replicated. %%------------------------------------------------------------------------------ add_table_copies(FromNode, ToNode, Tables) -> [ok = add_table_copy(FromNode, ToNode, T) || T <- Tables]. %%------------------------------------------------------------------------------ %% @private %% Add a RAM copy of `Table' at `FromNode' to `ToNode'. %%------------------------------------------------------------------------------ add_table_copy(FromNode, ToNode, Table) -> case rpc_mnesia(FromNode, add_table_copy, [Table, ToNode, ram_copies]) of {atomic, ok} -> ok; {aborted, {already_exists, Table, ToNode}} -> ok; {aborted, Reason} -> {error, Reason}; Error -> Error end. %%------------------------------------------------------------------------------ %% @private %% Returns the tables on `Node' that are managed by `mnkv'. %%------------------------------------------------------------------------------ relevant_tables(Node) -> Tables = rpc_mnesia(Node, system_info, [tables]), [T || T <- Tables, is_relevant(Node, T)]. %%------------------------------------------------------------------------------ %% @private %% Returns whether a table is managed by `mnkv'. %%------------------------------------------------------------------------------ is_relevant(Node, Table) -> get_cookie(Node, Table) =:= ?MNKV_COOKIE. %%------------------------------------------------------------------------------ %% @private %% Returns the cookie of `Table'. %%------------------------------------------------------------------------------ get_cookie(Node, Table) -> rpc_mnesia(Node, table_info, [Table, cookie]). %%------------------------------------------------------------------------------ %% @private %% Returns the current `running_db_nodes' as seen by `Node'. %%------------------------------------------------------------------------------ get_running_nodes(Node) -> rpc_mnesia(Node, system_info, [running_db_nodes]). %%------------------------------------------------------------------------------ %% @private %% Returns the current `db_nodes' as seen by `Node'. %%------------------------------------------------------------------------------ get_db_nodes(Node) -> rpc_mnesia(Node, system_info, [db_nodes]). %%------------------------------------------------------------------------------ %% @private %% Make an RPC call to the mnesia module on node `Node'. The `rpc' module knows %% when a call is local and optimizes that. %%------------------------------------------------------------------------------ rpc_mnesia(Node, Function, Args) -> Timeout = application:get_env(mnkv, rpc_timeout, ?MNKV_RPC_TIMEOUT), case rpc:call(Node, mnesia, Function, Args, Timeout) of {badrpc, Reason} -> {error, Reason}; Result -> Result end. %%------------------------------------------------------------------------------ %% @private %% Final conflict resolver for inconsistent database states, e.g. after %% netsplits. This will simply restart the local node. %%------------------------------------------------------------------------------ handle_unresolved_conflict() -> ?ERR("Final conflict resolution necessary, restarting ~s...~n", [node()]), init:restart().