You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by ch...@apache.org on 2025/03/21 00:20:35 UTC
(couchdb) branch couch-stats-resource-tracker-v3 updated: Finish baseline csrt_logger rework and tests
This is an automated email from the ASF dual-hosted git repository.
chewbranca pushed a commit to branch couch-stats-resource-tracker-v3
in repository https://gitbox.apache.org/repos/asf/couchdb.git
The following commit(s) were added to refs/heads/couch-stats-resource-tracker-v3 by this push:
new bba551359 Finish baseline csrt_logger rework and tests
bba551359 is described below
commit bba5513594db68040cb62bad069769f82b3d90d5
Author: Russell Branca <ch...@apache.org>
AuthorDate: Thu Mar 20 17:20:25 2025 -0700
Finish baseline csrt_logger rework and tests
---
src/couch_stats/CSRT.md | 1 +
.../src/couch_stats_resource_tracker.hrl | 5 +-
src/couch_stats/src/csrt_logger.erl | 26 ++-
src/couch_stats/src/csrt_util.erl | 91 +++++++-
src/couch_stats/test/eunit/csrt_logger_tests.erl | 260 ++++++++++++++++++---
5 files changed, 340 insertions(+), 43 deletions(-)
diff --git a/src/couch_stats/CSRT.md b/src/couch_stats/CSRT.md
new file mode 100644
index 000000000..073e89e09
--- /dev/null
+++ b/src/couch_stats/CSRT.md
@@ -0,0 +1 @@
+# Couch Stats Resource Tracker (CSRT)
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.hrl b/src/couch_stats/src/couch_stats_resource_tracker.hrl
index 5e92e609c..1fd1a99a1 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.hrl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.hrl
@@ -164,5 +164,8 @@
-type maybe_delta() :: delta() | undefined.
-type tagged_delta() :: {delta, maybe_delta()}.
--type matchers() :: #{atom() => {ets:match_spec(), ets:comp_match_spec()}}.
+-type matcher_name() :: string(). %% TODO: switch to string to allow dynamic options
+-type matcher() :: {ets:match_spec(), ets:comp_match_spec()}.
+-type maybe_matcher() :: matcher() | undefined.
+-type matchers() :: #{matcher_name() => matcher()}.
-type maybe_matchers() :: matchers() | undefined.
diff --git a/src/couch_stats/src/csrt_logger.erl b/src/couch_stats/src/csrt_logger.erl
index 4497d8690..4059efdaf 100644
--- a/src/couch_stats/src/csrt_logger.erl
+++ b/src/couch_stats/src/csrt_logger.erl
@@ -48,13 +48,18 @@
%% Matchers
-export([
+ get_matcher/1,
+ get_matchers/0,
+ is_match/1,
+ is_match/2,
matcher_on_dbname/1,
matcher_on_docs_read/1,
matcher_on_docs_written/1,
matcher_on_rows_read/1,
matcher_on_worker_changes_processed/1,
matcher_on_ioq_calls/1,
- matcher_on_nonce/1
+ matcher_on_nonce/1,
+ reload_matchers/0
]).
-include_lib("stdlib/include/ms_transform.hrl").
@@ -122,10 +127,18 @@ find_matches(Rctxs, Matchers) when is_list(Rctxs) andalso is_map(Matchers) ->
Matchers
).
+-spec reload_matchers() -> ok.
+reload_matchers() ->
+ ok = gen_server:call(?MODULE, reload_matchers, infinity).
+
-spec get_matchers() -> matchers().
get_matchers() ->
persistent_term:get(?MATCHERS_KEY, #{}).
+-spec get_matcher(Name :: matcher_name()) -> maybe_matcher().
+get_matcher(Name) ->
+ maps:get(Name, get_matchers(), undefined).
+
-spec is_match(Rctx :: maybe_rctx()) -> boolean().
is_match(undefined) ->
false;
@@ -136,6 +149,8 @@ is_match(#rctx{}=Rctx) ->
-spec is_match(Rctx :: maybe_rctx(), Matchers :: [any()]) -> boolean().
is_match(undefined, _Matchers) ->
false;
+is_match(_Rctx, undefined) ->
+ false;
is_match(#rctx{}=Rctx, Matchers) when is_map(Matchers) ->
maps:size(find_matches([Rctx], Matchers)) > 0.
@@ -229,17 +244,19 @@ matcher_on_dbname(DbName)
-spec matcher_on_docs_read(Threshold :: pos_integer()) -> ets:match_spec().
matcher_on_docs_read(Threshold)
when is_integer(Threshold) andalso Threshold > 0 ->
- ets:fun2ms(fun(#rctx{type=#coordinator{}, docs_read=DocsRead} = R) when DocsRead >= Threshold -> R end).
+ %%ets:fun2ms(fun(#rctx{type=#coordinator{}, docs_read=DocsRead} = R) when DocsRead >= Threshold -> R end).
+ ets:fun2ms(fun(#rctx{docs_read=DocsRead} = R) when DocsRead >= Threshold -> R end).
-spec matcher_on_docs_written(Threshold :: pos_integer()) -> ets:match_spec().
matcher_on_docs_written(Threshold)
when is_integer(Threshold) andalso Threshold > 0 ->
- ets:fun2ms(fun(#rctx{type=#coordinator{}, docs_written=DocsRead} = R) when DocsRead >= Threshold -> R end).
+ %%ets:fun2ms(fun(#rctx{type=#coordinator{}, docs_written=DocsRead} = R) when DocsRead >= Threshold -> R end).
+ ets:fun2ms(fun(#rctx{docs_written=DocsWritten} = R) when DocsWritten >= Threshold -> R end).
-spec matcher_on_rows_read(Threshold :: pos_integer()) -> ets:match_spec().
matcher_on_rows_read(Threshold)
when is_integer(Threshold) andalso Threshold > 0 ->
- ets:fun2ms(fun(#rctx{rows_read=DocsRead} = R) when DocsRead >= Threshold -> R end).
+ ets:fun2ms(fun(#rctx{rows_read=RowsRead} = R) when RowsRead >= Threshold -> R end).
-spec matcher_on_nonce(Nonce :: nonce()) -> ets:match_spec().
matcher_on_nonce(Nonce) ->
@@ -318,7 +335,6 @@ initialize_matchers() ->
persistent_term:put(?MATCHERS_KEY, Matchers),
ok.
-
-spec matcher_enabled(Name :: string()) -> boolean().
matcher_enabled(Name) when is_list(Name) ->
%% TODO: fix
diff --git a/src/couch_stats/src/csrt_util.erl b/src/couch_stats/src/csrt_util.erl
index dbc73066a..c386cf08e 100644
--- a/src/couch_stats/src/csrt_util.erl
+++ b/src/couch_stats/src/csrt_util.erl
@@ -55,7 +55,9 @@
%% Extra niceties and testing facilities
-export([
set_fabric_init_p/2,
- set_fabric_init_p/3
+ set_fabric_init_p/3,
+ map_to_rctx/1,
+ field/2
]).
@@ -184,9 +186,95 @@ to_json(#rctx{}=Rctx) ->
write_kp_node => Rctx#rctx.write_kp_node,
write_kv_node => Rctx#rctx.write_kv_node,
changes_returned => Rctx#rctx.changes_returned,
+ changes_processed => Rctx#rctx.changes_processed,
ioq_calls => Rctx#rctx.ioq_calls
}.
+%% NOTE: this does not do the inverse of to_json, should it conver types?
+-spec map_to_rctx(Map :: map()) -> rctx().
+map_to_rctx(Map) ->
+ maps:fold(fun map_to_rctx_field/3, #rctx{}, Map).
+
+-spec map_to_rctx_field(Field :: rctx_field(), Val :: any(), Rctx :: rctx()) -> rctx().
+map_to_rctx_field(updated_at, Val, Rctx) ->
+ Rctx#rctx{updated_at = Val};
+map_to_rctx_field(started_at, Val, Rctx) ->
+ Rctx#rctx{started_at = Val};
+map_to_rctx_field(pid_ref, Val, Rctx) ->
+ Rctx#rctx{pid_ref = Val};
+map_to_rctx_field(nonce, Val, Rctx) ->
+ Rctx#rctx{nonce = Val};
+map_to_rctx_field(dbname, Val, Rctx) ->
+ Rctx#rctx{dbname = Val};
+map_to_rctx_field(username, Val, Rctx) ->
+ Rctx#rctx{username = Val};
+map_to_rctx_field(db_open, Val, Rctx) ->
+ Rctx#rctx{db_open = Val};
+map_to_rctx_field(docs_read, Val, Rctx) ->
+ Rctx#rctx{docs_read = Val};
+map_to_rctx_field(docs_written, Val, Rctx) ->
+ Rctx#rctx{docs_written = Val};
+map_to_rctx_field(js_filter, Val, Rctx) ->
+ Rctx#rctx{js_filter = Val};
+map_to_rctx_field(js_filtered_docs, Val, Rctx) ->
+ Rctx#rctx{js_filtered_docs = Val};
+map_to_rctx_field(rows_read, Val, Rctx) ->
+ Rctx#rctx{rows_read = Val};
+map_to_rctx_field(type, Val, Rctx) ->
+ Rctx#rctx{type = Val};
+map_to_rctx_field(get_kp_node, Val, Rctx) ->
+ Rctx#rctx{get_kp_node = Val};
+map_to_rctx_field(get_kv_node, Val, Rctx) ->
+ Rctx#rctx{get_kv_node = Val};
+map_to_rctx_field(write_kp_node, Val, Rctx) ->
+ Rctx#rctx{write_kp_node = Val};
+map_to_rctx_field(write_kv_node, Val, Rctx) ->
+ Rctx#rctx{write_kv_node = Val};
+map_to_rctx_field(changes_returned, Val, Rctx) ->
+ Rctx#rctx{changes_returned = Val};
+map_to_rctx_field(changes_processed, Val, Rctx) ->
+ Rctx#rctx{changes_processed = Val};
+map_to_rctx_field(ioq_calls, Val, Rctx) ->
+ Rctx#rctx{ioq_calls = Val}.
+
+-spec field(Field :: rctx_field(), Rctx :: rctx()) -> any().
+field(updated_at, #rctx{updated_at = Val}) ->
+ Val;
+field(started_at, #rctx{started_at = Val}) ->
+ Val;
+field(pid_ref, #rctx{pid_ref = Val}) ->
+ Val;
+field(nonce, #rctx{nonce = Val}) ->
+ Val;
+field(dbname, #rctx{dbname = Val}) ->
+ Val;
+field(username, #rctx{username = Val}) ->
+ Val;
+field(db_open, #rctx{db_open = Val}) ->
+ Val;
+field(docs_read, #rctx{docs_read = Val}) ->
+ Val;
+field(docs_written, #rctx{docs_written = Val}) ->
+ Val;
+field(js_filter, #rctx{js_filter = Val}) ->
+ Val;
+field(js_filtered_docs, #rctx{js_filtered_docs = Val}) ->
+ Val;
+field(rows_read, #rctx{rows_read = Val}) ->
+ Val;
+field(type, #rctx{type = Val}) ->
+ Val;
+field(get_kp_node, #rctx{get_kp_node = Val}) ->
+ Val;
+field(get_kv_node, #rctx{get_kv_node = Val}) ->
+ Val;
+field(changes_returned, #rctx{changes_returned = Val}) ->
+ Val;
+field(changes_processed, #rctx{changes_processed = Val}) ->
+ Val;
+field(ioq_calls, #rctx{ioq_calls = Val}) ->
+ Val.
+
add_delta({A}, Delta) -> {A, Delta};
add_delta({A, B}, Delta) -> {A, B, Delta};
add_delta({A, B, C}, Delta) -> {A, B, C, Delta};
@@ -253,6 +341,7 @@ rctx_delta(#rctx{}=TA, #rctx{}=TB) ->
js_filtered_docs => TB#rctx.js_filtered_docs - TA#rctx.js_filtered_docs,
rows_read => TB#rctx.rows_read - TA#rctx.rows_read,
changes_returned => TB#rctx.changes_returned - TA#rctx.changes_returned,
+ changes_processed => TB#rctx.changes_processed - TA#rctx.changes_processed,
get_kp_node => TB#rctx.get_kp_node - TA#rctx.get_kp_node,
get_kv_node => TB#rctx.get_kv_node - TA#rctx.get_kv_node,
db_open => TB#rctx.db_open - TA#rctx.db_open,
diff --git a/src/couch_stats/test/eunit/csrt_logger_tests.erl b/src/couch_stats/test/eunit/csrt_logger_tests.erl
index bd155f7be..47393052e 100644
--- a/src/couch_stats/test/eunit/csrt_logger_tests.erl
+++ b/src/couch_stats/test/eunit/csrt_logger_tests.erl
@@ -16,6 +16,18 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
+-define(RCTX_RANGE, 1000).
+-define(RCTX_COUNT, 10000).
+
+%% Dirty hack for hidden records as .hrl is only in src/
+-define(RCTX_RPC, {rpc_worker, foo, bar, {self(), make_ref()}}).
+-define(RCTX_COORDINATOR, {coordinator, foo, bar, 'GET', "/foo/_all_docs"}).
+
+-define(THRESHOLD_DBNAME, <<"foo">>).
+-define(THRESHOLD_DOCS_READ, 123).
+-define(THRESHOLD_IOQ_CALLS, 439).
+-define(THRESHOLD_ROWS_READ, 143).
+-define(THRESHOLD_CHANGES, 79).
csrt_logger_works_test_() ->
{
@@ -29,6 +41,23 @@ csrt_logger_works_test_() ->
]
}.
+
+csrt_logger_matchers_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(t_matcher_on_dbname),
+ ?TDEF_FE(t_matcher_on_docs_read),
+ ?TDEF_FE(t_matcher_on_docs_written),
+ ?TDEF_FE(t_matcher_on_rows_read),
+ ?TDEF_FE(t_matcher_on_worker_changes_processed),
+ ?TDEF_FE(t_matcher_on_ioq_calls),
+ ?TDEF_FE(t_matcher_on_nonce)
+ ]
+ }.
+
make_docs(Count) ->
lists:map(
fun(I) ->
@@ -54,57 +83,190 @@ setup() ->
MArgs = #mrargs{include_docs = false},
_Res = fabric:all_docs(DbName, [?ADMIN_CTX], fun view_cb/2, [], MArgs),
Rctx = load_rctx(PidRef),
- #{ctx => Ctx, dbname => DbName, rctx => Rctx}.
+ ok = config:set("csrt_logger.matchers_threshold", "docs_read", integer_to_list(?THRESHOLD_DOCS_READ), false),
+ ok = config:set("csrt_logger.matchers_threshold", "ioq_calls", integer_to_list(?THRESHOLD_IOQ_CALLS), false),
+ ok = config:set("csrt_logger.matchers_threshold", "rows_read", integer_to_list(?THRESHOLD_ROWS_READ), false),
+ ok = config:set("csrt_logger.matchers_threshold", "worker_changes_processed", integer_to_list(?THRESHOLD_CHANGES), false),
+ csrt_logger:reload_matchers(),
+ #{ctx => Ctx, dbname => DbName, rctx => Rctx, rctxs => rctxs()}.
teardown(#{ctx := Ctx, dbname := DbName}) ->
ok = fabric:delete_db(DbName, [?ADMIN_CTX]),
test_util:stop_couch(Ctx).
-t_do_report(#{rctx := RctxA}) ->
- LRctxA = lists:sort(maps:to_list(csrt_util:to_json(RctxA))),
- ReportNameA = "foo",
+rctx_gen() ->
+ rctx_gen(#{}).
+
+rctx_gen(Opts0) ->
+ DbnameGen = one_of([<<"foo">>, <<"bar">>, ?tempdb]),
+ TypeGen = one_of([?RCTX_RPC, ?RCTX_COORDINATOR]),
+ R = fun() -> rand:uniform(?RCTX_RANGE) end,
+ R10 = fun() -> 3 + rand:uniform(round(?RCTX_RANGE / 10)) end,
+ Occasional = one_of([0, 0, 0, 0, 0, R]),
+ Nonce = one_of(["9c54fa9283", "foobar7799", lists:duplicate(10, fun nonce/0)]),
+ Base = #{
+ dbname => DbnameGen,
+ db_open => R10,
+ docs_read => R,
+ docs_written => Occasional,
+ get_kp_node => R10,
+ get_kv_node => R,
+ nonce => Nonce,
+ pid_ref => {self(), make_ref()},
+ ioq_calls => R,
+ rows_read => R,
+ type => TypeGen,
+ '_do_changes' => true %% Hack because we need to modify both fields
+ },
+ Opts = maps:merge(Base, Opts0),
+ csrt_util:map_to_rctx(maps:fold(
+ fun
+ %% Hack for changes because we need to modify both changes_processed
+ %% and changes_returned but the latter must be <= the former
+ ('_do_changes', V, Acc) ->
+ case V of
+ true ->
+ Processed = R(),
+ Returned = (one_of([0, 0, 1, Processed, rand:uniform(Processed)]))(),
+ maps:put(
+ changes_processed,
+ Processed,
+ maps:put(changes_returned, Returned, Acc));
+ _ ->
+ Acc
+ end;
+ (K, F, Acc) when is_function(F) ->
+ maps:put(K, F(), Acc);
+ (K, V, Acc) ->
+ maps:put(K, V, Acc)
+ end, #{}, Opts
+ )).
+
+rctxs() ->
+ [rctx_gen() || _ <- lists:seq(1, ?RCTX_COUNT)].
+
+t_do_report(#{rctx := Rctx}) ->
+ JRctx = csrt_util:to_json(Rctx),
+ ReportName = "foo",
ok = meck:new(couch_log),
- AssertDoReport = fun(ReportNameB, RctxB) ->
- LRctxB = lists:sort(maps:to_list(RctxB)),
- ?assertEqual(ReportNameA, ReportNameB, "Report Name"),
- ?assertEqual(LRctxA, LRctxB, "Rctx Report Body"),
- true
- end,
- ok = meck:expect(couch_log, report, AssertDoReport),
- ?assert(csrt_logger:do_report(ReportNameA, RctxA)),
+ ok = meck:expect(couch_log, report, fun(_, _) -> true end),
+ ?assert(csrt_logger:do_report(ReportName, Rctx), "CSRT _logger:do_report " ++ ReportName),
?assert(meck:validate(couch_log), "CSRT do_report"),
+ ?assert(meck:validate(couch_log), "CSRT validate couch_log"),
+ ?assert(
+ meck:called(couch_log, report, [ReportName, JRctx]),
+ "CSRT couch_log:report"
+ ),
ok = meck:unload(couch_log).
-t_do_lifetime_report(#{rctx := RctxA}) ->
- LRctxA = lists:sort(maps:to_list(csrt_util:to_json(RctxA))),
- ReportNameA = "csrt-pid-usage-lifetime",
+t_do_lifetime_report(#{rctx := Rctx}) ->
+ JRctx = csrt_util:to_json(Rctx),
+ ReportName = "csrt-pid-usage-lifetime",
ok = meck:new(couch_log),
- AssertDoReport = fun(ReportNameB, RctxB) ->
- LRctxB = lists:sort(maps:to_list(RctxB)),
- ?assertEqual(ReportNameA, ReportNameB, "Report Name"),
- ?assertEqual(LRctxA, LRctxB, "Rctx Report Body"),
- true
- end,
- ok = meck:expect(couch_log, report, AssertDoReport),
- ?assert(csrt_logger:do_lifetime_report(RctxA)),
- ?assert(meck:validate(couch_log), "CSRT do_report"),
+ ok = meck:expect(couch_log, report, fun(_, _) -> true end),
+ ?assert(
+ csrt_logger:do_lifetime_report(Rctx),
+ "CSRT _logger:do_report " ++ ReportName
+ ),
+ ?assert(meck:validate(couch_log), "CSRT validate couch_log"),
+ ?assert(
+ meck:called(couch_log, report, [ReportName, JRctx]),
+ "CSRT couch_log:report"
+ ),
ok = meck:unload(couch_log).
-t_do_status_report(#{rctx := RctxA}) ->
- LRctxA = lists:sort(maps:to_list(csrt_util:to_json(RctxA))),
- ReportNameA = "csrt-pid-usage-status",
+t_do_status_report(#{rctx := Rctx}) ->
+ JRctx = csrt_util:to_json(Rctx),
+ ReportName = "csrt-pid-usage-status",
ok = meck:new(couch_log),
- AssertDoReport = fun(ReportNameB, RctxB) ->
- LRctxB = lists:sort(maps:to_list(RctxB)),
- ?assertEqual(ReportNameA, ReportNameB, "Report Name"),
- ?assertEqual(LRctxA, LRctxB, "Rctx Report Body"),
- true
- end,
- ok = meck:expect(couch_log, report, AssertDoReport),
- ?assert(csrt_logger:do_status_report(RctxA)),
- ?assert(meck:validate(couch_log), "CSRT do_report"),
+ ok = meck:expect(couch_log, report, fun(_, _) -> true end),
+ ?assert(csrt_logger:do_status_report(Rctx), "csrt_logger:do_ " ++ ReportName),
+ ?assert(meck:validate(couch_log), "CSRT validate couch_log"),
+ ?assert(
+ meck:called(couch_log, report, [ReportName, JRctx]),
+ "CSRT couch_log:report"
+ ),
ok = meck:unload(couch_log).
+t_matcher_on_dbname(#{rctx := _Rctx, rctxs := Rctxs0}) ->
+ %% Make sure we have at least one match
+ Rctxs = [rctx_gen(#{dbname => <<"foo">>}) | Rctxs0],
+ ?assertEqual(
+ lists:sort(lists:filter(matcher_on(dbname, <<"foo">>), Rctxs)),
+ lists:sort(lists:filter(matcher_for_csrt("dbname"), Rctxs)),
+ "Dbname matcher on <<\"foo\">>"
+ ).
+
+t_matcher_on_docs_read(#{rctxs := Rctxs0}) ->
+ Threshold = ?THRESHOLD_DOCS_READ,
+ %% Make sure we have at least one match
+ Rctxs = [rctx_gen(#{docs_read => Threshold + 10}) | Rctxs0],
+ ?assertEqual(
+ lists:sort(lists:filter(matcher_gte(docs_read, Threshold), Rctxs)),
+ lists:sort(lists:filter(matcher_for_csrt("docs_read"), Rctxs)),
+ "Docs read matcher"
+ ).
+
+t_matcher_on_docs_written(#{rctxs := Rctxs0}) ->
+ %% Make sure we have at least one match
+ Rctxs = [rctx_gen(#{docs_written => 73}) | Rctxs0],
+ ?assertEqual(
+ lists:sort(lists:filter(matcher_gte(docs_written, 1), Rctxs)),
+ lists:sort(lists:filter(matcher_for_csrt("docs_written"), Rctxs)),
+ "Docs written matcher"
+ ).
+
+t_matcher_on_rows_read(#{rctxs := Rctxs0}) ->
+ Threshold = ?THRESHOLD_ROWS_READ,
+ %% Make sure we have at least one match
+ Rctxs = [rctx_gen(#{rows_read => Threshold + 10}) | Rctxs0],
+ ?assertEqual(
+ lists:sort(lists:filter(matcher_gte(rows_read, Threshold), Rctxs)),
+ lists:sort(lists:filter(matcher_for_csrt("rows_read"), Rctxs)),
+ "Rows read matcher"
+ ).
+
+t_matcher_on_worker_changes_processed(#{rctxs := Rctxs0}) ->
+ Threshold = ?THRESHOLD_CHANGES,
+ %% Make sure we have at least one match
+ Rctxs = [rctx_gen(#{changes_processed => Threshold + 10}) | Rctxs0],
+ ChangesFilter = fun(R) ->
+ Ret = csrt_util:field(changes_returned, R),
+ Proc = csrt_util:field(changes_processed, R),
+ (Proc - Ret) >= Threshold
+ end,
+ ?assertEqual(
+ lists:sort(lists:filter(ChangesFilter, Rctxs)),
+ lists:sort(lists:filter(matcher_for_csrt("worker_changes_processed"), Rctxs)),
+ "Changes processed matcher"
+ ).
+
+t_matcher_on_ioq_calls(#{rctxs := Rctxs0}) ->
+ Threshold = ?THRESHOLD_IOQ_CALLS,
+ %% Make sure we have at least one match
+ Rctxs = [rctx_gen(#{ioq_calls => Threshold + 10}) | Rctxs0],
+ ?assertEqual(
+ lists:sort(lists:filter(matcher_gte(ioq_calls, Threshold), Rctxs)),
+ lists:sort(lists:filter(matcher_for_csrt("ioq_calls"), Rctxs)),
+ "IOQ calls matcher"
+ ).
+
+t_matcher_on_nonce(#{rctxs := Rctxs0}) ->
+ Nonce = "foobar7799",
+ %% Make sure we have at least one match
+ Rctxs = [rctx_gen(#{nonce => Nonce}) | Rctxs0],
+ %% Nonce requires dynamic matcher as it's a static match
+ %% TODO: add pattern based nonce matching
+ MSpec = csrt_logger:matcher_on_nonce(Nonce),
+ CompMSpec = ets:match_spec_compile(MSpec),
+ Matchers = #{"nonce" => {MSpec, CompMSpec}},
+ IsMatch = fun(ARctx) -> csrt_logger:is_match(ARctx, Matchers) end,
+ ?assertEqual(
+ lists:sort(lists:filter(matcher_on(nonce, Nonce), Rctxs)),
+ lists:sort(lists:filter(IsMatch, Rctxs)),
+ "Rows read matcher"
+ ).
+
load_rctx(PidRef) ->
timer:sleep(50), %% Add slight delay to accumulate RPC response deltas
csrt:get_resource(PidRef).
@@ -113,3 +275,29 @@ view_cb({row, Row}, Acc) ->
{ok, [Row | Acc]};
view_cb(_Msg, Acc) ->
{ok, Acc}.
+
+matcher_gte(Field, Value) ->
+ matcher_for(Field, Value, fun erlang:'>='/2).
+
+matcher_on(Field, Value) ->
+ matcher_for(Field, Value, fun erlang:'=:='/2).
+
+matcher_for(Field, Value, Op) ->
+ fun(Rctx) -> Op(csrt_util:field(Field, Rctx), Value) end.
+
+matcher_for_csrt(MatcherName) ->
+ Matchers = #{MatcherName => {_, _} = csrt_logger:get_matcher(MatcherName)},
+ fun(Rctx) -> csrt_logger:is_match(Rctx, Matchers) end.
+
+nonce() ->
+ couch_util:to_hex(crypto:strong_rand_bytes(5)).
+
+one_of(L) ->
+ fun() ->
+ case lists:nth(rand:uniform(length(L)), L) of
+ F when is_function(F) ->
+ F();
+ N ->
+ N
+ end
+ end.