diff options
| -rw-r--r-- | lib/diameter/doc/src/diameter.xml | 24 | ||||
| -rw-r--r-- | lib/diameter/doc/src/diameter_transport.xml | 31 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter.erl | 4 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_callback.erl | 4 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_dist.erl | 525 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_misc_sup.erl | 3 | ||||
| -rw-r--r-- | lib/diameter/src/base/diameter_traffic.erl | 97 | ||||
| -rw-r--r-- | lib/diameter/src/modules.mk | 3 | ||||
| -rw-r--r-- | lib/diameter/test/diameter_dist_SUITE.erl | 332 | ||||
| -rw-r--r-- | lib/diameter/test/diameter_distribution_SUITE.erl | 7 | ||||
| -rw-r--r-- | lib/diameter/test/diameter_pool_SUITE.erl | 3 | ||||
| -rw-r--r-- | lib/diameter/test/diameter_traffic_SUITE.erl | 5 | ||||
| -rw-r--r-- | lib/diameter/test/modules.mk | 3 |
13 files changed, 985 insertions, 56 deletions
diff --git a/lib/diameter/doc/src/diameter.xml b/lib/diameter/doc/src/diameter.xml index 0a0194af2d..85522c99b2 100644 --- a/lib/diameter/doc/src/diameter.xml +++ b/lib/diameter/doc/src/diameter.xml @@ -1,7 +1,9 @@ <?xml version="1.0" encoding="utf-8" ?> <!DOCTYPE erlref SYSTEM "erlref.dtd" [ - <!ENTITY spawn_opt + <!ENTITY spawn_opt2 '<seealso marker="erts:erlang#spawn_opt-2">erlang:spawn_opt/2</seealso>'> + <!ENTITY spawn_opt5 + '<seealso marker="erts:erlang#spawn_opt-5">erlang:spawn_opt/5</seealso>'> <!ENTITY nodes '<seealso marker="erts:erlang#nodes-0">erlang:nodes/0</seealso>'> <!ENTITY make_ref @@ -21,7 +23,7 @@ <copyright> <year>2011</year> -<year>2017</year> +<year>2019</year> <holder>Ericsson AB. All Rights Reserved.</holder> </copyright> <legalnotice> @@ -1384,12 +1386,22 @@ the same peer.</p> </item> <tag> -<marker id="spawn_opt"/><c>{spawn_opt, [term()]}</c></tag> +<marker id="spawn_opt"/><c>{spawn_opt, [term()] | {M,F,A}}</c></tag> <item> <p> -Options passed to &spawn_opt; when spawning a process for an -incoming Diameter request. -Options <c>monitor</c> and <c>link</c> are ignored.</p> +An options list passed to &spawn_opt2; to spawn a handler process for an +incoming Diameter request on the local node, or an MFA that returns +the pid of a handler process.</p> + +<p> +Options <c>monitor</c> and <c>link</c> are ignored in the list-valued +case. +An MFA is applied with an additional term prepended to its argument +list, and should return either the pid of the handler process that +invokes <c>diameter_traffic:request/1</c> on the term in order to +process the request, or the atom <c>discard</c>. +The handler process need not be local, but diameter must be started on +the remote node.</p> <p> Defaults to the empty list.</p> diff --git a/lib/diameter/doc/src/diameter_transport.xml b/lib/diameter/doc/src/diameter_transport.xml index 67fd54bc56..0a8ef321c6 100644 --- a/lib/diameter/doc/src/diameter_transport.xml +++ b/lib/diameter/doc/src/diameter_transport.xml @@ -14,7 +14,8 @@ <erlref> <header> <copyright> -<year>2011</year><year>2016</year> +<year>2011</year> +<year>2019</year> <holder>Ericsson AB. All Rights Reserved.</holder> </copyright> <legalnotice> @@ -174,10 +175,13 @@ its parent.</p> <taglist> -<tag><c>{diameter, {send, &message;}}</c></tag> +<tag><c>{diameter, {send, &message; | false}}</c></tag> <item> <p> -An outbound Diameter message.</p> +An outbound Diameter message. +The atom <c>false</c> can only be received when request +acknowledgements have been requests: see the <c>ack</c> message +below.</p> </item> <tag><c>{diameter, {close, Pid}}</c></tag> @@ -246,6 +250,27 @@ A <c>LocalAddr</c> list has the same semantics as one returned from &start;.</p> </item> +<tag><c>{diameter, ack}</c></tag> +<item> +<p> +Request acknowledgements of unanswered requests. +A transport process should send this once before passing incoming +Diameter messages into diameter. +As a result, every Diameter request passed into diameter with a +<c>recv</c> message (below) will be answered with a +<c>send</c> message (above), either a &message; for the transport +process to send or the atom <c>false</c> if the request has been +discarded or otherwise not answered.</p> + +<p> +This is to allow a transport process to keep count of the number +of incoming request messages that have not yet been answered or +discarded, to allow it to regulate the amount of incoming traffic. +Both diameter_tcp and diameter_sctp request acknowledgements when a +<c>message_cb</c> is configured, turning send/recv message into +callbacks that can be used to regulate traffic.</p> +</item> + <tag><c>{diameter, {recv, &message;}}</c></tag> <item> <p> diff --git a/lib/diameter/src/base/diameter.erl b/lib/diameter/src/base/diameter.erl index b90b794611..7f172e1fa1 100644 --- a/lib/diameter/src/base/diameter.erl +++ b/lib/diameter/src/base/diameter.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2017. All Rights Reserved. +%% Copyright Ericsson AB 2010-2019. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -365,7 +365,7 @@ call(SvcName, App, Message) -> | {connect_timer, 'Unsigned32'()} | {watchdog_timer, 'Unsigned32'() | {module(), atom(), list()}} | {watchdog_config, [{okay|suspect, non_neg_integer()}]} - | {spawn_opt, list()}. + | {spawn_opt, list() | mfa()}. %% Options passed to start_service/2 diff --git a/lib/diameter/src/base/diameter_callback.erl b/lib/diameter/src/base/diameter_callback.erl index d04a416bef..3bcf550cd8 100644 --- a/lib/diameter/src/base/diameter_callback.erl +++ b/lib/diameter/src/base/diameter_callback.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2017. All Rights Reserved. +%% Copyright Ericsson AB 2010-2019. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -70,7 +70,7 @@ -module(diameter_callback). -%% Default callbacks when no aleternate is specified. +%% Default callbacks when no alternate is specified. -export([peer_up/3, peer_down/3, pick_peer/4, diff --git a/lib/diameter/src/base/diameter_dist.erl b/lib/diameter/src/base/diameter_dist.erl new file mode 100644 index 0000000000..5c29ea95a4 --- /dev/null +++ b/lib/diameter/src/base/diameter_dist.erl @@ -0,0 +1,525 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2019. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% + +-module(diameter_dist). + +-behaviour(gen_server). + +%% +%% Implements callbacks that can be configured as a spawn_opt +%% transport configuration, to be able to distribute incoming Diameter +%% requests to handler processes (local or remote) in various ways. +%% + +%% spawn_opt callbacks +-export([spawn_local/2, + spawn_local/1, + route_session/2, + route_session/1]). + +%% signal availability for handling incoming requests to route_sesssion/2 +-export([attach/1, + detach/1]). + +%% consistent hashing +-export([hash/3, %% for use as default MFA in route_session/2 options map + hash/2]). %% arbitrary key/values + +-include_lib("diameter/include/diameter.hrl"). + +%% server start +-export([start_link/0]). + +%% gen_server callbacks +-export([init/1, + handle_info/2, + handle_cast/2, + handle_call/3, + code_change/3, + terminate/2]). + +-type request() :: tuple(). %% callback argument from diameter_traffic + +-define(SERVER, ?MODULE). %% server monitoring node connections + +%% Maps a node name binary to the corresponding atom. Used by +%% route_session/2 to map the optional value of a Session-Id to +%% node(). +-define(NODE_TABLE, diameter_dist_node). + +%% Maps a diameter:service_name() to a node() that has called attach/1 +%% to declare its willingness to handle incoming requests for the +%% service. Use by route_session/2 in case the optional value mapping +%% has failed. +-define(SERVICE_TABLE, diameter_dist_service). + +-define(B(A), atom_to_binary(A, utf8)). +-define(ORCOND(List), list_to_tuple(['orelse', false | List])). +-define(HASH(T), erlang:phash2(T, 16#100000000)). + +%% spawn_local/2 +%% +%% Callback that is equivalent to an options list. That is, the +%% following are equivalent when passed as options to +%% diameter:add_transport/2. +%% +%% {spawn_opt, Opts} +%% {spawn_opt, {diameter_dist, spawn_local, [Opts]}} + +-spec spawn_local(ReqT :: request(), Opts :: list()) + -> pid(). + +spawn_local(ReqT, Opts) -> + spawn_opt(diameter_traffic, request, [ReqT], Opts). + +%% spawn_local/1 + +spawn_local(ReqT) -> + spawn_local(ReqT, []). + +%% route_session/2 +%% +%% Callback that maps the Session-Id of an incoming request to a +%% handler node. +%% +%% With an options list, maps an id whose optional value is the name +%% of a connected node to the same node, to handle the case that the +%% session id has been returned from diameter:session_id/1; otherwise +%% to a node that has called diameter_dist:attach/1 using the +%% consistent hashing provided by hash/3, or to the local node() if a +%% session id could not be extracted or there are no attached nodes. A +%% handler process is spawned on the selected node using +%% erlang:spawn_opt/4. +%% +%% Different behaviour can be configured by supplying an options map +%% of the following form: +%% +%% #{search => non_neg_integer(), +%% id => [binary()], +%% default => discard | local | mfa(), +%% dispatch => list() | mfa()} +%% +%% The search member limits the number of AVPs that are examined in +%% the message (from the front), to avoid searching entire message in +%% case it's known that peers follow RFC 6733's recommendation that +%% Session-Id be placed at the head of a message. The default is to +%% search the entire message. +%% +%% The id member restricts the optional value mapping to session ids +%% whose DiamterIdentity is one of those specified. Set this to the +%% list of Diameter identities advertised by the service in question +%% (typically one) to ensure that only locally generated session ids +%% are mapped; or to the empty list to disable the mapping. +%% +%% The default member determines where to handle a message whose +%% Session-Id isn't found or whose optional value isn't mapped to the +%% name of a connected node. The atom local says the local node, an +%% MFA is invoked on Session-Id | false, the name of the diameter +%% service, and the message binary, and should return either a node() +%% or false to discard the message. Defaults to {diameter_dist, hash, []}. +%% +%% The dispatch member determines how the pid() of the request handler +%% process is retrieved. An MFA is applied to a previously selected +%% node(), and the module, function, and arguments list to apply in +%% the handler process to handle the request, the MFA being supplied +%% by diameter, and returns pid() | discard. A list is equivalent to +%% {erlang, spawn_opt, []}. Defaults to []. +%% +%% This can be used with search = 0 to route on something other than +%% Session-Id, but this is probably no simpler than just implementing +%% an own spawn_opt callback. (Except with the default dispatch possibly.) +%% +%% Note that if the peer is also implemented with OTP diameter and +%% generating session ids with diameter:session_id/1 then +%% route_session/2 can map an optional value to a local node that +%% happens to have the same name as one of the peer's nodes. This +%% could lead to an uneven distribution; for example, if the peer +%% nodes are a subset of the local nodes. In practice, it's typically +%% known if it's peers or the local node originating sessions; if the +%% former then setting id = [] disables the optional value mapping, if +%% the latter then setting default = local disables the hashing. +-spec route_session(ReqT :: request(), Opts) + -> discard + | pid() + when Opts :: pos_integer() %% aka #{search => N} + | list() %% aka #{dispatch => Opts} + | #{search => non_neg_integer(), %% limit number of examined AVPs + id => [binary()], %% restrict optional value map on DiamIdent + default => local %% handle locally + | discard + | mfa(), %% return node() | false + dispatch => list() %% spawn options + | mfa()}. %% (Node, M, F, A) -> pid() | discard + +route_session(ReqT, Opts) -> + {_, Bin} = Info = diameter_traffic:request_info(ReqT), + Sid = session_id(avps(Bin), search(Opts)), + Node = default(node_of_session_id(Sid, Opts), Sid, Opts, Info), + dispatch(Node, ReqT, dispatch(Opts)). + +%% avps/1 + +avps(<<_:20/binary, Bin/binary>>) -> + Bin; + +avps(_) -> + false. + +%% dispatch/3 + +dispatch(false, _, _) -> + discard; + +dispatch(Node, ReqT, {M,F,A}) -> + apply(M, F, [Node, diameter_traffic, request, [ReqT] | A]); + +dispatch(Node, ReqT, Opts) -> + spawn_opt(Node, diameter_traffic, request, [ReqT], Opts). + +%% route_session/1 + +route_session(ReqT) -> + route_session(ReqT, []). + +%% node_of_session_id/2 +%% +%% Return the node name encoded as optional value in a Session-Id, +%% assuming the id has been created with diameter:session_id/0. Lookup +%% the node name to ensure we don't convert arbitrary binaries to +%% atom. + +node_of_session_id([Id, _, _, Bin], #{id := Ids}) -> + lists:member(Id, Ids) andalso nodemap(Bin); + +node_of_session_id([_, _, _, Bin], _) -> + nodemap(Bin); + +node_of_session_id(_, _) -> + false. + +%% nodemap/1 + +nodemap(Bin) -> + try + ets:lookup_element(?NODE_TABLE, Bin, 2) + catch + error: badarg -> false + end. + +%% session_id/2 + +session_id(_, 0) -> %% give up + false; + +%% Session-Id = Command Code 263, V-bit = 0. +session_id(<<263:32, 0:1, _:7, Len:24, _/binary>> = Bin, _) -> + case Bin of + <<Avp:Len/binary, _/binary>> -> + <<_:8/binary, Sid/binary>> = Avp, + split(Sid); + _ -> + false + end; + +%% Jump to the next AVP. This is potentially costly for a message with +%% many AVPs and no Session-Id, which an attacker is prone to send. +%% 8.8 or RFC 6733 says that Session-Id SHOULD (but not MUST) appear +%% immediately following the Diameter Header, so there is no +%% guarantee. +session_id(<<_:40, Len:24, _/binary>> = Bin, N) -> + Pad = (4 - (Len rem 4)) rem 4, + case Bin of + <<_:Len/binary, _:Pad/binary, Rest/binary>> -> + session_id(Rest, if N == infinity -> N; true -> N-1 end); + _ -> + false + end; + +session_id(_, _) -> + false. + +%% split/1 +%% +%% Split a Session-Id at no more than three semicolons: the optional +%% value (if any) follows the third. binary:split/2 does better than +%% matching character by character, especially when the pattern is +%% compiled. + +split(Bin) -> + split(3, Bin, pattern()). + +%% split/3 + +split(0, Bin, _) -> + [Bin]; + +split(N, Bin, Pattern) -> + [H|T] = binary:split(Bin, Pattern), + [H | case T of + [] -> + T; + [Rest] -> + split(N-1, Rest, Pattern) + end]. + +%% pattern/0 +%% +%% Since this is being called in a watchdog process, compile the +%% pattern once and maintain it in the process dictionary. + +pattern() -> + case get(?MODULE) of + undefined -> + CP = binary:compile_pattern(<<$;>>), %% tuple + put(?MODULE, CP), + CP; + CP -> + CP + end. + +%% dispatch/1 + +dispatch(#{} = Opts) -> + maps:get(dispatch, Opts, []); + +dispatch(Opts) + when is_list(Opts) -> + Opts; + +dispatch(_) -> + []. + +%% search/1 +%% +%% Bound number of AVPs examined when looking for Session-Id. + +search(#{search := N}) + when is_integer(N), 0 =< N -> + N; + +search(N) + when is_integer(N), 0 =< N -> + N; + +search(_) -> + infinity. + +%% default/3 +%% +%% Choose a node when Session-Id lookup has failed. + +default(false, _, #{default := discard}, _) -> + false; + +default(false, _, #{default := local}, _) -> + node(); + +default(false, Sid, #{default := {M,F,A}}, Info) -> + {ServiceName, Bin} = Info, + apply(M, F, [Sid, ServiceName, Bin | A]); %% node() | false + +default(false, Sid, _, Info) -> %% aka {?MODULE, hash, []} + {ServiceName, Bin} = Info, + hash(Sid, ServiceName, Bin); + +default(Node, _, _, _) -> + Node. + +%% =========================================================================== + +%% hash/3 +%% +%% Consistent hashing of Session-Id to an attached node, or the local +%% node if Session-Id = false or no attached nodes. + +hash(Sid, ServiceName, _) -> + case false /= Sid andalso attached(ServiceName) of + [_|_] = Nodes -> + hash(Sid, Nodes); + _ -> + node() + end. + +%% hash/2 +%% +%% Consistent hashing on arbitrary key/values. Returns false if the +%% list is empty. + +%% No key or no values. +hash(_, []) -> + false; + +%% Not much choice. +hash(_, [Value]) -> + Value; + +%% Hash on a circle and choose the closest predecessor. +hash(Key, Values) -> + Hash = ?HASH(Key), + tl(lists:foldl(fun(V,A) -> + choose(Hash, [?HASH({Key, V}) | V], A) + end, + false, %% < list() + Values)). + +%% choose/3 + +choose(Hash, [Hash1 | _] = T, [Hash2 | _]) + when Hash1 =< Hash, Hash < Hash2 -> + T; + +choose(Hash, [Hash1 | _], [Hash2 | _] = T) + when Hash2 =< Hash, Hash < Hash1 -> + T; + +choose(_, T1, T2) -> + max(T1, T2). + +%% =========================================================================== + +%% attach/1 +%% +%% Register the local node as a handler of incoming requests for the +%% specified services when using the route_session/2 spawn_opt +%% callback. + +attach(ServiceNames) -> + abcast({attach, node(), ServiceNames}). + +%% detach/1 +%% +%% Deregister the local node as a handler of incoming requests. + +detach(ServiceNames) -> + abcast({detach, node(), ServiceNames}). + +%% abcast/1 + +abcast(T) -> + gen_server:abcast([node() | nodes()], ?SERVER, T), + ok. + +%% attached/1 + +attached(ServiceName) -> + try + ets:lookup_element(?SERVICE_TABLE, ServiceName, 2) + catch + error: badarg -> [] + end. + +%% cast/2 + +cast(Node, T) -> + gen_server:cast({?SERVER, Node}, T). + +%% attach/2 + +attach(Node, S) -> + case sets:to_list(S) of + [] -> + ok; + Services -> + cast(Node, {attach, node(), Services}) + end. + +%% =========================================================================== + +start_link() -> + gen_server:start_link({local, ?SERVER}, ?MODULE, _Args = [], _Opts = []). + +%% init/1 +%% +%% Maintain [node() | nodes()] in a table that maps from binary-valued +%% names, so we can lookup the corresponding atoms rather than convert +%% binaries that aren't necessarily node names. + +init([]) -> + ets:new(?NODE_TABLE, [set, named_table]), + ets:new(?SERVICE_TABLE, [bag, named_table]), + ok = net_kernel:monitor_nodes(true, [{node_type, all}, nodedown_reason]), + ets:insert(?NODE_TABLE, [{?B(N), N} || N <- [node() | nodes()]]), + abcast({attach, node()}), + {ok, sets:new()}. + +%% handle_call/3 + +handle_call(_, _From, S) -> + {reply, nok, S}. + +%% handle_cast/2 + +%% Remote node is asking which services the local node wants to handle. +handle_cast({attach, Node}, S) + when Node /= node() -> + attach(Node, S), + {noreply, S}; + +%% Node wants to handle incoming requests ... +handle_cast({attach, Node, ServiceNames}, S) -> + ets:insert(?SERVICE_TABLE, [{N, Node} || N <- ServiceNames]), + {noreply, case node() of + Node -> + sets:union(S, sets:from_list(ServiceNames)); + _ -> + S + end}; + +%% ... or not. +handle_cast({detach, Node, ServiceNames}, S) -> + ets:select_delete(?SERVICE_TABLE, [{{'$1', Node}, + [?ORCOND([{'==', '$1', {const, N}} + || N <- ServiceNames])], + [true]}]), + {noreply, case node() of + Node -> + sets:subtract(S, sets:from_list(ServiceNames)); + _ -> + S + end}; + +handle_cast(_, S) -> + {noreply, S}. + +%% handle_info/2 + +handle_info({nodeup, Node, _}, S) -> + ets:insert(?NODE_TABLE, {?B(Node), Node}), + cast(Node, {attach, node()}), %% ask which services remote node handles + attach(Node, S), %% say which service local node handles + {noreply, S}; + +handle_info({nodedown, Node, _}, S) -> + ets:delete(?NODE_TABLE, ?B(Node)), + ets:select_delete(?SERVICE_TABLE, [{{'_', Node}, [], [true]}]), + {noreply, S}; + +handle_info(_, S) -> + {noreply, S}. + +%% terminate/2 + +terminate(_, _) -> + ok. + +%% code_change/3 + +code_change(_OldVsn, State, _Extra) -> + {ok, State}. diff --git a/lib/diameter/src/base/diameter_misc_sup.erl b/lib/diameter/src/base/diameter_misc_sup.erl index 343688be23..fec5a41b5c 100644 --- a/lib/diameter/src/base/diameter_misc_sup.erl +++ b/lib/diameter/src/base/diameter_misc_sup.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2016. All Rights Reserved. +%% Copyright Ericsson AB 2010-2019. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -35,6 +35,7 @@ diameter_stats, %% statistics counter management diameter_reg, %% service/property publishing diameter_peer, %% remote peer manager + diameter_dist, %% request distribution diameter_config]). %% configuration/restart %% start_link/0 diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl index 2d3e4a2ac9..8423e30269 100644 --- a/lib/diameter/src/base/diameter_traffic.erl +++ b/lib/diameter/src/base/diameter_traffic.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2013-2017. All Rights Reserved. +%% Copyright Ericsson AB 2013-2019. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -42,8 +42,12 @@ peer_up/1, peer_down/1]). +%% towards diameter_dist +-export([request_info/1]). + %% internal -export([send/1, %% send from remote node + request/1, %% process request in handler process init/1]). %% monitor process start -include_lib("diameter/include/diameter.hrl"). @@ -232,7 +236,7 @@ incr_rc(Dir, Pkt, TPid, MsgDict, AppDict, Dict0) -> -spec receive_message(pid(), Route, #diameter_packet{}, module(), RecvData) -> pid() %% request handler | boolean() %% answer, known request or not - | discard %% request discarded by MFA + | discard %% request discarded when Route :: {Handler, RequestRef, TPid} | Ack, RecvData :: {[SpawnOpt], #recvdata{}}, @@ -252,7 +256,8 @@ receive_message(TPid, Route, Pkt, Dict0, RecvData) -> recv(true, Ack, TPid, Pkt, Dict0, T) when is_boolean(Ack) -> {Opts, RecvData} = T, - spawn_request(Ack, TPid, Pkt, Dict0, RecvData, Opts); + AppT = find_app(TPid, Pkt, RecvData), + ack(Ack, TPid, spawn_request(AppT, Opts, Ack, TPid, Pkt, Dict0, RecvData)); %% ... answer to known request ... recv(false, {Pid, Ref, TPid}, _, Pkt, Dict0, _) -> @@ -274,58 +279,73 @@ recv(false, false, TPid, Pkt, _, _) -> incr(TPid, {{unknown, 0}, recv, discarded}), false. -%% spawn_request/6 +%% spawn_request/7 + +spawn_request(false, _, _, _, _, _, _) -> %% no transport + discard; -%% An MFA should return a pid() or the atom 'discard'. The latter -%% results in an acknowledgment back to the transport process when -%% appropriate, to ensure that send/recv callbacks can count -%% outstanding requests. Acknowledgement is implicit if the +%% An MFA should return the pid() of a process in which the argument +%% fun in applied, or the atom 'discard' if the fun is not applied. +%% The latter results in an acknowledgment back to the transport +%% process when appropriate, to ensure that send/recv callbacks can +%% count outstanding requests. Acknowledgement is implicit if the %% handler process dies (in a handle_request callback for example). -spawn_request(Ack, TPid, Pkt, Dict0, RecvData, {M,F,A}) -> - ReqF = fun() -> - ack(Ack, TPid, recv_request(Ack, TPid, Pkt, Dict0, RecvData)) - end, - ack(Ack, TPid, apply(M, F, [ReqF | A])); +spawn_request(AppT, {M,F,A}, Ack, TPid, Pkt, Dict0, RecvData) -> + %% Term to pass to request/1 in an appropriate process. Module + %% diameter_dist implements callbacks. + ReqT = {Pkt, AppT, Ack, TPid, Dict0, RecvData}, + apply(M, F, [ReqT | A]); %% A spawned process acks implicitly when it dies, so there's no need %% to handle 'discard'. -spawn_request(Ack, TPid, Pkt, Dict0, RecvData, Opts) -> +spawn_request(AppT, Opts, Ack, TPid, Pkt, Dict0, RecvData) -> spawn_opt(fun() -> - recv_request(Ack, TPid, Pkt, Dict0, RecvData) + recv_request(Ack, TPid, Pkt, Dict0, RecvData, AppT) end, Opts). +%% request_info/1 +%% +%% Limited request information for diameter_dist. + +request_info({Pkt, _AppT, _Ack, _TPid, _Dict0, RecvData} = _ReqT) -> + {RecvData#recvdata.service_name, Pkt#diameter_packet.bin}. + +%% request/1 +%% +%% Called from a handler process chosen by a transport spawn_opt MFA +%% to process an incoming request. + +request({Pkt, AppT, Ack, TPid, Dict0, RecvData} = _ReqT) -> + ack(Ack, TPid, recv_request(Ack, TPid, Pkt, Dict0, RecvData, AppT)). + %% ack/3 ack(Ack, TPid, RC) -> - RC == discard andalso Ack andalso (TPid ! {send, false}), + RC == discard + andalso Ack + andalso (TPid ! {send, false}), RC. %% --------------------------------------------------------------------------- -%% recv_request/5 +%% recv_request/6 %% --------------------------------------------------------------------------- -spec recv_request(Ack :: boolean(), TPid :: pid(), #diameter_packet{}, Dict0 :: module(), - #recvdata{}) + #recvdata{}, + AppT :: {#diameter_app{}, #diameter_caps{}} + | #diameter_caps{}) %% no suitable app -> ok %% answer was sent - | discard %% or not - | false. %% no transport - -recv_request(Ack, - TPid, - #diameter_packet{header = #diameter_header{application_id = Id}} - = Pkt, - Dict0, - #recvdata{peerT = PeerT, - apps = Apps, - counters = Count} - = RecvData) -> + | discard. %% or not + +recv_request(Ack, TPid, Pkt, Dict0, RecvData, AppT) -> Ack andalso (TPid ! {handler, self()}), - case diameter_service:find_incoming_app(PeerT, TPid, Id, Apps) of + case AppT of {#diameter_app{id = Aid, dictionary = AppDict} = App, Caps} -> + Count = RecvData#recvdata.counters, Count andalso incr(recv, Pkt, TPid, AppDict), DecPkt = decode(Aid, AppDict, RecvData, Pkt), Count andalso incr_error(recv, DecPkt, TPid, AppDict), @@ -349,11 +369,20 @@ recv_request(Ack, Dict0, RecvData, DecPkt, - [[]]); - false = No -> %% transport has gone down - No + [[]]) end. +%% find_app/3 +%% +%% Lookup the application of a received Diameter request on the node +%% on which it's received. + +find_app(TPid, + #diameter_packet{header = #diameter_header{application_id = Id}}, + #recvdata{peerT = PeerT, + apps = Apps}) -> + diameter_service:find_incoming_app(PeerT, TPid, Id, Apps). + %% decode/4 decode(Id, Dict, #recvdata{codec = Opts}, Pkt) -> diff --git a/lib/diameter/src/modules.mk b/lib/diameter/src/modules.mk index bb86de016a..d16292bb88 100644 --- a/lib/diameter/src/modules.mk +++ b/lib/diameter/src/modules.mk @@ -1,7 +1,7 @@ # %CopyrightBegin% # -# Copyright Ericsson AB 2010-2017. All Rights Reserved. +# Copyright Ericsson AB 2010-2019. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -40,6 +40,7 @@ RT_MODULES = \ base/diameter_config \ base/diameter_config_sup \ base/diameter_codec \ + base/diameter_dist \ base/diameter_gen \ base/diameter_lib \ base/diameter_misc_sup \ diff --git a/lib/diameter/test/diameter_dist_SUITE.erl b/lib/diameter/test/diameter_dist_SUITE.erl new file mode 100644 index 0000000000..b2e4c35b9a --- /dev/null +++ b/lib/diameter/test/diameter_dist_SUITE.erl @@ -0,0 +1,332 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2019. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% + +%% +%% Tests of traffic between two Diameter nodes, the server being +%% spread across three Erlang nodes. +%% + +-module(diameter_dist_SUITE). + +-export([suite/0, + all/0]). + +%% testcases +-export([enslave/1, enslave/0, + ping/1, + start/1, + connect/1, + send/1, + stop/1, stop/0]). + +%% diameter callbacks +-export([peer_up/3, + peer_down/3, + pick_peer/4, + prepare_request/3, + prepare_retransmit/3, + handle_answer/4, + handle_error/4, + handle_request/3]). + +-export([call/1]). + +-include("diameter.hrl"). +-include("diameter_gen_base_rfc6733.hrl"). + +%% =========================================================================== + +-define(util, diameter_util). + +-define(CLIENT, 'CLIENT'). +-define(SERVER, 'SERVER'). +-define(REALM, "erlang.org"). +-define(DICT, diameter_gen_base_rfc6733). +-define(ADDR, {127,0,0,1}). + +%% Config for diameter:start_service/2. +-define(SERVICE(Host), + [{'Origin-Host', Host ++ [$.|?REALM]}, + {'Origin-Realm', ?REALM}, + {'Host-IP-Address', [?ADDR]}, + {'Vendor-Id', 12345}, + {'Product-Name', "OTP/diameter"}, + {'Auth-Application-Id', [?DICT:id()]}, + {'Origin-State-Id', origin()}, + {spawn_opt, {diameter_dist, route_session, [#{id => []}]}}, + {sequence, fun sequence/0}, + {string_decode, false}, + {application, [{dictionary, ?DICT}, + {module, ?MODULE}, + {request_errors, callback}, + {answer_errors, callback}]}]). + +-define(SUCCESS, 2001). +-define(BUSY, 3004). +-define(LOGOUT, ?'DIAMETER_BASE_TERMINATION-CAUSE_LOGOUT'). +-define(MOVED, ?'DIAMETER_BASE_TERMINATION-CAUSE_USER_MOVED'). +-define(TIMEOUT, ?'DIAMETER_BASE_TERMINATION-CAUSE_SESSION_TIMEOUT'). + +-define(L, atom_to_list). +-define(A, list_to_atom). + +%% The order here is significant and causes the server to listen +%% before the clients connect. The server listens on the first node, +%% and distributes requests to the other two. +-define(NODES, [{server0, ?SERVER}, + {server1, ?SERVER}, + {server2, ?SERVER}, + {client, ?CLIENT}]). + +%% Options to ct_slave:start/2. +-define(TIMEOUTS, [{T, 15000} || T <- [boot_timeout, + init_timeout, + start_timeout]]). + +%% =========================================================================== + +suite() -> + [{timetrap, {seconds, 60}}]. + +all() -> + [enslave, + ping, + start, + connect, + send, + stop]. + +%% =========================================================================== +%% start/stop testcases + +%% enslave/1 +%% +%% Start four slave nodes, three to implement a Diameter server, +%% one to implement a client. + +enslave() -> + [{timetrap, {seconds, 30*length(?NODES)}}]. + +enslave(Config) -> + Here = filename:dirname(code:which(?MODULE)), + Ebin = filename:join([Here, "..", "ebin"]), + Dirs = [Here, Ebin], + Nodes = [{N,S} || {M,S} <- ?NODES, N <- [slave(M, Dirs)]], + ?util:write_priv(Config, nodes, [{N,S} || {{N,ok},S} <- Nodes]), + [] = [{T,S} || {{_,E} = T, S} <- Nodes, E /= ok]. + +slave(Name, Dirs) -> + add_pathsa(Dirs, ct_slave:start(Name, ?TIMEOUTS)). + +add_pathsa(Dirs, {ok, Node}) -> + {Node, rpc:call(Node, code, add_pathsa, [Dirs])}; +add_pathsa(_, No) -> + {No, error}. + +%% ping/1 +%% +%% Ensure the server nodes are connected so that diameter_dist can attach. + +ping({S, Nodes}) -> + ?SERVER = S, + [N || {N,_} <- Nodes, + node() /= N, + pang <- [net_adm:ping(N)]]; + +ping(Config) -> + Nodes = lists:droplast(?util:read_priv(Config, nodes)), + [] = [{N,RC} || {N,S} <- Nodes, + RC <- [rpc:call(N, ?MODULE, ping, [{S,Nodes}])], + RC /= []]. + +%% start/1 +%% +%% Start diameter services. + +start(SvcName) + when is_atom(SvcName) -> + ok = diameter:start(), + ok = diameter:start_service(SvcName, ?SERVICE((?L(SvcName)))); + +start(Config) -> + Nodes = ?util:read_priv(Config, nodes), + [] = [{N,RC} || {N,S} <- Nodes, + RC <- [rpc:call(N, ?MODULE, start, [S])], + RC /= ok]. + +sequence() -> + sequence(sname()). + +sequence(client) -> + {0,32}; +sequence(Server) -> + "server" ++ N = ?L(Server), + {list_to_integer(N), 30}. + +origin() -> + origin(sname()). + +origin(client) -> + 99; +origin(Server) -> + "server" ++ N = ?L(Server), + list_to_integer(N). + +%% connect/1 +%% +%% Establish one connection from the client, terminated on the first +%% server node, the others handling requests. + +connect({?SERVER, Config, [{Node, _} | _]}) -> + if Node == node() -> %% server0 + ?util:write_priv(Config, lref, {Node, ?util:listen(?SERVER, tcp)}); + true -> + diameter_dist:attach([?SERVER]) + end, + ok; + +connect({?CLIENT, Config, _}) -> + ?util:connect(?CLIENT, tcp, ?util:read_priv(Config, lref)), + ok; + +connect(Config) -> + Nodes = ?util:read_priv(Config, nodes), + [] = [{N,RC} || {N,S} <- Nodes, + RC <- [rpc:call(N, ?MODULE, connect, [{S, Config, Nodes}])], + RC /= ok]. + +%% stop/1 +%% +%% Stop the slave nodes. + +stop() -> + [{timetrap, {seconds, 30*length(?NODES)}}]. + +stop(_Config) -> + [] = [{N,E} || {N,_} <- ?NODES, + {error, _, _} = E <- [ct_slave:stop(N)]]. + +%% =========================================================================== +%% traffic testcases + +%% send/1 +%% +%% Send 100 requests and ensure the node name sent as User-Name isn't +%% the node terminating transport. + +send(Config) -> + send(Config, 100, dict:new()). + +%% send/2 + +send(Config, 0, Dict) -> + [{Server0, _} | _] = ?util:read_priv(Config, nodes) , + Node = atom_to_binary(Server0, utf8), + {false, _} = {dict:is_key(Node, Dict), dict:to_list(Dict)}; + +send(Config, N, Dict) -> + #diameter_base_STA{'Result-Code' = ?SUCCESS, + 'User-Name' = [ServerNode]} + = send(Config, str(?LOGOUT)), + true = is_binary(ServerNode), + send(Config, N-1, dict:update_counter(ServerNode, 1, Dict)). + +%% =========================================================================== + +str(Cause) -> + #diameter_base_STR{'Destination-Realm' = ?REALM, + 'Auth-Application-Id' = ?DICT:id(), + 'Termination-Cause' = Cause}. + +%% send/2 + +send(Config, Req) -> + {Node, _} = lists:last(?util:read_priv(Config, nodes)), + rpc:call(Node, ?MODULE, call, [Req]). + +%% call/1 + +call(Req) -> + diameter:call(?CLIENT, ?DICT, Req, []). + +%% sname/0 + +sname() -> + ?A(hd(string:tokens(?L(node()), "@"))). + +%% =========================================================================== +%% diameter callbacks + +%% peer_up/3 + +peer_up(_SvcName, _Peer, State) -> + State. + +%% peer_down/3 + +peer_down(_SvcName, _Peer, State) -> + State. + +%% pick_peer/4 + +pick_peer([Peer], [], ?CLIENT, _State) -> + {ok, Peer}. + +%% prepare_request/3 + +prepare_request(Pkt, ?CLIENT, {_Ref, Caps}) -> + #diameter_packet{msg = Req} + = Pkt, + #diameter_caps{origin_host = {OH, _}, + origin_realm = {OR, _}} + = Caps, + {send, Req#diameter_base_STR{'Origin-Host' = OH, + 'Origin-Realm' = OR, + 'Session-Id' = diameter:session_id(OH)}}. + +%% prepare_retransmit/3 + +prepare_retransmit(_, ?CLIENT, _) -> + discard. + +%% handle_answer/5 + +handle_answer(Pkt, _Req, ?CLIENT, _Peer) -> + #diameter_packet{msg = Rec, errors = []} = Pkt, + Rec. + +%% handle_error/5 + +handle_error(Reason, _Req, ?CLIENT, _Peer) -> + {error, Reason}. + +%% handle_request/3 + +handle_request(Pkt, ?SERVER, {_, Caps}) -> + #diameter_packet{msg = #diameter_base_STR{'Session-Id' = SId}} + = Pkt, + #diameter_caps{origin_host = {OH, _}, + origin_realm = {OR, _}} + = Caps, + {reply, #diameter_base_STA{'Result-Code' = ?SUCCESS, + 'Session-Id' = SId, + 'Origin-Host' = OH, + 'Origin-Realm' = OR, + 'User-Name' = [atom_to_binary(node(), utf8)]}}. diff --git a/lib/diameter/test/diameter_distribution_SUITE.erl b/lib/diameter/test/diameter_distribution_SUITE.erl index 5146f68ff1..5fe02284ae 100644 --- a/lib/diameter/test/diameter_distribution_SUITE.erl +++ b/lib/diameter/test/diameter_distribution_SUITE.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2013-2016. All Rights Reserved. +%% Copyright Ericsson AB 2013-2019. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -76,6 +76,7 @@ {share_peers, peers()}, {use_shared_peers, peers()}, {restrict_connections, false}, + {spawn_opt, {diameter_dist, spawn_local, []}}, {sequence, fun sequence/0}, {application, [{dictionary, ?DICT}, {module, ?MODULE}, @@ -125,7 +126,7 @@ all() -> %% enslave/1 %% %% Start four slave nodes, one to implement a Diameter server, -%% two three to implement a client. +%% three to implement a client. enslave() -> [{timetrap, {seconds, 30*length(?NODES)}}]. @@ -331,6 +332,8 @@ prepare_request(Pkt, ?CLIENT, {_Ref, Caps}, {_, client0}) -> 'Origin-Realm' = OR, 'Session-Id' = diameter:session_id(OH)}}. +%% prepare_retransmit/4 + prepare_retransmit(Pkt, ?CLIENT, _, {_, client0}) -> #diameter_packet{msg = #diameter_base_STR{'Termination-Cause' = ?MOVED}} = Pkt, %% assert diff --git a/lib/diameter/test/diameter_pool_SUITE.erl b/lib/diameter/test/diameter_pool_SUITE.erl index 97c16940ff..a36a4fa17a 100644 --- a/lib/diameter/test/diameter_pool_SUITE.erl +++ b/lib/diameter/test/diameter_pool_SUITE.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2015-2017. All Rights Reserved. +%% Copyright Ericsson AB 2015-2019. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -51,6 +51,7 @@ {'Auth-Application-Id', [0]}, %% common {'Acct-Application-Id', [3]}, %% accounting {restrict_connections, false}, + {spawn_opt, {diameter_dist, route_session, []}}, {application, [{alias, common}, {dictionary, diameter_gen_base_rfc6733}, {module, diameter_callback}]}, diff --git a/lib/diameter/test/diameter_traffic_SUITE.erl b/lib/diameter/test/diameter_traffic_SUITE.erl index 434aef01dd..47b00c25a2 100644 --- a/lib/diameter/test/diameter_traffic_SUITE.erl +++ b/lib/diameter/test/diameter_traffic_SUITE.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2010-2018. All Rights Reserved. +%% Copyright Ericsson AB 2010-2019. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -539,8 +539,7 @@ add_transports(Config) -> ++ [{unordered, unordered()} || T == sctp], [{capabilities_cb, fun capx/2}, {pool_size, 8} - | server_apps()] - ++ [{spawn_opt, {erlang, spawn, []}} || CS]), + | server_apps()]), Cs = [?util:connect(CN, [T, {sender, CS} | client_opts(T)], LRef, diff --git a/lib/diameter/test/modules.mk b/lib/diameter/test/modules.mk index 0c73adca12..90b0a25d5f 100644 --- a/lib/diameter/test/modules.mk +++ b/lib/diameter/test/modules.mk @@ -1,7 +1,7 @@ # %CopyrightBegin% # -# Copyright Ericsson AB 2010-2017. All Rights Reserved. +# Copyright Ericsson AB 2010-2019. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -31,6 +31,7 @@ MODULES = \ diameter_codec_test \ diameter_config_SUITE \ diameter_compiler_SUITE \ + diameter_dist_SUITE \ diameter_distribution_SUITE \ diameter_dpr_SUITE \ diameter_event_SUITE \ |