Fix traffic counters for remote request handler processes

Commit f1cdd721 documented the possibility of starting a handler process
for an incoming Diameter request with a configured MFA, but the counter
implementation (ie. {traffic_counters, true}) was broken when this
process was not on the same node as the transport configuration:
incrementing a counter was a noop since the transport pid was not
associated with the transport reference in diameter_stats on the remote
node.

This was a good thing though, since distributing the counters across
handler nodes has a number of problems: not losing counters if the
connection to a handler node is lost, or not being able to remove
counters on a handler node when the transport is removed, keeping track
of where counters are located if they have to be retrieved from multiple
nodes, and so on. Moreover, it's the service/transport configuration on
the local node that orders the counters, so incrementing them on remote
nodes is questionable.

Avoid the problems by incrementing counters where the transport in
question is configured, the node that terminates any associated peer
connection. Instead of calling diameter_stats to increment the counter
when the diameter_peer_fsm process it's being incremented for is not
local, bang the counter to that process to let it call diameter_stats.
The only requirement on the handler node is now that diameter and
application callbacks reside on the code path, and the documentation has
been updated to reflect this.

Compare this to the case of outgoing requests, where counters are
incremented on the originating node. Similarly, it's the local
configuration that orders the counters, and there can be multiple
remote nodes through which the request can be sent on a peer connection
just like there can be multiple handler nodes in the case of an incoming
request.

4 files changed, 57 insertions, 20 deletions

diff --git a/lib/diameter/doc/src/diameter.xml b/lib/diameter/doc/src/diameter.xml
index 85522c99b2..0ab6a122cd 100644
--- a/lib/diameter/doc/src/diameter.xml
+++ b/lib/diameter/doc/src/diameter.xml
@@ -23,7 +23,7 @@
 
 <copyright>
 <year>2011</year>
-<year>2019</year>
+<year>2020</year>
 <holder>Ericsson AB. All Rights Reserved.</holder>
 </copyright>
 <legalnotice>
@@ -1398,10 +1398,11 @@ 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
+invokes <c>diameter_traffic:request/1</c> on the argument 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>
+The handler process need not be local, and diameter need not be
+started on the remote node, but diameter and relevant application
+callbacks must be on the code path.</p>
 
 <p>
 Defaults to the empty list.</p>
diff --git a/lib/diameter/src/base/diameter_peer_fsm.erl b/lib/diameter/src/base/diameter_peer_fsm.erl
index cf5e7f21d3..6ea8fbd571 100644
--- a/lib/diameter/src/base/diameter_peer_fsm.erl
+++ b/lib/diameter/src/base/diameter_peer_fsm.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2010-2018. All Rights Reserved.
+%% Copyright Ericsson AB 2010-2020. 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.
@@ -341,6 +341,11 @@ handle_cast(_, State) ->
 
 %% handle_info/1
 
+%% Counter increment from a remote handler process.
+handle_info({incr, Counter}, State) ->
+    diameter_stats:incr(Counter, self(), 1),
+    {noreply, State};
+
 handle_info(T, #state{} = State) ->
     try transition(T, State) of
         ok ->
diff --git a/lib/diameter/src/base/diameter_traffic.erl b/lib/diameter/src/base/diameter_traffic.erl
index 8423e30269..0336f2392c 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-2019. All Rights Reserved.
+%% Copyright Ericsson AB 2013-2020. 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.
@@ -1238,8 +1238,12 @@ is_result(RC, true, _) ->
 
 %% incr/2
 
+incr(TPid, Counter)
+  when node(TPid) == node() ->
+    diameter_stats:incr(Counter, TPid, 1);
+
 incr(TPid, Counter) ->
-    diameter_stats:incr(Counter, TPid, 1).
+    TPid ! {incr, Counter}.
 
 %% rcc/1
 
diff --git a/lib/diameter/test/diameter_dist_SUITE.erl b/lib/diameter/test/diameter_dist_SUITE.erl
index b2e4c35b9a..2fda2830ae 100644
--- a/lib/diameter/test/diameter_dist_SUITE.erl
+++ b/lib/diameter/test/diameter_dist_SUITE.erl
@@ -1,7 +1,7 @@
 %%
 %% %CopyrightBegin%
 %%
-%% Copyright Ericsson AB 2019. All Rights Reserved.
+%% Copyright Ericsson AB 2019-2020. 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.
@@ -160,15 +160,32 @@ ping(Config) ->
 %%
 %% Start diameter services.
 
-start(SvcName)
-  when is_atom(SvcName) ->
+%% There's no need to start diameter on a node that only services
+%% diameter_dist as a handler of incoming requests, but the
+%% diameter_dist server must be started since the servers communicate
+%% to determine who services what. The typical case is probably that
+%% handler nodes also want to be able to send Diameter requests, in
+%% which case the application needs to be started and diameter_dist is
+%% started as a part of this, but only start the server here to ensure
+%% everything still works as expected.
+start({_SvcName, [_, {S1, _}, {S2, _}, _]}) 
+  when node() == S1;    %% server1
+       node() == S2 ->  %% server2
+    Mod = diameter_dist,
+    {ok, _} = gen_server:start({local, Mod}, Mod, _Args = [], _Opts  = []),
+    ok;
+
+start({SvcName, [{S0, _}, _, _, {C, _}]})
+  when node() == S0;    %% server0
+       node() == C ->   %% client
     ok = diameter:start(),
     ok = diameter:start_service(SvcName, ?SERVICE((?L(SvcName))));
 
-start(Config) ->
+start(Config)
+  when is_list(Config) ->
     Nodes = ?util:read_priv(Config, nodes),
     [] = [{N,RC} || {N,S} <- Nodes,
-                    RC <- [rpc:call(N, ?MODULE, start, [S])],
+                    RC <- [rpc:call(N, ?MODULE, start, [{S, Nodes}])],
                     RC /= ok].
 
 sequence() ->
@@ -194,13 +211,12 @@ origin(Server) ->
 %% 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({?SERVER, Config, [{Node, _} | _]})
+  when Node == node() ->  %% server0
+    ok = ?util:write_priv(Config, lref, {Node, ?util:listen(?SERVER, tcp)});
+
+connect({?SERVER, _Config, _}) -> %% server[12]: register to receive requests
+    ok = diameter_dist:attach([?SERVER]);
 
 connect({?CLIENT, Config, _}) ->
     ?util:connect(?CLIENT, tcp, ?util:read_priv(Config, lref)),
@@ -239,7 +255,18 @@ send(Config) ->
 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)};
+    {false, _} = {dict:is_key(Node, Dict), dict:to_list(Dict)},
+    %% Check that counters have been incremented as expected on server0.
+    [Info] = rpc:call(Server0, diameter, service_info, [?SERVER, connections]),
+    {[Stats], _} = {[S || {statistics, S} <- Info], Info},
+    {[{recv, 1, 100}, {send, 0, 100}], _}
+        = {[{D,R,N} || T <- [recv, send],
+                       {{{0,275,R}, D}, N} <- Stats,
+                       D == T],
+           Stats},
+    {[{send, 0, 100, 2001}], _}
+        = {[{D,R,N,C} || {{{0,275,R}, D, {'Result-Code', C}}, N} <- Stats],
+           Stats};
 
 send(Config, N, Dict) ->
     #diameter_base_STA{'Result-Code' = ?SUCCESS,