Merge pull request #4455 from esl/s2s/listeners

S2S listeners rework

big_tests/default.spec

@@ -98,7 +98,6 @@
 {suites, "tests", xep_0352_csi_SUITE}.
 {suites, "tests", domain_isolation_SUITE}.
 {suites, "tests", domain_removal_SUITE}.
-{suites, "tests", tcp_listener_SUITE}.
 {suites, "tests", cets_disco_SUITE}.
 {suites, "tests", start_node_id_SUITE}.
 {suites, "tests", tr_util_SUITE}.

big_tests/dynamic_domains.spec

@@ -98,7 +98,6 @@
 {suites, "tests", xep_0352_csi_SUITE}.
 {suites, "tests", domain_removal_SUITE}.
 {suites, "tests", local_iq_SUITE}.
-{suites, "tests", tcp_listener_SUITE}.
 {suites, "tests", cets_disco_SUITE}.
 {suites, "tests", start_node_id_SUITE}.
 {suites, "tests", tr_util_SUITE}.

big_tests/tests/bosh_SUITE.erl

@@ -953,14 +953,12 @@ wait_for_zero_bosh_sessions() ->
 
 instrumentation_events() ->
     instrument_helper:declared_events(mod_bosh, [])
-    ++ instrument_helper:declared_events(mongoose_c2s, [global])
-    ++ [{c2s_message_processed, #{host_type => host_type()}}].
+    ++ [{c2s_message_processed, #{host_type => domain_helper:host_type()}}
+        | instrument_helper:declared_events(mongoose_c2s_listener, [])]
+    -- negative_instrumentation_events().
 
 negative_instrumentation_events() ->
-    [{Name, #{}} || Name <- negative_instrumentation_events_names()].
-
-negative_instrumentation_events_names() ->
-    [c2s_tcp_data_out,
-     c2s_tcp_data_in,
-     c2s_tls_data_out,
-     c2s_tls_data_in].
+    [{c2s_tcp_data_out, #{}},
+     {c2s_tcp_data_in, #{}},
+     {c2s_tls_data_out, #{}},
+     {c2s_tls_data_in, #{}}].

big_tests/tests/component_SUITE.erl

@@ -589,7 +589,7 @@ restore_domain(Config) ->
     Config.
 
 events() ->
-    instrument_helper:declared_events(mongoose_component_listener, []).
+    instrument_helper:declared_events(mongoose_component_listener, [#{}]).
 
 %%--------------------------------------------------------------------
 %% Stanzas

big_tests/tests/connect_SUITE.erl

@@ -863,6 +863,5 @@ proxy_info() ->
      }.
 
 instrumentation_events() ->
-    instrument_helper:declared_events(mongoose_c2s_listener, [#{}])
-    ++ instrument_helper:declared_events(mongoose_c2s, [global])
-    ++ [{c2s_message_processed, #{host_type => domain_helper:host_type()}}].
+    [{c2s_message_processed, #{host_type => domain_helper:host_type()}}
+     | instrument_helper:declared_events(mongoose_c2s_listener, [])].

big_tests/tests/mim_c2s_SUITE.erl

@@ -236,9 +236,8 @@ escalus_start(Cfg, FlatCDs) ->
     Clients.
 
 instrumentation_events() ->
-    instrument_helper:declared_events(mongoose_c2s_listener, [#{}])
-    ++ instrument_helper:declared_events(mongoose_c2s, [global])
-    ++ [{c2s_message_processed, #{host_type => domain_helper:host_type()}}].
+    [{c2s_message_processed, #{host_type => domain_helper:host_type()}}
+     | instrument_helper:declared_events(mongoose_c2s_listener, [])].
 
 tcp_instrumentation_events() ->
     [{c2s_tcp_data_out, #{}},

big_tests/tests/s2s_SUITE.erl

@@ -14,7 +14,6 @@
 -include_lib("kernel/include/inet.hrl").
 
 %% Module aliases
--define(dh, distributed_helper).
 -import(distributed_helper, [mim/0, rpc_spec/1, rpc/4]).
 
 %%%===================================================================
@@ -43,7 +42,7 @@ all() ->
     ].
 
 groups() ->
-    [{both_plain, [sequence], all_tests()},
+    [{both_plain, [], all_tests()},
      {both_tls_optional, [], essentials()},
      {both_tls_required, [], essentials()},
 
@@ -89,23 +88,19 @@ connection_cases() ->
 suite() ->
     distributed_helper:require_rpc_nodes([mim, mim2, fed]) ++ escalus:suite().
 
-users() ->
-    [alice2, alice, bob].
-
 %%%===================================================================
 %%% Init & teardown
 %%%===================================================================
 
-init_per_suite(Config) ->
+init_per_suite(Config0) ->
     instrument_helper:start(tested_events()),
     mongoose_helper:inject_module(?MODULE, reload),
-    Config1 = s2s_helper:init_s2s(escalus:init_per_suite(Config)),
-    escalus:create_users(Config1, escalus:get_users(users())).
+    Config1 = escalus:init_per_suite(Config0),
+    s2s_helper:init_s2s(Config1).
 
 end_per_suite(Config) ->
     escalus_fresh:clean(),
     s2s_helper:end_s2s(Config),
-    escalus:delete_users(Config, escalus:get_users(users())),
     escalus:end_per_suite(Config),
     instrument_helper:stop().
 
@@ -195,8 +190,8 @@ connections_info(Config) ->
     simple_message(Config),
     FedDomain = ct:get_config({hosts, fed, domain}),
     %% there should be at least one in and at least one out connection
-    [_ | _] = get_s2s_connections(?dh:mim(), FedDomain, in),
-    [_ | _] = get_s2s_connections(?dh:mim(), FedDomain, out),
+    [_ | _] = get_s2s_connections(mim(), FedDomain, in),
+    [_ | _] = get_s2s_connections(mim(), FedDomain, out),
     ok.
 
 dns_discovery(Config) ->
@@ -222,7 +217,7 @@ dns_discovery_ip_fail(Config) ->
     end).
 
 get_s2s_connections(RPCSpec, Domain, Type) ->
-    AllS2SConnections = ?dh:rpc(RPCSpec, mongoose_s2s_info, get_connections, [Type]),
+    AllS2SConnections = rpc(RPCSpec, mongoose_s2s_info, get_connections, [Type]),
     DomainS2SConnections =
         [Connection || Connection <- AllS2SConnections,
                        Type =/= in orelse [Domain] =:= maps:get(domains, Connection),
@@ -584,6 +579,7 @@ group_with_tls(node1_tls_optional_node2_tls_required_trusted_with_cachain) -> tr
 group_with_tls(_GN) -> false.
 
 tested_events() ->
-    Names = [s2s_xmpp_element_size_in, s2s_xmpp_element_size_out,
-             s2s_tcp_data_in, s2s_tcp_data_out],
-    [{Name, #{}} || Name <- Names].
+    [{s2s_xmpp_element_size_in, #{}},
+     {s2s_xmpp_element_size_out, #{}},
+     {s2s_tcp_data_in, #{}},
+     {s2s_tcp_data_out, #{}}].

big_tests/tests/s2s_helper.erl

@@ -1,45 +1,64 @@
 -module(s2s_helper).
--export([init_s2s/1]).
--export([end_s2s/1]).
--export([configure_s2s/2]).
--export([has_inet_errors/2]).
--export([has_xmpp_server/2]).
+
+-export([init_s2s/1, end_s2s/1, configure_s2s/2, has_inet_errors/2, has_xmpp_server/2]).
 
 -import(distributed_helper, [rpc_spec/1, rpc/4]).
--import(domain_helper, [host_type/1]).
 
 init_s2s(Config) ->
     [{{s2s, NodeKey}, get_s2s_opts(NodeKey)} || NodeKey <- node_keys()] ++
-        [{escalus_user_db, xmpp} | Config].
-
-node_keys() ->
-    [mim, fed].
-
-get_s2s_opts(NodeKey) ->
-    RPCSpec = rpc_spec(NodeKey),
-    S2SOpts = rpc(RPCSpec, mongoose_config, get_opt, [{s2s, host_type(NodeKey)}]),
-    S2SPort = ct:get_config({hosts, NodeKey, incoming_s2s_port}),
-    [S2SListener] = mongoose_helper:get_listeners(RPCSpec, #{port => S2SPort,
-                                                             module => ejabberd_s2s_in}),
-    #{opts => S2SOpts, listener => S2SListener}.
+    [{escalus_user_db, xmpp} | Config].
 
 end_s2s(Config) ->
     [configure_and_restart_s2s(NodeKey, S2SOrig) || {{s2s, NodeKey}, S2SOrig} <- Config],
     ok.
 
+node_keys() ->
+    [mim, fed].
+
 configure_s2s(Group, Config) ->
     TLSPreset = tls_preset(Group),
     [configure_and_restart_s2s(NodeKey, s2s_config(maps:get(NodeKey, TLSPreset), S2SOrig, Config))
      || {{s2s, NodeKey}, S2SOrig} <- Config],
     Config.
 
-s2s_config(plain, S2S = #{opts := Opts}, _) ->
-    S2S#{opts := maps:remove(certfile, Opts#{use_starttls := false})};
+has_inet_errors(History, Server) ->
+    Inet = lists:any(
+        fun({_, {inet, getaddr, [Server1, inet]}, {error, nxdomain}})
+            when Server1 == Server -> true;
+           (_) -> false
+        end, History),
+    Inet6 = lists:any(
+        fun({_, {inet, getaddr, [Server1, inet6]}, {error, nxdomain}})
+            when Server1 == Server -> true;
+           (_) -> false
+        end, History),
+    Inet andalso Inet6.
+
+has_xmpp_server(History, Server) ->
+    lists:all(
+        fun({_, _, {ok, {hostent, "_xmpp-server._tcp." ++ Server1, _, srv, _, _}}})
+            when Server1 == Server -> true;
+           (_) -> false
+        end, History).
+
+get_s2s_opts(NodeKey) ->
+    RPCSpec = rpc_spec(NodeKey),
+    S2SOpts = rpc(RPCSpec, mongoose_config, get_opt, [{s2s, domain_helper:host_type(NodeKey)}]),
+    S2SPort = ct:get_config({hosts, NodeKey, incoming_s2s_port}),
+    [S2SListener] = mongoose_helper:get_listeners(RPCSpec, #{port => S2SPort, module => mongoose_s2s_listener}),
+    #{opts => S2SOpts, listener => S2SListener}.
+
 s2s_config(required_trusted_with_cachain, S2S = #{opts := Opts, listener := Listener}, Config) ->
     #{tls := TLSOpts} = Listener,
     CACertFile = filename:join([path_helper:repo_dir(Config), "tools", "ssl", "ca", "cacert.pem"]),
-    NewTLSOpts = TLSOpts#{cacertfile => CACertFile},
+    NewTLSOpts = TLSOpts#{cacertfile => CACertFile, verify_mode => peer},
+    S2S#{opts := Opts#{use_starttls := required_trusted}, listener := Listener#{tls := NewTLSOpts}};
+s2s_config(required_trusted, S2S = #{opts := Opts, listener := Listener}, _) ->
+    #{tls := TLSOpts} = Listener,
+    NewTLSOpts = TLSOpts#{verify_mode => peer},
     S2S#{opts := Opts#{use_starttls := required_trusted}, listener := Listener#{tls := NewTLSOpts}};
+s2s_config(plain, S2S = #{opts := Opts}, _) ->
+    S2S#{opts := maps:remove(certfile, Opts#{use_starttls := false})};
 s2s_config(StartTLS, S2S = #{opts := Opts}, _) ->
     S2S#{opts := Opts#{use_starttls := StartTLS}}.
 
@@ -67,7 +86,7 @@ tls_preset(node1_tls_optional_node2_tls_required_trusted_with_cachain) ->
     #{mim => optional, fed => required_trusted_with_cachain}.
 
 configure_and_restart_s2s(NodeKey, #{opts := Opts, listener := Listener}) ->
-    HostType = host_type(NodeKey),
+    HostType = domain_helper:host_type(NodeKey),
     set_opt(rpc_spec(NodeKey), [{s2s, HostType}], Opts),
     restart_s2s(rpc_spec(NodeKey), Listener).
 
@@ -79,29 +98,9 @@ restart_s2s(#{} = Spec, S2SListener) ->
     [rpc(Spec, ejabberd_s2s_out, stop_connection, [Pid]) ||
      {_, Pid, _, _} <- Children],
 
-    ChildrenIn = rpc(Spec, supervisor, which_children, [ejabberd_s2s_in_sup]),
-    [rpc(Spec, erlang, exit, [Pid, kill]) ||
-     {_, Pid, _, _} <- ChildrenIn],
+    Children0 = rpc(Spec, supervisor, which_children, [mongoose_listener_sup]),
+    Listeners = [Ref || {Ref, _, _, [mongoose_s2s_listener | _]} <- Children],
+    ChildrenIn = lists:flatten([ranch:procs(Ref, connections) || Ref <- Listeners]),
+    [rpc(Spec, erlang, exit, [Pid, kill]) || {_, Pid, _, _} <- ChildrenIn],
 
     mongoose_helper:restart_listener(Spec, S2SListener).
-
-has_inet_errors(History, Server) ->
-    Inet = lists:any(
-        fun({_, {inet, getaddr, [Server1, inet]}, {error, nxdomain}})
-            when Server1 == Server -> true;
-           (_) -> false
-        end, History),
-    Inet6 = lists:any(
-        fun({_, {inet, getaddr, [Server1, inet6]}, {error, nxdomain}})
-            when Server1 == Server -> true;
-           (_) -> false
-        end, History),
-    Inet andalso Inet6.
-
-has_xmpp_server(History, Server) ->
-    lists:all(
-        fun({_, _, {ok, {hostent, "_xmpp-server._tcp." ++ Server1, _, srv, _, _}}})
-            when Server1 == Server -> true;
-           (_) -> false
-        end, History).
-

big_tests/tests/sasl_external_SUITE.erl

@@ -158,7 +158,10 @@ modify_config_and_restart(CyrsaslExternalConfig, Config) ->
                                     "  tls.certfile = \"priv/ssl/fake_server.pem\"\n"
                                     "  tls.cacertfile = \"" ++ CACertFile ++ "\""
                                     ++ SSLOpts},
-		       {https_config, "tls.certfile = \"priv/ssl/fake_cert.pem\"\n"
+                       {s2s_tls_config, "tls.certfile = \"priv/ssl/fake_server.pem\"\n"
+                                    "  tls.cacertfile = \"" ++ CACertFile ++ "\""
+                                    ++ SSLOpts},
+                       {https_config, "tls.certfile = \"priv/ssl/fake_cert.pem\"\n"
                                       "  tls.keyfile = \"priv/ssl/fake_key.pem\"\n"
                                       "  tls.password = \"\"\n"
                                       "  tls.cacertfile = \"" ++ CACertFile ++ "\""

big_tests/tests/websockets_SUITE.erl

@@ -170,14 +170,12 @@ escape_attrs(Config) ->
 
 instrumentation_events() ->
     instrument_helper:declared_events(mod_websockets, [])
-    ++ instrument_helper:declared_events(mongoose_c2s, [global])
-    ++ [{c2s_message_processed, #{host_type => domain_helper:host_type()}}].
+    ++ [{c2s_message_processed, #{host_type => domain_helper:host_type()}}
+        | instrument_helper:declared_events(mongoose_c2s_listener, [])]
+    -- negative_instrumentation_events().
 
 negative_instrumentation_events() ->
-    [{Name, #{}} || Name <- negative_instrumentation_events_names()].
-
-negative_instrumentation_events_names() ->
-    [c2s_tcp_data_out,
-     c2s_tcp_data_in,
-     c2s_tls_data_out,
-     c2s_tls_data_in].
+    [{c2s_tcp_data_out, #{}},
+     {c2s_tcp_data_in, #{}},
+     {c2s_tls_data_out, #{}},
+     {c2s_tls_data_in, #{}}].