Traffic monitor

big_tests/tests/traffic_monitor_SUITE.erl

@@ -0,0 +1,135 @@
+-module(traffic_monitor_SUITE).
+-compile([export_all, nowarn_export_all]).
+
+-include_lib("common_test/include/ct.hrl").
+-include_lib("eunit/include/eunit.hrl").
+-include_lib("escalus/include/escalus_xmlns.hrl").
+-include_lib("exml/include/exml.hrl").
+-include_lib("jid/include/jid.hrl").
+
+%%--------------------------------------------------------------------
+%% Suite configuration
+%%--------------------------------------------------------------------
+
+
+all() ->
+    [
+        {group, regular}
+    ].
+
+groups() ->
+    [{regular, [], [get_page, tracing_from_websocket]}].
+
+suite() ->
+    escalus:suite().
+
+init_per_suite(Config) ->
+    mongoose_helper:inject_module(?MODULE),
+    escalus:init_per_suite(Config).
+
+end_per_suite(Config) ->
+    escalus:end_per_suite(Config).
+
+init_per_group(GroupName, Config) when GroupName =:= regular; GroupName =:= async_pools ->
+    HostType = domain_helper:host_type(),
+    SecHostType = domain_helper:secondary_host_type(),
+    Config1 = dynamic_modules:save_modules(HostType, Config),
+    Config2 = dynamic_modules:save_modules(SecHostType, Config1),
+    ok = dynamic_modules:ensure_modules(HostType, [{mongoose_traffic, #{}}]),
+    init_listeners(),
+    Config2.
+
+init_per_testcase(get_page, Config) ->
+    {ok, Client} = fusco:start("http://localhost:5111", []),
+    init_per_testcase(generic, [{http_client, Client} | Config]);
+init_per_testcase(tracing_from_websocket, Config) ->
+    {ok, Wsc} = gun:open("localhost", 5111, #{transport => tcp, protocols => [http], retry => 1}),
+    StreamRef = gun:ws_upgrade(Wsc, "/ws-traffic", [], #{}),
+    receive
+        {gun_upgrade, Wsc, StreamRef, [<<"websocket">>], _} -> ok
+    after 1000 ->
+        ct:fail("gun did not fire")
+    end,
+    init_per_testcase(generic, [{ws_client, {Wsc, StreamRef}} | Config]);
+init_per_testcase(CaseName, Config) ->
+    escalus:init_per_testcase(CaseName, Config).
+
+end_per_testcase(get_page, Config) ->
+    C = proplists:get_value(http_client, Config),
+    fusco:disconnect(C),
+    end_per_testcase(generic, Config);
+end_per_testcase(tracing_from_websocket, Config) ->
+    {C, _} = proplists:get_value(ws_client, Config),
+    gun:close(C),
+    end_per_testcase(generic, Config);
+end_per_testcase(CaseName, Config) ->
+    escalus:end_per_testcase(CaseName, Config).
+
+end_per_group(GroupName, Config) when GroupName =:= regular ->
+    escalus_fresh:clean(),
+    dynamic_modules:restore_modules(Config).
+
+get_page(Config) ->
+    % just to make sure listener works
+    C = proplists:get_value(http_client, Config),
+    {ok, Res} = fusco:request(C, <<"/traffic">>, <<"GET">>, [], [], 5000),
+    {{<<"200">>, <<"OK">>}, _, _, _, _} = Res,
+    ok.
+
+tracing_from_websocket(Config) ->
+    C = proplists:get_value(ws_client, Config),
+    send(C, <<"get_status">>, #{}),
+    receive_status(false),
+    send(C, <<"trace_flag">>, #{<<"value">>  => true}),
+    receive_status(true),
+    send(C, <<"get_status">>, #{}),
+    receive_status(true),
+    escalus:fresh_story(Config, [{carol, 1}], fun(Alice) ->
+            {<<"new_trace">>, #{<<"bare_jid">> := <<>>}} = receive_msg(),
+            {<<"new_trace">>, #{<<"full_jid">> := <<>>, <<"pid">> := BPid}} = receive_msg(),
+            send(C, <<"get_trace">>, #{<<"pid">>  => BPid}),
+            {<<"get_trace">>, Trace} = receive_msg(),
+            [_ | _] = maps:get(<<"trace">>, Trace),
+            Server = escalus_client:server(Alice),
+            escalus:send(Alice, escalus_stanza:to(escalus_stanza:iq_get(?NS_DISCO_INFO, []), Server)),
+            {<<"message">>, _} = receive_msg(),
+            ok
+        end),
+    ok.
+
+init_listeners() ->
+    Opts = #{connection_type => undefined,
+             handlers =>
+                 [#{host => '_',module => mongoose_traffic, path => "/traffic/[...]"},
+                  #{host => '_',module => mongoose_traffic_channel, path => "/ws-traffic"}],
+             ip_address => "0",
+             ip_tuple => {0,0,0,0},
+             ip_version => 4,
+             module => ejabberd_cowboy,
+             port => 5111,
+             proto => tcp,
+             protocol => #{compress => false},
+             transport => #{max_connections => 1024,num_acceptors => 2}},
+    distributed_helper:rpc(distributed_helper:mim(), ejabberd_cowboy,
+        start_listener, [Opts]),
+    ok.
+
+receive_status(Expected) ->
+    {Evt, Data} = receive_msg(),
+    ?assertEqual(<<"status">>, Evt),
+    ?assertEqual(Expected, maps:get(<<"trace_flag">>, Data, missing)).
+
+receive_msg() ->
+    receive
+        {gun_ws, _, _, {text, Bin}} ->
+            {Data} = jiffy:decode(Bin),
+            Event = proplists:get_value(<<"event">>, Data),
+            {Payload} = proplists:get_value(<<"payload">>, Data),
+            {Event, maps:from_list(Payload)}
+    after 100 ->
+        ct:fail("message not received")
+    end.
+
+send({C, Stream}, Event, Payload) ->
+    Data = #{event => Event, payload => Payload},
+    gun:ws_send(C, Stream, {text, jiffy:encode(Data)}).

doc/operation-and-maintenance/Logging-&-monitoring.md

@@ -146,3 +146,9 @@ Parts of the names are indexed from `0`.
 Time-based metrics in MongooseIM are given in **microseconds**, so to display human-readable values in graph's legend, the Y-axis unit has to be edited on the `Axes` tab.
 
 [Logger]: https://erlang.org/doc/apps/kernel/logger_chapter.html#handlers
+
+## Traffic monitor
+
+For debugging purposes, especially while developing MIM additional modules or client applications,
+you may want to use browser-based traffic monitor. It is downloaded as a dependency in installed
+in /web/traffic subdirectory. Consult README for details.

rebar.config

@@ -118,7 +118,8 @@
   {jwerl, "1.2.0"},
   {cpool, "0.1.0"}, %% Do not upgrade cpool to version 0.1.1, it has bugs
   {nkpacket, {git, "https://github.com/esl/nkpacket.git", {branch, "mongooseim-ranch-compatibility"}}},
-  {nksip, {git, "https://github.com/arcusfelis/nksip.git", {branch, "mu-fix-dialyzer"}}}
+  {nksip, {git, "https://github.com/arcusfelis/nksip.git", {branch, "mu-fix-dialyzer"}}},
+  {traffic_monitor, {git, "https://github.com/bartekgorny/mongooseim_traffic_monitor.git", {branch, "master"}}}
  ]}.
 
 {relx, [{release, { mongooseim, {cmd, "cat VERSION | tr -d '\r\n'"} },
@@ -147,6 +148,8 @@
                    {copy, "rel/files/scripts",   "./"},
                    {copy, "rel/files/templates", "./"},
                    {copy, "rel/files/templates.ini", "etc/templates.ini"},
+                   {mkdir, "web"},
+                   {copy, "_build/default/lib/traffic_monitor", "web/traffic"},
 
                    {template, "rel/files/nodetool",     "erts-\{\{erts_vsn\}\}/bin/nodetool"},
 

rebar.lock

@@ -125,6 +125,10 @@
  {<<"thoas">>,{pkg,<<"thoas">>,<<"1.0.0">>},2},
  {<<"tirerl">>,{pkg,<<"tirerl">>,<<"1.2.0">>},0},
  {<<"tomerl">>,{pkg,<<"tomerl">>,<<"0.5.0">>},0},
+ {<<"traffic_monitor">>,
+  {git,"https://github.com/bartekgorny/mongooseim_traffic_monitor.git",
+       {ref,"f80a45dad104fd72eed7096c788962dab7139fea"}},
+  0},
  {<<"trails">>,{pkg,<<"trails">>,<<"2.3.0">>},0},
  {<<"unicode_util_compat">>,{pkg,<<"unicode_util_compat">>,<<"0.7.0">>},1},
  {<<"uuid">>,{pkg,<<"uuid_erl">>,<<"2.0.5">>},0},

rel/fed1.vars-toml.config

@@ -12,6 +12,7 @@
 {http_graphql_api_user_endpoint_port, 5566}.
 {http_api_endpoint_port, 5294}.
 {http_api_client_endpoint_port, 8095}.
+{traffic_channel_port, 5115}.
 
 %% This node is for s2s testing.
 %% "localhost" host should NOT be defined.

rel/files/mongooseim.toml

@@ -38,6 +38,18 @@
     host = "_"
     path = "/ws-xmpp"
 
+# [[listen.http]]
+#   port = {{traffic_channel_port}}
+#   transport.num_acceptors = 2
+#
+#   [[listen.http.handlers.mongoose_traffic_channel]]
+#     host = "_"
+#     path = "/ws-traffic"
+#
+#   [[listen.http.handlers.mongoose_traffic]]
+#     host = "_"
+#     path = "/traffic/[...]"
+
 [[listen.http]]
   port = {{{https_port}}}
   transport.num_acceptors = 10
@@ -197,6 +209,8 @@
 {{/service_domain_db}}
 [modules.mod_adhoc]
 
+# [modules.mongoose_traffic]
+
 {{#mod_amp}}
 [modules.mod_amp]
 {{{mod_amp}}}

rel/mim1.vars-toml.config

@@ -16,6 +16,7 @@
 {http_graphql_api_user_endpoint_port, 5561}.
 {http_api_endpoint_port, 8088}.
 {http_api_client_endpoint_port, 8089}.
+{traffic_channel_port, 5111}.
 
 {hosts, "\"localhost\", \"anonymous.localhost\", \"localhost.bis\""}.
 {host_types, "\"test type\", \"dummy auth\", \"anonymous\""}.

rel/mim2.vars-toml.config

@@ -14,6 +14,7 @@
 {http_graphql_api_admin_endpoint_port, 5552}.
 {http_graphql_api_domain_admin_endpoint_port, 5542}.
 {http_graphql_api_user_endpoint_port, 5562}.
+{traffic_channel_port, 5112}.
 
 {hosts, "\"localhost\", \"anonymous.localhost\", \"localhost.bis\""}.
 {host_types, "\"test type\", \"dummy auth\""}.

rel/mim3.vars-toml.config

@@ -14,6 +14,7 @@
 {http_graphql_api_user_endpoint_port, 5563}.
 {http_api_endpoint_port, 8092}.
 {http_api_client_endpoint_port, 8193}.
+{traffic_channel_port, 5113}.
 
 "./vars-toml.config".
 

rel/reg1.vars-toml.config

@@ -13,6 +13,7 @@
 {http_graphql_api_user_endpoint_port, 5564}.
 {http_api_endpoint_port, 8074}.
 {http_api_client_endpoint_port, 8075}.
+{traffic_channel_port, 5114}.
 
 %% This node is for global distribution testing.
 %% reg is short for region.

src/c2s/mongoose_c2s.erl

@@ -239,6 +239,10 @@ handle_socket_packet(StateData = #c2s_data{parser = Parser}, Packet) ->
             NextEvent = {next_event, internal, #xmlstreamerror{name = iolist_to_binary(Reason)}},
             {keep_state, StateData, NextEvent};
         {ok, NewParser, XmlElements} ->
+            lists:foreach(fun(Element) ->
+                              mongoose_hooks:c2s_debug(no_acc,
+                                                       {client_to_server, StateData#c2s_data.jid, Element})
+                          end, XmlElements),
             Size = iolist_size(Packet),
             NewStateData = StateData#c2s_data{parser = NewParser},
             handle_socket_elements(NewStateData, XmlElements, Size)
@@ -1009,9 +1013,11 @@ maybe_send_xml(StateData, Acc, ToSend) ->
 
 -spec do_send_element(data(), mongoose_acc:t(), exml:element()) -> mongoose_acc:t().
 do_send_element(StateData = #c2s_data{host_type = undefined}, Acc, El) ->
+    mongoose_hooks:c2s_debug(Acc, {server_to_client, StateData#c2s_data.jid, El}),
     send_xml(StateData, El),
     Acc;
 do_send_element(StateData = #c2s_data{host_type = HostType}, Acc, #xmlel{} = El) ->
+    mongoose_hooks:c2s_debug(Acc, {server_to_client, StateData#c2s_data.jid, El}),
     Res = send_xml(StateData, El),
     Acc1 = mongoose_acc:set(c2s, send_result, Res, Acc),
     mongoose_hooks:xmpp_send_element(HostType, Acc1, El).

src/hooks/mongoose_hooks.erl

@@ -47,6 +47,7 @@
 -export([get_pep_recipients/2,
          filter_pep_recipient/3,
          c2s_stream_features/3,
+         c2s_debug/2,
          check_bl_c2s/1,
          forbidden_session_hook/3,
          session_opening_allowed_for_user/2]).
@@ -535,6 +536,12 @@ sasl2_start(HostType, Acc, Element) ->
 sasl2_success(HostType, Acc, Params) ->
     run_hook_for_host_type(sasl2_success, HostType, Acc, Params).
 
+-spec c2s_debug(Acc, Arg) -> mongoose_acc:t() when
+    Acc :: mongoose_acc:t() | no_acc,
+    Arg :: mongoose_debug:debug_entry().
+c2s_debug(Acc, Arg) ->
+    run_global_hook(c2s_debug, Acc, #{arg => Arg}).
+
 -spec check_bl_c2s(IP) -> Result when
     IP ::  inet:ip_address(),
     Result :: boolean().

src/mongoose_debug.erl

@@ -0,0 +1,67 @@
+-module(mongoose_debug).
+
+%% The most simple use case possible is:
+%% - add [modules.mongoose_debug] to mongooseim.toml
+%% - from erlang shell, run recon_trace:calls([{mongoose_debug, traffic, '_'}], 100, [{scope, local}]).
+%% - watch all the traffic coming in and out
+
+-behaviour(gen_mod).
+
+-include("mongoose.hrl").
+-include("jlib.hrl").
+-include_lib("exml/include/exml_stream.hrl").
+
+-type debug_entry() :: {client_to_server, jid:jid() | undefined, exml:element()}| {server_to_client, jid:jid(), exml:element()}.
+-type direction() :: client_to_server | server_to_client.
+-export_type([debug_entry/0, direction/0]).
+
+%% API
+-export([start/2, stop/1]).
+-export([trace_traffic/3]).
+-export([supported_features/0]).
+
+-spec start(mongooseim:host_type(), gen_mod:module_opts()) -> ok.
+start(Host, _Opts) ->
+    gen_hook:add_handlers(hooks(Host)),
+    ok.
+
+-spec stop(mongooseim:host_type()) -> ok.
+stop(Host) ->
+    gen_hook:delete_handlers(hooks(Host)),
+    ok.
+
+hooks(_Host) ->
+    [{c2s_debug, global, fun ?MODULE:trace_traffic/3, #{}, 50}].
+
+
+-spec trace_traffic(mongoose_acc:t(), #{arg => debug_entry()}, term()) ->
+    {ok, mongoose_acc:t()}.
+trace_traffic(Acc, #{arg := {client_to_server, From, El}}, _) ->
+    Sfrom = binary_to_list(maybe_jid_to_binary(From)),
+    Sto = binary_to_list(get_attr(El, <<"to">>)),
+    St = exml:to_binary(El),
+    Marker = " C >>>> MiM ",
+    traffic(Sfrom, Marker, Sto, St),
+    {ok, Acc};
+trace_traffic(Acc, #{arg := {server_to_client, To, El}}, _) ->
+    Sto = binary_to_list(maybe_jid_to_binary(To)),
+    Sfrom = binary_to_list(get_attr(El, <<"from">>)),
+    St = exml:to_binary(El),
+    Marker = " C <<<< MiM ",
+    traffic(Sfrom, Marker, Sto, St),
+    {ok, Acc}.
+
+traffic(_Sender, _Marker, _Recipient, _Stanza) -> ok.
+
+-spec supported_features() -> [atom()].
+supported_features() -> [dynamic_domains].
+
+maybe_jid_to_binary(undefined) -> <<" ">>;
+maybe_jid_to_binary(J) -> jid:to_binary(J).
+
+get_attr(#xmlstreamstart{attrs = AttrList}, AttrName) ->
+    proplists:get_value(AttrName, AttrList, <<" ">>);
+get_attr(#xmlstreamend{}, _) ->
+    <<" ">>;
+get_attr(El, AttrName) ->
+    exml_query:attr(El, AttrName, <<" ">>).