%% @private %% @doc A server that keeps track of handler nodes. %% Handler nodes should call {@link eradius:modules_ready/2} from their application master %% as soon as they are ready, which makes them available for request processing. %% The node_mon server monitors the application master and removes it from %% request processing when it goes down. -module(eradius_node_mon). -export([start_link/0, modules_ready/2, set_nodes/1, get_module_nodes/1, get_remote_version/1]). -behaviour(gen_server). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(NODE_TAB, eradius_node_mon). -define(NODE_INFO_TAB, eradius_node_info). -define(PING_INTERVAL, 3000). % 3 sec -define(PING_TIMEOUT, 300). % 0.3 sec -define(SERVER, ?MODULE). %% ------------------------------------------------------------------------------------------ %% -- API start_link() -> gen_server:start_link({local, ?SERVER}, ?MODULE, [], []). -spec modules_ready(pid(), list(module())) -> ok. modules_ready(ApplicationMaster, Modules) when is_pid(ApplicationMaster), is_list(Modules) -> gen_server:cast(?SERVER, {modules_ready, ApplicationMaster, Modules}). -spec set_nodes(list(node())) -> ok. set_nodes(Nodes) -> gen_server:call(?SERVER, {set_nodes, Nodes}). -spec get_module_nodes(module()) -> [node()]. get_module_nodes(Module) -> try ets:lookup_element(?NODE_TAB, Module, 2) catch error:badarg -> [] end. -spec get_remote_version(node()) -> {integer(), integer()} | undefined. get_remote_version(Node) -> try ets:lookup_element(?NODE_INFO_TAB, Node, 2) catch error:badarg -> undefined end. %% ------------------------------------------------------------------------------------------ %% -- gen_server callbacks -record(state, { live_registrar_nodes = sets:new() :: sets:set(), dead_registrar_nodes = sets:new() :: sets:set(), app_masters = maps:new() :: map(), ping_timer :: reference() }). init([]) -> ets:new(?NODE_TAB, [bag, named_table, protected, {read_concurrency, true}]), ets:new(?NODE_INFO_TAB, [set, named_table, protected, {read_concurrency, true}]), PingTimer = erlang:send_after(?PING_INTERVAL, self(), ping_dead_nodes), {ok, #state{ping_timer = PingTimer}}. handle_call(remote_get_regs_v1, From, State) -> check_eradius_version(From), Registrations = maps:to_list(State#state.app_masters), {reply, {ok, Registrations}, State}; handle_call({set_nodes, Nodes}, _From, State) -> NewState = State#state{live_registrar_nodes = sets:new(), dead_registrar_nodes = sets:from_list(Nodes)}, self() ! ping_dead_nodes, {reply, ok, NewState}. handle_cast({remote_modules_ready_v1, ApplicationMaster, Modules}, State) -> NewState = State#state{app_masters = register_locally({ApplicationMaster, Modules}, State#state.app_masters)}, {noreply, NewState}; handle_cast({modules_ready, ApplicationMaster, Modules}, State) -> NewState = State#state{app_masters = register_locally({ApplicationMaster, Modules}, State#state.app_masters)}, lists:foreach(fun (Node) -> check_eradius_version(Node), gen_server:cast({?SERVER, Node}, {remote_modules_ready_v1, ApplicationMaster, Modules}) end, nodes()), {noreply, NewState}. handle_info({'DOWN', _MRef, process, {?SERVER, Node}, _Reason}, State = #state{live_registrar_nodes = LiveRegistrars}) -> case sets:is_element(Node, LiveRegistrars) of false -> %% ignore the 'DOWN', it's from a node we don't really want to monitor anymore %% and that shouldn't get into dead_registrar_nodes {noreply, State}; true -> {noreply, State#state{live_registrar_nodes = sets:del_element(Node, LiveRegistrars), dead_registrar_nodes = sets:add_element(Node, State#state.dead_registrar_nodes)}} end; handle_info({'DOWN', _MRef, process, Pid, _Reason}, State = #state{app_masters = AppMasters}) when is_pid(Pid) -> case maps:find(Pid, AppMasters) of error -> {noreply, State}; {ok, Modules} -> ServerNode = node(Pid), lists:foreach(fun (Mod) -> ets:delete_object(?NODE_TAB, {Mod, ServerNode}) end, Modules), NewState = State#state{app_masters = maps:remove(Pid, AppMasters)}, {noreply, NewState} end; handle_info(ping_dead_nodes, State = #state{app_masters = AppMasters, live_registrar_nodes = LiveRegistrars}) -> erlang:cancel_timer(State#state.ping_timer), {NewLive, NewDead, NewAppMasters} = sets:fold(fun (Node, {Live, Dead, AppMastersAcc}) -> case (catch gen_server:call({?SERVER, Node}, remote_get_regs_v1, ?PING_TIMEOUT)) of {ok, Registrations} -> NewAppMastersAcc = lists:foldl(fun register_locally/2, AppMastersAcc, Registrations), erlang:monitor(process, {?SERVER, Node}), {sets:add_element(Node, Live), Dead, NewAppMastersAcc}; {'EXIT', _Reason} -> {Live, sets:add_element(Node, Dead), AppMastersAcc} end end, {LiveRegistrars, sets:new(), AppMasters}, State#state.dead_registrar_nodes), NewPingTimer = erlang:send_after(?PING_INTERVAL, self(), ping_dead_nodes), NewState = State#state{live_registrar_nodes = NewLive, dead_registrar_nodes = NewDead, app_masters = NewAppMasters, ping_timer = NewPingTimer}, {noreply, NewState}; handle_info(_Info, State) -> {noreply, State}. terminate(_Reason, _State) -> ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %% ------------------------------------------------------------------------------------------ %% -- helpers -spec dict_prepend(term(), list(term()), map()) -> map(). dict_prepend(Key, List, Map) -> update_with(Key, fun (Old) -> List ++ Old end, List, Map). % NOTE: % copy-pasted from maps.erl to have backward compatability with OTP 18 % it can be rewmoved if minimal version of OTP will be set to 19. update_with(Key,Fun,Init,Map) when is_function(Fun,1), is_map(Map) -> case maps:find(Key,Map) of {ok,Val} -> maps:update(Key,Fun(Val),Map); error -> maps:put(Key,Init,Map) end; update_with(Key,Fun,Init,Map) -> erlang:error(error_type(Map),[Key,Fun,Init,Map]). -define(IS_ITERATOR(I), is_tuple(I) andalso tuple_size(I) == 3; I == none; is_integer(hd(I)) andalso is_map(tl(I))). error_type(M) when is_map(M); ?IS_ITERATOR(M) -> badarg; error_type(V) -> {badmap, V}. register_locally({ApplicationMaster, Modules}, AppMasters) -> case maps:is_key(ApplicationMaster, AppMasters) of true -> ok; %% already monitored false -> monitor(process, ApplicationMaster) end, ServerNode = node(ApplicationMaster), ets:insert(?NODE_TAB, [{Mod, ServerNode} || Mod <- Modules]), dict_prepend(ApplicationMaster, Modules, AppMasters). check_eradius_version({Pid, _}) when is_pid(Pid) -> check_eradius_version(Pid); check_eradius_version(Pid) when is_pid(Pid) -> check_eradius_version(node(Pid)); check_eradius_version(Node) -> case rpc:call(Node, application, get_key, [eradius, vsn]) of {ok, Vsn} -> try interpret_vsn(Vsn) of Version -> ets:insert(?NODE_INFO_TAB, {Node, Version}) catch _:_ -> lager:warning("unknown eradius version format ~p on node ~p", [Vsn, Node]) end; _ -> lager:warning("eradius version do not known on node ~p", [Node]) end. interpret_vsn(Vsn) -> BinVsn = list_to_binary(Vsn), [MajorVsn, MinorVsn | _] = binary:split(BinVsn, <<".">>, [global]), {binary_to_integer(MajorVsn), binary_to_integer(MinorVsn)}.