diff options
Diffstat (limited to 'lib/diameter/src/app/diameter_config.erl')
| -rw-r--r-- | lib/diameter/src/app/diameter_config.erl | 681 |
1 files changed, 681 insertions, 0 deletions
diff --git a/lib/diameter/src/app/diameter_config.erl b/lib/diameter/src/app/diameter_config.erl new file mode 100644 index 0000000000..42c70890b3 --- /dev/null +++ b/lib/diameter/src/app/diameter_config.erl @@ -0,0 +1,681 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2010-2011. All Rights Reserved. +%% +%% 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 online 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. +%% +%% %CopyrightEnd% +%% + +%% +%% This module writes service/transport configuration to the table +%% diameter_config, so that the config will survive service process +%% death, and then turns it into calls towards diameter_service. It +%% also restarts services upon their death. +%% +%% The table diameter_config is only written here while +%% diameter_service reads. This is all somewhat after the fact. Once +%% upon a time the config was only stored in the service process, +%% causing much grief if these processes died (which they did with +%% some regularity) and one was forced to reconfigure. This module was +%% then inserted into the service start in order to keep a more +%% permanent record of the config. That said, service processes are +%% now much more robust than they once were and crashing is a thing of +%% the past. +%% + +-module(diameter_config). +-compile({no_auto_import, [monitor/2]}). + +-behaviour(gen_server). + +-export([start_service/2, + stop_service/1, + add_transport/2, + remove_transport/2, + have_transport/2, + lookup/1]). + +%% child server start +-export([start_link/0]). + +%% gen_server callbacks +-export([init/1, + terminate/2, + handle_call/3, + handle_cast/2, + handle_info/2, + code_change/3]). + +%% diameter_sync requests. +-export([sync/1]). + +%% debug +-export([state/0, + uptime/0]). + +-include_lib("diameter/include/diameter.hrl"). +-include("diameter_internal.hrl"). + +%% Server state. +-record(state, {id = now()}). + +%% Registered name of the server. +-define(SERVER, ?MODULE). + +%% Table config is written to. +-define(TABLE, ?MODULE). + +%% Workaround for dialyzer's lack of understanding of match specs. +-type match(T) + :: T | '_' | '$1' | '$2' | '$3' | '$4'. + +%% Configuration records in ?TABLE. + +-record(service, + {name, + rec :: match(#diameter_service{}), + options :: match(list())}). + +-record(transport, + {service, %% name + ref = make_ref() :: match(reference()), + type :: match(connect | listen), + options :: match(list())}). + +%% Monitor entry in ?TABLE. +-record(monitor, {mref = make_ref() :: reference(), + service}). %% name + +%% Time to lay low before restarting a dead service. +-define(RESTART_SLEEP, 2000). + +%% A minimal diameter_caps for checking for valid capabilities values. +-define(EXAMPLE_CAPS, + #diameter_caps{origin_host = "TheHost", + origin_realm = "TheRealm", + host_ip_address = [{127,0,0,1}], + vendor_id = 42, + product_name = "TheProduct"}). + +-define(VALUES(Rec), tl(tuple_to_list(Rec))). + +%%% The return values below assume the server diameter_config is started. +%%% The functions will exit if it isn't. + +%% -------------------------------------------------------------------------- +%% # start_service(SvcName, Opts) +%% +%% Output: ok | {error, Reason} +%% -------------------------------------------------------------------------- + +start_service(SvcName, Opts) + when is_list(Opts) -> + start_rc(sync(SvcName, {start_service, SvcName, Opts})). + +start_rc({ok = T, _Pid}) -> + T; +start_rc({error, _} = No) -> + No; +start_rc(timeout) -> + {error, application_not_started}. + +%% -------------------------------------------------------------------------- +%% # stop_service(SvcName) +%% +%% Output: ok +%% -------------------------------------------------------------------------- + +stop_service(SvcName) -> + sync(SvcName, {stop_service, SvcName}). + +%% -------------------------------------------------------------------------- +%% # add_transport(SvcName, {Type, Opts}) +%% +%% Input: Type = connect | listen +%% +%% Output: {ok, Ref} | {error, Reason} +%% -------------------------------------------------------------------------- + +add_transport(SvcName, {T, Opts}) + when is_list(Opts), (T == connect orelse T == listen) -> + sync(SvcName, {add, SvcName, T, Opts}). + +%% -------------------------------------------------------------------------- +%% # remove_transport(SvcName, Pred) +%% +%% Input: Pred = arity 3 fun on transport ref, connect|listen and Opts, +%% returning true if the transport is to be removed, false if +%% not +%% | arity 2 fun on Ref and Opts only +%% | arity 1 fun on Opts only +%% | Opts matching all transports that have all of the specified +%% options +%% | Ref matching only the transport with this reference. +%% | {M,F,A} applied to Ref, connect|listen and Opts +%% | boolean() +%% +%% Output: ok | {error, Reason} +%% -------------------------------------------------------------------------- + +remove_transport(SvcName, Pred) -> + try + sync(SvcName, {remove, SvcName, pred(Pred)}) + catch + ?FAILURE(Reason) -> + {error, Reason} + end. + +pred(Pred) + when is_function(Pred, 3) -> + Pred; +pred(Pred) + when is_function(Pred, 2) -> + fun(R,_,O) -> Pred(R,O) end; +pred(Pred) + when is_function(Pred, 1) -> + fun(_,_,O) -> Pred(O) end; +pred(Opts) + when is_list(Opts) -> + fun(_,_,O) -> [] == Opts -- O end; +pred(Ref) + when is_reference(Ref) -> + fun(R,_,_) -> R == Ref end; +pred({M,F,A}) + when is_atom(M), is_atom(F), is_list(A) -> + fun(R,T,O) -> apply(M,F,[R,T,O|A]) end; +pred({Type, Pred}) -> %% backwards compatibility + P = pred(Pred), + fun(R,T,O) -> T == Type andalso P(R,T,O) end; +pred(B) + when is_boolean(B) -> + fun(_,_,_) -> B end; +pred(_) -> + ?THROW(pred). + +%% -------------------------------------------------------------------------- +%% # have_transport/2 +%% +%% Output: true | false +%% -------------------------------------------------------------------------- + +have_transport(SvcName, Ref) -> + member([{#transport{service = '$1', + ref = '$2', + _ = '_'}, + [{'andalso', {'=:=', '$1', {const, SvcName}}, + {'=:=', '$2', {const, Ref}}}], + [true]}]). + +%% -------------------------------------------------------------------------- +%% # lookup/1 +%% -------------------------------------------------------------------------- + +lookup(SvcName) -> + select([{#service{name = '$1', rec = '$2', options = '$3'}, + [{'=:=', '$1', {const, SvcName}}], + [{{'$1', '$2', '$3'}}]}, + {#transport{service = '$1', + ref = '$2', + type = '$3', + options = '$4'}, + [{'=:=', '$1', {const, SvcName}}], + [{{'$2', '$3', '$4'}}]}]). + +%% --------------------------------------------------------- +%% EXPORTED INTERNAL FUNCTIONS +%% --------------------------------------------------------- + +start_link() -> + ServerName = {local, ?SERVER}, + Module = ?MODULE, + Args = [], + Options = [{spawn_opt, diameter_lib:spawn_opts(server, [])}], + gen_server:start_link(ServerName, Module, Args, Options). + +state() -> + call(state). + +uptime() -> + call(uptime). + +%%% ---------------------------------------------------------- +%%% # init/1 +%%% ---------------------------------------------------------- + +init([]) -> + {ok, #state{}}. + +%%% ---------------------------------------------------------- +%%% # handle_call/2 +%%% ---------------------------------------------------------- + +handle_call(state, _, State) -> + {reply, State, State}; + +handle_call(uptime, _, #state{id = Time} = State) -> + {reply, diameter_lib:now_diff(Time), State}; + +handle_call(Req, From, State) -> + warning_msg("received unexpected request from ~p:~n~w", [From, Req]), + Reply = {error, {bad_request, Req}}, + {reply, Reply, State}. + +%%% ---------------------------------------------------------- +%%% # handle_cast/2 +%%% ---------------------------------------------------------- + +handle_cast(Msg, State) -> + warning_msg("received unexpected message:~n~w", [Msg]), + {noreply, State}. + +%%% ---------------------------------------------------------- +%%% # handle_info/2 +%%% ---------------------------------------------------------- + +%% A service process has died. This is most likely a consequence of +%% stop_service, in which case the restart will find no config for the +%% service and do nothing. The entry keyed on the monitor ref is only +%% removed as a result of the 'DOWN' notification however. +handle_info({'DOWN', MRef, process, _, Reason}, State) -> + [#monitor{service = SvcName} = T] = select([{#monitor{mref = MRef, + _ = '_'}, + [], + ['$_']}]), + queue_restart(Reason, SvcName), + delete_object(T), + {noreply, State}; + +handle_info({monitor, SvcName, Pid}, State) -> + monitor(Pid, SvcName), + {noreply, State}; + +handle_info({restart, SvcName}, State) -> + restart(SvcName), + {noreply, State}; + +handle_info(restart, State) -> + restart(), + {noreply, State}; + +handle_info(Info, State) -> + warning_msg("received unknown info:~n~w", [Info]), + {noreply, State}. + +%%-------------------------------------------------------------------- +%% # terminate/2 +%%-------------------------------------------------------------------- + +terminate(_Reason, _State) -> + ok. + +%%% ---------------------------------------------------------- +%%% # code_change/3 +%%% ---------------------------------------------------------- + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%% --------------------------------------------------------- +%% INTERNAL FUNCTIONS +%% --------------------------------------------------------- + +insert(T) -> + ets:insert(?TABLE, T). + +%% ?TABLE is a bag: check only for a service entry. +have_service(SvcName) -> + member([{#service{name = '$1', _ = '_'}, + [{'=:=', '$1', {const, SvcName}}], + [true]}]). + +member(MatchSpec) -> + '$end_of_table' =/= ets:select(?TABLE, MatchSpec, 1). + +delete_object(T) -> + ets:delete_object(?TABLE, T). + +delete(Key) -> + ets:delete(?TABLE, Key). + +select(MatchSpec) -> + ets:select(?TABLE, MatchSpec). + +select_delete(MatchSpec) -> + ets:select_delete(?TABLE, MatchSpec). + +%% sync/2 +%% +%% Interface functions used to be implemented as calls to ?SERVER but +%% now serialize things per service instead since stopping a service +%% can take time if the server doesn't answer DPR. A caller who wants +%% to stop multiple services can then improve performance by spawning +%% processes to stop them concurrently. + +sync(SvcName, T) -> + diameter_sync:call({?MODULE, SvcName}, + {?MODULE, sync, [T]}, + infinity, + infinity). + +%% sync/1 + +sync({restart, SvcName}) -> + have_service(SvcName) andalso start(SvcName); + +sync({start_service, SvcName, Opts}) -> + try + start(have_service(SvcName), SvcName, Opts) + catch + ?FAILURE(Reason) -> {error, Reason} + end; + +sync({stop_service, SvcName}) -> + stop(SvcName); + +sync({add, SvcName, Type, Opts}) -> + try + add(SvcName, Type, Opts) + catch + ?FAILURE(Reason) -> {error, Reason} + end; + +sync({remove, SvcName, Pred}) -> + remove(select([{#transport{service = '$1', _ = '_'}, + [{'=:=', '$1', {const, SvcName}}], + ['$_']}]), + SvcName, + Pred). + +%% start/3 + +start(true, _, _) -> + {error, already_started}; +start(false, SvcName, Opts) -> + insert(make_config(SvcName, Opts)), + start(SvcName). + +%% start/1 + +start(SvcName) -> + RC = diameter_service:start(SvcName), + startmon(SvcName, RC), + RC. + +startmon(SvcName, {ok, Pid}) -> + ?SERVER ! {monitor, SvcName, Pid}; +startmon(_, {error, _}) -> + ok. + +monitor(Pid, SvcName) -> + MRef = erlang:monitor(process, Pid), + insert(#monitor{mref = MRef, service = SvcName}). + +%% queue_restart/2 + +%% Service has gone down on monitor death. Note that all service-related +%% config is deleted. +queue_restart({shutdown, {monitor, _}}, SvcName) -> + delete(SvcName); + +%% Application shutdown: ignore. +queue_restart(shutdown, _) -> + ok; + +%% Or not. +queue_restart(_, SvcName) -> + erlang:send_after(?RESTART_SLEEP, self(), {restart, SvcName}). + +%% restart/1 + +restart(SvcName) -> + sync(SvcName, {restart, SvcName}). + +%% restart/0 +%% +%% Start anything configured as required. Bang 'restart' to the server +%% to kick things into gear manually. (Not that it should be required +%% but it's been useful for test.) + +restart() -> + MatchSpec = [{#service{name = '$1', _ = '_'}, + [], + ['$1']}], + lists:foreach(fun restart/1, select(MatchSpec)). + +%% stop/1 + +stop(SvcName) -> + %% If the call to the service returns error for any reason other + %% than the process not being alive then deleting the config from + %% under it will surely bring it down. + diameter_service:stop(SvcName), + %% Delete only the service entry, not everything keyed on the name, + select_delete([{#service{name = '$1', _ = '_'}, + [{'=:=', '$1', {const, SvcName}}], + [true]}]), + ok. +%% Note that a transport has to be removed for its statistics to be +%% deleted. + +%% add/3 + +add(SvcName, Type, Opts) -> + %% Ensure usable capabilities. diameter_service:merge_service/2 + %% depends on this. + lists:foreach(fun(Os) -> + is_list(Os) orelse ?THROW({capabilities, Os}), + ok = encode_CER(Os) + end, + [Os || {capabilities, Os} <- Opts, is_list(Os)]), + + Ref = make_ref(), + T = {Ref, Type, Opts}, + %% The call to the service returns error if the service isn't + %% started yet, which is harmless. The transport will be started + %% when the service is in that case. + case start_transport(SvcName, T) of + ok -> + insert(#transport{service = SvcName, + ref = Ref, + type = Type, + options = Opts}), + {ok, Ref}; + {error, _} = No -> + No + end. + +start_transport(SvcName, T) -> + case diameter_service:start_transport(SvcName, T) of + {ok, _Pid} -> + ok; + {error, no_service} -> + ok; + {error, _} = No -> + No + end. + +%% remove/3 + +remove(L, SvcName, Pred) -> + rm(SvcName, lists:filter(fun(#transport{ref = R, type = T, options = O}) -> + Pred(R,T,O) + end, + L)). + +rm(_, []) -> + ok; +rm(SvcName, L) -> + Refs = lists:map(fun(#transport{ref = R}) -> R end, L), + case stop_transport(SvcName, Refs) of + ok -> + lists:foreach(fun delete_object/1, L); + {error, _} = No -> + No + end. + +stop_transport(SvcName, Refs) -> + case diameter_service:stop_transport(SvcName, Refs) of + ok -> + ok; + {error, no_service} -> + ok; + {error, _} = No -> + No + end. + +%% make_config/2 + +make_config(SvcName, Opts) -> + Apps = init_apps(Opts), + [] == Apps andalso ?THROW(no_apps), + + %% Use the fact that diameter_caps has the same field names as CER. + Fields = diameter_gen_base_rfc3588:'#info-'(diameter_base_CER) -- ['AVP'], + + COpts = [T || {K,_} = T <- Opts, lists:member(K, Fields)], + Caps = make_caps(#diameter_caps{}, COpts), + + ok = encode_CER(COpts), + + Os = split(Opts, [{[fun erlang:is_boolean/1], false, share_peers}, + {[fun erlang:is_boolean/1], false, use_shared_peers}, + {[fun erlang:is_pid/1, false], false, monitor}]), + %% share_peers and use_shared_peers are currently undocumented. + + #service{name = SvcName, + rec = #diameter_service{applications = Apps, + capabilities = Caps}, + options = Os}. + +make_caps(Caps, Opts) -> + case diameter_capx:make_caps(Caps, Opts) of + {ok, T} -> + T; + {error, {Reason, _}} -> + ?THROW(Reason) + end. + +%% Validate types by encoding a CER. +encode_CER(Opts) -> + {ok, CER} = diameter_capx:build_CER(make_caps(?EXAMPLE_CAPS, Opts)), + + Hdr = #diameter_header{version = ?DIAMETER_VERSION, + end_to_end_id = 0, + hop_by_hop_id = 0}, + + try + diameter_codec:encode(?BASE, #diameter_packet{header = Hdr, + msg = CER}), + ok + catch + exit: Reason -> + ?THROW(Reason) + end. + +init_apps(Opts) -> + lists:foldl(fun app_acc/2, [], lists:reverse(Opts)). + +app_acc({application, Opts}, Acc) -> + is_list(Opts) orelse ?THROW({application, Opts}), + + [Dict, Mod] = get_opt([dictionary, module], Opts), + Alias = get_opt(alias, Opts, Dict), + ModS = get_opt(state, Opts, Alias), + M = get_opt(call_mutates_state, Opts, false), + A = get_opt(answer_errors, Opts, report), + [#diameter_app{alias = Alias, + dictionary = Dict, + id = cb(Dict, id), + module = init_mod(Mod), + init_state = ModS, + mutable = init_mutable(M), + answer_errors = init_answers(A)} + | Acc]; +app_acc(_, Acc) -> + Acc. + +init_mod(M) + when is_atom(M) -> + [M]; +init_mod([M|_] = L) + when is_atom(M) -> + L; +init_mod(M) -> + ?THROW({module, M}). + +init_mutable(M) + when M == true; + M == false -> + M; +init_mutable(M) -> + ?THROW({call_mutates_state, M}). + +init_answers(A) + when callback == A; + report == A; + discard == A -> + A; +init_answers(A) -> + ?THROW({answer_errors, A}). + +%% Get a single value at the specified key. +get_opt(Keys, List) + when is_list(Keys) -> + [get_opt(K, List) || K <- Keys]; +get_opt(Key, List) -> + case [V || {K,V} <- List, K == Key] of + [V] -> V; + _ -> ?THROW({arity, Key}) + end. + +%% Get an optional value at the specified key. +get_opt(Key, List, Def) -> + case [V || {K,V} <- List, K == Key] of + [] -> Def; + [V] -> V; + _ -> ?THROW({arity, Key}) + end. + +split(Opts, Defs) -> + [{K, value(D, Opts)} || {_,_,K} = D <- Defs]. + +value({Preds, Def, Key}, Opts) -> + V = get_opt(Key, Opts, Def), + lists:any(fun(P) -> pred(P,V) end, Preds) + orelse ?THROW({value, Key}), + V. + +pred(F, V) + when is_function(F) -> + F(V); +pred(T, V) -> + T == V. + +cb(M,F) -> + try M:F() of + V -> V + catch + E: Reason -> + ?THROW({callback, E, Reason, ?STACK}) + end. + +%% call/1 + +call(Request) -> + gen_server:call(?SERVER, Request, infinity). + +%% warning_msg/2 + +warning_msg(F, A) -> + ?diameter_warning("~p: " ++ F, [?MODULE | A]). |