Customise telemetry event name

Author: NelsonVides

src/segmented_cache.erl

@@ -47,6 +47,7 @@ For more information, see the README, and the function documentation.
 -type merger_fun(Value) :: fun((Value, Value) -> Value).
 ?DOC("Configuration values for the cache.").
 -type opts() :: #{
+    prefix => telemetry:event_name(),
     scope => scope(),
     strategy => strategy(),
     entries_limit => entries_limit(),
@@ -94,6 +95,8 @@ Start and link a cache entity in the local node.
 and an entry in persistent_term will be created and the worker will join a pg group of
 the same name.
 `Opts` is a map containing the configuration.
+- `prefix` is a `telemetry` event name to prefix events raised by this library.
+    Defaults to `[segmented_cache, Name]`.
 - `scope` is a `pg` scope. Defaults to `pg`.
 - `strategy` can be fifo or lru. Default is `fifo`.
 - `segment_num` is the number of segments for the cache. Default is `3`
@@ -110,27 +113,25 @@ start_link(Name, Opts) when is_atom(Name), is_map(Opts) ->
 Check if Key is cached.
 
 Raises a telemetry span:
-- name: `[segmented_cache, Name, request, _]`
+- name: `Prefix ++ [request]`
 - start metadata: `#{name => atom()}`
 - stop metadata: `t:hit/0`
 """).
 -spec is_member(name(), key()) -> boolean().
 is_member(Name, Key) when is_atom(Name) ->
-    Span = segmented_cache_helpers:is_member_span(Name, Key),
-    telemetry:span([segmented_cache, Name, request], #{name => Name, type => is_member}, Span).
+    segmented_cache_helpers:is_member(Name, Key).
 
 ?DOC("""
 Get the entry for Key in cache.
 
 Raises telemetry span:
-- name: `[segmented_cache, Name, request, _]`
+- name: `Prefix ++ [request]`
 - start metadata: `#{name => atom()}`
 - stop metadata: `t:hit/0`
 """).
 -spec get_entry(name(), key()) -> value() | not_found.
 get_entry(Name, Key) when is_atom(Name) ->
-    Span = segmented_cache_helpers:get_entry_span(Name, Key),
-    telemetry:span([segmented_cache, Name, request], #{name => Name, type => get_entry}, Span).
+    segmented_cache_helpers:get_entry(Name, Key).
 
 ?DOC("""
 Add an entry to the first table in the segments.
@@ -182,7 +183,7 @@ merge_entry(Name, Key, Value) when is_atom(Name) ->
 Delete an entry in all ets segments.
 
 Might raise a telemetry error if the request fails:
-- name: `[segmented_cache, Name, delete_error]`
+- name: `Prefix ++ [request, delete_error]`
 - measurements: `#{}`
 - metadata: `t:delete_error/1`
 """).

src/segmented_cache_helpers.erl

@@ -10,22 +10,22 @@
 -compile({inline, [next/2]}).
 
 -export([init_cache_config/2, get_cache_scope/1, erase_cache_config/1]).
--export([is_member_span/2, get_entry_span/2, put_entry_front/3, merge_entry/3]).
+-export([is_member/2, get_entry/2]).
+-export([put_entry_front/3, merge_entry/3]).
 -export([delete_entry/2, delete_pattern/2]).
 -export([purge_last_segment_and_rotate/1]).
 
 -record(segmented_cache, {
     scope :: segmented_cache:scope(),
     name :: segmented_cache:name(),
+    telemetry_name :: [segmented_cache:name()],
     strategy = fifo :: segmented_cache:strategy(),
     entries_limit = infinity :: segmented_cache:entries_limit(),
     index :: atomics:atomics_ref(),
     segments :: tuple(),
     merger_fun :: merger_fun(dynamic())
 }).
 
--type span() :: fun(() -> {dynamic(), span_metadata()}).
--type span_metadata() :: #{hit := boolean()}.
 -type merger_fun(Value) :: fun((Value, Value) -> Value).
 -type iterative_fun(Key, Value) :: fun((ets:tid(), Key) -> {continue | stop, Value}).
 -type config() :: #segmented_cache{}.
@@ -43,15 +43,18 @@ init_cache_config(Name, Opts0) ->
         entries_limit := EntriesLimit,
         segment_num := N,
         ttl := TTL,
-        merger_fun := MergerFun
-    } = Opts = assert_parameters(Opts0),
+        merger_fun := MergerFun,
+        prefix := TelemetryPrefix
+    } = Opts = assert_parameters(Name, Opts0),
     SegmentOpts = ets_settings(Opts),
     SegmentsList = lists:map(fun(_) -> ets:new(undefined, SegmentOpts) end, lists:seq(1, N)),
     Segments = list_to_tuple(SegmentsList),
     Index = atomics:new(1, [{signed, false}]),
     atomics:put(Index, 1, 1),
+    TelemetryEventName = TelemetryPrefix ++ [request],
     Config = #segmented_cache{
         scope = Scope,
+        telemetry_name = TelemetryEventName,
         name = Name,
         strategy = Strategy,
         index = Index,
@@ -83,19 +86,28 @@ persist_cache_config(Name, Config) ->
 %% ETS checks
 %%====================================================================
 
--spec is_member_span(segmented_cache:name(), segmented_cache:key()) -> span().
-is_member_span(Name, Key) when is_atom(Name) ->
-    fun() ->
-        Value = iterate_fun_in_tables(Name, Key, fun segmented_cache_callbacks:is_member_ets_fun/2),
-        {Value, #{hit => Value =:= true}}
-    end.
-
--spec get_entry_span(segmented_cache:name(), segmented_cache:key()) -> span().
-get_entry_span(Name, Key) when is_atom(Name) ->
-    fun() ->
-        Value = iterate_fun_in_tables(Name, Key, fun segmented_cache_callbacks:get_entry_ets_fun/2),
-        {Value, #{hit => Value =/= not_found}}
-    end.
+-spec is_member(segmented_cache:name(), segmented_cache:key()) -> boolean().
+is_member(Name, Key) when is_atom(Name) ->
+    #segmented_cache{telemetry_name = Prefix} = SegmentRecord = get_cache_config(Name),
+    Span = fun() ->
+        Value = iterate_fun_in_tables(
+            SegmentRecord, Key, fun segmented_cache_callbacks:is_member_ets_fun/2
+        ),
+        {Value, #{name => Name, type => is_member, hit => Value =:= true}}
+    end,
+    telemetry:span(Prefix, #{name => Name, type => is_member}, Span).
+
+-spec get_entry(segmented_cache:name(), segmented_cache:key()) ->
+    segmented_cache:value() | not_found.
+get_entry(Name, Key) when is_atom(Name) ->
+    #segmented_cache{telemetry_name = Prefix} = SegmentRecord = get_cache_config(Name),
+    Span = fun() ->
+        Value = iterate_fun_in_tables(
+            SegmentRecord, Key, fun segmented_cache_callbacks:get_entry_ets_fun/2
+        ),
+        {Value, #{name => Name, type => get_entry, hit => Value =/= not_found}}
+    end,
+    telemetry:span(Prefix, #{name => Name, type => get_entry}, Span).
 
 %% Atomically compare_and_swap an entry, attempt three times, post-check front insert
 -spec put_entry_front(segmented_cache:name(), segmented_cache:key(), segmented_cache:value()) ->
@@ -115,7 +127,7 @@ merge_entry(Name, Key, Value) when is_atom(Name) ->
             false -> {continue, false}
         end
     end,
-    case iterate_fun_in_tables(Name, Key, F) of
+    case iterate_fun_in_tables(SegmentRecord, Key, F) of
         true -> true;
         false -> do_put_entry_front(SegmentRecord, Key, Value, 3)
     end.
@@ -133,24 +145,24 @@ delete_pattern(Name, Pattern) when is_atom(Name) ->
     Type :: entry | pattern,
     IterativeFun :: iterative_fun(Key, dynamic()).
 delete_request(Name, Value, Type, Fun) ->
+    #segmented_cache{telemetry_name = Prefix} = SegmentRecord = get_cache_config(Name),
     try
-        iterate_fun_in_tables(Name, Value, Fun)
+        iterate_fun_in_tables(SegmentRecord, Value, Fun)
     catch
         Class:Reason ->
             Metadata = #{
                 name => Name, delete_type => Type, value => Value, class => Class, reason => Reason
             },
-            telemetry:execute([segmented_cache, Name, delete_error], #{}, Metadata)
+            telemetry:execute(Prefix ++ [delete_error], #{}, Metadata)
     end.
 
 %%====================================================================
 %% Internals
 %%====================================================================
 
--spec iterate_fun_in_tables(segmented_cache:name(), Key, IterativeFun) -> Value when
+-spec iterate_fun_in_tables(config(), Key, IterativeFun) -> Value when
     IterativeFun :: iterative_fun(Key, Value).
-iterate_fun_in_tables(Name, Key, IterativeFun) ->
-    SegmentRecord = get_cache_config(Name),
+iterate_fun_in_tables(SegmentRecord, Key, IterativeFun) ->
     Segments = SegmentRecord#segmented_cache.segments,
     Size = tuple_size(Segments),
     CurrentIndex = atomics:get(SegmentRecord#segmented_cache.index, 1),
@@ -318,8 +330,8 @@ purge_last_segment_and_rotate(Name) ->
     atomics:put(SegmentRecord#segmented_cache.index, 1, NewIndex),
     NewIndex.
 
--spec assert_parameters(segmented_cache:opts()) -> segmented_cache:opts().
-assert_parameters(Opts0) when is_map(Opts0) ->
+-spec assert_parameters(segmented_cache:name(), segmented_cache:opts()) -> segmented_cache:opts().
+assert_parameters(Name, Opts0) when is_map(Opts0) ->
     #{
         scope := Scope,
         strategy := Strategy,
@@ -328,6 +340,8 @@ assert_parameters(Opts0) when is_map(Opts0) ->
         ttl := TTL0,
         merger_fun := MergerFun
     } = Opts = maps:merge(defaults(), Opts0),
+    TelemetryEventName = maps:get(prefix, Opts0, [segmented_cache, Name]),
+    true = is_list(TelemetryEventName),
     TTL =
         case TTL0 of
             infinity -> infinity;
@@ -343,7 +357,7 @@ assert_parameters(Opts0) when is_map(Opts0) ->
     true = (Strategy =:= fifo) orelse (Strategy =:= lru),
     true = is_function(MergerFun, 2),
     true = (undefined =/= whereis(Scope)),
-    Opts#{ttl := TTL}.
+    Opts#{ttl := TTL, prefix => TelemetryEventName}.
 
 is_pos_int_or_infinity(Value) ->
     (Value =:= infinity) orelse (is_integer(Value) andalso 0 < Value).

test/segmented_cache_SUITE.erl

@@ -19,6 +19,7 @@
 -include_lib("proper/include/proper.hrl").
 
 -define(CACHE_NAME, test).
+-define(CACHE_NAME_B, [custom_test, lru]).
 -define(CMD_MODULE, segmented_cache_proper_commands).
 
 all() ->
@@ -45,6 +46,7 @@ groups() ->
             put_entry_wait_and_check_false
         ]},
         {lru, [sequence], [
+            get_entry_with_custom_prefix,
             put_entry_and_verify_it_stays,
             put_entry_and_verify_it_stays_under_load,
             stateful_property
@@ -58,33 +60,39 @@ init_per_suite(Config) ->
     ct:pal("Online schedulers ~p~n", [erlang:system_info(schedulers_online)]),
     application:ensure_all_started(telemetry),
     cnt_pt_new(?CACHE_NAME),
-    ok = telemetry:attach(
+    cnt_pt_new(?CACHE_NAME_B),
+    ok = telemetry:attach_many(
         <<"cache-request-handler">>,
-        [segmented_cache, ?CACHE_NAME, request, stop],
+        [
+            [segmented_cache, ?CACHE_NAME, request, stop],
+            ?CACHE_NAME_B ++ [request, stop]
+        ],
         fun ?MODULE:handle_event/4,
         []
     ),
     pg:start(pg),
     Config.
 
 end_per_suite(_Config) ->
-    print_and_restart_counters(),
+    print_and_restart_counters(?CACHE_NAME),
+    print_and_restart_counters(?CACHE_NAME_B),
     ok.
 
 %%%===================================================================
 %%% Group specific setup/teardown
 %%%===================================================================
 init_per_group(lru, Config) ->
-    print_and_restart_counters(),
+    print_and_restart_counters(?CACHE_NAME_B),
     Opts = #{
+        prefix => ?CACHE_NAME_B,
         strategy => lru,
         segment_num => 2,
         ttl => {milliseconds, 100}
     },
     {ok, Cleaner} = segmented_cache:start(?CACHE_NAME, Opts),
     [{cleaner, Cleaner} | Config];
 init_per_group(short_fifo, Config) ->
-    print_and_restart_counters(),
+    print_and_restart_counters(?CACHE_NAME),
     Opts = #{
         strategy => fifo,
         segment_num => 2,
@@ -93,7 +101,7 @@ init_per_group(short_fifo, Config) ->
     {ok, Cleaner} = segmented_cache:start(?CACHE_NAME, Opts),
     [{cleaner, Cleaner} | Config];
 init_per_group(cache_limits, Config) ->
-    print_and_restart_counters(),
+    print_and_restart_counters(?CACHE_NAME),
     Opts = #{
         entries_limit => 1,
         strategy => fifo,
@@ -103,7 +111,7 @@ init_per_group(cache_limits, Config) ->
     {ok, Cleaner} = segmented_cache:start(?CACHE_NAME, Opts),
     [{cleaner, Cleaner} | Config];
 init_per_group(_Groupname, Config) ->
-    print_and_restart_counters(),
+    print_and_restart_counters(?CACHE_NAME),
     {ok, Cleaner} = segmented_cache:start(?CACHE_NAME),
     [{cleaner, Cleaner} | Config].
 
@@ -118,7 +126,7 @@ init_per_testcase(TestCase, Config) when
     TestCase =:= put_entry_and_verify_it_stays;
     TestCase =:= put_entry_and_verify_it_stays_under_load
 ->
-    print_and_restart_counters(),
+    print_and_restart_counters(?CACHE_NAME_B),
     Config;
 init_per_testcase(_TestCase, Config) ->
     Config.
@@ -226,6 +234,18 @@ put_entry_wait_and_check_false(_) ->
     ),
     run_prop(?FUNCTION_NAME, Prop).
 
+get_entry_with_custom_prefix(_) ->
+    Prop = ?FORALL(
+        {Key0, Value},
+        {non_empty(binary()), union([char(), binary(), integer()])},
+        begin
+            Key = {Key0, make_ref()},
+            segmented_cache:put_entry(?CACHE_NAME, Key, Value),
+            Value =:= segmented_cache:get_entry(?CACHE_NAME, Key)
+        end
+    ),
+    run_prop(?FUNCTION_NAME, Prop).
+
 put_entry_and_verify_it_stays(_) ->
     Prop = ?FORALL(
         {Key0, Value},
@@ -266,7 +286,7 @@ put_entry_and_verify_it_stays_under_load(_) ->
         end
     ),
     run_prop(?FUNCTION_NAME, Prop, 10_000, 128),
-    {Hits, Misses} = print_and_restart_counters(),
+    {Hits, Misses} = print_and_restart_counters(?CACHE_NAME_B),
     ct:pal("Hits ~p; Misses ~p~n", [Hits, Misses]),
     Total = Hits + Misses,
     case Misses / Total < 0.001 of
@@ -341,6 +361,11 @@ handle_event([segmented_cache, CacheName, request, stop], _, #{hit := Hit}, _) -
     case Hit of
         true -> cnt_pt_incr_hits(CacheName);
         false -> cnt_pt_incr_misses(CacheName)
+    end;
+handle_event([custom_test, lru, request, stop], _, #{hit := Hit}, _) ->
+    case Hit of
+        true -> cnt_pt_incr_hits(?CACHE_NAME_B);
+        false -> cnt_pt_incr_misses(?CACHE_NAME_B)
     end.
 
 cnt_pt_new(Counter) ->
@@ -358,9 +383,9 @@ cnt_pt_read_hits(Counter) ->
 cnt_pt_read_misses(Counter) ->
     counters:get(persistent_term:get({?MODULE, Counter}), 2).
 
-print_and_restart_counters() ->
-    Hits = cnt_pt_read_hits(?CACHE_NAME),
-    Misses = cnt_pt_read_misses(?CACHE_NAME),
-    counters:put(persistent_term:get({?MODULE, ?CACHE_NAME}), 1, 0),
-    counters:put(persistent_term:get({?MODULE, ?CACHE_NAME}), 2, 0),
+print_and_restart_counters(CacheName) ->
+    Hits = cnt_pt_read_hits(CacheName),
+    Misses = cnt_pt_read_misses(CacheName),
+    counters:put(persistent_term:get({?MODULE, CacheName}), 1, 0),
+    counters:put(persistent_term:get({?MODULE, CacheName}), 2, 0),
     {Hits, Misses}.