fix: set load config done after update tnx_id

This commit is contained in:
zhongwencool 2023-07-07 11:05:26 +08:00
parent 9f57ba510e
commit f7513b900a
3 changed files with 29 additions and 37 deletions

View File

@ -17,7 +17,7 @@
-behaviour(gen_server). -behaviour(gen_server).
%% API %% API
-export([start_link/1, mnesia/1]). -export([start_link/0, mnesia/1]).
%% Note: multicall functions are statically checked by %% Note: multicall functions are statically checked by
%% `emqx_bapi_trans' and `emqx_bpapi_static_checks' modules. Don't %% `emqx_bapi_trans' and `emqx_bpapi_static_checks' modules. Don't
@ -30,7 +30,8 @@
skip_failed_commit/1, skip_failed_commit/1,
fast_forward_to_commit/2, fast_forward_to_commit/2,
on_mria_stop/1, on_mria_stop/1,
wait_for_cluster_rpc/0 wait_for_cluster_rpc/0,
maybe_init_tnx_id/2
]). ]).
-export([ -export([
commit/2, commit/2,
@ -69,12 +70,6 @@
-compile(export_all). -compile(export_all).
-compile(nowarn_export_all). -compile(nowarn_export_all).
start_link() ->
start_link(-1).
start_link(Node, Name, RetryMs) ->
start_link(-1, Node, Name, RetryMs).
-endif. -endif.
-define(INITIATE(MFA), {initiate, MFA}). -define(INITIATE(MFA), {initiate, MFA}).
@ -115,11 +110,11 @@ mnesia(boot) ->
{attributes, record_info(fields, cluster_rpc_commit)} {attributes, record_info(fields, cluster_rpc_commit)}
]). ]).
start_link(TnxId) -> start_link() ->
start_link(TnxId, node(), ?MODULE, get_retry_ms()). start_link(node(), ?MODULE, get_retry_ms()).
start_link(TnxId, Node, Name, RetryMs) -> start_link(Node, Name, RetryMs) ->
case gen_server:start_link({local, Name}, ?MODULE, [TnxId, Node, RetryMs], []) of case gen_server:start_link({local, Name}, ?MODULE, [Node, RetryMs], []) of
{ok, Pid} -> {ok, Pid} ->
{ok, Pid}; {ok, Pid};
{error, {already_started, Pid}} -> {error, {already_started, Pid}} ->
@ -303,26 +298,22 @@ wait_for_cluster_rpc() ->
%%%=================================================================== %%%===================================================================
%% @private %% @private
init([TnxId, Node, RetryMs]) -> init([Node, RetryMs]) ->
register_mria_stop_cb(fun ?MODULE:on_mria_stop/1), register_mria_stop_cb(fun ?MODULE:on_mria_stop/1),
{ok, _} = mnesia:subscribe({table, ?CLUSTER_MFA, simple}), {ok, _} = mnesia:subscribe({table, ?CLUSTER_MFA, simple}),
State = #{node => Node, retry_interval => RetryMs, is_leaving => false}, State = #{node => Node, retry_interval => RetryMs, is_leaving => false},
%% Now continue with the normal catch-up process %% Now continue with the normal catch-up process
%% That is: apply the missing transactions after the config %% That is: apply the missing transactions after the config
%% was copied until now. %% was copied until now.
{ok, State, {continue, {?CATCH_UP, TnxId}}}. {ok, State, {continue, {?CATCH_UP, init}}}.
%% @private %% @private
handle_continue({?CATCH_UP, TnxId}, State = #{node := Node}) -> handle_continue({?CATCH_UP, init}, State) ->
%% emqx app must be started before %% emqx app must be started before
%% trying to catch up the rpc commit logs %% trying to catch up the rpc commit logs
ok = wait_for_emqx_ready(), ok = wait_for_emqx_ready(),
ok = wait_for_cluster_rpc(), ok = wait_for_cluster_rpc(),
%% The init transaction ID is set in emqx_conf_app after
%% it has fetched the latest config from one of the core nodes
ok = maybe_init_tnx_id(Node, TnxId),
{noreply, State, catch_up(State)}; {noreply, State, catch_up(State)};
%% @private
handle_continue(?CATCH_UP, State) -> handle_continue(?CATCH_UP, State) ->
{noreply, State, catch_up(State)}. {noreply, State, catch_up(State)}.

View File

@ -31,17 +31,16 @@
-define(DEFAULT_INIT_TXN_ID, -1). -define(DEFAULT_INIT_TXN_ID, -1).
start(_StartType, _StartArgs) -> start(_StartType, _StartArgs) ->
{ok, TnxId} = try
try ok = init_conf()
{ok, _} = init_conf() catch
catch C:E:St ->
C:E:St -> %% logger is not quite ready.
%% logger is not quite ready. io:format(standard_error, "Failed to load config~n~p~n~p~n~p~n", [C, E, St]),
io:format(standard_error, "Failed to load config~n~p~n~p~n~p~n", [C, E, St]), init:stop(1)
init:stop(1) end,
end,
ok = emqx_config_logger:refresh_config(), ok = emqx_config_logger:refresh_config(),
emqx_conf_sup:start_link(TnxId). emqx_conf_sup:start_link().
stop(_State) -> stop(_State) ->
ok. ok.
@ -94,10 +93,12 @@ sync_data_from_node() ->
%% Internal functions %% Internal functions
%% ------------------------------------------------------------------------------ %% ------------------------------------------------------------------------------
init_load() -> init_load(TnxId) ->
case emqx_app:get_config_loader() of case emqx_app:get_config_loader() of
Module when Module == emqx; Module == emqx_conf -> Module when Module == emqx; Module == emqx_conf ->
ok = emqx_config:init_load(emqx_conf:schema_module()), ok = emqx_config:init_load(emqx_conf:schema_module()),
%% Set load config done after update(init) tnx_id.
ok = emqx_cluster_rpc:maybe_init_tnx_id(node(), TnxId),
ok = emqx_app:set_config_loader(emqx_conf), ok = emqx_app:set_config_loader(emqx_conf),
ok; ok;
Module -> Module ->
@ -115,8 +116,8 @@ init_load_done() ->
init_conf() -> init_conf() ->
emqx_cluster_rpc:wait_for_cluster_rpc(), emqx_cluster_rpc:wait_for_cluster_rpc(),
{ok, TnxId} = sync_cluster_conf(), {ok, TnxId} = sync_cluster_conf(),
ok = init_load(), ok = init_load(TnxId),
{ok, TnxId}. ok.
cluster_nodes() -> cluster_nodes() ->
mria:cluster_nodes(cores) -- [node()]. mria:cluster_nodes(cores) -- [node()].

View File

@ -18,16 +18,16 @@
-behaviour(supervisor). -behaviour(supervisor).
-export([start_link/1]). -export([start_link/0]).
-export([init/1]). -export([init/1]).
-define(SERVER, ?MODULE). -define(SERVER, ?MODULE).
start_link(TnxId) -> start_link() ->
supervisor:start_link({local, ?SERVER}, ?MODULE, [TnxId]). supervisor:start_link({local, ?SERVER}, ?MODULE, []).
init([TnxId]) -> init([]) ->
SupFlags = #{ SupFlags = #{
strategy => one_for_all, strategy => one_for_all,
intensity => 10, intensity => 10,
@ -35,7 +35,7 @@ init([TnxId]) ->
}, },
ChildSpecs = ChildSpecs =
[ [
child_spec(emqx_cluster_rpc, [TnxId]), child_spec(emqx_cluster_rpc, []),
child_spec(emqx_cluster_rpc_cleaner, []) child_spec(emqx_cluster_rpc_cleaner, [])
], ],
{ok, {SupFlags, ChildSpecs}}. {ok, {SupFlags, ChildSpecs}}.