Merge pull request #4323 from esl/instrument/user-cache

Instrument `mongoose_user_cache`

big_tests/tests/muc_light_SUITE.erl

@@ -93,7 +93,10 @@
                            user_leave/3,
                            stanza_blocking_set/1,
                            default_config/1,
-                           default_schema/0
+                           default_schema/0,
+                           eq_bjid/2,
+                           pos_int/1,
+                           cache_name/0
                           ]).
 -import(config_parser_helper, [mod_config/2]).
 
@@ -203,14 +206,17 @@ suite() ->
 init_per_suite(Config) ->
     Config1 = dynamic_modules:save_modules(host_type(), Config),
     Config2 = escalus:init_per_suite(Config1),
+    instrument_helper:start([{user_cache_lookup, #{host_type => host_type(),
+                                                   cache_name => cache_name()}}]),
     escalus:create_users(Config2, escalus:get_users([alice, bob, kate, mike])).
 
 end_per_suite(Config) ->
     HostType = host_type(),
     muc_light_helper:clear_db(HostType),
     Config1 = escalus:delete_users(Config, escalus:get_users([alice, bob, kate, mike])),
     dynamic_modules:restore_modules(Config),
-    escalus:end_per_suite(Config1).
+    escalus:end_per_suite(Config1),
+    instrument_helper:stop().
 
 init_per_group(_GroupName, Config) ->
     Config.
@@ -515,11 +521,14 @@ unauthorized_stanza(Config) ->
 
 send_message(Config) ->
     escalus:story(Config, [{alice, 1}, {bob, 1}, {kate, 1}], fun(Alice, Bob, Kate) ->
+            TS = instrument_helper:timestamp(),
             Msg = <<"Heyah!">>,
             Id = <<"MyID">>,
             Stanza = escalus_stanza:set_id(
                        escalus_stanza:groupchat_to(room_bin_jid(?ROOM), Msg), Id),
-            foreach_occupant([Alice, Bob, Kate], Stanza, gc_message_verify_fun(?ROOM, Msg, Id))
+            foreach_occupant([Alice, Bob, Kate], Stanza, gc_message_verify_fun(?ROOM, Msg, Id)),
+            assert_cache_miss_event(TS, 1, room_bin_jid(?ROOM)),
+            assert_cache_hit_event(TS, 2, room_bin_jid(?ROOM))
         end).
 
 change_subject(Config) ->
@@ -714,8 +723,10 @@ destroy_room_get_disco_items_one_left(Config) ->
 
 set_config(Config) ->
     escalus:story(Config, [{alice, 1}, {bob, 1}, {kate, 1}], fun(Alice, Bob, Kate) ->
+            TS = instrument_helper:timestamp(),
             ConfigChange = [{<<"roomname">>, <<"The Coven">>}],
-            change_room_config(Alice, [Alice, Bob, Kate], ConfigChange)
+            change_room_config(Alice, [Alice, Bob, Kate], ConfigChange),
+            assert_cache_miss_event(TS, 1, room_bin_jid(?ROOM))
         end).
 
 set_partial_config(Config) ->
@@ -768,14 +779,17 @@ assorted_config_doesnt_lead_to_duplication(Config) ->
 
 remove_and_add_users(Config) ->
     escalus:story(Config, [{alice, 1}, {bob, 1}, {kate, 1}], fun(Alice, Bob, Kate) ->
+            TS = instrument_helper:timestamp(),
             AffUsersChanges1 = [{Bob, none}, {Kate, none}],
             escalus:send(Alice, stanza_aff_set(?ROOM, AffUsersChanges1)),
             verify_aff_bcast([{Alice, owner}], AffUsersChanges1),
             escalus:assert(is_iq_result, escalus:wait_for_stanza(Alice)),
             AffUsersChanges2 = [{Bob, member}, {Kate, member}],
             escalus:send(Alice, stanza_aff_set(?ROOM, AffUsersChanges2)),
             verify_aff_bcast([{Alice, owner}, {Bob, member}, {Kate, member}], AffUsersChanges2),
-            escalus:assert(is_iq_result, escalus:wait_for_stanza(Alice))
+            escalus:assert(is_iq_result, escalus:wait_for_stanza(Alice)),
+            assert_cache_miss_event(TS, 1, room_bin_jid(?ROOM)),
+            assert_cache_hit_event(TS, 1, room_bin_jid(?ROOM))
         end).
 
 explicit_owner_change(Config) ->
@@ -1140,6 +1154,24 @@ assert_process_memory_not_growing(Pid, OldMemory, Counter) ->
                  end,
   assert_process_memory_not_growing(Pid, Memory, NewCounter).
 
+assert_cache_hit_event(TS, Count, Jid) ->
+    assert_cache_event(TS, Count, Jid, hits).
+
+assert_cache_miss_event(TS, Count, Jid) ->
+    assert_cache_event(TS, Count, Jid, misses).
+
+assert_cache_event(TS, Count, BinJid, CacheResult) ->
+    F = fun(#{jid := Jid, CacheResult := 1, latency := T}) ->
+            eq_bjid(Jid, BinJid) andalso pos_int(T)
+        end,
+    instrument_helper:assert(
+        user_cache_lookup,
+        #{host_type => host_type(), cache_name => cache_name()},
+        F,
+        % Due to implementation, for every lookup the event is raised two times
+        #{min_timestamp => TS, expected_count => Count * 2}
+    ).
+
 -spec custom_schema() -> [muc_light_helper:schema_item()].
 custom_schema() ->
     % This list needs to be sorted

big_tests/tests/muc_light_helper.erl

@@ -32,6 +32,14 @@ muc_host() ->
 muc_host_pattern() ->
     ct:get_config({hosts, mim, muc_light_service_pattern}).
 
+cache_name() ->
+    case domain_helper:host_type() of
+        <<"localhost">> ->
+            mod_muc_light_cache_localhost;
+        <<"test type">> ->
+            'mod_muc_light_cache_test type'
+    end.
+
 create_room(RoomU, MUCHost, Owner, Members, Config, Version) ->
     DefaultConfig = default_internal_config(MUCHost),
     RoomUS = {RoomU, MUCHost},
@@ -299,3 +307,7 @@ stanza_blocking_set(BlocklistChanges) ->
                      children = [#xmlcdata{ content = Who }] }
              || {What, Action, Who} <- BlocklistChanges],
     escalus_stanza:to(escalus_stanza:iq_set(?NS_MUC_LIGHT_BLOCKING, Items), muc_host()).
+
+eq_bjid(Jid, BinJid) -> Jid =:= jid:from_binary(BinJid).
+
+pos_int(T) -> is_integer(T) andalso T > 0.

src/instrument/mongoose_instrument.erl

@@ -24,8 +24,9 @@
 -type event_name() :: atom().
 -type labels() :: #{host_type => mongooseim:host_type(),
                     function => atom(),
+                    cache_name => atom(),
                     pool_tag => mongoose_wpool:tag()}. % to be extended
--type label_key() :: host_type | function | pool_tag. % to be extended
+-type label_key() :: host_type | function | cache_name | pool_tag. % to be extended
 -type label_value() :: mongooseim:host_type() | atom() | mongoose_wpool:tag(). % to be extended
 -type metrics() :: #{metric_name() => metric_type()}.
 -type metric_name() :: atom().

src/mongoose_user_cache.erl

@@ -2,20 +2,34 @@
 
 -include("mongoose_config_spec.hrl").
 
--export([start_new_cache/3, stop_cache/2, handle_telemetry_event/4,
+-export([start_new_cache/3, stop_cache/2,
          config_spec/0, process_cache_config/1]).
 
 -export([is_member/3, get_entry/3, merge_entry/4, delete_user/3, delete_domain/3]).
 
 -spec is_member(mongooseim:host_type(), module(), jid:jid()) -> boolean().
 is_member(HostType, Module, Jid) ->
     CacheName = cache_name(HostType, Module),
-    segmented_cache:is_member(CacheName, key(Jid)).
+    mongoose_instrument:span(user_cache_lookup, #{host_type => HostType, cache_name => CacheName},
+                             fun segmented_cache:is_member/2, [CacheName, key(Jid)],
+                             fun(Time, Result) -> handle_is_member_result(Time, Result, Jid) end).
+
+handle_is_member_result(Time, false, Jid) ->
+    #{misses => 1, latency => Time, jid => Jid};
+handle_is_member_result(Time, true, Jid) ->
+    #{hits => 1, latency => Time, jid => Jid}.
 
 -spec get_entry(mongooseim:host_type(), module(), jid:jid()) -> term() | not_found.
 get_entry(HostType, Module, Jid) ->
     CacheName = cache_name(HostType, Module),
-    segmented_cache:get_entry(CacheName, key(Jid)).
+    mongoose_instrument:span(user_cache_lookup, #{host_type => HostType, cache_name => CacheName},
+                             fun segmented_cache:get_entry/2, [CacheName, key(Jid)],
+                             fun(Time, Result) -> handle_get_entry_result(Time, Result, Jid) end).
+
+handle_get_entry_result(Time, not_found, Jid) ->
+    #{misses => 1, latency => Time, jid => Jid};
+handle_get_entry_result(Time, _Entry, Jid) ->
+    #{hits => 1, latency => Time, jid => Jid}.
 
 -spec merge_entry(mongooseim:host_type(), module(), jid:jid(), map()) -> boolean().
 merge_entry(HostType, Module, Jid, Entry) ->
@@ -74,31 +88,17 @@ do_start_new_cache(HostType, Module, Opts) ->
              restart => permanent, shutdown => 5000,
              type => worker, modules => [segmented_cache]},
     {ok, _} = ejabberd_sup:start_child(Spec),
-    create_metrics(HostType, Module, CacheName),
+    mongoose_instrument:set_up(instrumentation(HostType, CacheName)),
     ok.
 
-create_metrics(HostType, Module, CacheName) ->
-    telemetry:attach(CacheName,
-                     [segmented_cache, CacheName, request, stop],
-                     fun ?MODULE:handle_telemetry_event/4,
-                     #{host_type => HostType, cache_name => CacheName, module => Module}),
-    mongoose_metrics:ensure_metric(HostType, [Module, hit], counter),
-    mongoose_metrics:ensure_metric(HostType, [Module, miss], counter),
-    mongoose_metrics:ensure_metric(HostType, [Module, latency], histogram).
-
-handle_telemetry_event([segmented_cache, CacheName, request, stop],
-                       #{duration := Latency},
-                       #{hit := Hit},
-                       #{host_type := HostType, cache_name := CacheName, module := Module}) ->
-    case Hit of
-        true -> mongoose_metrics:update(HostType, [Module, hit], 1);
-        false -> mongoose_metrics:update(HostType, [Module, miss], 1)
-    end,
-    mongoose_metrics:update(HostType, [Module, latency], Latency),
-    ok.
+instrumentation(HostType, CacheName) ->
+    [{user_cache_lookup, #{cache_name => CacheName, host_type => HostType},
+      #{metrics => #{hits => counter, misses => counter, latency => histogram}}}].
 
 -spec stop_cache(mongooseim:host_type(), module()) -> ok.
 stop_cache(HostType, Module) ->
+    CacheName = gen_mod:get_module_proc(HostType, Module),
+    mongoose_instrument:tear_down(instrumentation(HostType, CacheName)),
     case gen_mod:get_module_opt(HostType, Module, module, internal) of
         internal -> ok = ejabberd_sup:stop_child(cache_name(HostType, Module));
         _ConfiguredModule -> ok

test/batches_SUITE.erl

@@ -43,11 +43,13 @@ groups() ->
     ].
 
 init_per_suite(Config) ->
-    application:ensure_all_started(telemetry),
     meck:new(mongoose_metrics, [stub_all, no_link]),
-    Config.
+    mongoose_config:set_opts(opts()),
+    async_helper:start(Config, mongoose_instrument, start_link, []).
 
-end_per_suite(_Config) ->
+end_per_suite(Config) ->
+    async_helper:stop_all(Config),
+    mongoose_config:erase_opts(),
     meck:unload().
 
 init_per_group(_, Config) ->
@@ -66,6 +68,9 @@ end_per_testcase(_TestCase, _Config) ->
     meck:unload(gen_mod),
     ok.
 
+opts() ->
+    #{instrumentation => config_parser_helper:default_config([instrumentation])}.
+
 cache_config() ->
     config_parser_helper:default_mod_config(mod_cache_users).