diff options
Diffstat (limited to 'lib/diameter/src/base/diameter_service.erl')
| -rw-r--r-- | lib/diameter/src/base/diameter_service.erl | 1311 |
1 files changed, 897 insertions, 414 deletions
diff --git a/lib/diameter/src/base/diameter_service.erl b/lib/diameter/src/base/diameter_service.erl index 3dfdcee2b2..a4a0b80348 100644 --- a/lib/diameter/src/base/diameter_service.erl +++ b/lib/diameter/src/base/diameter_service.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2011. All Rights Reserved. +%% Copyright Ericsson AB 2010-2012. 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 @@ -43,8 +43,7 @@ subscriptions/0, services/0, services/1, - whois/1, - flush_stats/1]). + whois/1]). %% test/debug -export([call_module/3, @@ -65,13 +64,33 @@ -include_lib("diameter/include/diameter.hrl"). -include("diameter_internal.hrl"). +%% The states mirrored by peer_up/peer_down callbacks. -define(STATE_UP, up). -define(STATE_DOWN, down). +-type op_state() :: ?STATE_UP + | ?STATE_DOWN. + +%% The RFC 3539 watchdog states that are now maintained, albeit +%% along with the old up/down. okay = up, else down. +-define(WD_INITIAL, initial). +-define(WD_OKAY, okay). +-define(WD_SUSPECT, suspect). +-define(WD_DOWN, down). +-define(WD_REOPEN, reopen). + +-type wd_state() :: ?WD_INITIAL + | ?WD_OKAY + | ?WD_SUSPECT + | ?WD_DOWN + | ?WD_REOPEN. + -define(DEFAULT_TC, 30000). %% RFC 3588 ch 2.1 -define(DEFAULT_TIMEOUT, 5000). %% for outgoing requests -define(RESTART_TC, 1000). %% if restart was this recent +-define(RELAY, ?DIAMETER_DICT_RELAY). + %% Used to be able to swap this with anything else dict-like but now %% rely on the fact that a service's #state{} record does not change %% in storing in it ?STATE table and not always going through the @@ -88,6 +107,12 @@ %% process. -define(STATE_TABLE, ?MODULE). +%% The default sequence mask. +-define(NOMASK, {0,32}). + +%% The default restrict_connections. +-define(RESTRICT, nodes). + %% Workaround for dialyzer's lack of understanding of match specs. -type match(T) :: T | '_' | '$1' | '$2' | '$3' | '$4'. @@ -95,15 +120,18 @@ %% State of service gen_server. -record(state, {id = now(), - service_name, %% as passed to start_service/2, key in ?STATE_TABLE + service_name, %% as passed to start_service/2, key in ?STATE_TABLE service :: #diameter_service{}, - peerT = ets_new(peers) :: ets:tid(), %% #peer{} at start_fsm - connT = ets_new(conns) :: ets:tid(), %% #conn{} at connection_up - share_peers = false :: boolean(), %% broadcast peers to remote nodes? - use_shared_peers = false :: boolean(), %% use broadcasted peers? + peerT = ets_new(peers) :: ets:tid(),%% #peer{} at start_fsm + connT = ets_new(conns) :: ets:tid(),%% #conn{} at connection_up/reopen shared_peers = ?Dict:new(), %% Alias -> [{TPid, Caps}, ...] local_peers = ?Dict:new(), %% Alias -> [{TPid, Caps}, ...] - monitor = false :: false | pid()}). %% process to die with + monitor = false :: false | pid(), %% process to die with + options + :: [{sequence, diameter:sequence()} %% sequence mask + | {restrict_connections, diameter:restriction()} + | {share_peers, boolean()} %% broadcast peers to remote nodes? + | {use_shared_peers, boolean()}]}).%% use broadcasted peers? %% shared_peers reflects the peers broadcast from remote nodes. Note %% that the state term itself doesn't change, which is relevant for %% the stateless application callbacks since the state is retrieved @@ -111,18 +139,30 @@ %% service record is used to determine whether or not we need to call %% the process for a pick_peer callback. -%% Record representing a watchdog process. +%% Record representing a watchdog process as implemented by +%% diameter_watchdog. The term "peer" here is historical, made +%% especially confusing by the fact that a peer_ref() in the +%% documentation is the key of a #conn{} record, not a #peer{} record. +%% The name is also unfortunate given the meaning of peer in the +%% Diameter sense. -record(peer, {pid :: match(pid()), type :: match(connect | accept), ref :: match(reference()), %% key into diameter_config options :: match([diameter:transport_opt()]),%% from start_transport - op_state = ?STATE_DOWN :: match(?STATE_DOWN | ?STATE_UP), + op_state = {?STATE_DOWN, ?WD_INITIAL} + :: match(op_state() | {op_state(), wd_state()}), started = now(), %% at process start conn = false :: match(boolean() | pid())}). - %% true at accept, pid() at connection_up (connT key) - -%% Record representing a peer_fsm process. + %% true at accepted, pid() at connection_up or reopen + +%% Record representing a peer process as implemented by +%% diameter_peer_fsm. The term "conn" is historical. Despite the name +%% here, comments refer to watchdog and peer processes, that are keys +%% in #peer{} and #conn{} records respectively. To add to the +%% confusion, a #request.transport is a peer process = key in a +%% #conn{} record. The actual transport process (that the peer process +%% knows about and that has a transport connection) isn't seen here. -record(conn, {pid :: pid(), apps :: [{0..16#FFFFFFFF, diameter:app_alias()}], %% {Id, Alias} @@ -136,10 +176,9 @@ handler :: match(pid()), %% request process transport :: match(pid()), %% peer process caps :: match(#diameter_caps{}), - app :: match(diameter:app_alias()), %% #diameter_app.alias - dictionary :: match(module()), %% #diameter_app.dictionary - module :: match([module() | list()]), - %% #diameter_app.module + app :: match(diameter:app_alias()),%% #diameter_app.alias + dictionary :: match(module()), %% #diameter_app.dictionary + module :: match([module() | list()]), %% #diameter_app.module filter :: match(diameter:peer_filter()), packet :: match(#diameter_packet{})}). @@ -150,20 +189,6 @@ timeout = ?DEFAULT_TIMEOUT :: 0..16#FFFFFFFF, detach = false :: boolean()}). -%% Since RFC 3588 requires that a Diameter agent not modify End-to-End -%% Identifiers, the possibility of explicitly setting an End-to-End -%% Identifier would be needed to be able to implement an agent in -%% which one side of the communication is not implemented on top of -%% diameter. For example, Diameter being sent or received encapsulated -%% in some other protocol, or even another Diameter stack in a -%% non-Erlang environment. (Not that this is likely to be a normal -%% case.) -%% -%% The implemented solution is not an option but to respect any header -%% values set in a diameter_header record returned from a -%% prepare_request callback. A call to diameter:call/4 can communicate -%% values to the callback using the 'extra' option if so desired. - %%% --------------------------------------------------------------------------- %%% # start(SvcName) %%% --------------------------------------------------------------------------- @@ -216,20 +241,20 @@ stop_transport(SvcName, [_|_] = Refs) -> %%% --------------------------------------------------------------------------- info(SvcName, Item) -> - info_rc(call_service_by_name(SvcName, {info, Item})). - -info_rc({error, _}) -> - undefined; -info_rc(Info) -> - Info. + case find_state(SvcName) of + #state{} = S -> + service_info(Item, S); + false -> + undefined + end. %%% --------------------------------------------------------------------------- %%% # receive_message(TPid, Pkt, MessageData) %%% --------------------------------------------------------------------------- -%% Handle an incoming message in the watchdog process. This used to -%% come through the service process but this avoids that becoming a -%% bottleneck. +%% Handle an incoming Diameter message in the watchdog process. This +%% used to come through the service process but this avoids that +%% becoming a bottleneck. receive_message(TPid, Pkt, T) when is_pid(TPid) -> @@ -309,21 +334,39 @@ call_rc(_, _, Sent) -> %% In the process spawned for the outgoing request. call(SvcName, App, Msg, Opts, Caller) -> - c(ets:lookup(?STATE_TABLE, SvcName), App, Msg, Opts, Caller). + c(find_state(SvcName), App, Msg, Opts, Caller). -c([#state{service_name = SvcName} = S], App, Msg, Opts, Caller) -> +c(#state{service_name = Svc, options = [{_, Mask} | _]} = S, + App, + Msg, + Opts, + Caller) -> case find_transport(App, Msg, Opts, S) of {_,_,_} = T -> - send_request(T, Msg, Opts, Caller, SvcName); + send_request(T, Mask, Msg, Opts, Caller, Svc); false -> {error, no_connection}; {error, _} = No -> No end; -c([], _, _, _, _) -> +c(false, _, _, _, _) -> {error, no_service}. +%% find_state/1 + +find_state(SvcName) -> + fs(ets:lookup(?STATE_TABLE, SvcName)). + +fs([#state{} = S]) -> + S; + +fs([S]) -> %% inserted from old code + upgrade(S); + +fs([]) -> + false. + %% make_options/1 make_options(Options) -> @@ -388,15 +431,6 @@ whois(SvcName) -> undefined end. -%%% --------------------------------------------------------------------------- -%%% # flush_stats/1 -%%% -%%% Output: list of {{SvcName, Alias, Counter}, Value} -%%% --------------------------------------------------------------------------- - -flush_stats(TPid) -> - diameter_stats:flush(TPid). - %% =========================================================================== %% =========================================================================== @@ -428,6 +462,10 @@ i(_, false) -> %%% # handle_call(Req, From, State) %%% --------------------------------------------------------------------------- +handle_call(T, From, S) + when not is_record(S, state) -> + handle_call(T, From, upgrade(S)); + handle_call(state, _, S) -> {reply, S, S}; @@ -451,16 +489,25 @@ handle_call({pick_peer, Local, Remote, App}, _From, S) -> handle_call({call_module, AppMod, Req}, From, S) -> call_module(AppMod, Req, From, S); +%% Call from old code. handle_call({info, Item}, _From, S) -> {reply, service_info(Item, S), S}; handle_call(stop, _From, S) -> - shutdown(S), + shutdown(service, S), {stop, normal, ok, S}; %% The server currently isn't guaranteed to be dead when the caller %% gets the reply. We deal with this in the call to the server, %% stating a monitor that waits for DOWN before returning. +%% Watchdog is asking for the sequence mask. +handle_call(sequence, _From, #state{options = [{_, Mask} | _]} = S) -> + {reply, Mask, S}; + +%% Watchdog is asking for the nodes restriction. +handle_call(restriction, _From, #state{options = [_,_,_,{_,R} | _]} = S) -> + {reply, R, S}; + handle_call(Req, From, S) -> unexpected(handle_call, [Req, From], S), {reply, nok, S}. @@ -477,15 +524,16 @@ handle_cast(Req, S) -> %%% # handle_info(Req, State) %%% --------------------------------------------------------------------------- -handle_info(T,S) -> +handle_info(T, #state{} = S) -> case transition(T,S) of ok -> {noreply, S}; - #state{} = NS -> - {noreply, NS}; {stop, Reason} -> {stop, {shutdown, Reason}, S} - end. + end; + +handle_info(T, S) -> + handle_info(T, upgrade(S)). %% transition/2 @@ -496,15 +544,26 @@ transition({accepted, Pid, TPid}, S) -> %% Peer process has a new open connection. transition({connection_up, Pid, T}, S) -> - connection_up(Pid, T, S); + connection_up(Pid, T, S), + ok; + +%% Watchdog has a new connection that will be opened after DW[RA] +%% exchange. This message was added long after connection_up, to +%% communicate the information as soon as it's available. Leave +%% connection_up as is it for now, duplicated information and all. +transition({reopen, Pid, T}, S) -> + reopen(Pid, T, S), + ok; -%% Peer process has left state open. +%% Watchdog has left state OKAY. transition({connection_down, Pid}, S) -> - connection_down(Pid, S); + connection_down(Pid, S), + ok; -%% Peer process has returned to state open. +%% Watchdog has returned to state OKAY. transition({connection_up, Pid}, S) -> - connection_up(Pid, S); + connection_up(Pid, S), + ok; %% Accepting transport has lost connectivity. transition({close, Pid}, S) -> @@ -516,29 +575,61 @@ transition({reconnect, Pid}, S) -> reconnect(Pid, S), ok; +%% Watchdog is sending notification of a state transition. Note that +%% the connection_up/down messages are pre-date this message and are +%% still used. A watchdog message will follow these and communicate +%% the same state as was set in handling connection_up/down. +transition({watchdog, Pid, {TPid, From, To}}, #state{service_name = SvcName, + peerT = PeerT}) -> + #peer{ref = Ref, type = T, options = Opts, op_state = {OS,_}} + = P + = fetch(PeerT, Pid), + insert(PeerT, P#peer{op_state = {OS, To}}), + send_event(SvcName, {watchdog, Ref, TPid, {From, To}, {T, Opts}}), + ok; +%% Death of a watchdog process (#peer.pid) results in the removal of +%% it's peer and any associated conn record when 'DOWN' is received +%% (after this) but the states will be {?STATE_UP, ?WD_DOWN} for a +%% short time. (No real problem since ?WD_* is only used in +%% service_info.) We set ?WD_OKAY as a consequence of connection_up +%% since we know a watchdog is coming. We can't set anything at +%% connection_down since we don't know if the subsequent watchdog +%% message will be ?WD_DOWN or ?WD_SUSPECT. We don't (yet) set +%% ?STATE_* as a consequence of a watchdog message since this requires +%% changing some of the matching on ?STATE_*. +%% +%% Death of a peer process process (#conn.pid, #peer.conn) results in +%% connection_down followed by watchdog ?WD_DOWN. The latter doesn't +%% result in the conn record being deleted since 'DOWN' from death of +%% its watchdog doesn't (yet) deal with the record having been +%% removed. + %% Monitor process has died. Just die with a reason that tells %% diameter_config about the happening. If a cleaner shutdown is %% required then someone should stop us. transition({'DOWN', MRef, process, _, Reason}, #state{monitor = MRef}) -> {stop, {monitor, Reason}}; -%% Local peer process has died. +%% Local watchdog process has died. transition({'DOWN', _, process, Pid, Reason}, S) when node(Pid) == node() -> - peer_down(Pid, Reason, S); + peer_down(Pid, Reason, S), + ok; -%% Remote service wants to know about shared transports. +%% Remote service wants to know about shared peers. transition({service, Pid}, S) -> share_peers(Pid, S), ok; %% Remote service is communicating a shared peer. transition({peer, TPid, Aliases, Caps}, S) -> - remote_peer_up(TPid, Aliases, Caps, S); + remote_peer_up(TPid, Aliases, Caps, S), + ok; %% Remote peer process has died. transition({'DOWN', _, process, TPid, _}, S) -> - remote_peer_down(TPid, S); + remote_peer_down(TPid, S), + ok; %% Restart after tc expiry. transition({tc_timeout, T}, S) -> @@ -552,18 +643,48 @@ transition({failover, TRef, Seqs}, S) -> failover(TRef, Seqs, S), ok; +%% Ensure upgraded state is stored in state table. +transition(upgrade, _) -> + ok; + transition(Req, S) -> unexpected(handle_info, [Req], S), ok. +%% upgrade/1 + +upgrade({state, Id, Svc, Name, Svc, PT, CT, SB, UB, SD, LD, MPid}) -> + S = #state{id = Id, + service_name = Name, + service = Svc, + peerT = PT, + connT = CT, + shared_peers = SD, + local_peers = LD, + monitor = MPid, + options = [{sequence, ?NOMASK}, + {share_peers, SB}, + {use_shared_peers, UB}, + {restrict_connections, ?RESTRICT}]}, + upgrade_insert(S), + S. + +upgrade_insert(#state{service = #diameter_service{pid = Pid}} = S) -> + if Pid == self() -> + ets:insert(?STATE_TABLE, S); + true -> + Pid ! upgrade + end. + %%% --------------------------------------------------------------------------- %%% # terminate(Reason, State) %%% --------------------------------------------------------------------------- terminate(Reason, #state{service_name = Name} = S) -> + send_event(Name, stop), ets:delete(?STATE_TABLE, Name), shutdown == Reason %% application shutdown - andalso shutdown(S). + andalso shutdown(application, S). %%% --------------------------------------------------------------------------- %%% # code_change(FromVsn, State, Extra) @@ -646,41 +767,49 @@ mod_state(Alias, ModS) -> %%% # shutdown/2 %%% --------------------------------------------------------------------------- -shutdown(Refs, #state{peerT = PeerT}) -> - ets:foldl(fun(P,ok) -> s(P, Refs), ok end, ok, PeerT). - -s(#peer{ref = Ref, pid = Pid}, Refs) -> - s(lists:member(Ref, Refs), Pid); - -s(true, Pid) -> - Pid ! {shutdown, self()}; %% 'DOWN' will cleanup as usual -s(false, _) -> - ok. +%% remove_transport: ask watchdogs to terminate their transport. +shutdown(Refs, #state{peerT = PeerT}) + when is_list(Refs) -> + ets:foldl(fun(P,ok) -> sp(P, Refs), ok end, ok, PeerT); -%%% --------------------------------------------------------------------------- -%%% # shutdown/1 -%%% --------------------------------------------------------------------------- - -shutdown(#state{peerT = PeerT}) -> +%% application/service shutdown: ask transports to terminate themselves. +shutdown(Reason, #state{peerT = PeerT}) -> %% A transport might not be alive to receive the shutdown request %% but give those that are a chance to shutdown gracefully. - wait(fun st/2, PeerT), + shutdown(conn, Reason, PeerT), %% Kill the watchdogs explicitly in case there was no transport. - wait(fun sw/2, PeerT). + shutdown(peer, Reason, PeerT). -wait(Fun, T) -> - diameter_lib:wait(ets:foldl(Fun, [], T)). +%% sp/2 -st(#peer{conn = B}, Acc) - when is_boolean(B) -> - Acc; -st(#peer{conn = Pid}, Acc) -> - Pid ! shutdown, - [Pid | Acc]. +sp(#peer{ref = Ref, pid = Pid}, Refs) -> + lists:member(Ref, Refs) + andalso (Pid ! {shutdown, self()}). %% 'DOWN' cleans up + +%% shutdown/3 + +shutdown(Who, Reason, T) -> + diameter_lib:wait(ets:foldl(fun(X,A) -> shutdown(Who, X, Reason, A) end, + [], + T)). + +shutdown(conn = Who, #peer{op_state = {OS,_}} = P, Reason, Acc) -> + shutdown(Who, P#peer{op_state = OS}, Reason, Acc); -sw(#peer{pid = Pid}, Acc) -> +shutdown(conn, + #peer{pid = Pid, op_state = ?STATE_UP, conn = TPid}, + Reason, + Acc) -> + TPid ! {shutdown, Pid, Reason}, + [TPid | Acc]; + +shutdown(peer, #peer{pid = Pid}, _Reason, Acc) + when is_pid(Pid) -> exit(Pid, shutdown), - [Pid | Acc]. + [Pid | Acc]; + +shutdown(_, #peer{}, _, Acc) -> + Acc. %%% --------------------------------------------------------------------------- %%% # call_service/2 @@ -733,6 +862,7 @@ i(SvcName) -> lists:foreach(fun(T) -> start_fsm(T,S) end, CL), init_shared(S), + send_event(SvcName, start), S. cfg_acc({SvcName, #diameter_service{applications = Apps} = Rec, Opts}, @@ -740,9 +870,8 @@ cfg_acc({SvcName, #diameter_service{applications = Apps} = Rec, Opts}, lists:foreach(fun init_mod/1, Apps), S = #state{service_name = SvcName, service = Rec#diameter_service{pid = self()}, - share_peers = get_value(share_peers, Opts), - use_shared_peers = get_value(use_shared_peers, Opts), - monitor = mref(get_value(monitor, Opts))}, + monitor = mref(get_value(monitor, Opts)), + options = service_options(Opts)}, {S, Acc}; cfg_acc({_Ref, Type, _Opts} = T, {S, Acc}) @@ -750,15 +879,24 @@ cfg_acc({_Ref, Type, _Opts} = T, {S, Acc}) Type == listen -> {S, [T | Acc]}. +service_options(Opts) -> + [{sequence, proplists:get_value(sequence, Opts, ?NOMASK)}, + {share_peers, get_value(share_peers, Opts)}, + {use_shared_peers, get_value(use_shared_peers, Opts)}, + {restrict_connections, proplists:get_value(restrict_connections, + Opts, + ?RESTRICT)}]. +%% The order of options is significant since we match against the list. + mref(false = No) -> No; mref(P) -> erlang:monitor(process, P). -init_shared(#state{use_shared_peers = true, +init_shared(#state{options = [_, _, {_, true} | _], service_name = Svc}) -> diameter_peer:notify(Svc, {service, self()}); -init_shared(#state{use_shared_peers = false}) -> +init_shared(#state{options = [_, _, {_, false} | _]}) -> ok. init_mod(#diameter_app{alias = Alias, @@ -821,9 +959,8 @@ start(Ref, Type, Opts, #state{peerT = PeerT, Pid. %% Note that the service record passed into the watchdog is the merged -%% record so that each watchdog (and peer_fsm) may get a different -%% record. This record is what is passed back into application -%% callbacks. +%% record so that each watchdog may get a different record. This +%% record is what is passed back into application callbacks. s(Type, Ref, T) -> case diameter_watchdog:start({Type, Ref}, T) of @@ -874,20 +1011,25 @@ accepted(Pid, _TPid, #state{peerT = PeerT} = S) -> #peer{ref = Ref, type = accept = T, conn = false, options = Opts} = P = fetch(PeerT, Pid), - insert(PeerT, P#peer{conn = true}), %% mark replacement transport started - start(Ref, T, Opts, S). %% start new peer + insert(PeerT, P#peer{conn = true}), %% mark replacement as started + start(Ref, T, Opts, S). %% start new watchdog fetch(Tid, Key) -> [T] = ets:lookup(Tid, Key), - T. + case T of + #peer{op_state = ?STATE_UP} = P -> + P#peer{op_state = {?STATE_UP, ?WD_OKAY}}; + #peer{op_state = ?STATE_DOWN} = P -> + P#peer{op_state = {?STATE_DOWN, ?WD_DOWN}}; + _ -> + T + end. %%% --------------------------------------------------------------------------- %%% # connection_up/3 -%%% -%%% Output: #state{} %%% --------------------------------------------------------------------------- -%% Peer process has reached the open state. +%% Watchdog process has reached state OKAY. connection_up(Pid, {TPid, {Caps, SApps, Pkt}}, #state{peerT = PeerT, connT = ConnT} @@ -902,9 +1044,29 @@ connection_up(Pid, {TPid, {Caps, SApps, Pkt}}, #state{peerT = PeerT, connection_up([Pkt], P#peer{conn = TPid}, C, S). %%% --------------------------------------------------------------------------- +%%% # reopen/3 +%%% --------------------------------------------------------------------------- + +%% Note that this connection_up/3 rewrites the same #conn{} now +%% written here. Both do so in case reopen has not happened in old +%% code. + +reopen(Pid, {TPid, {Caps, SApps, _Pkt}}, #state{peerT = PeerT, + connT = ConnT}) -> + P = fetch(PeerT, Pid), + C = #conn{pid = TPid, + apps = SApps, + caps = Caps, + peer = Pid}, + + insert(ConnT, C), + #peer{op_state = {?STATE_DOWN, _}} + = P, + insert(PeerT, P#peer{op_state = {?STATE_DOWN, ?WD_REOPEN}, + conn = TPid}). + +%%% --------------------------------------------------------------------------- %%% # connection_up/2 -%%% -%%% Output: #state{} %%% --------------------------------------------------------------------------- %% Peer process has transitioned back into the open state. Note that there @@ -925,18 +1087,16 @@ connection_up(T, P, C, #state{peerT = PeerT, service = #diameter_service{applications = Apps}} = S) -> - #peer{conn = TPid, op_state = ?STATE_DOWN} + #peer{conn = TPid, op_state = {?STATE_DOWN, _}} = P, #conn{apps = SApps, caps = Caps} = C, - insert(PeerT, P#peer{op_state = ?STATE_UP}), + insert(PeerT, P#peer{op_state = {?STATE_UP, ?WD_OKAY}}), request_peer_up(TPid), - report_status(up, P, C, S, T), - S#state{local_peers = insert_local_peer(SApps, - {{TPid, Caps}, {SvcName, Apps}}, - LDict)}. + insert_local_peer(SApps, {{TPid, Caps}, {SvcName, Apps}}, LDict), + report_status(up, P, C, S, T). insert_local_peer(SApps, T, LDict) -> lists:foldl(fun(A,D) -> ilp(A, T, D) end, LDict, SApps). @@ -945,56 +1105,62 @@ ilp({Id, Alias}, {TC, SA}, LDict) -> init_conn(Id, Alias, TC, SA), ?Dict:append(Alias, TC, LDict). -init_conn(Id, Alias, TC, {SvcName, Apps}) -> +init_conn(Id, Alias, {TPid, _} = TC, {SvcName, Apps}) -> #diameter_app{module = ModX, id = Id} %% assert = find_app(Alias, Apps), - peer_cb({ModX, peer_up, [SvcName, TC]}, Alias). + peer_cb({ModX, peer_up, [SvcName, TC]}, Alias) + orelse exit(TPid, kill). %% fake transport failure + +%% find_app/2 find_app(Alias, Apps) -> - lists:keyfind(Alias, #diameter_app.alias, Apps). + case lists:keyfind(Alias, #diameter_app.alias, Apps) of + #diameter_app{options = E} = A when is_atom(E) -> %% upgrade + A#diameter_app{options = [{answer_errors, E}]}; + A -> + A + end. -%% A failing peer callback brings down the service. In the case of -%% peer_up we could just kill the transport and emit an error but for -%% peer_down we have no way to cleanup any state change that peer_up -%% may have introduced. +%% Don't bring down the service (and all associated connections) +%% regardless of what happens. peer_cb(MFA, Alias) -> try state_cb(MFA, Alias) of ModS -> - mod_state(Alias, ModS) + mod_state(Alias, ModS), + true catch - E: Reason -> - ?ERROR({E, Reason, MFA, ?STACK}) + E:R -> + diameter_lib:error_report({failure, {E, R, Alias, ?STACK}}, MFA), + false end. %%% --------------------------------------------------------------------------- %%% # connection_down/2 -%%% -%%% Output: #state{} %%% --------------------------------------------------------------------------- -%% Peer process has transitioned out of the open state. +%% Watchdog has transitioned out of state OKAY. connection_down(Pid, #state{peerT = PeerT, connT = ConnT} = S) -> - #peer{op_state = ?STATE_UP, %% assert + #peer{op_state = {?STATE_UP, WS}, %% assert conn = TPid} = P = fetch(PeerT, Pid), C = fetch(ConnT, TPid), - insert(PeerT, P#peer{op_state = ?STATE_DOWN}), + insert(PeerT, P#peer{op_state = {?STATE_DOWN, WS}}), connection_down(P,C,S). %% connection_down/3 -connection_down(#peer{op_state = ?STATE_DOWN}, _, S) -> - S; +connection_down(#peer{op_state = {?STATE_DOWN, _}}, _, _) -> + ok; connection_down(#peer{conn = TPid, - op_state = ?STATE_UP} + op_state = {?STATE_UP, _}} = P, #conn{caps = Caps, apps = SApps} @@ -1004,12 +1170,8 @@ connection_down(#peer{conn = TPid, local_peers = LDict} = S) -> report_status(down, P, C, S, []), - NewS = S#state{local_peers - = remove_local_peer(SApps, - {{TPid, Caps}, {SvcName, Apps}}, - LDict)}, - request_peer_down(TPid, NewS), - NewS. + remove_local_peer(SApps, {{TPid, Caps}, {SvcName, Apps}}, LDict), + request_peer_down(TPid, S). remove_local_peer(SApps, T, LDict) -> lists:foldl(fun(A,D) -> rlp(A, T, D) end, LDict, SApps). @@ -1028,11 +1190,9 @@ down_conn(Id, Alias, TC, {SvcName, Apps}) -> %%% --------------------------------------------------------------------------- %%% # peer_down/3 -%%% -%%% Output: #state{} %%% --------------------------------------------------------------------------- -%% Peer process has died. +%% Watchdog process has died. peer_down(Pid, Reason, #state{peerT = PeerT} = S) -> P = fetch(PeerT, Pid), @@ -1043,7 +1203,7 @@ peer_down(Pid, Reason, #state{peerT = PeerT} = S) -> %% Send an event at connection establishment failure. closed({shutdown, {close, _TPid, Reason}}, - #peer{op_state = ?STATE_DOWN, + #peer{op_state = {?STATE_DOWN, _}, ref = Ref, type = Type, options = Opts}, @@ -1052,12 +1212,12 @@ closed({shutdown, {close, _TPid, Reason}}, closed(_, _, _) -> ok. -%% The peer has never come up ... -peer_down(#peer{conn = B}, S) +%% The watchdog has never reached OKAY ... +peer_down(#peer{conn = B}, _) when is_boolean(B) -> - S; + ok; -%% ... or it has. +%% ... or maybe it has. peer_down(#peer{conn = TPid} = P, #state{connT = ConnT} = S) -> #conn{} = C = fetch(ConnT, TPid), ets:delete_object(ConnT, C), @@ -1085,7 +1245,7 @@ restart(#peer{ref = Ref, started = Time}) -> {Time, {Ref, T, Opts}}; -%% ... or it has: a replacement transport has already been spawned. +%% ... or it has: a replacement has already been spawned. restart(#peer{type = accept}) -> false. @@ -1111,8 +1271,8 @@ default_tc(connect, Opts) -> default_tc(accept, _) -> 0. -%% Bound tc below if the peer was restarted recently to avoid -%% continuous in case of faulty config or other problems. +%% Bound tc below if the watchdog was restarted recently to avoid +%% continuous restarted in case of faulty config or other problems. tc(Time, Tc) -> choose(Tc > ?RESTART_TC orelse timer:now_diff(now(), Time) > 1000*?RESTART_TC, @@ -1234,7 +1394,7 @@ cm([_,_|_], _, _, _) -> multiple. %%% --------------------------------------------------------------------------- -%%% # send_request/5 +%%% # send_request/6 %%% --------------------------------------------------------------------------- %% Send an outgoing request in its dedicated process. @@ -1247,71 +1407,89 @@ cm([_,_|_], _, _, _) -> %% The mod field of the #diameter_app{} here includes any extra %% arguments passed to diameter:call/2. -send_request({TPid, Caps, App}, Msg, Opts, Caller, SvcName) -> +send_request({TPid, Caps, App} = T, Mask, Msg, Opts, Caller, SvcName) -> #diameter_app{module = ModX} = App, - Pkt = make_request_packet(Msg), - - case cb(ModX, prepare_request, [Pkt, SvcName, {TPid, Caps}]) of - {send, P} -> - send_request(make_request_packet(P, Pkt), - TPid, - Caps, - App, - Opts, - Caller, - SvcName); - {discard, Reason} -> - {error, Reason}; - discard -> - {error, discarded}; - T -> - ?ERROR({invalid_return, prepare_request, App, T}) - end. + Pkt = make_prepare_packet(Mask, Msg), + + send_req(cb(ModX, prepare_request, [Pkt, SvcName, {TPid, Caps}]), + Pkt, + T, + Opts, + Caller, + SvcName, + []). + +send_req({send, P}, Pkt, T, Opts, Caller, SvcName, Fs) -> + send_req(make_request_packet(P, Pkt), T, Opts, Caller, SvcName, Fs); + +send_req({discard, Reason} , _, _, _, _, _, _) -> + {error, Reason}; + +send_req(discard, _, _, _, _, _, _) -> + {error, discarded}; -%% make_request_packet/1 +send_req({eval_packet, RC, F}, Pkt, T, Opts, Caller, SvcName, Fs) -> + send_req(RC, Pkt, T, Opts, Caller, SvcName, [F|Fs]); + +send_req(E, _, {_, _, App}, _, _, _, _) -> + ?ERROR({invalid_return, prepare_request, App, E}). + +%% make_prepare_packet/2 %% %% Turn an outgoing request as passed to call/4 into a diameter_packet %% record in preparation for a prepare_request callback. -make_request_packet(Bin) +make_prepare_packet(_, Bin) when is_binary(Bin) -> #diameter_packet{header = diameter_codec:decode_header(Bin), bin = Bin}; -make_request_packet(#diameter_packet{msg = [#diameter_header{} = Hdr | Avps]} - = Pkt) -> - Pkt#diameter_packet{msg = [make_request_header(Hdr) | Avps]}; +make_prepare_packet(Mask, #diameter_packet{msg = [#diameter_header{} = Hdr + | Avps]} + = Pkt) -> + Pkt#diameter_packet{msg = [make_prepare_header(Mask, Hdr) | Avps]}; -make_request_packet(#diameter_packet{header = Hdr} = Pkt) -> - Pkt#diameter_packet{header = make_request_header(Hdr)}; +make_prepare_packet(Mask, #diameter_packet{header = Hdr} = Pkt) -> + Pkt#diameter_packet{header = make_prepare_header(Mask, Hdr)}; -make_request_packet(Msg) -> - make_request_packet(#diameter_packet{msg = Msg}). +make_prepare_packet(Mask, Msg) -> + make_prepare_packet(Mask, #diameter_packet{msg = Msg}). -%% make_request_header/1 +%% make_prepare_header/1 + +make_prepare_header(Mask, undefined) -> + Seq = diameter_session:sequence(Mask), + make_prepare_header(#diameter_header{end_to_end_id = Seq, + hop_by_hop_id = Seq}); -make_request_header(undefined) -> - Seq = diameter_session:sequence(), - make_request_header(#diameter_header{end_to_end_id = Seq, +make_prepare_header(Mask, #diameter_header{end_to_end_id = undefined, + hop_by_hop_id = undefined}) -> + Seq = diameter_session:sequence(Mask), + make_prepare_header(#diameter_header{end_to_end_id = Seq, hop_by_hop_id = Seq}); -make_request_header(#diameter_header{version = undefined} = Hdr) -> - make_request_header(Hdr#diameter_header{version = ?DIAMETER_VERSION}); +make_prepare_header(Mask, #diameter_header{end_to_end_id = undefined} = H) -> + Seq = diameter_session:sequence(Mask), + make_prepare_header(H#diameter_header{end_to_end_id = Seq}); -make_request_header(#diameter_header{end_to_end_id = undefined} = H) -> - Seq = diameter_session:sequence(), - make_request_header(H#diameter_header{end_to_end_id = Seq}); +make_prepare_header(Mask, #diameter_header{hop_by_hop_id = undefined} = H) -> + Seq = diameter_session:sequence(Mask), + make_prepare_header(H#diameter_header{hop_by_hop_id = Seq}); -make_request_header(#diameter_header{hop_by_hop_id = undefined} = H) -> - Seq = diameter_session:sequence(), - make_request_header(H#diameter_header{hop_by_hop_id = Seq}); +make_prepare_header(_, Hdr) -> + make_prepare_header(Hdr). -make_request_header(#diameter_header{} = Hdr) -> +%% make_prepare_header/1 + +make_prepare_header(#diameter_header{version = undefined} = Hdr) -> + make_prepare_header(Hdr#diameter_header{version = ?DIAMETER_VERSION}); + +make_prepare_header(#diameter_header{} = Hdr) -> Hdr; -make_request_header(T) -> +make_prepare_header(T) -> ?ERROR({invalid_header, T}). %% make_request_packet/2 @@ -1321,7 +1499,7 @@ make_request_header(T) -> make_request_packet(Bin, _) when is_binary(Bin) -> - make_request_packet(Bin); + make_prepare_packet(false, Bin); make_request_packet(#diameter_packet{msg = [#diameter_header{} | _]} = Pkt, @@ -1333,7 +1511,7 @@ make_request_packet(#diameter_packet{msg = [#diameter_header{} | _]} %% This is primarily so that the end to end and hop by hop identifiers %% are retained. make_request_packet(#diameter_packet{header = Hdr} = Pkt, - #diameter_packet{header = Hdr0}) -> + #diameter_packet{header = Hdr0}) -> Pkt#diameter_packet{header = fold_record(Hdr0, Hdr)}; make_request_packet(Msg, Pkt) -> @@ -1346,16 +1524,16 @@ fold_record(undefined, R) -> fold_record(Rec, R) -> diameter_lib:fold_tuple(2, Rec, R). -%% send_request/7 +%% send_req/6 -send_request(Pkt, TPid, Caps, App, Opts, Caller, SvcName) -> +send_req(Pkt, {TPid, Caps, App}, Opts, Caller, SvcName, Fs) -> #diameter_app{alias = Alias, dictionary = Dict, module = ModX, - answer_errors = AE} + options = [{answer_errors, AE} | _]} = App, - EPkt = encode(Dict, Pkt), + EPkt = encode(Dict, Pkt, Fs), #options{filter = Filter, timeout = Timeout} @@ -1436,6 +1614,13 @@ msg(#diameter_packet{msg = undefined, bin = Bin}) -> msg(#diameter_packet{msg = Msg}) -> Msg. +%% encode/3 + +encode(Dict, Pkt, Fs) -> + P = encode(Dict, Pkt), + eval_packet(P, Fs), + P. + %% encode/2 %% Note that prepare_request can return a diameter_packet containing @@ -1517,38 +1702,47 @@ send(Pid, Pkt) -> %% retransmit/4 -retransmit({TPid, Caps, #diameter_app{alias = Alias} = App}, - #request{app = Alias, - packet = Pkt} +retransmit({TPid, Caps, #diameter_app{alias = Alias} = App} = T, + #request{app = Alias, packet = Pkt} = Req, SvcName, Timeout) -> have_request(Pkt, TPid) %% Don't failover to a peer we've andalso ?THROW(timeout), %% already sent to. - case cb(App, prepare_retransmit, [Pkt, SvcName, {TPid, Caps}]) of - {send, P} -> - retransmit(make_request_packet(P, Pkt), TPid, Caps, Req, Timeout); - {discard, Reason} -> - ?THROW(Reason); - discard -> - ?THROW(discarded); - T -> - ?ERROR({invalid_return, prepare_retransmit, App, T}) - end. + resend_req(cb(App, prepare_retransmit, [Pkt, SvcName, {TPid, Caps}]), + T, + Req, + Timeout, + []). + +resend_req({send, P}, T, #request{packet = Pkt} = Req, Timeout, Fs) -> + retransmit(make_request_packet(P, Pkt), T, Req, Timeout, Fs); + +resend_req({discard, Reason}, _, _, _, _) -> + ?THROW(Reason); + +resend_req(discard, _, _, _, _) -> + ?THROW(discarded); -%% retransmit/5 +resend_req({eval_packet, RC, F}, T, Req, Timeout, Fs) -> + resend_req(RC, T, Req, Timeout, [F|Fs]); + +resend_req(T, {_, _, App}, _, _, _) -> + ?ERROR({invalid_return, prepare_retransmit, App, T}). -retransmit(Pkt, TPid, Caps, #request{dictionary = Dict} = Req, Timeout) -> - EPkt = encode(Dict, Pkt), +%% retransmit/6 - NewReq = Req#request{transport = TPid, - packet = Pkt, - caps = Caps}, +retransmit(Pkt, {TPid, Caps, _}, #request{dictionary = D} = Req0, Tmo, Fs) -> + EPkt = encode(D, Pkt, Fs), - ?LOG(retransmission, NewReq), - TRef = send_request(TPid, EPkt, NewReq, Timeout), - {TRef, NewReq}. + Req = Req0#request{transport = TPid, + packet = Pkt, + caps = Caps}, + + ?LOG(retransmission, Req), + TRef = send_request(TPid, EPkt, Req, Tmo), + {TRef, Req}. %% store_request/4 @@ -1620,10 +1814,13 @@ request_peer_down(TPid, S) -> %%% recv_request/3 %%% --------------------------------------------------------------------------- -recv_request(TPid, Pkt, {ConnT, SvcName, Apps}) -> +recv_request(TPid, Pkt, {ConnT, SvcName, Apps}) -> %% upgrade + recv_request(TPid, Pkt, {ConnT, SvcName, Apps, ?NOMASK}); + +recv_request(TPid, Pkt, {ConnT, SvcName, Apps, Mask}) -> try ets:lookup(ConnT, TPid) of [C] -> - recv_request(C, TPid, Pkt, SvcName, Apps); + recv_request(C, TPid, Pkt, SvcName, Apps, Mask); [] -> %% transport has gone down ok catch @@ -1633,7 +1830,12 @@ recv_request(TPid, Pkt, {ConnT, SvcName, Apps}) -> %% recv_request/5 -recv_request(#conn{apps = SApps, caps = Caps}, TPid, Pkt, SvcName, Apps) -> +recv_request(#conn{apps = SApps, caps = Caps}, + TPid, + Pkt, + SvcName, + Apps, + Mask) -> #diameter_caps{origin_host = {OH,_}, origin_realm = {OR,_}} = Caps, @@ -1645,6 +1847,7 @@ recv_request(#conn{apps = SApps, caps = Caps}, TPid, Pkt, SvcName, Apps) -> {SvcName, OH, OR}, TPid, Apps, + Mask, Caps, Pkt). @@ -1670,20 +1873,24 @@ keyfind([Key | Rest], Pos, L) -> T end. -%% recv_request/6 +%% recv_request/7 -recv_request({Id, Alias}, T, TPid, Apps, Caps, Pkt) -> +recv_request({Id, Alias}, T, TPid, Apps, Mask, Caps, Pkt) -> #diameter_app{dictionary = Dict} = A = find_app(Alias, Apps), - recv_request(T, {TPid, Caps}, A, diameter_codec:decode(Id, Dict, Pkt)); + recv_request(T, + {TPid, Caps}, + A, + Mask, + diameter_codec:decode(Id, Dict, Pkt)); %% Note that the decode is different depending on whether or not Id is %% ?APP_ID_RELAY. %% DIAMETER_APPLICATION_UNSUPPORTED 3007 %% A request was sent for an application that is not supported. -recv_request(false, T, TPid, _, _, Pkt) -> +recv_request(false, T, TPid, _, _, _, Pkt) -> As = collect_avps(Pkt), protocol_error(3007, T, TPid, Pkt#diameter_packet{avps = As}). @@ -1695,7 +1902,7 @@ collect_avps(Pkt) -> As end. -%% recv_request/4 +%% recv_request/5 %% Wrong number of bits somewhere in the message: reply. %% @@ -1704,7 +1911,7 @@ collect_avps(Pkt) -> %% set to an unrecognized value, or that is inconsistent with the %% AVP's definition. %% -recv_request(T, {TPid, _}, _, #diameter_packet{errors = [Bs | _]} = Pkt) +recv_request(T, {TPid, _}, _, _, #diameter_packet{errors = [Bs | _]} = Pkt) when is_bitstring(Bs) -> protocol_error(3009, T, TPid, Pkt); @@ -1719,6 +1926,7 @@ recv_request(T, {TPid, _}, _, #diameter_packet{errors = [Bs | _]} = Pkt) recv_request(T, {TPid, _}, #diameter_app{id = Id}, + _, #diameter_packet{header = #diameter_header{is_proxiable = P}, msg = M} = Pkt) @@ -1736,6 +1944,7 @@ recv_request(T, recv_request(T, {TPid, _}, _, + _, #diameter_packet{header = #diameter_header{is_error = true}} = Pkt) -> protocol_error(3008, T, TPid, Pkt); @@ -1744,14 +1953,20 @@ recv_request(T, %% in the relay application. Don't distinguish between the two since %% each application has its own callback config. That is, the user can %% easily distinguish between the two cases. -recv_request(T, TC, App, Pkt) -> - request_cb(T, TC, App, examine(Pkt)). +recv_request(T, TC, App, Mask, Pkt) -> + request_cb(T, TC, App, Mask, examine(Pkt)). %% Note that there may still be errors but these aren't protocol %% (3xxx) errors that lead to an answer-message. -request_cb({SvcName, _OH, _OR} = T, TC, App, Pkt) -> - request_cb(cb(App, handle_request, [Pkt, SvcName, TC]), App, T, TC, Pkt). +request_cb({SvcName, _OH, _OR} = T, TC, App, Mask, Pkt) -> + request_cb(cb(App, handle_request, [Pkt, SvcName, TC]), + App, + Mask, + T, + TC, + [], + Pkt). %% examine/1 %% @@ -1771,7 +1986,7 @@ examine(#diameter_packet{errors = Es} = Pkt) -> Pkt#diameter_packet{errors = [5011 | Es]}. %% It's odd/unfortunate that this isn't a protocol error. -%% request_cb/5 +%% request_cb/7 %% A reply may be an answer-message, constructed either here or by %% the handle_request callback. The header from the incoming request @@ -1781,21 +1996,23 @@ examine(#diameter_packet{errors = Es} = Pkt) -> request_cb({reply, Ans}, #diameter_app{dictionary = Dict}, _, + _, {TPid, _}, + Fs, Pkt) -> - reply(Ans, Dict, TPid, Pkt); + reply(Ans, Dict, TPid, Fs, Pkt); %% An 3xxx result code, for which the E-bit is set in the header. -request_cb({protocol_error, RC}, _, T, {TPid, _}, Pkt) +request_cb({protocol_error, RC}, _, _, T, {TPid, _}, Fs, Pkt) when 3000 =< RC, RC < 4000 -> - protocol_error(RC, T, TPid, Pkt); + protocol_error(RC, T, TPid, Fs, Pkt); %% RFC 3588 says we must reply 3001 to anything unrecognized or %% unsupported. 'noreply' is undocumented (and inappropriately named) %% backwards compatibility for this, protocol_error the documented %% alternative. -request_cb(noreply, _, T, {TPid, _}, Pkt) -> - protocol_error(3001, T, TPid, Pkt); +request_cb(noreply, _, _, T, {TPid, _}, Fs, Pkt) -> + protocol_error(3001, T, TPid, Fs, Pkt); %% Relay a request to another peer. This is equivalent to doing an %% explicit call/4 with the message in question except that (1) a loop @@ -1815,38 +2032,51 @@ request_cb(noreply, _, T, {TPid, _}, Pkt) -> request_cb({A, Opts}, #diameter_app{id = Id} = App, + Mask, T, TC, + Fs, Pkt) when A == relay, Id == ?APP_ID_RELAY; A == proxy, Id /= ?APP_ID_RELAY; A == resend -> - resend(Opts, App, T, TC, Pkt); + resend(Opts, App, Mask, T, TC, Fs, Pkt); -request_cb(discard, _, _, _, _) -> +request_cb(discard, _, _, _, _, _, _) -> ok; -request_cb({eval, RC, F}, App, T, TC, Pkt) -> - request_cb(RC, App, T, TC, Pkt), +request_cb({eval_packet, RC, F}, App, Mask, T, TC, Fs, Pkt) -> + request_cb(RC, App, Mask, T, TC, [F|Fs], Pkt); + +request_cb({eval, RC, F}, App, Mask, T, TC, Fs, Pkt) -> + request_cb(RC, App, Mask, T, TC, Pkt, Fs), diameter_lib:eval(F). -%% protocol_error/4 +%% protocol_error/5 -protocol_error(RC, {_, OH, OR}, TPid, #diameter_packet{avps = Avps} = Pkt) -> +protocol_error(RC, {_, OH, OR}, TPid, Fs, Pkt) -> + #diameter_packet{avps = Avps} = Pkt, ?LOG({error, RC}, Pkt), - reply(answer_message({OH, OR, RC}, Avps), ?BASE, TPid, Pkt). + reply(answer_message({OH, OR, RC}, Avps), ?BASE, TPid, Fs, Pkt). + +%% protocol_error/4 -%% resend/5 +protocol_error(RC, T, TPid, Pkt) -> + protocol_error(RC, T, TPid, [], Pkt). + +%% resend/7 %% %% Resend a message as a relay or proxy agent. resend(Opts, #diameter_app{} = App, + Mask, {_SvcName, OH, _OR} = T, {_TPid, _Caps} = TC, + Fs, #diameter_packet{avps = Avps} = Pkt) -> {Code, _Flags, Vid} = ?BASE:avp_header('Route-Record'), - resend(is_loop(Code, Vid, OH, Avps), Opts, App, T, TC, Pkt). + resend(is_loop(Code, Vid, OH, Avps), Opts, App, Mask, T, TC, Fs, Pkt). %% DIAMETER_LOOP_DETECTED 3005 %% An agent detected a loop while trying to get the message to the @@ -1854,8 +2084,8 @@ resend(Opts, %% if one is available, but the peer reporting the error has %% identified a configuration problem. -resend(true, _, _, T, {TPid, _}, Pkt) -> %% Route-Record loop - protocol_error(3005, T, TPid, Pkt); +resend(true, _, _, _, T, {TPid, _}, Fs, Pkt) -> %% Route-Record loop + protocol_error(3005, T, TPid, Fs, Pkt); %% 6.1.8. Relaying and Proxying Requests %% @@ -1866,16 +2096,18 @@ resend(true, _, _, T, {TPid, _}, Pkt) -> %% Route-Record loop resend(false, Opts, App, + Mask, {SvcName, _, _} = T, {TPid, #diameter_caps{origin_host = {_, OH}}}, + Fs, #diameter_packet{header = Hdr0, avps = Avps} = Pkt) -> Route = #diameter_avp{data = {?BASE, 'Route-Record', OH}}, - Seq = diameter_session:sequence(), + Seq = diameter_session:sequence(Mask), Hdr = Hdr0#diameter_header{hop_by_hop_id = Seq}, Msg = [Hdr, Route | Avps], - resend(call(SvcName, App, Msg, Opts), T, TPid, Pkt). + resend(call(SvcName, App, Msg, Opts), T, TPid, Fs, Pkt). %% The incoming request is relayed with the addition of a %% Route-Record. Note the requirement on the return from call/4 below, %% which places a requirement on the value returned by the @@ -1901,15 +2133,18 @@ resend(#diameter_packet{bin = B} = Pkt, _, TPid, + Fs, #diameter_packet{header = #diameter_header{hop_by_hop_id = Id}, transport_data = TD}) -> - send(TPid, Pkt#diameter_packet{bin = diameter_codec:hop_by_hop_id(Id, B), - transport_data = TD}); + P = Pkt#diameter_packet{bin = diameter_codec:hop_by_hop_id(Id, B), + transport_data = TD}, + eval_packet(P, Fs), + send(TPid, P); %% TODO: counters %% Or not: DIAMETER_UNABLE_TO_DELIVER. -resend(_, T, TPid, Pkt) -> - protocol_error(3002, T, TPid, Pkt). +resend(_, T, TPid, Fs, Pkt) -> + protocol_error(3002, T, TPid, Fs, Pkt). %% is_loop/4 %% @@ -1931,46 +2166,68 @@ is_loop(Code, Vid, OH, [_ | Avps]) is_loop(Code, Vid, OH, Avps) -> is_loop(Code, Vid, ?BASE:avp(encode, OH, 'Route-Record'), Avps). -%% reply/4 +%% reply/5 %% %% Send a locally originating reply. +%% Skip the setting of Result-Code and Failed-AVP's below. +reply([Msg], Dict, TPid, Fs, Pkt) + when is_list(Msg); + is_tuple(Msg) -> + reply(Msg, Dict, TPid, Fs, Pkt#diameter_packet{errors = []}); + %% No errors or a diameter_header/avp list. -reply(Msg, Dict, TPid, #diameter_packet{errors = Es, - transport_data = TD} - = ReqPkt) +reply(Msg, Dict, TPid, Fs, #diameter_packet{errors = Es} = ReqPkt) when [] == Es; is_record(hd(Msg), diameter_header) -> Pkt = diameter_codec:encode(Dict, make_answer_packet(Msg, ReqPkt)), - incr(send, Pkt, TPid), %% count result codes in sent answers - send(TPid, Pkt#diameter_packet{transport_data = TD}); + eval_packet(Pkt, Fs), + incr(send, Pkt, Dict, TPid), %% count result codes in sent answers + send(TPid, Pkt); %% Or not: set Result-Code and Failed-AVP AVP's. -reply(Msg, Dict, TPid, #diameter_packet{errors = [H|_] = Es} = Pkt) -> +reply(Msg, Dict, TPid, Fs, #diameter_packet{errors = [H|_] = Es} = Pkt) -> reply(rc(Msg, rc(H), [A || {_,A} <- Es], Dict), Dict, TPid, + Fs, Pkt#diameter_packet{errors = []}). +eval_packet(Pkt, Fs) -> + lists:foreach(fun(F) -> diameter_lib:eval([F,Pkt]) end, Fs). + %% make_answer_packet/2 +%% A reply message clears the R and T flags and retains the P flag. +%% The E flag will be set at encode. 6.2 of 3588 requires the same P +%% flag on an answer as on the request. A #diameter_packet{} returned +%% from a handle_request callback can circumvent this by setting its +%% own header values. +make_answer_packet(#diameter_packet{header = Hdr, + msg = Msg, + transport_data = TD}, + #diameter_packet{header = ReqHdr}) -> + Hdr0 = ReqHdr#diameter_header{version = ?DIAMETER_VERSION, + is_request = false, + is_error = undefined, + is_retransmitted = false}, + #diameter_packet{header = fold_record(Hdr0, Hdr), + msg = Msg, + transport_data = TD}; + %% Binaries and header/avp lists are sent as-is. -make_answer_packet(Bin, _) +make_answer_packet(Bin, #diameter_packet{transport_data = TD}) when is_binary(Bin) -> - #diameter_packet{bin = Bin}; -make_answer_packet([#diameter_header{} | _] = Msg, _) -> - #diameter_packet{msg = Msg}; - -%% Otherwise a reply message clears the R and T flags and retains the -%% P flag. The E flag will be set at encode. 6.2 of 3588 requires the -%% same P flag on an answer as on the request. -make_answer_packet(Msg, #diameter_packet{header = ReqHdr}) -> - Hdr = ReqHdr#diameter_header{version = ?DIAMETER_VERSION, - is_request = false, - is_error = undefined, - is_retransmitted = false}, - #diameter_packet{header = Hdr, - msg = Msg}. + #diameter_packet{bin = Bin, + transport_data = TD}; +make_answer_packet([#diameter_header{} | _] = Msg, + #diameter_packet{transport_data = TD}) -> + #diameter_packet{msg = Msg, + transport_data = TD}; + +%% Otherwise, preserve transport_data. +make_answer_packet(Msg, #diameter_packet{transport_data = TD} = Pkt) -> + make_answer_packet(#diameter_packet{msg = Msg, transport_data = TD}, Pkt). %% rc/1 @@ -1983,7 +2240,10 @@ rc(RC) -> rc(Rec, RC, Failed, Dict) when is_integer(RC) -> - set(Rec, [{'Result-Code', RC} | failed_avp(Rec, Failed, Dict)], Dict). + set(Rec, + lists:append([rc(Rec, {'Result-Code', RC}, Dict), + failed_avp(Rec, Failed, Dict)]), + Dict). %% Reply as name and tuple list ... set([_|_] = Ans, Avps, _) -> @@ -1993,6 +2253,22 @@ set([_|_] = Ans, Avps, _) -> set(Rec, Avps, Dict) -> Dict:'#set-'(Avps, Rec). +%% rc/3 +%% +%% Turn the result code into a list if its optional and only set it if +%% the arity is 1 or {0,1}. In other cases (which probably shouldn't +%% exist in practise) we can't know what's appropriate. + +rc([MsgName | _], {'Result-Code' = K, RC} = T, Dict) -> + case Dict:avp_arity(MsgName, 'Result-Code') of + 1 -> [T]; + {0,1} -> [{K, [RC]}]; + _ -> [] + end; + +rc(Rec, T, Dict) -> + rc([Dict:rec2msg(element(1, Rec))], T, Dict). + %% failed_avp/3 failed_avp(_, [] = No, _) -> @@ -2200,44 +2476,39 @@ handle_answer(SvcName, _, {error, Req, Reason}) -> handle_answer(SvcName, AnswerErrors, {answer, #request{dictionary = Dict} = Req, Pkt}) -> - a(examine(diameter_codec:decode(Dict, Pkt)), - SvcName, - AnswerErrors, - Req). + answer(examine(diameter_codec:decode(Dict, Pkt)), + SvcName, + AnswerErrors, + Req). %% We don't really need to do a full decode if we're a relay and will %% just resend with a new hop by hop identifier, but might a proxy %% want to examine the answer? -a(#diameter_packet{errors = []} - = Pkt, - SvcName, - AE, - #request{transport = TPid, - caps = Caps, - packet = P} - = Req) -> +answer(Pkt, SvcName, AE, #request{transport = TPid, + dictionary = Dict} + = Req) -> try - incr(in, Pkt, TPid) + incr(recv, Pkt, Dict, TPid) of - _ -> - cb(Req, handle_answer, [Pkt, msg(P), SvcName, {TPid, Caps}]) + _ -> a(Pkt, SvcName, AE, Req) catch exit: {invalid_error_bit, _} = E -> - e(Pkt#diameter_packet{errors = [E]}, SvcName, AE, Req) - end; + a(Pkt#diameter_packet{errors = [E]}, SvcName, AE, Req) + end. -a(#diameter_packet{} = Pkt, SvcName, AE, Req) -> - e(Pkt, SvcName, AE, Req). +a(#diameter_packet{errors = Es} = Pkt, SvcName, AE, #request{transport = TPid, + caps = Caps, + packet = P} + = Req) + when [] == Es; + callback == AE -> + cb(Req, handle_answer, [Pkt, msg(P), SvcName, {TPid, Caps}]); -e(Pkt, SvcName, callback, #request{transport = TPid, - caps = Caps, - packet = Pkt} - = Req) -> - cb(Req, handle_answer, [Pkt, msg(Pkt), SvcName, {TPid, Caps}]); -e(Pkt, SvcName, report, Req) -> +a(Pkt, SvcName, report, Req) -> x(errors, handle_answer, [SvcName, Req, Pkt]); -e(Pkt, SvcName, discard, Req) -> + +a(Pkt, SvcName, discard, Req) -> x({errors, handle_answer, [SvcName, Req, Pkt]}). %% Note that we don't check that the application id in the answer's @@ -2249,17 +2520,19 @@ e(Pkt, SvcName, discard, Req) -> %% Increment a stats counter for an incoming or outgoing message. %% TODO: fix -incr(_, #diameter_packet{msg = undefined}, _) -> +incr(_, #diameter_packet{msg = undefined}, _, _) -> ok; -incr(Dir, Pkt, TPid) - when is_pid(TPid) -> +incr(recv = D, #diameter_packet{header = H, errors = [_|_]}, _, TPid) -> + incr(TPid, {diameter_codec:msg_id(H), D, error}); + +incr(Dir, Pkt, Dict, TPid) -> #diameter_packet{header = #diameter_header{is_error = E} = Hdr, msg = Rec} = Pkt, - RC = int(get_avp_value(?BASE, 'Result-Code', Rec)), + RC = int(get_avp_value(Dict, 'Result-Code', Rec)), PE = is_protocol_error(RC), %% Check that the E bit is set only for 3xxx result codes. @@ -2267,15 +2540,21 @@ incr(Dir, Pkt, TPid) orelse (E andalso PE) orelse x({invalid_error_bit, RC}, answer, [Dir, Pkt]), - Ctr = rc_counter(Rec, RC), - is_tuple(Ctr) - andalso incr(TPid, {diameter_codec:msg_id(Hdr), Dir, Ctr}). + irc(TPid, Hdr, Dir, rc_counter(Dict, Rec, RC)). + +irc(_, _, _, undefined) -> + false; + +irc(TPid, Hdr, Dir, Ctr) -> + incr(TPid, {diameter_codec:msg_id(Hdr), Dir, Ctr}). %% incr/2 incr(TPid, Counter) -> diameter_stats:incr(Counter, TPid, 1). +%% error_counter/2 + %% RFC 3588, 7.6: %% %% All Diameter answer messages defined in vendor-specific @@ -2285,26 +2564,27 @@ incr(TPid, Counter) -> %% Maintain statistics assuming one or the other, not both, which is %% surely the intent of the RFC. -rc_counter(_, RC) - when is_integer(RC) -> - {'Result-Code', RC}; -rc_counter(Rec, _) -> - rcc(get_avp_value(?BASE, 'Experimental-Result', Rec)). +rc_counter(Dict, Rec, undefined) -> + er(get_avp_value(Dict, 'Experimental-Result', Rec)); +rc_counter(_, _, RC) -> + {'Result-Code', RC}. %% Outgoing answers may be in any of the forms messages can be sent %% in. Incoming messages will be records. We're assuming here that the %% arity of the result code AVP's is 0 or 1. -rcc([{_,_,RC} = T]) - when is_integer(RC) -> +er([{_,_,N} = T | _]) + when is_integer(N) -> T; -rcc({_,_,RC} = T) - when is_integer(RC) -> +er({_,_,N} = T) + when is_integer(N) -> T; -rcc(_) -> +er(_) -> undefined. -int([N]) +%% Extract the first good looking integer. There's no guarantee +%% that what we're looking for has arity 1. +int([N|_]) when is_integer(N) -> N; int(N) @@ -2349,8 +2629,11 @@ rt(#request{packet = #diameter_packet{msg = undefined}}, _) -> false; %% TODO: Not what we should do. %% ... or not. -rt(#request{packet = #diameter_packet{msg = Msg}} = Req, S) -> - find_transport(get_destination(Msg), Req, S). +rt(#request{packet = #diameter_packet{msg = Msg}, + dictionary = Dict} + = Req, + S) -> + find_transport(get_destination(Dict, Msg), Req, S). %%% --------------------------------------------------------------------------- %%% # report_status/5 @@ -2383,7 +2666,7 @@ send_event(#diameter_event{service = SvcName} = E) -> %%% # share_peer/5 %%% --------------------------------------------------------------------------- -share_peer(up, Caps, Aliases, TPid, #state{share_peers = true, +share_peer(up, Caps, Aliases, TPid, #state{options = [_, {_, true} | _], service_name = Svc}) -> diameter_peer:notify(Svc, {peer, TPid, Aliases, Caps}); @@ -2394,11 +2677,11 @@ share_peer(_, _, _, _, _) -> %%% # share_peers/2 %%% --------------------------------------------------------------------------- -share_peers(Pid, #state{share_peers = true, +share_peers(Pid, #state{options = [_, {_, true} | _], local_peers = PDict}) -> ?Dict:fold(fun(A,Ps,ok) -> sp(Pid, A, Ps), ok end, ok, PDict); -share_peers(_, #state{share_peers = false}) -> +share_peers(_, _) -> ok. sp(Pid, Alias, Peers) -> @@ -2408,39 +2691,31 @@ sp(Pid, Alias, Peers) -> %%% # remote_peer_up/4 %%% --------------------------------------------------------------------------- -remote_peer_up(Pid, Aliases, Caps, #state{use_shared_peers = true, +remote_peer_up(Pid, Aliases, Caps, #state{options = [_, _, {_, true} | _], service = Svc, - shared_peers = PDict} - = S) -> + shared_peers = PDict}) -> #diameter_service{applications = Apps} = Svc, - Update = lists:filter(fun(A) -> - lists:keymember(A, #diameter_app.alias, Apps) - end, - Aliases), - S#state{shared_peers = rpu(Pid, Caps, PDict, Update)}; + Key = #diameter_app.alias, + As = lists:filter(fun(A) -> lists:keymember(A, Key, Apps) end, Aliases), + rpu(Pid, Caps, PDict, As); -remote_peer_up(_, _, _, #state{use_shared_peers = false} = S) -> - S. +remote_peer_up(_, _, _, #state{options = [_, _, {_, false} | _]}) -> + ok. rpu(_, _, PDict, []) -> PDict; rpu(Pid, Caps, PDict, Aliases) -> erlang:monitor(process, Pid), T = {Pid, Caps}, - lists:foldl(fun(A,D) -> ?Dict:append(A, T, D) end, - PDict, - Aliases). + lists:foreach(fun(A) -> ?Dict:append(A, T, PDict) end, Aliases). %%% --------------------------------------------------------------------------- %%% # remote_peer_down/2 %%% --------------------------------------------------------------------------- -remote_peer_down(Pid, #state{use_shared_peers = true, - shared_peers = PDict} - = S) -> - S#state{shared_peers = lists:foldl(fun(A,D) -> rpd(Pid, A, D) end, - PDict, - ?Dict:fetch_keys(PDict))}. +remote_peer_down(Pid, #state{options = [_, _, {_, true} | _], + shared_peers = PDict}) -> + lists:foreach(fun(A) -> rpd(Pid, A, PDict) end, ?Dict:fetch_keys(PDict)). rpd(Pid, Alias, PDict) -> ?Dict:update(Alias, fun(Ps) -> lists:keydelete(Pid, 1, Ps) end, PDict). @@ -2462,12 +2737,12 @@ find_transport({alias, Alias}, Msg, Opts, #state{service = Svc} = S) -> find_transport(#diameter_app{} = App, Msg, Opts, S) -> ft(App, Msg, Opts, S). -ft(#diameter_app{module = Mod} = App, Msg, Opts, S) -> +ft(#diameter_app{module = Mod, dictionary = Dict} = App, Msg, Opts, S) -> #options{filter = Filter, extra = Xtra} = Opts, pick_peer(App#diameter_app{module = Mod ++ Xtra}, - get_destination(Msg), + get_destination(Dict, Msg), Filter, S); ft(false = No, _, _, _) -> @@ -2503,11 +2778,11 @@ find_transport([_,_] = RH, Filter, S). -%% get_destination/1 +%% get_destination/2 -get_destination(Msg) -> - [str(get_avp_value(?BASE, 'Destination-Realm', Msg)), - str(get_avp_value(?BASE, 'Destination-Host', Msg))]. +get_destination(Dict, Msg) -> + [str(get_avp_value(Dict, 'Destination-Realm', Msg)), + str(get_avp_value(Dict, 'Destination-Host', Msg))]. %% This is not entirely correct. The avp could have an arity 1, in %% which case an empty list is a DiameterIdentity of length 0 rather @@ -2531,6 +2806,15 @@ str(T) -> %% question. The third form allows messages to be sent as is, without %% a dictionary, which is needed in the case of relay agents, for one. +%% Messages will be header/avps list as a relay and the only AVP's we +%% look for are in the common dictionary. This is required since the +%% relay dictionary doesn't inherit the common dictionary (which maybe +%% it should). +get_avp_value(?RELAY, Name, Msg) -> + get_avp_value(?BASE, Name, Msg); + +%% Message sent as a header/avps list, probably a relay case but not +%% necessarily. get_avp_value(Dict, Name, [#diameter_header{} | Avps]) -> try {Code, _, VId} = Dict:avp_header(Name), @@ -2544,6 +2828,7 @@ get_avp_value(Dict, Name, [#diameter_header{} | Avps]) -> undefined end; +%% Outgoing message as a name/values list. get_avp_value(_, Name, [_MsgName | Avps]) -> case lists:keyfind(Name, 1, Avps) of {_, V} -> @@ -2552,6 +2837,11 @@ get_avp_value(_, Name, [_MsgName | Avps]) -> undefined end; +%% Record might be an answer message in the common dictionary. +get_avp_value(Dict, Name, Rec) + when Dict /= ?BASE, element(1, Rec) == 'diameter_base_answer-message' -> + get_avp_value(?BASE, Name, Rec); + %% Message is typically a record but not necessarily: diameter:call/4 %% can be passed an arbitrary term. get_avp_value(Dict, Name, Rec) -> @@ -2746,20 +3036,59 @@ transports(#state{peerT = PeerT}) -> 'Vendor-Specific-Application-Id', 'Firmware-Revision']). +%% The config returned by diameter:service_info(SvcName, all). -define(ALL_INFO, [capabilities, applications, transport, pending, - statistics]). + options]). + +%% The rest. +-define(OTHER_INFO, [connections, + name, + peers, + statistics]). -service_info(Items, S) - when is_list(Items) -> - [{complete(I), service_info(I,S)} || I <- Items]; service_info(Item, S) when is_atom(Item) -> - service_info(Item, S, true). + case tagged_info(Item, S) of + {_, T} -> T; + undefined = No -> No + end; + +service_info(Items, S) -> + tagged_info(Items, S). + +tagged_info(Item, S) + when is_atom(Item) -> + case complete(Item) of + {value, I} -> + {I, complete_info(I,S)}; + false -> + undefined + end; -service_info(Item, #state{service = Svc} = S, Complete) -> +tagged_info(TPid, #state{peerT = PT, connT = CT}) + when is_pid(TPid) -> + try + [#conn{peer = Pid}] = ets:lookup(CT, TPid), + [#peer{ref = Ref, type = Type, options = Opts}] = ets:lookup(PT, Pid), + [{ref, Ref}, + {type, Type}, + {options, Opts}] + catch + error:_ -> + [] + end; + +tagged_info(Items, S) + when is_list(Items) -> + [T || I <- Items, T <- [tagged_info(I,S)], T /= undefined, T /= []]; + +tagged_info(_, _) -> + undefined. + +complete_info(Item, #state{service = Svc} = S) -> case Item of name -> S#state.service_name; @@ -2802,85 +3131,176 @@ service_info(Item, #state{service = Svc} = S, Complete) -> capabilities -> service_info(?CAP_INFO, S); applications -> info_apps(S); transport -> info_transport(S); + options -> info_options(S); pending -> info_pending(S); - statistics -> info_stats(S); - keys -> ?ALL_INFO ++ ?CAP_INFO; %% mostly for test + keys -> ?ALL_INFO ++ ?CAP_INFO ++ ?OTHER_INFO; all -> service_info(?ALL_INFO, S); - _ when Complete -> service_info(complete(Item), S, false); - _ -> undefined + statistics -> info_stats(S); + connections -> info_connections(S); + peers -> info_peers(S) end. +complete(I) + when I == keys; + I == all -> + {value, I}; complete(Pre) -> P = atom_to_list(Pre), - case [I || I <- [name | ?ALL_INFO] ++ ?CAP_INFO, + case [I || I <- ?ALL_INFO ++ ?CAP_INFO ++ ?OTHER_INFO, lists:prefix(P, atom_to_list(I))] of - [I] -> I; - _ -> Pre + [I] -> {value, I}; + _ -> false end. +%% info_stats/1 + info_stats(#state{peerT = PeerT}) -> - Peers = ets:select(PeerT, [{#peer{ref = '$1', conn = '$2', _ = '_'}, - [{'is_pid', '$2'}], - [['$1', '$2']]}]), - diameter_stats:read(lists:append(Peers)). -%% TODO: include peer identities in return value - -info_transport(#state{peerT = PeerT, connT = ConnT}) -> - dict:fold(fun it/3, + MatchSpec = [{#peer{ref = '$1', conn = '$2', _ = '_'}, + [{'is_pid', '$2'}], + [['$1', '$2']]}], + try ets:select(PeerT, MatchSpec) of + L -> + diameter_stats:read(lists:append(L)) + catch + error: badarg -> [] %% service has gone down + end. + +%% info_transport/1 +%% +%% One entry per configured transport. Statistics for each entry are +%% the accumulated values for the ref and associated peer pids. + +info_transport(S) -> + PeerD = peer_dict(S, config_dict(S)), + RefsD = dict:map(fun(_, Ls) -> [P || L <- Ls, {peer, {P,_}} <- L] end, + PeerD), + Refs = lists:append(dict:fold(fun(R, Ps, A) -> [[R|Ps] | A] end, + [], + RefsD)), + Stats = diameter_stats:read(Refs), + dict:fold(fun(R, Ls, A) -> + Ps = dict:fetch(R, RefsD), + [[{ref, R} | transport(Ls)] ++ [stats([R|Ps], Stats)] + | A] + end, [], - ets:foldl(fun(T,A) -> it_acc(ConnT, A, T) end, - dict:new(), - PeerT)). - -it(Ref, [[{type, connect} | _] = L], Acc) -> - [[{ref, Ref} | L] | Acc]; -it(Ref, [[{type, accept}, {options, Opts} | _] | _] = L, Acc) -> - [[{ref, Ref}, - {type, listen}, - {options, Opts}, - {accept, [lists:nthtail(2,A) || A <- L]}] - | Acc]. -%% Each entry has the same Opts. (TODO) - -it_acc(ConnT, Acc, #peer{pid = Pid, - type = Type, - ref = Ref, - options = Opts, - op_state = OS, - started = T, - conn = TPid}) -> + PeerD). + +%% Only a config entry for a listening transport: use it. +transport([[{type, listen}, _] = L]) -> + L ++ [{accept, []}]; + +%% Only one config or peer entry for a connecting transport: use it. +transport([[{type, connect} | _] = L]) -> + L; + +%% Peer entries: discard config. Note that the peer entries have +%% length at least 3. +transport([[_,_] | L]) -> + transport(L); + +%% Possibly many peer entries for a listening transport. Note that all +%% have the same options by construction, which is not terribly space +%% efficient. (TODO: all entries for the same Ref should share options.) +transport([[{type, accept}, {options, Opts} | _] | _] = Ls) -> + [{type, listen}, + {options, Opts}, + {accept, [lists:nthtail(2,L) || L <- Ls]}]. + +peer_dict(#state{peerT = PeerT, connT = ConnT}, Dict0) -> + try ets:tab2list(PeerT) of + L -> + lists:foldl(fun(T,A) -> peer_acc(ConnT, A, T) end, Dict0, L) + catch + error: badarg -> Dict0 %% service has gone down + end. + +peer_acc(ConnT, Acc, #peer{pid = Pid, + type = Type, + ref = Ref, + options = Opts, + op_state = OS, + started = T, + conn = TPid}) -> + WS = wd_state(OS), dict:append(Ref, [{type, Type}, {options, Opts}, - {watchdog, {Pid, T, OS}} - | info_conn(ConnT, TPid)], + {watchdog, {Pid, T, WS}} + | info_conn(ConnT, TPid, WS /= ?WD_DOWN)], Acc). -info_conn(ConnT, TPid) -> - info_conn(ets:lookup(ConnT, TPid)). +info_conn(ConnT, TPid, true) + when is_pid(TPid) -> + try ets:lookup(ConnT, TPid) of + T -> info_conn(T) + catch + error: badarg -> [] %% service has gone down + end; +info_conn(_, _, _) -> + []. + +%% The point of extracting the config here is so that 'transport' info +%% has one entry for each transport ref, the peer table only +%% containing entries that have a living watchdog. + +config_dict(#state{service_name = SvcName}) -> + lists:foldl(fun config_acc/2, + dict:new(), + diameter_config:lookup(SvcName)). + +config_acc({Ref, T, Opts}, Dict) + when T == listen; + T == connect -> + dict:store(Ref, [[{type, T}, {options, Opts}]], Dict); +config_acc(_, Dict) -> + Dict. + +wd_state({_,S}) -> + S; +wd_state(?STATE_UP) -> + ?WD_OKAY; +wd_state(?STATE_DOWN) -> + ?WD_DOWN. info_conn([#conn{pid = Pid, apps = SApps, caps = Caps, started = T}]) -> [{peer, {Pid, T}}, {apps, SApps}, - {caps, info_caps(Caps)}]; + {caps, info_caps(Caps)} + | try [{port, info_port(Pid)}] catch _:_ -> [] end]; info_conn([] = No) -> No. +%% Extract information that the processes involved are expected to +%% "publish" in their process dictionaries. Simple but backhanded. +info_port(Pid) -> + {_, PD} = process_info(Pid, dictionary), + {_, T} = lists:keyfind({diameter_peer_fsm, start}, 1, PD), + {TPid, {_Type, TMod, _Cfg}} = T, + {_, TD} = process_info(TPid, dictionary), + {_, Data} = lists:keyfind({TMod, info}, 1, TD), + [{owner, TPid}, + {module, TMod} + | try TMod:info(Data) catch _:_ -> [] end]. + +%% Use the fields names from diameter_caps instead of +%% diameter_base_CER to distinguish between the 2-tuple values +%% compared to the single capabilities values. Note also that the +%% returned list is tagged 'caps' rather than 'capabilities' to +%% emphasize the difference. info_caps(#diameter_caps{} = C) -> lists:zip(record_info(fields, diameter_caps), tl(tuple_to_list(C))). info_apps(#state{service = #diameter_service{applications = Apps}}) -> lists:map(fun mk_app/1, Apps). -mk_app(#diameter_app{alias = Alias, - dictionary = Dict, - module = ModX, - id = Id}) -> - [{alias, Alias}, - {dictionary, Dict}, - {module, ModX}, - {id, Id}]. +mk_app(#diameter_app{} = A) -> + lists:zip(record_info(fields, diameter_app), tl(tuple_to_list(A))). + +%% info_pending/1 +%% +%% One entry for each outgoing request whose answer is outstanding. info_pending(#state{} = S) -> MatchSpec = [{{'$1', @@ -2894,4 +3314,67 @@ info_pending(#state{} = S) -> {{transport, '$2'}}, {{from, '$3'}}]}}]}], - ets:select(?REQUEST_TABLE, MatchSpec). + try + ets:select(?REQUEST_TABLE, MatchSpec) + catch + error: badarg -> [] %% service has gone down + end. + +%% info_connections/1 +%% +%% One entry per transport connection. Statistics for each entry are +%% for the peer pid only. + +info_connections(S) -> + ConnL = conn_list(S), + Stats = diameter_stats:read([P || L <- ConnL, {peer, {P,_}} <- L]), + [L ++ [stats([P], Stats)] || L <- ConnL, {peer, {P,_}} <- L]. + +conn_list(S) -> + lists:append(dict:fold(fun conn_acc/3, [], peer_dict(S, dict:new()))). + +conn_acc(Ref, Peers, Acc) -> + [[[{ref, Ref} | L] || L <- Peers, lists:keymember(peer, 1, L)] + | Acc]. + +stats(Refs, Stats) -> + {statistics, dict:to_list(lists:foldl(fun(R,D) -> + stats_acc(R, D, Stats) + end, + dict:new(), + Refs))}. + +stats_acc(Ref, Dict, Stats) -> + lists:foldl(fun({C,N}, D) -> dict:update_counter(C, N, D) end, + Dict, + proplists:get_value(Ref, Stats, [])). + +%% info_peers/1 +%% +%% One entry per peer Origin-Host. Statistics for each entry are +%% accumulated values for all peer pids. + +info_peers(S) -> + {PeerD, RefD} = lists:foldl(fun peer_acc/2, + {dict:new(), dict:new()}, + conn_list(S)), + Refs = lists:append(dict:fold(fun(_, Rs, A) -> [Rs|A] end, + [], + RefD)), + Stats = diameter_stats:read(Refs), + dict:fold(fun(OH, Cs, A) -> + Rs = dict:fetch(OH, RefD), + [{OH, [{connections, Cs}, stats(Rs, Stats)]} | A] + end, + [], + PeerD). + +peer_acc(Peer, {PeerD, RefD}) -> + [{TPid, _}, [{origin_host, {_, OH}} | _]] + = [proplists:get_value(K, Peer) || K <- [peer, caps]], + {dict:append(OH, Peer, PeerD), dict:append(OH, TPid, RefD)}. + +%% info_options/1 + +info_options(S) -> + S#state.options. |