mnesia_bup.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,170 行 · 第 1/3 页

%% ``The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved via the world wide web at http://www.erlang.org/.
%% 
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%% 
%% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
%% AB. All Rights Reserved.''
%% 
%%     $Id$
%%
-module(mnesia_bup).
-export([
	 %% Public interface
	 iterate/4,
	 read_schema/2,
	 fallback_bup/0,
	 fallback_exists/0,
	 tm_fallback_start/1,
	 create_schema/1,
	 install_fallback/1,
	 install_fallback/2,
	 uninstall_fallback/0,
	 uninstall_fallback/1,
	 traverse_backup/4,
	 traverse_backup/6,
	 make_initial_backup/3,
	 fallback_to_schema/0,
	 lookup_schema/2,
	 schema2bup/1,
	 refresh_cookie/2,

	 %% Internal
	 fallback_receiver/2,
	 install_fallback_master/2,
	 uninstall_fallback_master/2,
	 local_uninstall_fallback/2,
	 do_traverse_backup/7,
	 trav_apply/4
	]).

-include("mnesia.hrl").
-import(mnesia_lib, [verbose/2, dbg_out/2]).

-record(restore, {mode, bup_module, bup_data}).

-record(fallback_args, {opaque,
			scope = global,
			module = mnesia_monitor:get_env(backup_module),
			use_default_dir = true,
			mnesia_dir,
			fallback_bup,
			fallback_tmp,
			skip_tables = [],
			keep_tables = [],
			default_op = keep_tables		       
		       }).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Backup iterator

%% Reads schema section and iterates over all records in a backup.
%%
%% Fun(BunchOfRecords, Header, Schema, Acc) is applied when a suitable amount
%% of records has been collected.
%%
%% BunchOfRecords will be [] when the iteration is done.
iterate(Mod, Fun, Opaque, Acc) ->
    R = #restore{bup_module = Mod, bup_data = Opaque},
    case catch read_schema_section(R) of
	{error, Reason} ->
	    {error, Reason};
	{R2, {Header, Schema, Rest}} ->
	    case catch iter(R2, Header, Schema, Fun, Acc, Rest) of
		{ok, R3, Res} ->
		    catch safe_apply(R3, close_read, [R3#restore.bup_data]),
		    {ok, Res};
		{error, Reason} ->
		    catch safe_apply(R2, close_read, [R2#restore.bup_data]),
		    {error, Reason};
		{'EXIT', Pid, Reason} ->
		    catch safe_apply(R2, close_read, [R2#restore.bup_data]),
		    {error, {'EXIT', Pid, Reason}};
		{'EXIT', Reason} ->
		    catch safe_apply(R2, close_read, [R2#restore.bup_data]),
		    {error, {'EXIT', Reason}}
	    end
    end.

iter(R, Header, Schema, Fun, Acc, []) ->
    case safe_apply(R, read, [R#restore.bup_data]) of
	{R2, []} ->
	    Res = Fun([], Header, Schema, Acc),
	    {ok, R2, Res};
	{R2, BupItems} ->
	    iter(R2, Header, Schema, Fun, Acc, BupItems)
    end;
iter(R, Header, Schema, Fun, Acc, BupItems) ->
    Acc2 = Fun(BupItems, Header, Schema, Acc),
    iter(R, Header, Schema, Fun, Acc2, []).

safe_apply(R, write, [_, Items]) when Items == [] ->
    R;
safe_apply(R, What, Args) ->
    Abort = fun(Re) -> abort_restore(R, What, Args, Re) end,
    receive
	{'EXIT', Pid, Re} -> Abort({'EXIT', Pid, Re})
    after 0 ->
	    Mod = R#restore.bup_module,
	    case catch apply(Mod, What, Args) of
		{ok, Opaque, Items} when What == read ->
		    {R#restore{bup_data = Opaque}, Items};
		{ok, Opaque}  when What /= read->
		    R#restore{bup_data = Opaque};
		{error, Re} ->
		    Abort(Re);
		Re ->
		    Abort(Re)
	    end
    end.

abort_restore(R, What, Args, Reason) ->
    Mod = R#restore.bup_module,
    Opaque = R#restore.bup_data,
    dbg_out("Restore aborted. ~p:~p~p -> ~p~n",
	    [Mod, What, Args, Reason]),
    catch apply(Mod, close_read, [Opaque]),
    throw({error, Reason}).
	    
fallback_to_schema() ->
    Fname = fallback_bup(),
    fallback_to_schema(Fname).

fallback_to_schema(Fname) ->
    Mod = mnesia_backup,
    case read_schema(Mod, Fname) of
	{error, Reason} ->
	    {error, Reason};
	Schema ->
	    case catch lookup_schema(schema, Schema) of
		{error, _} -> 
		    {error, "No schema in fallback"};
		List ->
		    {ok, fallback, List}
	    end
    end.

%% Opens Opaque reads schema and then close 
read_schema(Mod, Opaque) ->
    R = #restore{bup_module = Mod, bup_data = Opaque},
    case catch read_schema_section(R) of
	{error, Reason} ->
	    {error, Reason};
	{R2, {_Header, Schema, _}} ->
	    catch safe_apply(R2, close_read, [R2#restore.bup_data]),
	    Schema
    end.
    
%% Open backup media and extract schema
%% rewind backup media and leave it open
%% Returns {R, {Header, Schema}}
read_schema_section(R) ->
    case catch do_read_schema_section(R) of
	{'EXIT', Reason} ->
	    catch safe_apply(R, close_read, [R#restore.bup_data]),
	    {error, {'EXIT', Reason}};
	{error, Reason} ->
	    catch safe_apply(R, close_read, [R#restore.bup_data]),
	    {error, Reason};
	{R2, {H, Schema, Rest}} ->
	    Schema2 = convert_schema(H#log_header.log_version, Schema),
	    {R2, {H, Schema2, Rest}}
    end.

do_read_schema_section(R) ->
    R2 = safe_apply(R, open_read, [R#restore.bup_data]),
    {R3, RawSchema} = safe_apply(R2, read, [R2#restore.bup_data]),
    do_read_schema_section(R3, verify_header(RawSchema), []).

do_read_schema_section(R, {ok, B, C, []}, Acc) ->
    case safe_apply(R, read, [R#restore.bup_data]) of
	{R2, []} ->
	    {R2, {B, Acc, []}};
	{R2, RawSchema} ->
	    do_read_schema_section(R2, {ok, B, C, RawSchema}, Acc)
    end;

do_read_schema_section(R, {ok, B, C, [Head | Tail]}, Acc)
        when element(1, Head) == schema ->
    do_read_schema_section(R, {ok, B, C, Tail}, Acc ++ [Head]);

do_read_schema_section(R, {ok, B, _C, Rest}, Acc) ->
    {R, {B, Acc, Rest}};

do_read_schema_section(_R, {error, Reason}, _Acc) ->
    {error, Reason}.

verify_header([H | RawSchema])  when record(H, log_header) ->
    Current = mnesia_log:backup_log_header(),
    if
	H#log_header.log_kind == Current#log_header.log_kind ->
	    Versions = ["0.1", "1.1", Current#log_header.log_version],
	    case lists:member(H#log_header.log_version, Versions) of
		true ->
		    {ok, H, Current, RawSchema};
		false ->
		    {error, {"Bad header version. Cannot be used as backup.", H}}
	    end;
	true ->
	    {error, {"Bad kind of header. Cannot be used as backup.", H}}
    end;
verify_header(RawSchema) ->
    {error, {"Missing header. Cannot be used as backup.", catch hd(RawSchema)}}.

refresh_cookie(Schema, NewCookie) ->
    case lists:keysearch(schema, 2, Schema) of
	{value, {schema, schema, List}} ->
	    Cs = mnesia_schema:list2cs(List),
	    Cs2 = Cs#cstruct{cookie = NewCookie},
	    Item = {schema, schema, mnesia_schema:cs2list(Cs2)},
	    lists:keyreplace(schema, 2, Schema, Item);
	
	false ->
	    Reason = "No schema found. Cannot be used as backup.",
	    throw({error, {Reason, Schema}})
    end.

%% Convert schema items from an external backup
%% If backup format is the latest, no conversion is needed
%% All supported backup formats should have their converters
%% here as separate function clauses.
convert_schema("0.1", Schema) ->
    convert_0_1(Schema);
convert_schema("1.1", Schema) ->
    %% The new backup format is a pure extension of the old one
    Current = mnesia_log:backup_log_header(),
    convert_schema(Current#log_header.log_version, Schema);
convert_schema(Latest, Schema) ->
    H = mnesia_log:backup_log_header(),
    if
	H#log_header.log_version == Latest ->
	    Schema;
	true ->
	    Reason = "Bad backup header version. Cannot convert schema.",
	    throw({error, {Reason, H}})
    end.

%% Backward compatibility for 0.1
convert_0_1(Schema) ->
    case lists:keysearch(schema, 2, Schema) of
	{value, {schema, schema, List}} ->
	    Schema2 = lists:keydelete(schema, 2, Schema),
	    Cs = mnesia_schema:list2cs(List),
	    convert_0_1(Schema2, [], Cs);
	false ->
	    List = mnesia_schema:get_initial_schema(disc_copies, [node()]),
	    Cs = mnesia_schema:list2cs(List),
	    convert_0_1(Schema, [], Cs)
    end.

convert_0_1([{schema, cookie, Cookie} | Schema], Acc, Cs) ->
    convert_0_1(Schema, Acc, Cs#cstruct{cookie = Cookie});
convert_0_1([{schema, db_nodes, DbNodes} | Schema], Acc, Cs) ->
    convert_0_1(Schema, Acc, Cs#cstruct{disc_copies = DbNodes});
convert_0_1([{schema, version, Version} | Schema], Acc, Cs) ->
    convert_0_1(Schema, Acc, Cs#cstruct{version = Version});
convert_0_1([{schema, Tab, Def} | Schema], Acc, Cs) ->
    Head = 
	case lists:keysearch(index, 1, Def) of
	    {value, {index, PosList}} ->
		%% Remove the snmp "index"
		P = PosList -- [snmp],
		Def2 = lists:keyreplace(index, 1, Def, {index, P}),
		{schema, Tab, Def2};
	    false ->
		{schema, Tab, Def}
	end,
    convert_0_1(Schema, [Head | Acc], Cs);
convert_0_1([Head | Schema], Acc, Cs) ->
    convert_0_1(Schema, [Head | Acc], Cs);
convert_0_1([], Acc, Cs) ->
    [schema2bup({schema, schema, Cs}) | Acc].

%% Returns Val or throw error
lookup_schema(Key, Schema) ->
    case lists:keysearch(Key, 2, Schema) of
	{value, {schema, Key, Val}} -> Val;
	false -> throw({error, {"Cannot lookup", Key}})
    end.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Backup compatibility

%% Convert internal schema items to backup dito
schema2bup({schema, Tab}) ->
    {schema, Tab};
schema2bup({schema, Tab, TableDef}) ->
    {schema, Tab, mnesia_schema:cs2list(TableDef)}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Create schema on the given nodes
%% Requires that old schemas has been deleted
%% Returns ok | {error, Reason}
create_schema([]) ->
    create_schema([node()]);
create_schema(Ns) when list(Ns) ->
    case is_set(Ns) of
	true ->
	    create_schema(Ns, mnesia_schema:ensure_no_schema(Ns));
	false ->
	    {error, {combine_error, Ns}}
    end;
create_schema(Ns) ->
    {error, {badarg, Ns}}.

is_set(List) when list(List) ->
    ordsets:is_set(lists:sort(List));
is_set(_) ->
    false.

create_schema(Ns, ok) ->
    %% Ensure that we access the intended Mnesia
    %% directory. This function may not be called
    %% during startup since it will cause the
    %% application_controller to get into deadlock
    case mnesia_lib:ensure_loaded(?APPLICATION) of
	ok ->
	    case mnesia_monitor:get_env(schema_location) of
		ram -> 
		    {error, {has_no_disc, node()}};
		_ ->
		    case mnesia_schema:opt_create_dir(true, mnesia_lib:dir()) of
			{error, What} ->
			    {error, What};
			ok ->
			    Mod = mnesia_backup,
			    Str = mk_str(),
			    File = mnesia_lib:dir(Str),
			    file:delete(File),
			    case catch make_initial_backup(Ns, File, Mod) of
				{ok, _Res} ->
				    case do_install_fallback(File, Mod) of
					ok ->
					    file:delete(File),
					    ok;
					{error, Reason} ->
					    {error, Reason}
				    end;
				{error, Reason} ->
				    {error, Reason}
			    end
		    end
	    end;	
	{error, Reason} ->
	    {error, Reason}
    end;
create_schema(_Ns, {error, Reason}) ->
    {error, Reason};
create_schema(_Ns, Reason) ->
    {error, Reason}.

mk_str() ->
    Now = [integer_to_list(I) || I <- tuple_to_list(now())],
    lists:concat([node()] ++ Now ++ ".TMP").

make_initial_backup(Ns, Opaque, Mod) ->
    Schema = [{schema, schema, mnesia_schema:get_initial_schema(disc_copies, Ns)}],
    O2 = do_apply(Mod, open_write, [Opaque], Opaque),
    O3 = do_apply(Mod, write, [O2, [mnesia_log:backup_log_header()]], O2),
    O4 = do_apply(Mod, write, [O3, Schema], O3),
    O5 = do_apply(Mod, commit_write, [O4], O4),
    {ok, O5}.

do_apply(_, write, [_, Items], Opaque) when Items == [] ->
    Opaque;
do_apply(Mod, What, Args, _Opaque) ->
    case catch apply(Mod, What, Args) of
	{ok, Opaque2} ->  Opaque2;
	{error, Reason} -> throw({error, Reason});
	{'EXIT', Reason} -> throw({error, {'EXIT', Reason}})
    end.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%% Restore