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/25 16:11:57 UTC
(couchdb) branch couch-stats-resource-tracker-v3-rebase created (now 175a76dec)
This is an automated email from the ASF dual-hosted git repository.
chewbranca pushed a change to branch couch-stats-resource-tracker-v3-rebase
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at 175a76dec Add Couch Stats Resource Tracker (CSRT)
This branch includes the following new commits:
new 175a76dec Add Couch Stats Resource Tracker (CSRT)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(couchdb) 01/01: Add Couch Stats Resource Tracker (CSRT)
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chewbranca pushed a commit to branch couch-stats-resource-tracker-v3-rebase
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 175a76dec5bdb0a55c1fe845ad13d433b74d7d76
Author: Russell Branca <ch...@apache.org>
AuthorDate: Fri Jun 7 15:46:17 2024 -0700
Add Couch Stats Resource Tracker (CSRT)
---
rel/overlay/etc/default.ini | 30 ++
src/chttpd/src/chttpd.erl | 10 +
src/chttpd/src/chttpd_db.erl | 2 +
src/chttpd/src/chttpd_httpd_handlers.erl | 1 +
src/chttpd/src/chttpd_misc.erl | 105 ++++
src/chttpd/test/eunit/chttpd_db_doc_size_tests.erl | 7 +-
src/config/src/config_listener_mon.erl | 2 +
src/couch/include/couch_db.hrl | 2 +
src/couch/priv/stats_descriptions.cfg | 33 ++
src/couch/src/couch_btree.erl | 3 +
src/couch/src/couch_db.erl | 2 +
src/couch/src/couch_os_process.erl | 11 +-
src/couch/src/couch_query_servers.erl | 8 +
src/couch/src/couch_server.erl | 2 +
src/couch_stats/CSRT.md | 1 +
src/couch_stats/src/couch_stats.app.src | 9 +-
src/couch_stats/src/couch_stats.erl | 26 +
.../src/couch_stats_resource_tracker.hrl | 171 ++++++
src/couch_stats/src/couch_stats_sup.erl | 2 +
src/couch_stats/src/csrt.erl | 500 ++++++++++++++++++
src/couch_stats/src/csrt_logger.erl | 401 ++++++++++++++
src/couch_stats/src/csrt_query.erl | 174 +++++++
src/couch_stats/src/csrt_server.erl | 198 +++++++
src/couch_stats/src/csrt_util.erl | 470 +++++++++++++++++
src/couch_stats/test/eunit/csrt_logger_tests.erl | 345 +++++++++++++
src/couch_stats/test/eunit/csrt_server_tests.erl | 575 +++++++++++++++++++++
src/fabric/priv/stats_descriptions.cfg | 50 ++
src/fabric/src/fabric_rpc.erl | 16 +
src/fabric/src/fabric_util.erl | 40 +-
src/fabric/test/eunit/fabric_rpc_purge_tests.erl | 2 +
src/fabric/test/eunit/fabric_rpc_tests.erl | 11 +-
src/ioq/src/ioq.erl | 1 +
src/mango/src/mango_cursor_view.erl | 2 +
src/mango/src/mango_selector.erl | 1 +
src/mem3/src/mem3_rpc.erl | 34 +-
src/rexi/include/rexi.hrl | 1 +
src/rexi/src/rexi.erl | 10 +-
src/rexi/src/rexi_monitor.erl | 1 +
src/rexi/src/rexi_server.erl | 19 +-
src/rexi/src/rexi_utils.erl | 12 +-
src/rexi/test/rexi_tests.erl | 15 +-
41 files changed, 3261 insertions(+), 44 deletions(-)
diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index 1a0f318bf..a8663e358 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -1119,3 +1119,33 @@ url = {{nouveau_url}}
;mem3_shards = true
;nouveau_index_manager = true
;dreyfus_index_manager = true
+
+; Couch Stats Resource Tracker (CSRT)
+[csrt]
+enabled = true
+
+; CSRT Rexi Server Init P tracking
+; Enable these to enable additional metrics for RPC worker spawn rates
+; Mod and Function are separated by double underscores
+[csrt.init_p]
+enabled = false
+fabric_rpc__all_docs = true
+fabric_rpc__changes = true
+fabric_rpc__map_view = true
+fabric_rpc__reduce_view = true
+fabric_rpc__get_all_security = true
+fabric_rpc__open_doc = true
+fabric_rpc__update_docs = true
+fabric_rpc__open_shard = true
+
+;; CSRT dbname matchers
+;; Given a dbname and a positive integer, this will enable an IO matcher
+;; against the provided db for any requests that induce IO in quantities
+;; greater than the provided threshold on any one of: ioq_calls, rows_read
+;; docs_read, get_kp_node, get_kv_node, or changes_processed.
+;;
+[csrt_logger.dbnames_io]
+;; foo = 100
+;; _dbs = 123
+;; _users = 234
+;; foo/bar = 200
diff --git a/src/chttpd/src/chttpd.erl b/src/chttpd/src/chttpd.erl
index 57a3aeaea..e9631ce48 100644
--- a/src/chttpd/src/chttpd.erl
+++ b/src/chttpd/src/chttpd.erl
@@ -339,6 +339,10 @@ handle_request_int(MochiReq) ->
% Save client socket so that it can be monitored for disconnects
chttpd_util:mochiweb_client_req_set(MochiReq),
+ %% This is probably better in before_request, but having Path is nice
+ csrt:create_coordinator_context(HttpReq0, Path),
+ csrt:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
+
{HttpReq2, Response} =
case before_request(HttpReq0) of
{ok, HttpReq1} ->
@@ -369,6 +373,7 @@ handle_request_int(MochiReq) ->
before_request(HttpReq) ->
try
+ csrt:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
chttpd_stats:init(),
chttpd_plugin:before_request(HttpReq)
catch
@@ -388,6 +393,8 @@ after_request(HttpReq, HttpResp0) ->
HttpResp2 = update_stats(HttpReq, HttpResp1),
chttpd_stats:report(HttpReq, HttpResp2),
maybe_log(HttpReq, HttpResp2),
+ %% NOTE: do not set_context_handler_fun to preserve the Handler
+ csrt:destroy_context(),
HttpResp2.
process_request(#httpd{mochi_req = MochiReq} = HttpReq) ->
@@ -400,6 +407,7 @@ process_request(#httpd{mochi_req = MochiReq} = HttpReq) ->
RawUri = MochiReq:get(raw_path),
try
+ csrt:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
couch_httpd:validate_host(HttpReq),
check_request_uri_length(RawUri),
check_url_encoding(RawUri),
@@ -425,10 +433,12 @@ handle_req_after_auth(HandlerKey, HttpReq) ->
HandlerKey,
fun chttpd_db:handle_request/1
),
+ csrt:set_context_handler_fun(HandlerFun),
AuthorizedReq = chttpd_auth:authorize(
possibly_hack(HttpReq),
fun chttpd_auth_request:authorize_request/1
),
+ csrt:set_context_username(AuthorizedReq),
{AuthorizedReq, HandlerFun(AuthorizedReq)}
catch
ErrorType:Error:Stack ->
diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index b4c141f8c..2419e77d8 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -83,6 +83,7 @@
% Database request handlers
handle_request(#httpd{path_parts = [DbName | RestParts], method = Method} = Req) ->
+ csrt:set_context_dbname(DbName),
case {Method, RestParts} of
{'PUT', []} ->
create_db_req(Req, DbName);
@@ -103,6 +104,7 @@ handle_request(#httpd{path_parts = [DbName | RestParts], method = Method} = Req)
do_db_req(Req, fun db_req/2);
{_, [SecondPart | _]} ->
Handler = chttpd_handlers:db_handler(SecondPart, fun db_req/2),
+ csrt:set_context_handler_fun(Handler),
do_db_req(Req, Handler)
end.
diff --git a/src/chttpd/src/chttpd_httpd_handlers.erl b/src/chttpd/src/chttpd_httpd_handlers.erl
index 932b52e5f..e1b260222 100644
--- a/src/chttpd/src/chttpd_httpd_handlers.erl
+++ b/src/chttpd/src/chttpd_httpd_handlers.erl
@@ -20,6 +20,7 @@ url_handler(<<"_utils">>) -> fun chttpd_misc:handle_utils_dir_req/1;
url_handler(<<"_all_dbs">>) -> fun chttpd_misc:handle_all_dbs_req/1;
url_handler(<<"_dbs_info">>) -> fun chttpd_misc:handle_dbs_info_req/1;
url_handler(<<"_active_tasks">>) -> fun chttpd_misc:handle_task_status_req/1;
+url_handler(<<"_active_resources">>) -> fun chttpd_misc:handle_resource_status_req/1;
url_handler(<<"_scheduler">>) -> fun couch_replicator_httpd:handle_scheduler_req/1;
url_handler(<<"_node">>) -> fun chttpd_node:handle_node_req/1;
url_handler(<<"_reload_query_servers">>) -> fun chttpd_misc:handle_reload_query_servers_req/1;
diff --git a/src/chttpd/src/chttpd_misc.erl b/src/chttpd/src/chttpd_misc.erl
index 888111a64..0baf0b972 100644
--- a/src/chttpd/src/chttpd_misc.erl
+++ b/src/chttpd/src/chttpd_misc.erl
@@ -20,6 +20,7 @@
handle_replicate_req/1,
handle_reload_query_servers_req/1,
handle_task_status_req/1,
+ handle_resource_status_req/1,
handle_up_req/1,
handle_utils_dir_req/1,
handle_utils_dir_req/2,
@@ -219,6 +220,110 @@ handle_task_status_req(#httpd{method = 'GET'} = Req) ->
handle_task_status_req(Req) ->
send_method_not_allowed(Req, "GET,HEAD").
+handle_resource_status_req(#httpd{method = 'POST'} = Req) ->
+ ok = chttpd:verify_is_server_admin(Req),
+ chttpd:validate_ctype(Req, "application/json"),
+ {Props} = chttpd:json_body_obj(Req),
+ Action = proplists:get_value(<<"action">>, Props),
+ Key = proplists:get_value(<<"key">>, Props),
+ Val = proplists:get_value(<<"val">>, Props),
+
+ CountBy = fun csrt:count_by/1,
+ GroupBy = fun csrt:group_by/2,
+ SortedBy1 = fun csrt:sorted_by/1,
+ SortedBy2 = fun csrt:sorted_by/2,
+ ConvertEle = fun erlang:binary_to_existing_atom/1,
+ ConvertList = fun(L) -> [ConvertEle(E) || E <- L] end,
+ ToJson = fun csrt_util:to_json/1,
+ JsonKeys = fun(PL) -> [[ToJson(K), V] || {K, V} <- PL] end,
+
+ Fun = case {Action, Key, Val} of
+ {<<"count_by">>, Keys, undefined} when is_list(Keys) ->
+ Keys1 = [ConvertEle(K) || K <- Keys],
+ fun() -> CountBy(Keys1) end;
+ {<<"count_by">>, Key, undefined} ->
+ Key1 = ConvertEle(Key),
+ fun() -> CountBy(Key1) end;
+ {<<"group_by">>, Keys, Vals} when is_list(Keys) andalso is_list(Vals) ->
+ Keys1 = ConvertList(Keys),
+ Vals1 = ConvertList(Vals),
+ fun() -> GroupBy(Keys1, Vals1) end;
+ {<<"group_by">>, Key, Vals} when is_list(Vals) ->
+ Key1 = ConvertEle(Key),
+ Vals1 = ConvertList(Vals),
+ fun() -> GroupBy(Key1, Vals1) end;
+ {<<"group_by">>, Keys, Val} when is_list(Keys) ->
+ Keys1 = ConvertList(Keys),
+ Val1 = ConvertEle(Val),
+ fun() -> GroupBy(Keys1, Val1) end;
+ {<<"group_by">>, Key, Val} ->
+ Key1 = ConvertEle(Key),
+ Val1 = ConvertList(Val),
+ fun() -> GroupBy(Key1, Val1) end;
+
+ {<<"sorted_by">>, Key, undefined} ->
+ Key1 = ConvertEle(Key),
+ fun() -> JsonKeys(SortedBy1(Key1)) end;
+ {<<"sorted_by">>, Keys, undefined} when is_list(Keys) ->
+ Keys1 = [ConvertEle(K) || K <- Keys],
+ fun() -> JsonKeys(SortedBy1(Keys1)) end;
+ {<<"sorted_by">>, Keys, Vals} when is_list(Keys) andalso is_list(Vals) ->
+ Keys1 = ConvertList(Keys),
+ Vals1 = ConvertList(Vals),
+ fun() -> JsonKeys(SortedBy2(Keys1, Vals1)) end;
+ {<<"sorted_by">>, Key, Vals} when is_list(Vals) ->
+ Key1 = ConvertEle(Key),
+ Vals1 = ConvertList(Vals),
+ fun() -> JsonKeys(SortedBy2(Key1, Vals1)) end;
+ {<<"sorted_by">>, Keys, Val} when is_list(Keys) ->
+ Keys1 = ConvertList(Keys),
+ Val1 = ConvertEle(Val),
+ fun() -> JsonKeys(SortedBy2(Keys1, Val1)) end;
+ {<<"sorted_by">>, Key, Val} ->
+ Key1 = ConvertEle(Key),
+ Val1 = ConvertList(Val),
+ fun() -> JsonKeys(SortedBy2(Key1, Val1)) end;
+ _ ->
+ throw({badrequest, invalid_resource_request})
+ end,
+
+ Fun1 = fun() ->
+ case Fun() of
+ Map when is_map(Map) ->
+ {maps:fold(
+ fun
+ (_K,0,A) -> A; %% TODO: Skip 0 value entries?
+ (K,V,A) -> [{ToJson(K), V} | A]
+ end,
+ [], Map)};
+ List when is_list(List) ->
+ List
+ end
+ end,
+
+ {Resp, _Bad} = rpc:multicall(erlang, apply, [
+ fun() ->
+ {node(), Fun1()}
+ end,
+ []
+ ]),
+ %%io:format("{CSRT}***** GOT RESP: ~p~n", [Resp]),
+ send_json(Req, {Resp});
+handle_resource_status_req(#httpd{method = 'GET'} = Req) ->
+ ok = chttpd:verify_is_server_admin(Req),
+ {Resp, Bad} = rpc:multicall(erlang, apply, [
+ fun() ->
+ {node(), csrt:active()}
+ end,
+ []
+ ]),
+ %% TODO: incorporate Bad responses
+ send_json(Req, {Resp});
+handle_resource_status_req(Req) ->
+ ok = chttpd:verify_is_server_admin(Req),
+ send_method_not_allowed(Req, "GET,HEAD,POST").
+
+
handle_replicate_req(#httpd{method = 'POST', user_ctx = Ctx, req_body = PostBody} = Req) ->
chttpd:validate_ctype(Req, "application/json"),
%% see HACK in chttpd.erl about replication
diff --git a/src/chttpd/test/eunit/chttpd_db_doc_size_tests.erl b/src/chttpd/test/eunit/chttpd_db_doc_size_tests.erl
index 01ef16f23..da60e85e6 100644
--- a/src/chttpd/test/eunit/chttpd_db_doc_size_tests.erl
+++ b/src/chttpd/test/eunit/chttpd_db_doc_size_tests.erl
@@ -24,8 +24,9 @@
setup() ->
Hashed = couch_passwords:hash_admin_password(?PASS),
- ok = config:set("admins", ?USER, ?b2l(Hashed), _Persist = false),
- ok = config:set("couchdb", "max_document_size", "50"),
+ ok = config:set("admins", ?USER, ?b2l(Hashed), false),
+ ok = config:set("couchdb", "max_document_size", "50", false),
+
TmpDb = ?tempdb(),
Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
Port = mochiweb_socket_server:get(chttpd, port),
@@ -35,7 +36,7 @@ setup() ->
teardown(Url) ->
delete_db(Url),
- ok = config:delete("admins", ?USER, _Persist = false),
+ ok = config:delete("admins", ?USER, false),
ok = config:delete("couchdb", "max_document_size").
create_db(Url) ->
diff --git a/src/config/src/config_listener_mon.erl b/src/config/src/config_listener_mon.erl
index b74a61306..898ecd183 100644
--- a/src/config/src/config_listener_mon.erl
+++ b/src/config/src/config_listener_mon.erl
@@ -13,6 +13,8 @@
-module(config_listener_mon).
-behaviour(gen_server).
+-dialyzer({nowarn_function, init/1}).
+
-export([
subscribe/2,
start_link/2
diff --git a/src/couch/include/couch_db.hrl b/src/couch/include/couch_db.hrl
index 9c1df21b6..ba6bd38ca 100644
--- a/src/couch/include/couch_db.hrl
+++ b/src/couch/include/couch_db.hrl
@@ -53,6 +53,8 @@
-define(INTERACTIVE_EDIT, interactive_edit).
-define(REPLICATED_CHANGES, replicated_changes).
+-define(LOG_UNEXPECTED_MSG(Msg), couch_log:warning("[~p:~p:~p/~p]{~p[~p]} Unexpected message: ~w", [?MODULE, ?LINE, ?FUNCTION_NAME, ?FUNCTION_ARITY, self(), element(2, process_info(self(), message_queue_len)), Msg])).
+
-type branch() :: {Key::term(), Value::term(), Tree::term()}.
-type path() :: {Start::pos_integer(), branch()}.
-type update_type() :: replicated_changes | interactive_edit.
diff --git a/src/couch/priv/stats_descriptions.cfg b/src/couch/priv/stats_descriptions.cfg
index 6a7120f87..586c6e66a 100644
--- a/src/couch/priv/stats_descriptions.cfg
+++ b/src/couch/priv/stats_descriptions.cfg
@@ -306,6 +306,10 @@
{type, counter},
{desc, <<"number of couch_server LRU operations skipped">>}
]}.
+{[couchdb, couch_server, open], [
+ {type, counter},
+ {desc, <<"number of couch_server open operations invoked">>}
+]}.
{[couchdb, query_server, vdu_rejects], [
{type, counter},
{desc, <<"number of rejections by validate_doc_update function">>}
@@ -418,10 +422,39 @@
{type, counter},
{desc, <<"number of other requests">>}
]}.
+{[couchdb, query_server, volume, ddoc_filter], [
+ {type, counter},
+ {desc, <<"number of docs filtered by ddoc filters">>}
+]}.
{[couchdb, legacy_checksums], [
{type, counter},
{desc, <<"number of legacy checksums found in couch_file instances">>}
]}.
+{[couchdb, btree, folds], [
+ {type, counter},
+ {desc, <<"number of couch btree kv fold callback invocations">>}
+]}.
+{[couchdb, btree, get_node, kp_node], [
+ {type, counter},
+ {desc, <<"number of couch btree kp_nodes read">>}
+]}.
+{[couchdb, btree, get_node, kv_node], [
+ {type, counter},
+ {desc, <<"number of couch btree kv_nodes read">>}
+]}.
+{[couchdb, btree, write_node, kp_node], [
+ {type, counter},
+ {desc, <<"number of couch btree kp_nodes written">>}
+]}.
+{[couchdb, btree, write_node, kv_node], [
+ {type, counter},
+ {desc, <<"number of couch btree kv_nodes written">>}
+]}.
+%% CSRT (couch_stats_resource_tracker) stats
+{[couchdb, csrt, delta_missing_t0], [
+ {type, counter},
+ {desc, <<"number of csrt contexts without a proper startime">>}
+]}.
{[pread, exceed_eof], [
{type, counter},
{desc, <<"number of the attempts to read beyond end of db file">>}
diff --git a/src/couch/src/couch_btree.erl b/src/couch/src/couch_btree.erl
index b974a22ee..ba176cca2 100644
--- a/src/couch/src/couch_btree.erl
+++ b/src/couch/src/couch_btree.erl
@@ -472,6 +472,8 @@ reduce_tree_size(kp_node, NodeSize, [{_K, {_P, _Red, Sz}} | NodeList]) ->
get_node(#btree{fd = Fd}, NodePos) ->
{ok, {NodeType, NodeList}} = couch_file:pread_term(Fd, NodePos),
+ %% TODO: wire in csrt tracking
+ couch_stats:increment_counter([couchdb, btree, get_node, NodeType]),
{NodeType, NodeList}.
write_node(#btree{fd = Fd, compression = Comp} = Bt, NodeType, NodeList) ->
@@ -480,6 +482,7 @@ write_node(#btree{fd = Fd, compression = Comp} = Bt, NodeType, NodeList) ->
% now write out each chunk and return the KeyPointer pairs for those nodes
ToWrite = [{NodeType, Chunk} || Chunk <- Chunks],
WriteOpts = [{compression, Comp}],
+ couch_stats:increment_counter([couchdb, btree, write_node, NodeType]),
{ok, PtrSizes} = couch_file:append_terms(Fd, ToWrite, WriteOpts),
{ok, group_kps(Bt, NodeType, Chunks, PtrSizes)}.
diff --git a/src/couch/src/couch_db.erl b/src/couch/src/couch_db.erl
index e33e695c0..694e82936 100644
--- a/src/couch/src/couch_db.erl
+++ b/src/couch/src/couch_db.erl
@@ -298,6 +298,7 @@ open_doc(Db, IdOrDocInfo) ->
open_doc(Db, IdOrDocInfo, []).
open_doc(Db, Id, Options) ->
+ %% TODO: wire in csrt tracking
increment_stat(Db, [couchdb, database_reads]),
case open_doc_int(Db, Id, Options) of
{ok, #doc{deleted = true} = Doc} ->
@@ -1987,6 +1988,7 @@ increment_stat(#db{options = Options}, Stat, Count) when
->
case lists:member(sys_db, Options) of
true ->
+ %% TODO: we shouldn't leak resource usage just because it's a sys_db
ok;
false ->
couch_stats:increment_counter(Stat, Count)
diff --git a/src/couch/src/couch_os_process.erl b/src/couch/src/couch_os_process.erl
index 59ceeca13..003c3dc51 100644
--- a/src/couch/src/couch_os_process.erl
+++ b/src/couch/src/couch_os_process.erl
@@ -246,8 +246,9 @@ bump_cmd_time_stat(Cmd, USec) when is_list(Cmd), is_integer(USec) ->
bump_time_stat(ddoc_new, USec);
[<<"ddoc">>, _, [<<"validate_doc_update">> | _] | _] ->
bump_time_stat(ddoc_vdu, USec);
- [<<"ddoc">>, _, [<<"filters">> | _] | _] ->
- bump_time_stat(ddoc_filter, USec);
+ [<<"ddoc">>, _, [<<"filters">> | _], [Docs | _] | _] ->
+ bump_time_stat(ddoc_filter, USec),
+ bump_volume_stat(ddoc_filter, Docs);
[<<"ddoc">> | _] ->
bump_time_stat(ddoc_other, USec);
_ ->
@@ -258,6 +259,12 @@ bump_time_stat(Stat, USec) when is_atom(Stat), is_integer(USec) ->
couch_stats:increment_counter([couchdb, query_server, calls, Stat]),
couch_stats:increment_counter([couchdb, query_server, time, Stat], USec).
+bump_volume_stat(ddoc_filter=Stat, Docs) when is_atom(Stat), is_list(Docs) ->
+ couch_stats:increment_counter([couchdb, query_server, volume, Stat, length(Docs)]);
+bump_volume_stat(_, _) ->
+ %% TODO: handle other stats?
+ ok.
+
log_level("debug") ->
debug;
log_level("info") ->
diff --git a/src/couch/src/couch_query_servers.erl b/src/couch/src/couch_query_servers.erl
index 6789bfaef..2d5185806 100644
--- a/src/couch/src/couch_query_servers.erl
+++ b/src/couch/src/couch_query_servers.erl
@@ -542,6 +542,8 @@ filter_docs(Req, Db, DDoc, FName, Docs) ->
{ok, filter_docs_int(Db, DDoc, FName, JsonReq, JsonDocs)}
catch
throw:{os_process_error, {exit_status, 1}} ->
+ %% TODO: wire in csrt tracking
+ couch_stats:increment_counter([couchdb, query_server, js_filter_error]),
%% batch used too much memory, retry sequentially.
Fun = fun(JsonDoc) ->
filter_docs_int(Db, DDoc, FName, JsonReq, [JsonDoc])
@@ -550,6 +552,12 @@ filter_docs(Req, Db, DDoc, FName, Docs) ->
end.
filter_docs_int(Db, DDoc, FName, JsonReq, JsonDocs) ->
+ %% Count usage in _int version as this can be repeated for OS error
+ %% Pros & cons... might not have actually processed `length(JsonDocs)` docs
+ %% but it certainly undercounts if we count in `filter_docs/5` above
+ %% TODO: replace with couchdb.query_server.*.ddoc_filter stats once we can
+ %% funnel back the stats used in the couchjs process to this caller process
+ csrt:js_filtered(length(JsonDocs)),
[true, Passes] = ddoc_prompt(
Db,
DDoc,
diff --git a/src/couch/src/couch_server.erl b/src/couch/src/couch_server.erl
index ca12a56fa..42ba80b1e 100644
--- a/src/couch/src/couch_server.erl
+++ b/src/couch/src/couch_server.erl
@@ -114,6 +114,8 @@ sup_start_link(N) ->
gen_server:start_link({local, couch_server(N)}, couch_server, [N], []).
open(DbName, Options) ->
+ %% TODO: wire in csrt tracking
+ couch_stats:increment_counter([couchdb, couch_server, open]),
try
validate_open_or_create(DbName, Options),
open_int(DbName, Options)
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.app.src b/src/couch_stats/src/couch_stats.app.src
index a54fac734..da100e068 100644
--- a/src/couch_stats/src/couch_stats.app.src
+++ b/src/couch_stats/src/couch_stats.app.src
@@ -13,8 +13,13 @@
{application, couch_stats, [
{description, "Simple statistics collection"},
{vsn, git},
- {registered, [couch_stats_aggregator, couch_stats_process_tracker]},
- {applications, [kernel, stdlib]},
+ {registered, [
+ couch_stats_aggregator,
+ couch_stats_process_tracker,
+ csrt_server,
+ csrt_logger
+ ]},
+ {applications, [kernel, stdlib, couch_log]},
{mod, {couch_stats_app, []}},
{env, []}
]}.
diff --git a/src/couch_stats/src/couch_stats.erl b/src/couch_stats/src/couch_stats.erl
index 29a402449..f92950fa9 100644
--- a/src/couch_stats/src/couch_stats.erl
+++ b/src/couch_stats/src/couch_stats.erl
@@ -24,6 +24,11 @@
update_gauge/2
]).
+%% couch_stats_resource_tracker API
+-export([
+ maybe_track_rexi_init_p/1
+]).
+
-type response() :: ok | {error, unknown_metric} | {error, invalid_metric}.
-type stat() :: {any(), [{atom(), any()}]}.
@@ -49,6 +54,11 @@ increment_counter(Name) ->
-spec increment_counter(any(), pos_integer()) -> response().
increment_counter(Name, Value) ->
+ %% Should maybe_track_local happen before or after notify?
+ %% If after, only currently tracked metrics declared in the app's
+ %% stats_description.cfg will be trackable locally. Pros/cons.
+ %io:format("NOTIFY_EXISTING_METRIC: ~p || ~p || ~p~n", [Name, Op, Type]),
+ ok = maybe_track_local_counter(Name, Value),
case couch_stats_util:get_counter(Name, stats()) of
{ok, Ctx} -> couch_stats_counter:increment(Ctx, Value);
{error, Error} -> {error, Error}
@@ -100,6 +110,22 @@ stats() ->
now_sec() ->
erlang:monotonic_time(second).
+%% Only potentially track positive increments to counters
+-spec maybe_track_local_counter(any(), any()) -> ok.
+maybe_track_local_counter(Name, Val) when is_integer(Val) andalso Val > 0 ->
+ %%io:format("maybe_track_local[~p]: ~p~n", [Val, Name]),
+ csrt:maybe_inc(Name, Val),
+ ok;
+maybe_track_local_counter(_, _) ->
+ ok.
+
+maybe_track_rexi_init_p({M, F, _A}) ->
+ Metric = [M, F, spawned],
+ case csrt:should_track_init_p(Metric) of
+ true -> increment_counter(Metric);
+ false -> ok
+ end.
+
-ifdef(TEST).
-include_lib("couch/include/couch_eunit.hrl").
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.hrl b/src/couch_stats/src/couch_stats_resource_tracker.hrl
new file mode 100644
index 000000000..1fd1a99a1
--- /dev/null
+++ b/src/couch_stats/src/couch_stats_resource_tracker.hrl
@@ -0,0 +1,171 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-define(CSRT, "csrt").
+-define(CSRT_INIT_P, "csrt.init_p").
+
+%% CSRT pdict markers
+-define(DELTA_TA, csrt_delta_ta).
+-define(DELTA_TZ, csrt_delta_tz). %% T Zed instead of T0
+-define(PID_REF, csrt_pid_ref). %% track local ID
+-define(TRACKER_PID, csrt_tracker). %% tracker pid
+
+-define(MANGO_EVAL_MATCH, mango_eval_match).
+-define(DB_OPEN_DOC, docs_read).
+-define(DB_OPEN, db_open).
+-define(COUCH_SERVER_OPEN, db_open).
+-define(COUCH_BT_GET_KP_NODE, get_kp_node).
+-define(COUCH_BT_GET_KV_NODE, get_kv_node).
+-define(COUCH_BT_WRITE_KP_NODE, write_kp_node).
+-define(COUCH_BT_WRITE_KV_NODE, write_kv_node).
+-define(COUCH_JS_FILTER, js_filter).
+-define(COUCH_JS_FILTERED_DOCS, js_filtered_docs).
+-define(IOQ_CALLS, ioq_calls).
+-define(DOCS_WRITTEN, docs_written).
+-define(ROWS_READ, rows_read).
+
+%% TODO: overlap between this and couch btree fold invocations
+%% TODO: need some way to distinguish fols on views vs find vs all_docs
+-define(FRPC_CHANGES_ROW, changes_processed).
+-define(FRPC_CHANGES_RETURNED, changes_returned).
+
+-define(STATS_TO_KEYS, #{
+ [mango, evaluate_selector] => ?MANGO_EVAL_MATCH,
+ [couchdb, database_reads] => ?DB_OPEN_DOC,
+ [fabric_rpc, changes, processed] => ?FRPC_CHANGES_ROW,
+ [fabric_rpc, changes, returned] => ?FRPC_CHANGES_RETURNED,
+ [fabric_rpc, view, rows_read] => ?ROWS_READ,
+ [couchdb, couch_server, open] => ?DB_OPEN,
+ [couchdb, btree, get_node, kp_node] => ?COUCH_BT_GET_KP_NODE,
+ [couchdb, btree, get_node, kv_node] => ?COUCH_BT_GET_KV_NODE,
+ [couchdb, btree, write_node, kp_node] => ?COUCH_BT_WRITE_KP_NODE,
+ [couchdb, btree, write_node, kv_node] => ?COUCH_BT_WRITE_KV_NODE,
+ %% NOTE: these stats are not local to the RPC worker, need forwarding
+ [couchdb, query_server, calls, ddoc_filter] => ?COUCH_JS_FILTER,
+ [couchdb, query_server, volume, ddoc_filter] => ?COUCH_JS_FILTERED_DOCS
+}).
+
+-define(KEYS_TO_FIELDS, #{
+ ?DB_OPEN => #rctx.?DB_OPEN,
+ ?ROWS_READ => #rctx.?ROWS_READ,
+ ?FRPC_CHANGES_RETURNED => #rctx.?FRPC_CHANGES_RETURNED,
+ ?DOCS_WRITTEN => #rctx.?DOCS_WRITTEN,
+ ?IOQ_CALLS => #rctx.?IOQ_CALLS,
+ ?COUCH_JS_FILTER => #rctx.?COUCH_JS_FILTER,
+ ?COUCH_JS_FILTERED_DOCS => #rctx.?COUCH_JS_FILTERED_DOCS,
+ ?MANGO_EVAL_MATCH => #rctx.?MANGO_EVAL_MATCH,
+ ?DB_OPEN_DOC => #rctx.?DB_OPEN_DOC,
+ ?FRPC_CHANGES_ROW => #rctx.?ROWS_READ, %% TODO: rework double use of rows_read
+ ?COUCH_BT_GET_KP_NODE => #rctx.?COUCH_BT_GET_KP_NODE,
+ ?COUCH_BT_GET_KV_NODE => #rctx.?COUCH_BT_GET_KV_NODE,
+ ?COUCH_BT_WRITE_KP_NODE => #rctx.?COUCH_BT_WRITE_KP_NODE,
+ ?COUCH_BT_WRITE_KV_NODE => #rctx.?COUCH_BT_WRITE_KV_NODE
+}).
+
+-type pid_ref() :: {pid(), reference()}.
+-type maybe_pid_ref() :: pid_ref() | undefined.
+-type maybe_pid() :: pid() | undefined.
+
+-record(rpc_worker, {
+ mod :: atom() | '_',
+ func :: atom() | '_',
+ from :: pid_ref() | '_'
+}).
+
+-record(coordinator, {
+ mod :: atom() | '_',
+ func :: atom() | '_',
+ method :: atom() | '_',
+ path :: binary() | '_'
+}).
+
+-type coordinator() :: #coordinator{}.
+-type rpc_worker() :: #rpc_worker{}.
+-type rctx_type() :: coordinator() | rpc_worker().
+
+-record(rctx, {
+ %% Metadata
+ started_at = csrt_util:tnow(),
+ updated_at = csrt_util:tnow(),
+ pid_ref :: maybe_pid_ref() | {'_', '_'},
+ nonce,
+ type :: rctx_type() | undefined | '_',
+ dbname,
+ username,
+
+ %% Stats counters
+ db_open = 0,
+ docs_read = 0 :: non_neg_integer(),
+ docs_written = 0 :: non_neg_integer(),
+ rows_read = 0 :: non_neg_integer(),
+ changes_processed = 0 :: non_neg_integer(),
+ changes_returned = 0 :: non_neg_integer(),
+ ioq_calls = 0 :: non_neg_integer(),
+ io_bytes_read = 0 :: non_neg_integer(),
+ io_bytes_written = 0 :: non_neg_integer(),
+ js_evals = 0 :: non_neg_integer(),
+ js_filter = 0 :: non_neg_integer(),
+ js_filtered_docs = 0 :: non_neg_integer(),
+ mango_eval_match = 0 :: non_neg_integer(),
+ %% TODO: switch record definitions to be macro based, eg:
+ %% ?COUCH_BT_GET_KP_NODE = 0 :: non_neg_integer(),
+ get_kv_node = 0 :: non_neg_integer(),
+ get_kp_node = 0 :: non_neg_integer(),
+ write_kv_node = 0 :: non_neg_integer(),
+ write_kp_node = 0 :: non_neg_integer()
+}).
+
+-type rctx_field() ::
+ started_at
+ | updated_at
+ | pid_ref
+ | nonce
+ | type
+ | dbname
+ | username
+ | db_open
+ | docs_read
+ | docs_written
+ | rows_read
+ | changes_processed
+ | changes_returned
+ | ioq_calls
+ | io_bytes_read
+ | io_bytes_written
+ | js_evals
+ | js_filter
+ | js_filtered_docs
+ | mango_eval_match
+ | get_kv_node
+ | get_kp_node
+ | write_kv_node
+ | write_kp_node.
+
+-type coordinator_rctx() :: #rctx{type :: coordinator()}.
+-type rpc_worker_rctx() :: #rctx{type :: rpc_worker()}.
+-type rctx() :: #rctx{} | coordinator_rctx() | rpc_worker_rctx().
+-type maybe_rctx() :: rctx() | undefined.
+
+%% TODO: solidify nonce type
+-type nonce() :: any().
+-type dbname() :: iodata().
+-type username() :: iodata().
+
+-type delta() :: map().
+-type maybe_delta() :: delta() | undefined.
+-type tagged_delta() :: {delta, maybe_delta()}.
+
+-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/couch_stats_sup.erl b/src/couch_stats/src/couch_stats_sup.erl
index 325372c3e..826dfbac4 100644
--- a/src/couch_stats/src/couch_stats_sup.erl
+++ b/src/couch_stats/src/couch_stats_sup.erl
@@ -29,6 +29,8 @@ init([]) ->
{
{one_for_one, 5, 10}, [
?CHILD(couch_stats_server, worker),
+ ?CHILD(csrt_server, worker),
+ ?CHILD(csrt_logger, worker),
?CHILD(couch_stats_process_tracker, worker)
]
}}.
diff --git a/src/couch_stats/src/csrt.erl b/src/couch_stats/src/csrt.erl
new file mode 100644
index 000000000..a2b5f51d8
--- /dev/null
+++ b/src/couch_stats/src/csrt.erl
@@ -0,0 +1,500 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(csrt).
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_stats_resource_tracker.hrl").
+
+%% PidRef API
+-export([
+ destroy_pid_ref/0,
+ destroy_pid_ref/1,
+ create_pid_ref/0,
+ get_pid_ref/0,
+ get_pid_ref/1,
+ set_pid_ref/1
+]).
+
+%% Context API
+-export([
+ create_context/2,
+ create_coordinator_context/2,
+ create_worker_context/3,
+ destroy_context/0,
+ destroy_context/1,
+ get_resource/0,
+ get_resource/1,
+ set_context_dbname/1,
+ set_context_dbname/2,
+ set_context_handler_fun/1,
+ set_context_handler_fun/2,
+ set_context_username/1,
+ set_context_username/2
+]).
+
+%% Public API
+-export([
+ is_enabled/0,
+ is_enabled_init_p/0,
+ do_report/2,
+ maybe_report/2,
+ conf_get/1,
+ conf_get/2
+]).
+
+%% stats collection api
+-export([
+ accumulate_delta/1,
+ add_delta/2,
+ docs_written/1,
+ extract_delta/1,
+ get_delta/0,
+ inc/1,
+ inc/2,
+ ioq_called/0,
+ js_filtered/1,
+ make_delta/0,
+ rctx_delta/2,
+ maybe_add_delta/1,
+ maybe_add_delta/2,
+ maybe_inc/2,
+ should_track_init_p/1
+]).
+
+%% aggregate query api
+-export([
+ active/0,
+ active/1,
+ active_coordinators/0,
+ active_coordinators/1,
+ active_workers/0,
+ active_workers/1,
+ count_by/1,
+ find_by_nonce/1,
+ find_by_pid/1,
+ find_by_pidref/1,
+ find_workers_by_pidref/1,
+ group_by/2,
+ group_by/3,
+ sorted/1,
+ sorted_by/1,
+ sorted_by/2,
+ sorted_by/3
+]).
+
+%%
+%% PidRef operations
+%%
+
+-spec get_pid_ref() -> maybe_pid_ref().
+get_pid_ref() ->
+ csrt_util:get_pid_ref().
+
+-spec get_pid_ref(Rctx :: rctx()) -> pid_ref().
+get_pid_ref(Rctx) ->
+ csrt_util:get_pid_ref(Rctx).
+
+-spec set_pid_ref(PidRef :: pid_ref()) -> pid_ref().
+set_pid_ref(PidRef) ->
+ csrt_util:set_pid_ref(PidRef).
+
+-spec create_pid_ref() -> pid_ref().
+create_pid_ref() ->
+ csrt_server:create_pid_ref().
+
+-spec destroy_pid_ref() -> maybe_pid_ref().
+destroy_pid_ref() ->
+ destroy_pid_ref(get_pid_ref()).
+
+%%destroy_pid_ref(undefined) ->
+%% undefined;
+-spec destroy_pid_ref(PidRef :: maybe_pid_ref()) -> maybe_pid_ref().
+destroy_pid_ref(_PidRef) ->
+ erase(?PID_REF).
+
+%%
+%% Context lifecycle API
+%%
+
+%% TODO: shouldn't need this?
+%% create_resource(#rctx{} = Rctx) ->
+%% csrt_server:create_resource(Rctx).
+
+-spec create_worker_context(From, MFA, Nonce) -> pid_ref() | false when
+ From :: pid_ref(), MFA :: mfa(), Nonce :: term().
+create_worker_context(From, {M,F,_A}, Nonce) ->
+ case is_enabled() of
+ true ->
+ Type = #rpc_worker{from=From, mod=M, func=F},
+ create_context(Type, Nonce);
+ false ->
+ false
+ end.
+
+-spec create_coordinator_context(Httpd , Path) -> pid_ref() | false when
+ Httpd :: #httpd{}, Path :: list().
+create_coordinator_context(#httpd{method=Verb, nonce=Nonce}, Path0) ->
+ case is_enabled() of
+ true ->
+ Path = list_to_binary([$/ | Path0]),
+ Type = #coordinator{method=Verb, path=Path},
+ create_context(Type, Nonce);
+ false ->
+ false
+ end.
+
+-spec create_context(Type :: rctx_type(), Nonce :: term()) -> pid_ref().
+create_context(Type, Nonce) ->
+ Rctx = csrt_server:new_context(Type, Nonce),
+ %% TODO: which approach
+ %% PidRef = csrt_server:pid_ref(Rctx),
+ PidRef = get_pid_ref(Rctx),
+ set_pid_ref(PidRef),
+ csrt_util:set_delta_zero(Rctx),
+ csrt_util:set_delta_a(Rctx),
+ csrt_server:create_resource(Rctx),
+ csrt_logger:track(Rctx),
+ PidRef.
+
+-spec set_context_dbname(DbName :: binary()) -> boolean().
+set_context_dbname(DbName) ->
+ set_context_dbname(DbName, get_pid_ref()).
+
+-spec set_context_dbname(DbName, PidRef) -> boolean() when
+ DbName :: binary(), PidRef :: pid_ref() | undefined.
+set_context_dbname(_, undefined) ->
+ false;
+set_context_dbname(DbName, PidRef) ->
+ is_enabled() andalso csrt_server:set_context_dbname(DbName, PidRef).
+
+-spec set_context_handler_fun(Fun :: function()) -> boolean().
+set_context_handler_fun(Fun) when is_function(Fun) ->
+ case is_enabled() of
+ false ->
+ false;
+ true ->
+ FProps = erlang:fun_info(Fun),
+ Mod = proplists:get_value(module, FProps),
+ Func = proplists:get_value(name, FProps),
+ update_handler_fun(Mod, Func, get_pid_ref())
+ end.
+
+-spec set_context_handler_fun(Mod :: atom(), Func :: atom()) -> boolean().
+set_context_handler_fun(Mod, Func)
+ when is_atom(Mod) andalso is_atom(Func) ->
+ case is_enabled() of
+ false ->
+ false;
+ true ->
+ update_handler_fun(Mod, Func, get_pid_ref())
+ end.
+
+-spec update_handler_fun(Mod, Func, PidRef) -> boolean() when
+ Mod :: atom(), Func :: atom(), PidRef :: maybe_pid_ref().
+update_handler_fun(_, _, undefined) ->
+ false;
+update_handler_fun(Mod, Func, PidRef) ->
+ Rctx = get_resource(PidRef),
+ %% TODO: #coordinator{} assumption needs to adapt for other types
+ #coordinator{} = Coordinator0 = csrt_server:get_context_type(Rctx),
+ Coordinator = Coordinator0#coordinator{mod=Mod, func=Func},
+ csrt_server:set_context_type(Coordinator, PidRef),
+ ok.
+
+%% @equiv set_context_username(User, get_pid_ref())
+set_context_username(User) ->
+ set_context_username(User, get_pid_ref()).
+
+-spec set_context_username(User, PidRef) -> boolean() when
+ User :: null | undefined | #httpd{} | #user_ctx{} | binary(),
+ PidRef :: maybe_pid_ref().
+set_context_username(null, _) ->
+ false;
+set_context_username(_, undefined) ->
+ false;
+set_context_username(#httpd{user_ctx = Ctx}, PidRef) ->
+ set_context_username(Ctx, PidRef);
+set_context_username(#user_ctx{name = Name}, PidRef) ->
+ set_context_username(Name, PidRef);
+set_context_username(UserName, PidRef) ->
+ is_enabled() andalso csrt_server:set_context_username(UserName, PidRef).
+
+-spec destroy_context() -> ok.
+destroy_context() ->
+ destroy_context(get_pid_ref()).
+
+-spec destroy_context(PidRef :: maybe_pid_ref()) -> ok.
+destroy_context(undefined) ->
+ ok;
+destroy_context({_, _} = PidRef) ->
+ csrt_logger:stop_tracker(),
+ destroy_pid_ref(PidRef),
+ ok.
+
+%%
+%% Public API
+%%
+
+%% @equiv csrt_util:is_enabled().
+-spec is_enabled() -> boolean().
+is_enabled() ->
+ csrt_util:is_enabled().
+
+%% @equiv csrt_util:is_enabled_init_p().
+-spec is_enabled_init_p() -> boolean().
+is_enabled_init_p() ->
+ csrt_util:is_enabled_init_p().
+
+-spec get_resource() -> maybe_rctx().
+get_resource() ->
+ get_resource(get_pid_ref()).
+
+-spec get_resource(PidRef :: maybe_pid_ref()) -> maybe_rctx().
+get_resource(PidRef) ->
+ csrt_server:get_resource(PidRef).
+
+%% Log a CSRT report if any filters match
+-spec maybe_report(ReportName :: string(), PidRef :: pid_ref()) -> ok.
+maybe_report(ReportName, PidRef) ->
+ csrt_logger:maybe_report(ReportName, PidRef).
+
+%% Direct report logic skipping should log filters
+-spec do_report(ReportName :: string(), PidRef :: pid_ref()) -> boolean().
+do_report(ReportName, PidRef) ->
+ csrt_logger:do_report(ReportName, get_resource(PidRef)).
+
+%%
+%% Stat collection API
+%%
+
+-spec inc(Key :: rctx_field()) -> non_neg_integer().
+inc(Key) ->
+ is_enabled() andalso csrt_server:inc(get_pid_ref(), Key).
+
+-spec inc(Key :: rctx_field(), N :: non_neg_integer()) -> non_neg_integer().
+inc(Key, N) when is_integer(N) andalso N >= 0 ->
+ is_enabled() andalso csrt_server:inc(get_pid_ref(), Key, N).
+
+
+-spec maybe_inc(Stat :: atom(), Val :: non_neg_integer()) -> non_neg_integer().
+maybe_inc(Stat, Val) ->
+ case maps:is_key(Stat, ?STATS_TO_KEYS) of
+ true ->
+ inc(maps:get(Stat, ?STATS_TO_KEYS), Val);
+ false ->
+ 0
+ end.
+
+-spec should_track_init_p(Stat :: [atom()]) -> boolean().
+should_track_init_p([Mod, Func, spawned]) ->
+ is_enabled_init_p() andalso csrt_util:should_track_init_p(Mod, Func);
+should_track_init_p(_Metric) ->
+ false.
+
+-spec ioq_called() -> non_neg_integer().
+ioq_called() ->
+ inc(ioq_calls).
+
+%% we cannot yet use stats couchdb.query_server.*.ddoc_filter because those
+%% are collected in a dedicated process.
+%% TODO: funnel back stats from background worker processes to the RPC worker
+js_filtered(N) ->
+ inc(js_filter),
+ inc(js_filtered_docs, N).
+
+docs_written(N) ->
+ inc(docs_written, N).
+
+-spec accumulate_delta(Delta :: map() | undefined) -> ok.
+accumulate_delta(Delta) when is_map(Delta) ->
+ %% TODO: switch to creating a batch of updates to invoke a single
+ %% update_counter rather than sequentially invoking it for each field
+ is_enabled() andalso maps:foreach(fun inc/2, Delta),
+ ok;
+accumulate_delta(undefined) ->
+ ok.
+
+-spec make_delta() -> maybe_delta().
+make_delta() ->
+ case is_enabled() of
+ false ->
+ undefined;
+ true ->
+ csrt_util:make_delta(get_pid_ref())
+ end.
+
+-spec rctx_delta(TA :: maybe_rctx(), TB :: maybe_rctx()) -> maybe_delta().
+rctx_delta(TA, TB) ->
+ csrt_util:rctx_delta(TA, TB).
+
+%% TODO: cleanup return type
+%%-spec update_counter(Field :: rctx_field(), Count :: non_neg_integer()) -> false | ok.
+%%-spec update_counter(Field :: non_neg_integer(), Count :: non_neg_integer()) -> false | ok.
+%%update_counter(_Field, Count) when Count < 0 ->
+%% false;
+%%update_counter(Field, Count) when Count >= 0 ->
+%% is_enabled() andalso csrt_server:update_counter(get_pid_ref(), Field, Count).
+
+
+-spec conf_get(Key :: string()) -> string().
+conf_get(Key) ->
+ csrt_util:conf_get(Key).
+
+
+-spec conf_get(Key :: string(), Default :: string()) -> string().
+conf_get(Key, Default) ->
+ csrt_util:conf_get(Key, Default).
+
+%%
+%% aggregate query api
+%%
+
+-spec active() -> [rctx()].
+active() ->
+ csrt_query:active().
+
+%% TODO: ensure Type fields align with type specs
+%%-spec active(Type :: rctx_type()) -> [rctx()].
+-spec active(Type :: json) -> [rctx()].
+active(Type) ->
+ csrt_query:active(Type).
+
+-spec active_coordinators() -> [coordinator_rctx()].
+active_coordinators() ->
+ csrt_query:active_coordinators().
+
+%% TODO: cleanup json logic here
+-spec active_coordinators(Type :: json) -> [coordinator_rctx()].
+active_coordinators(Type) ->
+ csrt_query:active_coordinators(Type).
+
+-spec active_workers() -> [rpc_worker_rctx()].
+active_workers() ->
+ csrt_query:active_workers().
+
+-spec active_workers(Type :: json) -> [rpc_worker_rctx()].
+active_workers(Type) ->
+ csrt_query:active_workers(Type).
+
+-spec count_by(Key :: string()) -> map().
+count_by(Key) ->
+ csrt_query:count_by(Key).
+
+find_by_nonce(Nonce) ->
+ csrt_query:find_by_nonce(Nonce).
+
+find_by_pid(Pid) ->
+ csrt_query:find_by_pid(Pid).
+
+find_by_pidref(PidRef) ->
+ csrt_query:find_by_pidref(PidRef).
+
+find_workers_by_pidref(PidRef) ->
+ csrt_query:find_workers_by_pidref(PidRef).
+
+group_by(Key, Val) ->
+ csrt_query:group_by(Key, Val).
+
+group_by(Key, Val, Agg) ->
+ csrt_query:group_by(Key, Val, Agg).
+
+sorted(Map) ->
+ csrt_query:sorted(Map).
+
+sorted_by(Key) ->
+ csrt_query:sorted_by(Key).
+
+sorted_by(Key, Val) ->
+ csrt_query:sorted_by(Key, Val).
+
+sorted_by(Key, Val, Agg) ->
+ csrt_query:sorted_by(Key, Val, Agg).
+
+%%
+%% Delta API
+%%
+
+add_delta(T, Delta) ->
+ csrt_util:add_delta(T, Delta).
+
+extract_delta(T) ->
+ csrt_util:extract_delta(T).
+
+
+get_delta() ->
+ csrt_util:get_delta(get_pid_ref()).
+
+maybe_add_delta(T) ->
+ csrt_util:maybe_add_delta(T).
+
+maybe_add_delta(T, Delta) ->
+ csrt_util:maybe_add_delta(T, Delta).
+
+%%
+%% Internal Operations assuming is_enabled() == true
+%%
+
+
+-ifdef(TEST).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+couch_stats_resource_tracker_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(t_static_map_translations),
+ ?TDEF_FE(t_should_track_init_p),
+ ?TDEF_FE(t_should_not_track_init_p)
+ ]
+ }.
+
+setup() ->
+ test_util:start_couch().
+
+teardown(Ctx) ->
+ test_util:stop_couch(Ctx).
+
+t_static_map_translations(_) ->
+ ?assert(lists:all(fun(E) -> maps:is_key(E, ?KEYS_TO_FIELDS) end, maps:values(?STATS_TO_KEYS))),
+ %% TODO: properly handle ioq_calls field
+ ?assertEqual(lists:sort(maps:values(?STATS_TO_KEYS)), lists:delete(docs_written, lists:delete(ioq_calls, lists:sort(maps:keys(?KEYS_TO_FIELDS))))).
+
+t_should_track_init_p(_) ->
+ config:set(?CSRT_INIT_P, "enabled", "true", false),
+ Metrics = [
+ [fabric_rpc, all_docs, spawned],
+ [fabric_rpc, changes, spawned],
+ [fabric_rpc, map_view, spawned],
+ [fabric_rpc, reduce_view, spawned],
+ [fabric_rpc, get_all_security, spawned],
+ [fabric_rpc, open_doc, spawned],
+ [fabric_rpc, update_docs, spawned],
+ [fabric_rpc, open_shard, spawned]
+ ],
+ [csrt_util:set_fabric_init_p(F, true, false) || [_, F, _] <- Metrics],
+ [?assert(should_track_init_p(M), M) || M <- Metrics].
+
+t_should_not_track_init_p(_) ->
+ config:set(?CSRT_INIT_P, "enabled", "true", false),
+ Metrics = [
+ [couch_db, name, spawned],
+ [couch_db, get_db_info, spawned],
+ [couch_db, open, spawned],
+ [fabric_rpc, get_purge_seq, spawned]
+ ],
+ [?assert(should_track_init_p(M) =:= false, M) || M <- Metrics].
+
+-endif.
diff --git a/src/couch_stats/src/csrt_logger.erl b/src/couch_stats/src/csrt_logger.erl
new file mode 100644
index 000000000..2e692e888
--- /dev/null
+++ b/src/couch_stats/src/csrt_logger.erl
@@ -0,0 +1,401 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(csrt_logger).
+
+%% Process lifetime logging api
+-export([
+ get_tracker/0,
+ log_process_lifetime_report/1,
+ put_tracker/1,
+ stop_tracker/0,
+ stop_tracker/1,
+ track/1,
+ tracker/1
+]).
+
+%% Raw API that bypasses is_enabled checks
+-export([
+ do_lifetime_report/1,
+ do_status_report/1,
+ do_report/2,
+ maybe_report/2
+]).
+
+-export([
+ start_link/0,
+ init/1,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2
+]).
+
+%% Config update subscription API
+-export([
+ subscribe_changes/0,
+ handle_config_change/5,
+ handle_config_terminate/3
+]).
+
+%% 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,
+ reload_matchers/0
+]).
+
+-include_lib("stdlib/include/ms_transform.hrl").
+-include_lib("couch_stats_resource_tracker.hrl").
+
+-define(MATCHERS_KEY, {?MODULE, all_csrt_matchers}).
+-define(CONF_MATCHERS_ENABLED, "csrt_logger.matchers_enabled").
+-define(CONF_MATCHERS_THRESHOLD, "csrt_logger.matchers_threshold").
+-define(CONF_MATCHERS_DBNAMES, "csrt_logger.dbnames_io").
+
+-record(st, {
+ matchers = #{}
+}).
+
+-spec track(Rctx :: rctx()) -> pid().
+track(#rctx{pid_ref=PidRef}) ->
+ case get_tracker() of
+ undefined ->
+ Pid = spawn(?MODULE, tracker, [PidRef]),
+ put_tracker(Pid),
+ Pid;
+ Pid when is_pid(Pid) ->
+ Pid
+ end.
+
+-spec tracker(PidRef :: pid_ref()) -> ok.
+tracker({Pid, _Ref}=PidRef) ->
+ MonRef = erlang:monitor(process, Pid),
+ receive
+ stop ->
+ %% TODO: do we need cleanup here?
+ log_process_lifetime_report(PidRef),
+ csrt_server:destroy_resource(PidRef),
+ ok;
+ {'DOWN', MonRef, _Type, _0DPid, _Reason0} ->
+ %% TODO: should we pass reason to log_process_lifetime_report?
+ %% Reason = case Reason0 of
+ %% {shutdown, Shutdown0} ->
+ %% Shutdown = atom_to_binary(Shutdown0),
+ %% <<"shutdown: ", Shutdown/binary>>;
+ %% Reason0 ->
+ %% Reason0
+ %% end,
+ %% TODO: should we send the induced work delta to the coordinator?
+ log_process_lifetime_report(PidRef),
+ csrt_server:destroy_resource(PidRef),
+ ok
+ end.
+
+-spec log_process_lifetime_report(PidRef :: pid_ref()) -> ok.
+log_process_lifetime_report(PidRef) ->
+ case csrt_util:is_enabled() of
+ true ->
+ maybe_report("csrt-pid-usage-lifetime", PidRef);
+ false ->
+ ok
+ end.
+
+%% TODO: add Matchers spec
+-spec find_matches(Rctxs :: [rctx()], Matchers :: [any()]) -> matchers().
+find_matches(Rctxs, Matchers) when is_list(Rctxs) andalso is_map(Matchers) ->
+ maps:filter(
+ fun(_Name, {_MSpec, CompMSpec}) ->
+ catch [] =/= ets:match_spec_run(Rctxs, CompMSpec)
+ end,
+ 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;
+is_match(#rctx{}=Rctx) ->
+ is_match(Rctx, get_matchers()).
+
+%% TODO: add Matchers spec
+-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.
+
+-spec maybe_report(ReportName :: string(), PidRef :: maybe_pid_ref()) -> ok.
+maybe_report(ReportName, PidRef) ->
+ Rctx = csrt_server:get_resource(PidRef),
+ case is_match(Rctx) of
+ true ->
+ do_report(ReportName, Rctx),
+ ok;
+ false ->
+ ok
+ end.
+
+-spec do_lifetime_report(Rctx :: rctx()) -> boolean().
+do_lifetime_report(Rctx) ->
+ do_report("csrt-pid-usage-lifetime", Rctx).
+
+-spec do_status_report(Rctx :: rctx()) -> boolean().
+do_status_report(Rctx) ->
+ do_report("csrt-pid-usage-status", Rctx).
+
+-spec do_report(ReportName :: string(), Rctx :: rctx()) -> boolean().
+do_report(ReportName, #rctx{}=Rctx) ->
+ couch_log:report(ReportName, csrt_util:to_json(Rctx)).
+
+%%
+%% Process lifetime logging api
+%%
+
+-spec get_tracker() -> maybe_pid().
+get_tracker() ->
+ get(?TRACKER_PID).
+
+-spec put_tracker(Pid :: pid()) -> maybe_pid().
+put_tracker(Pid) when is_pid(Pid) ->
+ put(?TRACKER_PID, Pid).
+
+-spec stop_tracker() -> ok.
+stop_tracker() ->
+ stop_tracker(get_tracker()).
+
+-spec stop_tracker(Pid :: maybe_pid()) -> ok.
+stop_tracker(undefined) ->
+ ok;
+stop_tracker(Pid) when is_pid(Pid) ->
+ Pid ! stop,
+ ok.
+
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+init([]) ->
+ ok = initialize_matchers(),
+ ok = subscribe_changes(),
+ {ok, #st{}}.
+
+handle_call({register, Name, MSpec}, _From, #st{matchers=Matchers}=St) ->
+ case add_matcher(Name, MSpec, Matchers) of
+ {ok, Matchers1} ->
+ set_matchers_term(Matchers1),
+ {reply, ok, St#st{matchers=Matchers1}};
+ {error, badarg}=Error ->
+ {reply, Error, St}
+ end;
+handle_call(reload_matchers, _From, St) ->
+ couch_log:warning("Reloading persistent term matchers", []),
+ ok = initialize_matchers(),
+ {reply, ok, St};
+handle_call(_, _From, State) ->
+ {reply, ok, State}.
+
+handle_cast(_Msg, State) ->
+ {noreply, State, 0}.
+
+handle_info(restart_config_listener, State) ->
+ ok = config:listen_for_changes(?MODULE, nil),
+ {noreply, State};
+handle_info(_Msg, St) ->
+ {noreply, St}.
+
+%%
+%% Matchers
+%%
+
+-spec matcher_on_dbname(DbName :: dbname()) -> ets:match_spec().
+matcher_on_dbname(DbName)
+ when is_binary(DbName) ->
+ ets:fun2ms(fun(#rctx{dbname=DbName1} = R) when DbName =:= DbName1 -> R end).
+
+-spec matcher_on_dbname_io_threshold(DbName, Threshold) -> ets:match_spec() when
+ DbName :: dbname(), Threshold :: pos_integer().
+matcher_on_dbname_io_threshold(DbName, Threshold)
+ when is_binary(DbName) ->
+ ets:fun2ms(fun(#rctx{dbname=DbName1, ioq_calls=IOQ, get_kv_node=KVN, get_kp_node=KPN, docs_read=Docs, rows_read=Rows, changes_processed=Chgs} = R) when DbName =:= DbName1 andalso ((IOQ > Threshold) or (KVN >= Threshold) or (KPN >= Threshold) or (Docs >= Threshold) or (Rows >= Threshold) or (Chgs >= Threshold)) -> R end).
+
+-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{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{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=RowsRead} = R) when RowsRead >= Threshold -> R end).
+
+-spec matcher_on_nonce(Nonce :: nonce()) -> ets:match_spec().
+matcher_on_nonce(Nonce) ->
+ ets:fun2ms(fun(#rctx{nonce = Nonce1} = R) when Nonce =:= Nonce1 -> R end).
+
+-spec matcher_on_worker_changes_processed(Threshold :: pos_integer()) -> ets:match_spec().
+matcher_on_worker_changes_processed(Threshold)
+ when is_integer(Threshold) andalso Threshold > 0 ->
+ ets:fun2ms(
+ fun(
+ #rctx{
+ changes_processed=Processed,
+ changes_returned=Returned
+ } = R
+ ) when (Processed - Returned) >= Threshold ->
+ R
+ end
+ ).
+
+-spec matcher_on_ioq_calls(Threshold :: pos_integer()) -> ets:match_spec().
+matcher_on_ioq_calls(Threshold)
+ when is_integer(Threshold) andalso Threshold > 0 ->
+ ets:fun2ms(fun(#rctx{ioq_calls=IOQCalls} = R) when IOQCalls >= Threshold -> R end).
+
+-spec add_matcher(Name, MSpec, Matchers) -> {ok, matchers()} | {error, badarg} when
+ Name :: string(), MSpec :: ets:match_spec(), Matchers :: matchers().
+add_matcher(Name, MSpec, Matchers) ->
+ try ets:match_spec_compile(MSpec) of
+ CompMSpec ->
+ %% TODO: handle already registered name case
+ Matchers1 = maps:put(Name, {MSpec, CompMSpec}, Matchers),
+ {ok, Matchers1}
+ catch
+ error:badarg ->
+ {error, badarg}
+ end.
+
+-spec set_matchers_term(Matchers :: matchers()) -> maybe_matchers().
+set_matchers_term(Matchers) when is_map(Matchers) ->
+ persistent_term:put({?MODULE, all_csrt_matchers}, Matchers).
+
+-spec initialize_matchers() -> ok.
+initialize_matchers() ->
+ DefaultMatchers = [
+ {docs_read, fun matcher_on_docs_read/1, 100},
+ {dbname, fun matcher_on_dbname/1, <<"foo">>},
+ {rows_read, fun matcher_on_rows_read/1, 100},
+ {docs_written, fun matcher_on_docs_written/1, 1},
+ %%{view_rows_read, fun matcher_on_rows_read/1, 1000},
+ %%{slow_reqs, fun matcher_on_slow_reqs/1, 10000},
+ {worker_changes_processed, fun matcher_on_worker_changes_processed/1, 1000},
+ {ioq_calls, fun matcher_on_ioq_calls/1, 10000}
+ ],
+ Matchers = lists:foldl(
+ fun({Name0, MatchGenFunc, Threshold0}, Matchers0) when is_atom(Name0) ->
+ Name = atom_to_list(Name0),
+ case matcher_enabled(Name) of
+ true ->
+ Threshold = matcher_threshold(Name, Threshold0),
+ %% TODO: handle errors from Func
+ case add_matcher(Name, MatchGenFunc(Threshold), Matchers0) of
+ {ok, Matchers1} ->
+ Matchers1;
+ {error, badarg} ->
+ couch_log:warning("[~p] Failed to initialize matcher: ~p", [?MODULE, Name]),
+ Matchers0
+ end;
+ false ->
+ Matchers0
+ end
+ end,
+ #{},
+ DefaultMatchers
+ ),
+ Matchers1 = lists:foldl(
+ fun({Dbname, Value}, Matchers0) ->
+ try list_to_integer(Value) of
+ Threshold when Threshold > 0 ->
+ Name = "dbname_io__" ++ Dbname ++ "__" ++ Value,
+ DbnameB = list_to_binary(Dbname),
+ MSpec = matcher_on_dbname_io_threshold(DbnameB, Threshold),
+ case add_matcher(Name, MSpec, Matchers0) of
+ {ok, Matchers1} ->
+ Matchers1;
+ {error, badarg} ->
+ couch_log:warning("[~p] Failed to initialize matcher: ~p", [?MODULE, Name]),
+ Matchers0
+ end;
+ _ ->
+ Matchers0
+ catch error:badarg ->
+ couch_log:warning("[~p] Failed to initialize dbname io matcher on: ~p", [?MODULE, Dbname])
+ end
+ end,
+ Matchers,
+ config:get(?CONF_MATCHERS_DBNAMES)
+ ),
+
+ couch_log:notice("Initialized ~p CSRT Logger matchers", [maps:size(Matchers1)]),
+ persistent_term:put(?MATCHERS_KEY, Matchers1),
+ ok.
+
+-spec matcher_enabled(Name :: string()) -> boolean().
+matcher_enabled(Name) when is_list(Name) ->
+ %% TODO: fix
+ %% config:get_boolean(?CONF_MATCHERS_ENABLED, Name, false).
+ config:get_boolean(?CONF_MATCHERS_ENABLED, Name, true).
+
+-spec matcher_threshold(Name, Threshold) -> string() | integer() when
+ Name :: string(), Threshold :: pos_integer() | string().
+matcher_threshold("dbname", DbName) when is_binary(DbName) ->
+ %% TODO: toggle Default to undefined to disallow for particular dbname
+ %% TODO: sort out list vs binary
+ %%config:get_integer(?CONF_MATCHERS_THRESHOLD, binary_to_list(DbName), Default);
+ DbName;
+matcher_threshold(Name, Default)
+ when is_list(Name) andalso is_integer(Default) andalso Default > 0 ->
+ config:get_integer(?CONF_MATCHERS_THRESHOLD, Name, Default).
+
+subscribe_changes() ->
+ config:listen_for_changes(?MODULE, nil).
+
+handle_config_change(?CONF_MATCHERS_ENABLED, _Key, _Val, _Persist, St) ->
+ ok = gen_server:call(?MODULE, reload_matchers, infinity),
+ {ok, St};
+handle_config_change(?CONF_MATCHERS_THRESHOLD, _Key, _Val, _Persist, St) ->
+ ok = gen_server:call(?MODULE, reload_matchers, infinity),
+ {ok, St};
+handle_config_change(_Sec, _Key, _Val, _Persist, St) ->
+ {ok, St}.
+
+handle_config_terminate(_, stop, _) ->
+ ok;
+handle_config_terminate(_, _, _) ->
+ erlang:send_after(5000, whereis(?MODULE), restart_config_listener).
diff --git a/src/couch_stats/src/csrt_query.erl b/src/couch_stats/src/csrt_query.erl
new file mode 100644
index 000000000..a3580c58a
--- /dev/null
+++ b/src/couch_stats/src/csrt_query.erl
@@ -0,0 +1,174 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(csrt_query).
+
+-include_lib("stdlib/include/ms_transform.hrl").
+-include_lib("couch_stats_resource_tracker.hrl").
+
+%% aggregate query api
+-export([
+ active/0,
+ active/1,
+ active_coordinators/0,
+ active_coordinators/1,
+ active_workers/0,
+ active_workers/1,
+ count_by/1,
+ find_by_nonce/1,
+ find_by_pid/1,
+ find_by_pidref/1,
+ find_workers_by_pidref/1,
+ group_by/2,
+ group_by/3,
+ sorted/1,
+ sorted_by/1,
+ sorted_by/2,
+ sorted_by/3
+]).
+
+%%
+%% Aggregate query API
+%%
+
+active() ->
+ active_int(all).
+
+active_coordinators() ->
+ active_int(coordinators).
+
+active_workers() ->
+ active_int(workers).
+
+%% active_json() or active(json)?
+active(json) ->
+ to_json_list(active_int(all)).
+
+active_coordinators(json) ->
+ to_json_list(active_int(coordinators)).
+
+active_workers(json) ->
+ to_json_list(active_int(workers)).
+
+active_int(coordinators) ->
+ select_by_type(coordinators);
+active_int(workers) ->
+ select_by_type(workers);
+active_int(all) ->
+ select_by_type(all).
+
+select_by_type(coordinators) ->
+ ets:select(?MODULE, ets:fun2ms(fun(#rctx{type = #coordinator{}} = R) -> R end));
+select_by_type(workers) ->
+ ets:select(?MODULE, ets:fun2ms(fun(#rctx{type = #rpc_worker{}} = R) -> R end));
+select_by_type(all) ->
+ ets:tab2list(?MODULE).
+
+find_by_nonce(Nonce) ->
+ %%ets:match_object(?MODULE, ets:fun2ms(fun(#rctx{nonce = Nonce1} = R) when Nonce =:= Nonce1 -> R end)).
+ [R || R <- ets:match_object(?MODULE, #rctx{nonce=Nonce})].
+
+find_by_pid(Pid) ->
+ %%[R || #rctx{} = R <- ets:match_object(?MODULE, #rctx{pid_ref={Pid, '_'}, _ = '_'})].
+ [R || R <- ets:match_object(?MODULE, #rctx{pid_ref={Pid, '_'}})].
+
+find_by_pidref(PidRef) ->
+ %%[R || R <- ets:match_object(?MODULE, #rctx{pid_ref=PidRef, _ = '_'})].
+ [R || R <- ets:match_object(?MODULE, #rctx{pid_ref=PidRef})].
+
+find_workers_by_pidref(PidRef) ->
+ %%[R || #rctx{} = R <- ets:match_object(?MODULE, #rctx{type=#rpc_worker{from=PidRef}, _ = '_'})].
+ [R || R <- ets:match_object(?MODULE, #rctx{type=#rpc_worker{from=PidRef}})].
+
+field(#rctx{pid_ref=Val}, pid_ref) -> Val;
+%% NOTE: Pros and cons to doing these convert functions here
+%% Ideally, this would be done later so as to prefer the core data structures
+%% as long as possible, but we currently need the output of this function to
+%% be jiffy:encode'able. The tricky bit is dynamically encoding the group_by
+%% structure provided by the caller of *_by aggregator functions below.
+%% For now, we just always return jiffy:encode'able data types.
+field(#rctx{nonce=Val}, nonce) -> Val;
+%%field(#rctx{from=Val}, from) -> Val;
+%% TODO: fix this, perhaps move it all to csrt_util?
+field(#rctx{type=Val}, type) -> csrt_util:convert_type(Val);
+field(#rctx{dbname=Val}, dbname) -> Val;
+field(#rctx{username=Val}, username) -> Val;
+%%field(#rctx{path=Val}, path) -> Val;
+field(#rctx{db_open=Val}, db_open) -> Val;
+field(#rctx{docs_read=Val}, docs_read) -> Val;
+field(#rctx{rows_read=Val}, rows_read) -> Val;
+field(#rctx{changes_processed=Val}, changes_processed) -> Val;
+field(#rctx{changes_returned=Val}, changes_returned) -> Val;
+field(#rctx{ioq_calls=Val}, ioq_calls) -> Val;
+field(#rctx{io_bytes_read=Val}, io_bytes_read) -> Val;
+field(#rctx{io_bytes_written=Val}, io_bytes_written) -> Val;
+field(#rctx{js_evals=Val}, js_evals) -> Val;
+field(#rctx{js_filter=Val}, js_filter) -> Val;
+field(#rctx{js_filtered_docs=Val}, js_filtered_docs) -> Val;
+field(#rctx{mango_eval_match=Val}, mango_eval_match) -> Val;
+field(#rctx{get_kv_node=Val}, get_kv_node) -> Val;
+field(#rctx{get_kp_node=Val}, get_kp_node) -> Val.
+
+curry_field(Field) ->
+ fun(Ele) -> field(Ele, Field) end.
+
+count_by(KeyFun) ->
+ group_by(KeyFun, fun(_) -> 1 end).
+
+group_by(KeyFun, ValFun) ->
+ group_by(KeyFun, ValFun, fun erlang:'+'/2).
+
+%% eg: group_by(mfa, docs_read).
+%% eg: group_by(fun(#rctx{mfa=MFA,docs_read=DR}) -> {MFA, DR} end, ioq_calls).
+%% eg: ^^ or: group_by([mfa, docs_read], ioq_calls).
+%% eg: group_by([username, dbname, mfa], docs_read).
+%% eg: group_by([username, dbname, mfa], ioq_calls).
+%% eg: group_by([username, dbname, mfa], js_filters).
+group_by(KeyL, ValFun, AggFun) when is_list(KeyL) ->
+ KeyFun = fun(Ele) -> list_to_tuple([field(Ele, Key) || Key <- KeyL]) end,
+ group_by(KeyFun, ValFun, AggFun);
+group_by(Key, ValFun, AggFun) when is_atom(Key) ->
+ group_by(curry_field(Key), ValFun, AggFun);
+group_by(KeyFun, Val, AggFun) when is_atom(Val) ->
+ group_by(KeyFun, curry_field(Val), AggFun);
+group_by(KeyFun, ValFun, AggFun) ->
+ FoldFun = fun(Ele, Acc) ->
+ Key = KeyFun(Ele),
+ Val = ValFun(Ele),
+ CurrVal = maps:get(Key, Acc, 0),
+ NewVal = AggFun(CurrVal, Val),
+ %% TODO: should we skip here? how to make this optional?
+ case NewVal > 0 of
+ true ->
+ maps:put(Key, NewVal, Acc);
+ false ->
+ Acc
+ end
+ end,
+ ets:foldl(FoldFun, #{}, ?MODULE).
+
+%% Sorts largest first
+sorted(Map) when is_map(Map) ->
+ lists:sort(fun({_K1, A}, {_K2, B}) -> B < A end, maps:to_list(Map)).
+
+shortened(L) ->
+ lists:sublist(L, 10).
+
+%% eg: sorted_by([username, dbname, mfa], ioq_calls)
+%% eg: sorted_by([dbname, mfa], doc_reads)
+sorted_by(KeyFun) -> shortened(sorted(count_by(KeyFun))).
+sorted_by(KeyFun, ValFun) -> shortened(sorted(group_by(KeyFun, ValFun))).
+sorted_by(KeyFun, ValFun, AggFun) -> shortened(sorted(group_by(KeyFun, ValFun, AggFun))).
+
+to_json_list(List) when is_list(List) ->
+ lists:map(fun csrt_util:to_json/1, List).
+
diff --git a/src/couch_stats/src/csrt_server.erl b/src/couch_stats/src/csrt_server.erl
new file mode 100644
index 000000000..a3135b97e
--- /dev/null
+++ b/src/couch_stats/src/csrt_server.erl
@@ -0,0 +1,198 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(csrt_server).
+
+-behaviour(gen_server).
+
+-export([
+ start_link/0,
+ init/1,
+ handle_call/3,
+ handle_cast/2
+]).
+
+-export([
+ create_pid_ref/0,
+ create_resource/1,
+ destroy_resource/1,
+ get_resource/1,
+ get_context_type/1,
+ inc/2,
+ inc/3,
+ new_context/2,
+ set_context_dbname/2,
+ set_context_username/2,
+ set_context_type/2,
+ update_counter/3
+]).
+
+-include_lib("stdlib/include/ms_transform.hrl").
+-include_lib("couch_stats_resource_tracker.hrl").
+
+
+-record(st, {}).
+
+%%
+%% Public API
+%%
+
+-spec create_pid_ref() -> pid_ref().
+create_pid_ref() ->
+ {self(), make_ref()}.
+
+%%
+%%
+%% Context lifecycle API
+%%
+
+-spec new_context(Type :: rctx_type(), Nonce :: nonce()) -> rctx().
+new_context(Type, Nonce) ->
+ #rctx{
+ nonce = Nonce,
+ pid_ref = create_pid_ref(),
+ type = Type
+ }.
+
+-spec set_context_dbname(DbName, PidRef) -> boolean() when
+ DbName :: dbname(), PidRef :: maybe_pid_ref().
+set_context_dbname(_, undefined) ->
+ false;
+set_context_dbname(DbName, PidRef) ->
+ update_element(PidRef, [{#rctx.dbname, DbName}]).
+
+%%set_context_handler_fun(_, undefined) ->
+%% ok;
+%%set_context_handler_fun(Fun, PidRef) when is_function(Fun) ->
+%% FProps = erlang:fun_info(Fun),
+%% Mod = proplists:get_value(module, FProps),
+%% Func = proplists:get_value(name, FProps),
+%% #rctx{type=#coordinator{}=Coordinator} = get_resource(PidRef),
+%% Update = [{#rctx.type, Coordinator#coordinator{mod=Mod, func=Func}}],
+%% update_element(PidRef, Update).
+
+-spec set_context_username(UserName, PidRef) -> boolean() when
+ UserName :: username(), PidRef :: maybe_pid_ref().
+set_context_username(_, undefined) ->
+ ok;
+set_context_username(UserName, PidRef) ->
+ update_element(PidRef, [{#rctx.username, UserName}]).
+
+-spec get_context_type(Rctx :: rctx()) -> rctx_type().
+get_context_type(#rctx{type=Type}) ->
+ Type.
+
+-spec set_context_type(Type, PidRef) -> boolean() when
+ Type :: rctx_type(), PidRef :: maybe_pid_ref().
+set_context_type(Type, PidRef) ->
+ update_element(PidRef, [{#rctx.type, Type}]).
+
+-spec create_resource(Rctx :: rctx()) -> true.
+create_resource(#rctx{} = Rctx) ->
+ catch ets:insert(?MODULE, Rctx).
+
+-spec destroy_resource(PidRef :: maybe_pid_ref()) -> boolean().
+destroy_resource(undefined) ->
+ false;
+destroy_resource({_,_}=PidRef) ->
+ catch ets:delete(?MODULE, PidRef).
+
+-spec get_resource(PidRef :: maybe_pid_ref()) -> maybe_rctx().
+get_resource(undefined) ->
+ undefined;
+get_resource(PidRef) ->
+ catch case ets:lookup(?MODULE, PidRef) of
+ [#rctx{}=Rctx] ->
+ Rctx;
+ [] ->
+ undefined
+ end.
+
+-spec is_rctx_field(Field :: rctx_field() | atom()) -> boolean().
+is_rctx_field(Field) ->
+ maps:is_key(Field, ?KEYS_TO_FIELDS).
+
+-spec get_rctx_field(Field :: rctx_field()) -> non_neg_integer().
+get_rctx_field(Field) ->
+ maps:get(Field, ?KEYS_TO_FIELDS).
+
+-spec update_counter(PidRef, Field, Count) -> non_neg_integer() when
+ PidRef :: maybe_pid_ref(),
+ Field :: rctx_field(),
+ Count :: non_neg_integer().
+update_counter(undefined, _Field, _Count) ->
+ 0;
+update_counter({_Pid,_Ref}=PidRef, Field, Count) when Count >= 0 ->
+ %% TODO: mem3 crashes without catch, why do we lose the stats table?
+ case is_rctx_field(Field) of
+ true ->
+ Update = {get_rctx_field(Field), Count},
+ catch ets:update_counter(?MODULE, PidRef, Update, #rctx{pid_ref=PidRef});
+ false ->
+ 0
+ end.
+
+-spec inc(PidRef :: maybe_pid_ref(), Field :: rctx_field()) -> non_neg_integer().
+inc(PidRef, Field) ->
+ inc(PidRef, Field, 1).
+
+-spec inc(PidRef, Field, N) -> non_neg_integer() when
+ PidRef :: maybe_pid_ref(),
+ Field :: rctx_field(),
+ N :: non_neg_integer().
+inc(undefined, _Field, _) ->
+ 0;
+inc(_PidRef, _Field, 0) ->
+ 0;
+inc({_Pid,_Ref}=PidRef, Field, N) when is_integer(N) andalso N >= 0 ->
+ case is_rctx_field(Field) of
+ true ->
+ update_counter(PidRef, Field, N);
+ false ->
+ 0
+ end.
+
+%%
+%% gen_server callbacks
+%%
+
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+init([]) ->
+ ets:new(?MODULE, [
+ named_table,
+ public,
+ {decentralized_counters, true},
+ {write_concurrency, true},
+ {read_concurrency, true},
+ {keypos, #rctx.pid_ref}
+ ]),
+ {ok, #st{}}.
+
+handle_call(_, _From, State) ->
+ {reply, ok, State}.
+
+handle_cast(_Msg, State) ->
+ {noreply, State, 0}.
+
+%%
+%% private functions
+%%
+
+-spec update_element(PidRef :: maybe_pid_ref(), Updates :: [tuple()]) -> boolean().
+update_element(undefined, _Update) ->
+ false;
+update_element({_Pid,_Ref}=PidRef, Update) ->
+ %% TODO: should we take any action when the update fails?
+ catch ets:update_element(?MODULE, PidRef, Update).
+
diff --git a/src/couch_stats/src/csrt_util.erl b/src/couch_stats/src/csrt_util.erl
new file mode 100644
index 000000000..c386cf08e
--- /dev/null
+++ b/src/couch_stats/src/csrt_util.erl
@@ -0,0 +1,470 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(csrt_util).
+
+-export([
+ is_enabled/0,
+ is_enabled_init_p/0,
+ conf_get/1,
+ conf_get/2,
+ get_pid_ref/0,
+ get_pid_ref/1,
+ set_pid_ref/1,
+ should_track_init_p/2,
+ tnow/0,
+ tutc/0,
+ tutc/1
+]).
+
+%% JSON Conversion API
+-export([
+ convert_type/1,
+ convert_pidref/1,
+ convert_pid/1,
+ convert_ref/1,
+ to_json/1
+]).
+
+%% Delta API
+-export([
+ add_delta/2,
+ extract_delta/1,
+ get_delta/1,
+ get_delta_a/0,
+ get_delta_zero/0,
+ maybe_add_delta/1,
+ maybe_add_delta/2,
+ make_delta/1,
+ make_dt/2,
+ make_dt/3,
+ rctx_delta/2,
+ set_delta_a/1,
+ set_delta_zero/1
+]).
+
+%% Extra niceties and testing facilities
+-export([
+ set_fabric_init_p/2,
+ set_fabric_init_p/3,
+ map_to_rctx/1,
+ field/2
+]).
+
+
+-include_lib("couch_stats_resource_tracker.hrl").
+
+-spec is_enabled() -> boolean().
+is_enabled() ->
+ config:get_boolean(?CSRT, "enabled", true).
+
+-spec is_enabled_init_p() -> boolean().
+is_enabled_init_p() ->
+ config:get_boolean(?CSRT_INIT_P, "enabled", true).
+
+-spec should_track_init_p(Mod :: atom(), Func :: atom()) -> boolean().
+should_track_init_p(fabric_rpc, Func) ->
+ config:get_boolean(?CSRT_INIT_P, fabric_conf_key(Func), false);
+should_track_init_p(_Mod, _Func) ->
+ false.
+
+-spec conf_get(Key :: list()) -> list().
+conf_get(Key) ->
+ conf_get(Key, undefined).
+
+-spec conf_get(Key :: list(), Default :: list()) -> list().
+conf_get(Key, Default) ->
+ config:get(?CSRT, Key, Default).
+
+%% Monotnonic time now in native format using time forward only event tracking
+-spec tnow() -> integer().
+tnow() ->
+ erlang:monotonic_time().
+
+%% Get current system time in UTC RFC 3339 format
+-spec tutc() -> calendar:rfc3339_string().
+tutc() ->
+ tutc(tnow()).
+
+%% Convert a integer system time in milliseconds into UTC RFC 3339 format
+-spec tutc(Time :: integer()) -> calendar:rfc3339_string().
+tutc(Time0) when is_integer(Time0) ->
+ Unit = millisecond,
+ Time1 = Time0 + erlang:time_offset(),
+ Time = erlang:convert_time_unit(Time1, native, Unit),
+ calendar:system_time_to_rfc3339(Time, [{unit, Unit}, {offset, "z"}]).
+
+%% Returns dt (delta time) in microseconds
+%% @equiv make_dt(A, B, microsecond)
+-spec make_dt(A, B) -> pos_integer() when
+ A :: integer(),
+ B :: integer().
+make_dt(A, B) ->
+ make_dt(A, B, microsecond).
+
+%% Returns monotonic dt (delta time) in specified time_unit()
+-spec make_dt(A, B, Unit) -> pos_integer() when
+ A :: integer(),
+ B :: integer(),
+ Unit :: erlang:time_unit().
+make_dt(A, A, _Unit) when is_integer(A) ->
+ %% Handle edge case when monotonic_time()'s are equal
+ %% Always return a non zero value so we don't divide by zero
+ %% This always returns 1, independent of unit, as that's the smallest
+ %% possible positive integer value delta.
+ 1;
+make_dt(A, B, Unit) when is_integer(A) andalso is_integer(B) andalso B > A ->
+ A1 = erlang:convert_time_unit(A, native, Unit),
+ B1 = erlang:convert_time_unit(B, native, Unit),
+ B1 - A1.
+
+%%
+%% Conversion API for outputting JSON
+%%
+
+-spec convert_type(T) -> binary() | null when
+ T :: #coordinator{} | #rpc_worker{} | undefined.
+convert_type(#coordinator{method=Verb0, path=Path, mod=M0, func=F0}) ->
+ M = atom_to_binary(M0),
+ F = atom_to_binary(F0),
+ Verb = atom_to_binary(Verb0),
+ <<"coordinator-{", M/binary, ":", F/binary, "}:", Verb/binary, ":", Path/binary>>;
+convert_type(#rpc_worker{mod=M0, func=F0, from=From0}) ->
+ M = atom_to_binary(M0),
+ F = atom_to_binary(F0),
+ From = convert_pidref(From0),
+ <<"rpc_worker-{", From/binary, "}:", M/binary, ":", F/binary>>;
+convert_type(undefined) ->
+ null.
+
+-spec convert_pidref(PidRef) -> binary() | null when
+ PidRef :: {A :: pid(), B :: reference()} | undefined.
+convert_pidref({Parent0, ParentRef0}) ->
+ Parent = convert_pid(Parent0),
+ ParentRef = convert_ref(ParentRef0),
+ <<Parent/binary, ":", ParentRef/binary>>;
+%%convert_pidref(null) ->
+%% null;
+convert_pidref(undefined) ->
+ null.
+
+-spec convert_pid(Pid :: pid()) -> binary().
+convert_pid(Pid) when is_pid(Pid) ->
+ list_to_binary(pid_to_list(Pid)).
+
+-spec convert_ref(Ref :: reference()) -> binary().
+convert_ref(Ref) when is_reference(Ref) ->
+ list_to_binary(ref_to_list(Ref)).
+
+-spec to_json(Rctx :: rctx()) -> map().
+to_json(#rctx{}=Rctx) ->
+ #{
+ updated_at => tutc(Rctx#rctx.updated_at),
+ started_at => tutc(Rctx#rctx.started_at),
+ pid_ref => convert_pidref(Rctx#rctx.pid_ref),
+ nonce => Rctx#rctx.nonce,
+ dbname => Rctx#rctx.dbname,
+ username => Rctx#rctx.username,
+ db_open => Rctx#rctx.db_open,
+ docs_read => Rctx#rctx.docs_read,
+ docs_written => Rctx#rctx.docs_written,
+ js_filter => Rctx#rctx.js_filter,
+ js_filtered_docs => Rctx#rctx.js_filtered_docs,
+ rows_read => Rctx#rctx.rows_read,
+ type => convert_type(Rctx#rctx.type),
+ get_kp_node => Rctx#rctx.get_kp_node,
+ get_kv_node => Rctx#rctx.get_kv_node,
+ 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};
+add_delta({A, B, C, D}, Delta) -> {A, B, C, D, Delta};
+add_delta({A, B, C, D, E}, Delta) -> {A, B, C, D, E, Delta};
+add_delta({A, B, C, D, E, F}, Delta) -> {A, B, C, D, E, F, Delta};
+add_delta({A, B, C, D, E, F, G}, Delta) -> {A, B, C, D, E, F, G, Delta};
+add_delta(T, _Delta) -> T.
+
+extract_delta({A, {delta, Delta}}) -> {{A}, Delta};
+extract_delta({A, B, {delta, Delta}}) -> {{A, B}, Delta};
+extract_delta({A, B, C, {delta, Delta}}) -> {{A, B, C}, Delta};
+extract_delta({A, B, C, D, {delta, Delta}}) -> {{A, B, C, D}, Delta};
+extract_delta({A, B, C, D, E, {delta, Delta}}) -> {{A, B, C, D, E}, Delta};
+extract_delta({A, B, C, D, E, F, {delta, Delta}}) -> {{A, B, C, D, E, F}, Delta};
+extract_delta({A, B, C, D, E, F, G, {delta, Delta}}) -> {{A, B, C, D, E, F, G}, Delta};
+extract_delta(T) -> {T, undefined}.
+
+-spec get_delta(PidRef :: maybe_pid_ref()) -> tagged_delta().
+get_delta(PidRef) ->
+ {delta, make_delta(PidRef)}.
+
+maybe_add_delta(T) ->
+ case is_enabled() of
+ false ->
+ T;
+ true ->
+ maybe_add_delta_int(T, get_delta(get_pid_ref()))
+ end.
+
+%% Allow for externally provided Delta in error handling scenarios
+%% eg in cases like rexi_server:notify_caller/3
+maybe_add_delta(T, Delta) ->
+ case is_enabled() of
+ false ->
+ T;
+ true ->
+ maybe_add_delta_int(T, Delta)
+ end.
+
+maybe_add_delta_int(T, undefined) ->
+ T;
+maybe_add_delta_int(T, Delta) when is_map(Delta) ->
+ maybe_add_delta_int(T, {delta, Delta});
+maybe_add_delta_int(T, {delta, _} = Delta) ->
+ add_delta(T, Delta).
+
+-spec make_delta(PidRef :: maybe_pid_ref()) -> maybe_delta().
+make_delta(undefined) ->
+ undefined;
+make_delta(PidRef) ->
+ TA = get_delta_a(),
+ TB = csrt_server:get_resource(PidRef),
+ Delta = rctx_delta(TA, TB),
+ set_delta_a(TB),
+ Delta.
+
+-spec rctx_delta(TA :: Rctx, TB :: Rctx) -> map().
+rctx_delta(#rctx{}=TA, #rctx{}=TB) ->
+ Delta = #{
+ docs_read => TB#rctx.docs_read - TA#rctx.docs_read,
+ docs_written => TB#rctx.docs_written - TA#rctx.docs_written,
+ js_filter => TB#rctx.js_filter - TA#rctx.js_filter,
+ 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,
+ ioq_calls => TB#rctx.ioq_calls - TA#rctx.ioq_calls,
+ dt => make_dt(TA#rctx.updated_at, TB#rctx.updated_at)
+ },
+ %% TODO: reevaluate this decision
+ %% Only return non zero (and also positive) delta fields
+ %% NOTE: this can result in Delta's of the form #{dt => 1}
+ maps:filter(fun(_K,V) -> V > 0 end, Delta);
+rctx_delta(_, _) ->
+ undefined.
+
+-spec get_delta_a() -> maybe_rctx().
+get_delta_a() ->
+ erlang:get(?DELTA_TA).
+
+-spec get_delta_zero() -> maybe_rctx().
+get_delta_zero() ->
+ erlang:get(?DELTA_TZ).
+
+-spec set_delta_a(TA :: rctx()) -> maybe_rctx().
+set_delta_a(TA) ->
+ erlang:put(?DELTA_TA, TA).
+
+-spec set_delta_zero(TZ :: rctx()) -> maybe_rctx().
+set_delta_zero(TZ) ->
+ erlang:put(?DELTA_TZ, TZ).
+
+-spec get_pid_ref() -> maybe_pid_ref().
+get_pid_ref() ->
+ get(?PID_REF).
+
+-spec get_pid_ref(Rctx :: rctx()) -> pid_ref().
+get_pid_ref(#rctx{pid_ref=PidRef}) ->
+ PidRef;
+get_pid_ref(R) ->
+ throw({unexpected, R}).
+
+-spec set_pid_ref(PidRef :: pid_ref()) -> pid_ref().
+set_pid_ref(PidRef) ->
+ erlang:put(?PID_REF, PidRef),
+ PidRef.
+
+%% @equiv set_fabric_init_p(Func, Enabled, true).
+-spec set_fabric_init_p(Func :: atom(), Enabled :: boolean()) -> ok.
+set_fabric_init_p(Func, Enabled) ->
+ set_fabric_init_p(Func, Enabled, true).
+
+%% Expose Persist for use in test cases outside this module
+-spec set_fabric_init_p(Func, Enabled, Persist) -> ok when
+ Func :: atom(), Enabled :: boolean(), Persist :: boolean().
+set_fabric_init_p(Func, Enabled, Persist) ->
+ Key = fabric_conf_key(Func),
+ ok = config:set_boolean(?CSRT_INIT_P, Key, Enabled, Persist).
+
+-spec fabric_conf_key(Key :: atom()) -> string().
+fabric_conf_key(Key) ->
+ %% Double underscore to separate Mod and Func
+ "fabric_rpc__" ++ atom_to_list(Key).
+
+-ifdef(TEST).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+couch_stats_resource_tracker_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(t_should_track_init_p),
+ ?TDEF_FE(t_should_track_init_p_empty),
+ ?TDEF_FE(t_should_track_init_p_disabled),
+ ?TDEF_FE(t_should_not_track_init_p)
+ ]
+ }.
+
+setup() ->
+ test_util:start_couch().
+
+teardown(Ctx) ->
+ test_util:stop_couch(Ctx).
+
+t_should_track_init_p(_) ->
+ enable_init_p(),
+ [?assert(should_track_init_p(M, F), {M, F}) || [M, F] <- base_metrics()].
+
+t_should_track_init_p_empty(_) ->
+ config:set(?CSRT_INIT_P, "enabled", "true", false),
+ [?assert(should_track_init_p(M, F) =:= false, {M, F}) || [M, F] <- base_metrics()].
+
+t_should_track_init_p_disabled(_) ->
+ config:set(?CSRT_INIT_P, "enabled", "false", false),
+ [?assert(should_track_init_p(M, F) =:= false, {M, F}) || [M, F] <- base_metrics()].
+
+t_should_not_track_init_p(_) ->
+ enable_init_p(),
+ Metrics = [
+ [couch_db, name],
+ [couch_db, get_after_doc_read_fun],
+ [couch_db, open],
+ [fabric_rpc, get_purge_seq]
+ ],
+ [?assert(should_track_init_p(M, F) =:= false, {M, F}) || [M, F] <- Metrics].
+
+enable_init_p() ->
+ enable_init_p(base_metrics()).
+
+enable_init_p(Metrics) ->
+ config:set(?CSRT_INIT_P, "enabled", "true", false),
+ [set_fabric_init_p(F, true, false) || [_, F] <- Metrics].
+
+base_metrics() ->
+ [
+ [fabric_rpc, all_docs],
+ [fabric_rpc, changes],
+ [fabric_rpc, map_view],
+ [fabric_rpc, reduce_view],
+ [fabric_rpc, get_all_security],
+ [fabric_rpc, open_doc],
+ [fabric_rpc, update_docs],
+ [fabric_rpc, open_shard]
+ ].
+
+-endif.
diff --git a/src/couch_stats/test/eunit/csrt_logger_tests.erl b/src/couch_stats/test/eunit/csrt_logger_tests.erl
new file mode 100644
index 000000000..648b7601c
--- /dev/null
+++ b/src/couch_stats/test/eunit/csrt_logger_tests.erl
@@ -0,0 +1,345 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(csrt_logger_tests).
+
+-include_lib("couch/include/couch_db.hrl").
+-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_DBNAME_IO, 91).
+-define(THRESHOLD_DOCS_READ, 123).
+-define(THRESHOLD_IOQ_CALLS, 439).
+-define(THRESHOLD_ROWS_READ, 143).
+-define(THRESHOLD_CHANGES, 79).
+
+csrt_logger_works_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(t_do_report),
+ ?TDEF_FE(t_do_lifetime_report),
+ ?TDEF_FE(t_do_status_report)
+ ]
+ }.
+
+
+csrt_logger_matchers_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(t_matcher_on_dbname),
+ ?TDEF_FE(t_matcher_on_dbnames_io),
+ ?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) ->
+ #doc{
+ id = ?l2b("foo_" ++ integer_to_list(I)),
+ body={[{<<"value">>, I}]}
+ }
+ end,
+ lists:seq(1, Count)).
+
+setup() ->
+ Ctx = test_util:start_couch([fabric, couch_stats]),
+ DbName = ?tempdb(),
+ ok = fabric:create_db(DbName, [{q, 8}, {n, 1}]),
+ Docs = make_docs(100),
+ Opts = [],
+ {ok, _} = fabric:update_docs(DbName, Docs, Opts),
+ Method = 'GET',
+ Path = "/" ++ ?b2l(DbName) ++ "/_all_docs",
+ Nonce = couch_util:to_hex(crypto:strong_rand_bytes(5)),
+ Req = #httpd{method=Method, nonce=Nonce},
+ {_, _} = PidRef = csrt:create_coordinator_context(Req, Path),
+ MArgs = #mrargs{include_docs = false},
+ _Res = fabric:all_docs(DbName, [?ADMIN_CTX], fun view_cb/2, [], MArgs),
+ Rctx = load_rctx(PidRef),
+ 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),
+ ok = config:set("csrt_logger.dbnames_io", "foo", integer_to_list(?THRESHOLD_DBNAME_IO), false),
+ ok = config:set("csrt_logger.dbnames_io", "bar", integer_to_list(?THRESHOLD_DBNAME_IO), false),
+ ok = config:set("csrt_logger.dbnames_io", "foo/bar", integer_to_list(?THRESHOLD_DBNAME_IO), 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).
+
+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),
+ 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 := Rctx}) ->
+ JRctx = csrt_util:to_json(Rctx),
+ ReportName = "csrt-pid-usage-lifetime",
+ ok = meck:new(couch_log),
+ 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 := Rctx}) ->
+ JRctx = csrt_util:to_json(Rctx),
+ ReportName = "csrt-pid-usage-status",
+ ok = meck:new(couch_log),
+ 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"
+ ).
+
+t_matcher_on_dbnames_io(#{rctxs := Rctxs0}) ->
+ Threshold = ?THRESHOLD_DBNAME_IO,
+ SThreshold = integer_to_list(Threshold),
+ DbFoo = "foo",
+ DbBar = "bar",
+ MatcherFoo = matcher_for_csrt("dbname_io__" ++ DbFoo ++ "__" ++ SThreshold),
+ MatcherBar = matcher_for_csrt("dbname_io__" ++ DbBar ++ "__" ++ SThreshold),
+ MatcherFooBar = matcher_for_csrt("dbname_io__foo/bar__" ++ SThreshold),
+ %% Add an extra Rctx with dbname foo/bar to ensure correct naming matches
+ ExtraRctx = rctx_gen(#{dbname => <<"foo/bar">>, get_kp_node => Threshold + 10}),
+ %% Make sure we have at least one match
+ Rctxs = [ExtraRctx, rctx_gen(#{ioq_calls => Threshold + 10}) | Rctxs0],
+ ?assertEqual(
+ lists:sort(lists:filter(matcher_for_dbname_io(DbFoo, Threshold), Rctxs)),
+ lists:sort(lists:filter(MatcherFoo, Rctxs)),
+ "dbname_io foo matcher"
+ ),
+ ?assertEqual(
+ lists:sort(lists:filter(matcher_for_dbname_io(DbBar, Threshold), Rctxs)),
+ lists:sort(lists:filter(MatcherBar, Rctxs)),
+ "dbname_io bar matcher"
+ ),
+ ?assertEqual(
+ [ExtraRctx],
+ lists:sort(lists:filter(MatcherFooBar, Rctxs)),
+ "dbname_io foo/bar matcher"
+ ).
+
+load_rctx(PidRef) ->
+ timer:sleep(50), %% Add slight delay to accumulate RPC response deltas
+ csrt:get_resource(PidRef).
+
+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.
+
+matcher_for_dbname_io(Dbname0, Threshold) ->
+ Dbname = list_to_binary(Dbname0),
+ fun(Rctx) ->
+ DbnameA = csrt_util:field(dbname, Rctx),
+ Fields = [ioq_calls, get_kv_node, get_kp_node, docs_read, rows_read, changes_processed],
+ Vals = [{F, csrt_util:field(F, Rctx)} || F <- Fields],
+ Dbname =:= mem3:dbname(DbnameA) andalso lists:any(fun(V) -> V >= Threshold end, Vals)
+ 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.
diff --git a/src/couch_stats/test/eunit/csrt_server_tests.erl b/src/couch_stats/test/eunit/csrt_server_tests.erl
new file mode 100644
index 000000000..f3cf07a83
--- /dev/null
+++ b/src/couch_stats/test/eunit/csrt_server_tests.erl
@@ -0,0 +1,575 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(csrt_server_tests).
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+
+-define(DOCS_COUNT, 100).
+-define(DDOCS_COUNT, 1).
+-define(DB_Q, 8).
+
+-define(DEBUG_ENABLED, false).
+
+
+csrt_context_test_() ->
+ {
+ setup,
+ fun setup/0,
+ fun teardown/1,
+ with([
+ ?TDEF(t_context_setting)
+ ])
+ }.
+
+test_funs() ->
+ [
+ ?TDEF_FE(t_all_docs_include_false),
+ ?TDEF_FE(t_all_docs_include_true),
+ ?TDEF_FE(t_all_docs_limit_zero),
+ ?TDEF_FE(t_get_doc),
+ ?TDEF_FE(t_put_doc),
+ ?TDEF_FE(t_delete_doc),
+ ?TDEF_FE(t_update_docs),
+ ?TDEF_FE(t_changes),
+ ?TDEF_FE(t_changes_limit_zero),
+ ?TDEF_FE(t_changes_filtered),
+ ?TDEF_FE(t_view_query),
+ ?TDEF_FE(t_view_query_include_docs)
+ ].
+
+ddoc_test_funs() ->
+ [
+ ?TDEF_FE(t_changes_js_filtered)
+ | test_funs()
+ ].
+
+csrt_fabric_no_ddoc_test_() ->
+ {
+ "CSRT fabric tests with no DDoc present",
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ test_funs()
+ }.
+
+csrt_fabric_test_() ->
+ {
+ "CSRT fabric tests with a DDoc present",
+ foreach,
+ fun() -> setup_ddoc(<<"_design/foo">>, <<"bar">>) end,
+ fun teardown/1,
+ ddoc_test_funs()
+ }.
+
+make_docs(Count) ->
+ lists:map(
+ fun(I) ->
+ #doc{
+ id = ?l2b("foo_" ++ integer_to_list(I)),
+ body={[{<<"value">>, I}]}
+ }
+ end,
+ lists:seq(1, Count)).
+
+setup() ->
+ Ctx = test_util:start_couch([fabric, couch_stats]),
+ DbName = ?tempdb(),
+ ok = fabric:create_db(DbName, [{q, ?DB_Q}, {n, 1}]),
+ Docs = make_docs(?DOCS_COUNT),
+ Opts = [],
+ {ok, _} = fabric:update_docs(DbName, Docs, Opts),
+ {Ctx, DbName, undefined}.
+
+teardown({Ctx, DbName, _View}) ->
+ ok = fabric:delete_db(DbName, [?ADMIN_CTX]),
+ test_util:stop_couch(Ctx).
+
+setup_ddoc(DDocId, ViewName) ->
+ {Ctx, DbName, undefined} = setup(),
+ DDoc = couch_doc:from_json_obj(
+ {[
+ {<<"_id">>, DDocId},
+ {<<"language">>, <<"javascript">>},
+ {
+ <<"views">>,
+ {[{
+ ViewName,
+ {[
+ {<<"map">>, <<"function(doc) { emit(doc.value, null); }">>}
+ ]}
+ }]}
+ },
+ {
+ <<"filters">>,
+ {[{
+ <<"even">>,
+ <<"function(doc) { return (doc.value % 2 == 0); }">>
+ }]}
+ }
+ ]}
+ ),
+ {ok, _Rev} = fabric:update_doc(DbName, DDoc, [?ADMIN_CTX]),
+ {Ctx, DbName, {DDocId, ViewName}}.
+
+t_context_setting({_Ctx, _DbName, _View}) ->
+ false.
+
+t_all_docs_limit_zero({_Ctx, DbName, _View}) ->
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_all_docs"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ MArgs = #mrargs{include_docs = false, limit = 0},
+ _Res = fabric:all_docs(DbName, [?ADMIN_CTX], fun view_cb/2, [], MArgs),
+ Rctx = load_rctx(PidRef),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => 0,
+ docs_read => 0,
+ docs_written => 0,
+ ioq_calls => assert_gt(),
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_all_docs_include_false({_Ctx, DbName, View}) ->
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_all_docs"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ MArgs = #mrargs{include_docs = false},
+ _Res = fabric:all_docs(DbName, [?ADMIN_CTX], fun view_cb/2, [], MArgs),
+ Rctx = load_rctx(PidRef),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => docs_count(View),
+ docs_read => 0,
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_all_docs_include_true({_Ctx, DbName, View}) ->
+ pdebug(dbname, DbName),
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_all_docs"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ MArgs = #mrargs{include_docs = true},
+ _Res = fabric:all_docs(DbName, [?ADMIN_CTX], fun view_cb/2, [], MArgs),
+ Rctx = load_rctx(PidRef),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => docs_count(View),
+ docs_read => docs_count(View),
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_update_docs({_Ctx, DbName, View}) ->
+ pdebug(dbname, DbName),
+ Context = #{
+ method => 'POST',
+ path => "/" ++ ?b2l(DbName)
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ Docs = [#doc{id = ?l2b("bar_" ++ integer_to_list(I))} || I <- lists:seq(1, ?DOCS_COUNT)],
+ _Res = fabric:update_docs(DbName, Docs, [?ADMIN_CTX]),
+ Rctx = load_rctx(PidRef),
+ pdebug(rctx, Rctx),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => 0,
+ docs_read => 0,
+ docs_written => ?DOCS_COUNT,
+ pid_ref => PidRef
+ }),
+ ok = ddoc_dependent_local_io_assert(Rctx, View),
+ ok = assert_teardown(PidRef).
+
+t_get_doc({_Ctx, DbName, _View}) ->
+ pdebug(dbname, DbName),
+ DocId = "foo_17",
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/" ++ DocId
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ _Res = fabric:open_doc(DbName, DocId, [?ADMIN_CTX]),
+ Rctx = load_rctx(PidRef),
+ pdebug(rctx, Rctx),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => 1,
+ rows_read => 0,
+ docs_read => 1,
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx, io_sum),
+ ok = assert_teardown(PidRef).
+
+
+t_put_doc({_Ctx, DbName, View}) ->
+ pdebug(dbname, DbName),
+ DocId = "bar_put_1919",
+ Context = #{
+ method => 'PUT',
+ path => "/" ++ ?b2l(DbName) ++ "/" ++ DocId
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ Doc = #doc{id = ?l2b(DocId)},
+ _Res = fabric:update_doc(DbName, Doc, [?ADMIN_CTX]),
+ Rctx = load_rctx(PidRef),
+ pdebug(rctx, Rctx),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => 1,
+ rows_read => 0,
+ docs_read => 0,
+ docs_written => 1,
+ pid_ref => PidRef
+ }),
+ ok = ddoc_dependent_local_io_assert(Rctx, View),
+ ok = assert_teardown(PidRef).
+
+t_delete_doc({_Ctx, DbName, View}) ->
+ pdebug(dbname, DbName),
+ DocId = "foo_17",
+ {ok, Doc0} = fabric:open_doc(DbName, DocId, [?ADMIN_CTX]),
+ Doc = Doc0#doc{body = {[{<<"_deleted">>, true}]}},
+ Context = #{
+ method => 'DELETE',
+ path => "/" ++ ?b2l(DbName) ++ "/" ++ DocId
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ _Res = fabric:update_doc(DbName, Doc, [?ADMIN_CTX]),
+ Rctx = load_rctx(PidRef),
+ pdebug(rctx, Rctx),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => 1,
+ rows_read => 0,
+ docs_read => 0,
+ docs_written => 1,
+ pid_ref => PidRef
+ }),
+ ok = ddoc_dependent_local_io_assert(Rctx, View),
+ ok = assert_teardown(PidRef).
+
+t_changes({_Ctx, DbName, View}) ->
+ pdebug(dbname, DbName),
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_changes"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ _Res = fabric:changes(DbName, fun changes_cb/2, [], #changes_args{}),
+ Rctx = load_rctx(PidRef),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => assert_gte(?DB_Q),
+ changes_returned => docs_count(View),
+ docs_read => 0,
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ %% at least one rows_read and changes_returned per shard that has at least
+ %% one document in it
+ ?assert(maps:get(rows_read, Rctx) >= ?DB_Q, rows_read),
+ ?assert(maps:get(changes_returned, Rctx) >= ?DB_Q, changes_returned),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+
+t_changes_limit_zero({_Ctx, DbName, _View}) ->
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_changes"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ _Res = fabric:changes(DbName, fun changes_cb/2, [], #changes_args{limit=0}),
+ Rctx = load_rctx(PidRef),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => false,
+ changes_returned => false,
+ docs_read => 0,
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ?assert(maps:get(rows_read, Rctx) >= ?DB_Q, rows),
+ ?assert(maps:get(changes_returned, Rctx) >= ?DB_Q, rows),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+%% TODO: stub in non JS filter with selector
+t_changes_filtered({_Ctx, _DbName, _View}) ->
+ false.
+
+t_changes_js_filtered({_Ctx, DbName, {DDocId, _ViewName}=View}) ->
+ pdebug(dbname, DbName),
+ Method = 'GET',
+ Path = "/" ++ ?b2l(DbName) ++ "/_changes",
+ Context = #{
+ method => Method,
+ path => Path
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Req = {json_req, null},
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ Filter = configure_filter(DbName, DDocId, Req),
+ Args = #changes_args{filter_fun = Filter},
+ _Res = fabric:changes(DbName, fun changes_cb/2, [], Args),
+ Rctx = load_rctx(PidRef),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => assert_gte(?DB_Q),
+ rows_read => assert_gte(docs_count(View)),
+ changes_returned => round(?DOCS_COUNT / 2),
+ docs_read => assert_gte(docs_count(View)),
+ docs_written => 0,
+ pid_ref => PidRef,
+ js_filter => docs_count(View),
+ js_filtered_docs => docs_count(View)
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_view_query({_Ctx, DbName, View}) ->
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_design/foo/_view/bar"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ MArgs = #mrargs{include_docs = false},
+ _Res = fabric:all_docs(DbName, [?ADMIN_CTX], fun view_cb/2, [], MArgs),
+ Rctx = load_rctx(PidRef),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => docs_count(View),
+ docs_read => 0,
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_view_query_include_docs({_Ctx, DbName, View}) ->
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_design/foo/_view/bar"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = load_rctx(PidRef),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ MArgs = #mrargs{include_docs = true},
+ _Res = fabric:all_docs(DbName, [?ADMIN_CTX], fun view_cb/2, [], MArgs),
+ Rctx = load_rctx(PidRef),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => docs_count(View),
+ docs_read => docs_count(View),
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+assert_teardown(PidRef) ->
+ ?assertEqual(ok, csrt:destroy_context(PidRef)),
+ ?assertEqual(undefined, csrt:get_resource()),
+ %% Normally the tracker is responsible for destroying the resource
+ ?assertEqual(true, csrt_server:destroy_resource(PidRef)),
+ ?assertEqual(undefined, csrt:get_resource(PidRef)),
+ ok.
+
+view_cb({row, Row}, Acc) ->
+ {ok, [Row | Acc]};
+view_cb(_Msg, Acc) ->
+ {ok, Acc}.
+
+changes_cb({change, {Change}}, Acc) ->
+ {ok, [Change | Acc]};
+changes_cb(_Msg, Acc) ->
+ {ok, Acc}.
+
+pdebug(dbname, DbName) ->
+ case ?DEBUG_ENABLED =:= true of
+ true ->
+ ?debugFmt("DBNAME[~p]: ~p", [DbName, fabric:get_db_info(DbName)]);
+ false ->
+ ok
+ end;
+pdebug(rctx, Rctx) ->
+ ?DEBUG_ENABLED andalso ?debugFmt("GOT RCTX: ~p~n", [Rctx]).
+
+pdbg(Str, Args) ->
+ ?DEBUG_ENABLED andalso ?debugFmt(Str, Args).
+
+convert_pidref({_, _}=PidRef) ->
+ csrt_util:convert_pidref(PidRef);
+convert_pidref(PidRef) when is_binary(PidRef) ->
+ PidRef;
+convert_pidref(false) ->
+ false.
+
+rctx_assert(Rctx, Asserts0) ->
+ DefaultAsserts = #{
+ changes_returned => 0,
+ js_filter => 0,
+ js_filtered_docs => 0,
+ write_kp_node => 0,
+ write_kv_node => 0,
+ nonce => undefined,
+ db_open => 0,
+ rows_read => 0,
+ docs_read => 0,
+ docs_written => 0,
+ pid_ref => undefined
+ },
+ Asserts = maps:merge(
+ DefaultAsserts,
+ maps:update_with(pid_ref, fun convert_pidref/1, Asserts0)
+ ),
+ ok = maps:foreach(
+ fun
+ (_K, false) ->
+ ok;
+ (K, Fun) when is_function(Fun) ->
+ Fun(K, maps:get(K, Rctx));
+ (K, V) ->
+ case maps:get(K, Rctx) of
+ false ->
+ ok;
+ RV ->
+ pdbg("?assertEqual(~p, ~p, ~p)", [V, RV, K]),
+ ?assertEqual(V, RV, K)
+ end
+ end,
+ Asserts
+ ),
+ ok.
+
+%% Doc updates and others don't perform local IO, they funnel to another pid
+zero_local_io_assert(Rctx) ->
+ ?assertEqual(0, maps:get(ioq_calls, Rctx)),
+ ?assertEqual(0, maps:get(get_kp_node, Rctx)),
+ ?assertEqual(0, maps:get(get_kv_node, Rctx)),
+ ok.
+
+nonzero_local_io_assert(Rctx) ->
+ nonzero_local_io_assert(Rctx, io_separate).
+
+%% io_sum for when get_kp_node=0
+nonzero_local_io_assert(Rctx, io_sum) ->
+ ?assert(maps:get(ioq_calls, Rctx) > 0),
+ #{
+ get_kp_node := KPNodes,
+ get_kv_node := KVNodes
+ } = Rctx,
+ ?assert((KPNodes + KVNodes) > 0),
+ ok;
+nonzero_local_io_assert(Rctx, io_separate) ->
+ ?assert(maps:get(ioq_calls, Rctx) > 0),
+ ?assert(maps:get(get_kp_node, Rctx) > 0),
+ ?assert(maps:get(get_kv_node, Rctx) > 0),
+ ok.
+
+ddoc_dependent_local_io_assert(Rctx, undefined) ->
+ zero_local_io_assert(Rctx);
+ddoc_dependent_local_io_assert(Rctx, {_DDoc, _ViewName}) ->
+ nonzero_local_io_assert(Rctx, io_sum).
+
+coordinator_context(#{method := Method, path := Path}) ->
+ Nonce = couch_util:to_hex(crypto:strong_rand_bytes(5)),
+ Req = #httpd{method=Method, nonce=Nonce},
+ {_, _} = PidRef = csrt:create_coordinator_context(Req, Path),
+ {PidRef, Nonce}.
+
+fresh_rctx_assert(Rctx, PidRef, Nonce) ->
+ pdebug(rctx, Rctx),
+ FreshAsserts = #{
+ nonce => Nonce,
+ db_open => 0,
+ rows_read => 0,
+ docs_read => 0,
+ docs_written => 0,
+ pid_ref => PidRef
+ },
+ rctx_assert(Rctx, FreshAsserts).
+
+assert_gt() ->
+ assert_gt(0).
+
+assert_gt(N) ->
+ fun(K, RV) -> ?assert(RV > N, {K, RV, N}) end.
+
+assert_gte(N) ->
+ fun(K, RV) -> ?assert(RV >= N, {K, RV, N}) end.
+
+docs_count(undefined) ->
+ ?DOCS_COUNT;
+docs_count({_, _}) ->
+ ?DOCS_COUNT + ?DDOCS_COUNT.
+
+configure_filter(DbName, DDocId, Req) ->
+ configure_filter(DbName, DDocId, Req, <<"even">>).
+
+configure_filter(DbName, DDocId, Req, FName) ->
+ {ok, DDoc} = ddoc_cache:open_doc(DbName, DDocId),
+ DIR = fabric_util:doc_id_and_rev(DDoc),
+ Style = main_only,
+ {fetch, custom, Style, Req, DIR, FName}.
+
+load_rctx(PidRef) ->
+ timer:sleep(50), %% Add slight delay to accumulate RPC response deltas
+ csrt_util:to_json(csrt:get_resource(PidRef)).
diff --git a/src/fabric/priv/stats_descriptions.cfg b/src/fabric/priv/stats_descriptions.cfg
index d12aa0c84..9ab054bf0 100644
--- a/src/fabric/priv/stats_descriptions.cfg
+++ b/src/fabric/priv/stats_descriptions.cfg
@@ -26,3 +26,53 @@
{type, counter},
{desc, <<"number of write quorum errors">>}
]}.
+
+
+%% fabric_rpc worker stats
+%% TODO: decide on which naming scheme:
+%% {[fabric_rpc, get_all_security, spawned], [
+%% {[fabric_rpc, spawned, get_all_security], [
+{[fabric_rpc, get_all_security, spawned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker get_all_security spawns">>}
+]}.
+{[fabric_rpc, open_doc, spawned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker open_doc spawns">>}
+]}.
+{[fabric_rpc, all_docs, spawned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker all_docs spawns">>}
+]}.
+{[fabric_rpc, update_docs, spawned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker update_docs spawns">>}
+]}.
+{[fabric_rpc, map_view, spawned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker map_view spawns">>}
+]}.
+{[fabric_rpc, reduce_view, spawned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker reduce_view spawns">>}
+]}.
+{[fabric_rpc, open_shard, spawned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker open_shard spawns">>}
+]}.
+{[fabric_rpc, changes, spawned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker changes spawns">>}
+]}.
+{[fabric_rpc, changes, processed], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker changes row invocations">>}
+]}.
+{[fabric_rpc, changes, returned], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc worker changes rows returned">>}
+]}.
+{[fabric_rpc, view, rows_read], [
+ {type, counter},
+ {desc, <<"number of fabric_rpc view_cb row invocations">>}
+]}.
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 67f529e09..afa165600 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -284,6 +284,7 @@ get_missing_revs(DbName, IdRevsList, Options) ->
with_db(DbName, Options, {couch_db, get_missing_revs, [IdRevsList]}).
update_docs(DbName, Docs0, Options) ->
+ csrt:docs_written(length(Docs0)),
{Docs1, Type} =
case couch_util:get_value(read_repair, Options) of
NodeRevs when is_list(NodeRevs) ->
@@ -493,6 +494,11 @@ view_cb({meta, Meta}, Acc) ->
ok = rexi:stream2({meta, Meta}),
{ok, Acc};
view_cb({row, Props}, #mrargs{extra = Options} = Acc) ->
+ %% TODO: distinguish between rows and docs
+ %% TODO: wire in csrt tracking
+ %% TODO: distinguish between all_docs vs view call
+ couch_stats:increment_counter([fabric_rpc, view, rows_read]),
+ %%csrt:inc(rows_read),
% Adding another row
ViewRow = fabric_view_row:from_props(Props, Options),
ok = rexi:stream2(ViewRow),
@@ -529,6 +535,7 @@ changes_enumerator(#full_doc_info{} = FDI, Acc) ->
changes_enumerator(#doc_info{id = <<"_local/", _/binary>>, high_seq = Seq}, Acc) ->
{ok, Acc#fabric_changes_acc{seq = Seq, pending = Acc#fabric_changes_acc.pending - 1}};
changes_enumerator(DocInfo, Acc) ->
+ couch_stats:increment_counter([fabric_rpc, changes, processed]),
#fabric_changes_acc{
db = Db,
args = #changes_args{
@@ -569,6 +576,7 @@ changes_enumerator(DocInfo, Acc) ->
{ok, Acc#fabric_changes_acc{seq = Seq, pending = Pending - 1}}.
changes_row(Changes, Docs, DocInfo, Acc) ->
+ couch_stats:increment_counter([fabric_rpc, changes, returned]),
#fabric_changes_acc{db = Db, pending = Pending, epochs = Epochs} = Acc,
#doc_info{id = Id, high_seq = Seq, revs = [#rev_info{deleted = Del} | _]} = DocInfo,
{change, [
@@ -667,6 +675,14 @@ clean_stack(S) ->
).
set_io_priority(DbName, Options) ->
+ csrt:set_context_dbname(DbName),
+ %% TODO: better approach here than using proplists?
+ case proplists:get_value(user_ctx, Options) of
+ undefined ->
+ ok;
+ #user_ctx{name = UserName} ->
+ csrt:set_context_username(UserName)
+ end,
case lists:keyfind(io_priority, 1, Options) of
{io_priority, Pri} ->
erlang:put(io_priority, Pri);
diff --git a/src/fabric/src/fabric_util.erl b/src/fabric/src/fabric_util.erl
index 4acb65c73..f93644d2e 100644
--- a/src/fabric/src/fabric_util.erl
+++ b/src/fabric/src/fabric_util.erl
@@ -136,24 +136,36 @@ get_shard([#shard{node = Node, name = Name} | Rest], Opts, Timeout, Factor) ->
MFA = {fabric_rpc, open_shard, [Name, [{timeout, Timeout} | Opts]]},
Ref = rexi:cast(Node, self(), MFA, [sync]),
try
- receive
- {Ref, {ok, Db}} ->
- {ok, Db};
- {Ref, {'rexi_EXIT', {{unauthorized, _} = Error, _}}} ->
- throw(Error);
- {Ref, {'rexi_EXIT', {{forbidden, _} = Error, _}}} ->
- throw(Error);
- {Ref, Reason} ->
- couch_log:debug("Failed to open shard ~p because: ~p", [Name, Reason]),
- get_shard(Rest, Opts, Timeout, Factor)
- after Timeout ->
- couch_log:debug("Failed to open shard ~p after: ~p", [Name, Timeout]),
- get_shard(Rest, Opts, Factor * Timeout, Factor)
- end
+ await_shard_response(Ref, Name, Rest, Opts, Factor, Timeout)
after
rexi_monitor:stop(Mon)
end.
+await_shard_response(Ref, Name, Rest, Opts, Factor, Timeout) ->
+ receive
+ Msg0 ->
+ {Msg, Delta} = csrt:extract_delta(Msg0),
+ csrt:accumulate_delta(Delta),
+ case Msg of
+ {Ref, {ok, Db}} ->
+ {ok, Db};
+ {Ref, {'rexi_EXIT', {{unauthorized, _} = Error, _}}} ->
+ throw(Error);
+ {Ref, {'rexi_EXIT', {{forbidden, _} = Error, _}}} ->
+ throw(Error);
+ {Ref, Reason} ->
+ couch_log:debug("Failed to open shard ~p because: ~p", [Name, Reason]),
+ get_shard(Rest, Opts, Timeout, Factor);
+ %% {OldRef, {ok, Db}} -> <-- stale db resp that got here late, should we do something?
+ _ ->
+ %% Got a message from an old Ref that timed out, try again
+ await_shard_response(Ref, Name, Rest, Opts, Factor, Timeout)
+ end
+ after Timeout ->
+ couch_log:debug("Failed to open shard ~p after: ~p", [Name, Timeout]),
+ get_shard(Rest, Opts, Factor * Timeout, Factor)
+ end.
+
get_db_timeout(N, Factor, MinTimeout, infinity) ->
% MaxTimeout may be infinity so we just use the largest Erlang small int to
% avoid blowing up the arithmetic
diff --git a/src/fabric/test/eunit/fabric_rpc_purge_tests.erl b/src/fabric/test/eunit/fabric_rpc_purge_tests.erl
index 07e6b1d42..c7a36fbe3 100644
--- a/src/fabric/test/eunit/fabric_rpc_purge_tests.erl
+++ b/src/fabric/test/eunit/fabric_rpc_purge_tests.erl
@@ -263,6 +263,8 @@ rpc_update_doc(DbName, Doc, Opts) ->
Reply = test_util:wait(fun() ->
receive
{Ref, Reply} ->
+ Reply;
+ {Ref, Reply, {delta, _}} ->
Reply
after 0 ->
wait
diff --git a/src/fabric/test/eunit/fabric_rpc_tests.erl b/src/fabric/test/eunit/fabric_rpc_tests.erl
index 16bb66bad..f5e4e52f6 100644
--- a/src/fabric/test/eunit/fabric_rpc_tests.erl
+++ b/src/fabric/test/eunit/fabric_rpc_tests.erl
@@ -101,7 +101,16 @@ t_no_config_db_create_fails_for_shard_rpc(DbName) ->
receive
Resp0 -> Resp0
end,
- ?assertMatch({Ref, {'rexi_EXIT', {{error, missing_target}, _}}}, Resp).
+ case csrt:is_enabled() of
+ true ->
+ ?assertMatch( %% allow for {Ref, {rexi_EXIT, error}, {delta, D}}
+ {Ref, {'rexi_EXIT', {{error, missing_target}, _}}, _},
+ Resp);
+ false ->
+ ?assertMatch(
+ {Ref, {'rexi_EXIT', {{error, missing_target}, _}}},
+ Resp)
+ end.
t_db_create_with_config(DbName) ->
MDbName = mem3:dbname(DbName),
diff --git a/src/ioq/src/ioq.erl b/src/ioq/src/ioq.erl
index 8e38c2a00..e8862857f 100644
--- a/src/ioq/src/ioq.erl
+++ b/src/ioq/src/ioq.erl
@@ -60,6 +60,7 @@ call_search(Fd, Msg, Metadata) ->
call(Fd, Msg, Metadata).
call(Fd, Msg, Metadata) ->
+ csrt:ioq_called(),
case bypass(Msg, Metadata) of
true ->
gen_server:call(Fd, Msg, infinity);
diff --git a/src/mango/src/mango_cursor_view.erl b/src/mango/src/mango_cursor_view.erl
index 0928ae193..e11c69416 100644
--- a/src/mango/src/mango_cursor_view.erl
+++ b/src/mango/src/mango_cursor_view.erl
@@ -245,9 +245,11 @@ execute(#cursor{db = Db, index = Idx, execution_stats = Stats} = Cursor0, UserFu
Result =
case mango_idx:def(Idx) of
all_docs ->
+ couch_stats:increment_counter([mango_cursor, view, all_docs]),
CB = fun ?MODULE:handle_all_docs_message/2,
fabric:all_docs(Db, DbOpts, CB, Cursor, Args);
_ ->
+ couch_stats:increment_counter([mango_cursor, view, idx]),
CB = fun ?MODULE:handle_message/2,
% Normal view
DDoc = ddocid(Idx),
diff --git a/src/mango/src/mango_selector.erl b/src/mango/src/mango_selector.erl
index 42031b756..d8d2c913c 100644
--- a/src/mango/src/mango_selector.erl
+++ b/src/mango/src/mango_selector.erl
@@ -50,6 +50,7 @@ normalize(Selector) ->
% This assumes that the Selector has been normalized.
% Returns true or false.
match(Selector, D) ->
+ %% TODO: wire in csrt tracking
couch_stats:increment_counter([mango, evaluate_selector]),
match_int(Selector, D).
diff --git a/src/mem3/src/mem3_rpc.erl b/src/mem3/src/mem3_rpc.erl
index 70fc797da..0711cfb67 100644
--- a/src/mem3/src/mem3_rpc.erl
+++ b/src/mem3/src/mem3_rpc.erl
@@ -378,20 +378,34 @@ rexi_call(Node, MFA, Timeout) ->
Mon = rexi_monitor:start([rexi_utils:server_pid(Node)]),
Ref = rexi:cast(Node, self(), MFA, [sync]),
try
- receive
- {Ref, {ok, Reply}} ->
- Reply;
- {Ref, Error} ->
- erlang:error(Error);
- {rexi_DOWN, Mon, _, Reason} ->
- erlang:error({rexi_DOWN, {Node, Reason}})
- after Timeout ->
- erlang:error(timeout)
- end
+ wait_message(Node, Ref, Mon, Timeout)
after
rexi_monitor:stop(Mon)
end.
+wait_message(Node, Ref, Mon, Timeout) ->
+ receive
+ Msg ->
+ process_raw_message(Msg, Node, Ref, Mon, Timeout)
+ after Timeout ->
+ erlang:error(timeout)
+ end.
+
+process_raw_message(Msg0, Node, Ref, Mon, Timeout) ->
+ {Msg, Delta} = csrt:extract_delta(Msg0),
+ csrt:accumulate_delta(Delta),
+ case Msg of
+ {Ref, {ok, Reply}} ->
+ Reply;
+ {Ref, Error} ->
+ erlang:error(Error);
+ {rexi_DOWN, Mon, _, Reason} ->
+ erlang:error({rexi_DOWN, {Node, Reason}});
+ Other ->
+ ?LOG_UNEXPECTED_MSG(Other),
+ wait_message(Node, Ref, Mon, Timeout)
+ end.
+
get_or_create_db(DbName, Options) ->
mem3_util:get_or_create_db_int(DbName, Options).
diff --git a/src/rexi/include/rexi.hrl b/src/rexi/include/rexi.hrl
index a2d86b2ab..a962f3069 100644
--- a/src/rexi/include/rexi.hrl
+++ b/src/rexi/include/rexi.hrl
@@ -11,6 +11,7 @@
% the License.
-record(error, {
+ delta,
timestamp,
reason,
mfa,
diff --git a/src/rexi/src/rexi.erl b/src/rexi/src/rexi.erl
index 02d3a9e55..bb7c57037 100644
--- a/src/rexi/src/rexi.erl
+++ b/src/rexi/src/rexi.erl
@@ -104,7 +104,7 @@ kill_all(NodeRefs) when is_list(NodeRefs) ->
-spec reply(any()) -> any().
reply(Reply) ->
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {Ref, Reply}).
+ erlang:send(Caller, csrt:maybe_add_delta({Ref, Reply})).
%% Private function used by stream2 to initialize the stream. Message is of the
%% form {OriginalRef, {self(),reference()}, Reply}, which enables the
@@ -188,7 +188,7 @@ stream2(Msg, Limit, Timeout) ->
{ok, Count} ->
put(rexi_unacked, Count + 1),
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {Ref, self(), Msg}),
+ erlang:send(Caller, csrt:maybe_add_delta({Ref, self(), Msg})),
ok
catch
throw:timeout ->
@@ -222,7 +222,11 @@ stream_ack(Client) ->
%%
ping() ->
{Caller, _} = get(rexi_from),
- erlang:send(Caller, {rexi, '$rexi_ping'}).
+ %% It is essential ping/0 includes deltas as otherwise long running
+ %% filtered queries will be silent on usage until they finally return
+ %% a row or no results. This delay is proportional to the database size,
+ %% so instead we make sure ping/0 keeps live stats flowing.
+ erlang:send(Caller, csrt:maybe_add_delta({rexi, '$rexi_ping'})).
aggregate_server_queue_len() ->
rexi_server_mon:aggregate_queue_len(rexi_server).
diff --git a/src/rexi/src/rexi_monitor.erl b/src/rexi/src/rexi_monitor.erl
index 7fe66db71..72f0985df 100644
--- a/src/rexi/src/rexi_monitor.erl
+++ b/src/rexi/src/rexi_monitor.erl
@@ -35,6 +35,7 @@ start(Procs) ->
%% messages from our mailbox.
-spec stop(pid()) -> ok.
stop(MonitoringPid) ->
+ unlink(MonitoringPid),
MonitoringPid ! {self(), shutdown},
flush_down_messages().
diff --git a/src/rexi/src/rexi_server.erl b/src/rexi/src/rexi_server.erl
index b2df65c71..8ba1ee2e5 100644
--- a/src/rexi/src/rexi_server.erl
+++ b/src/rexi/src/rexi_server.erl
@@ -102,12 +102,12 @@ handle_info({'DOWN', Ref, process, Pid, Error}, #st{workers = Workers} = St) ->
case find_worker(Ref, Workers) of
#job{worker_pid = Pid, worker = Ref, client_pid = CPid, client = CRef} = Job ->
case Error of
- #error{reason = {_Class, Reason}, stack = Stack} ->
- notify_caller({CPid, CRef}, {Reason, Stack}),
+ #error{reason = {_Class, Reason}, stack = Stack, delta = Delta} ->
+ notify_caller({CPid, CRef}, {Reason, Stack}, Delta),
St1 = save_error(Error, St),
{noreply, remove_job(Job, St1)};
_ ->
- notify_caller({CPid, CRef}, Error),
+ notify_caller({CPid, CRef}, Error, undefined),
{noreply, remove_job(Job, St)}
end;
false ->
@@ -134,15 +134,20 @@ init_p(From, MFA) ->
string() | undefined
) -> any().
init_p(From, {M, F, A}, Nonce) ->
+ MFA = {M, F, length(A)},
put(rexi_from, From),
- put('$initial_call', {M, F, length(A)}),
+ put('$initial_call', MFA),
put(nonce, Nonce),
try
+ csrt:create_worker_context(From, MFA, Nonce),
+ couch_stats:maybe_track_rexi_init_p(MFA),
apply(M, F, A)
catch
exit:normal ->
+ csrt:destroy_context(),
ok;
Class:Reason:Stack0 ->
+ csrt:destroy_context(),
Stack = clean_stack(Stack0),
{ClientPid, _ClientRef} = From,
couch_log:error(
@@ -158,6 +163,7 @@ init_p(From, {M, F, A}, Nonce) ->
]
),
exit(#error{
+ delta = csrt:make_delta(),
timestamp = os:timestamp(),
reason = {Class, Reason},
mfa = {M, F, A},
@@ -200,8 +206,9 @@ find_worker(Ref, Tab) ->
[Worker] -> Worker
end.
-notify_caller({Caller, Ref}, Reason) ->
- rexi_utils:send(Caller, {Ref, {rexi_EXIT, Reason}}).
+notify_caller({Caller, Ref}, Reason, Delta) ->
+ Msg = csrt:maybe_add_delta({Ref, {rexi_EXIT, Reason}}, Delta),
+ rexi_utils:send(Caller, Msg).
kill_worker(FromRef, #st{clients = Clients} = St) ->
case find_worker(FromRef, Clients) of
diff --git a/src/rexi/src/rexi_utils.erl b/src/rexi/src/rexi_utils.erl
index 146d0238a..4ee2586ec 100644
--- a/src/rexi/src/rexi_utils.erl
+++ b/src/rexi/src/rexi_utils.erl
@@ -60,6 +60,16 @@ process_mailbox(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
receive
+ Msg ->
+ process_raw_message(Msg, RefList, Keypos, Fun, Acc0, TimeoutRef)
+ after PerMsgTO ->
+ {timeout, Acc0}
+ end.
+
+process_raw_message(Payload0, RefList, Keypos, Fun, Acc0, TimeoutRef) ->
+ {Payload, Delta} = csrt:extract_delta(Payload0),
+ csrt:accumulate_delta(Delta),
+ case Payload of
{timeout, TimeoutRef} ->
{timeout, Acc0};
{rexi, Ref, Msg} ->
@@ -95,6 +105,4 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
end;
{rexi_DOWN, _, _, _} = Msg ->
Fun(Msg, nil, Acc0)
- after PerMsgTO ->
- {timeout, Acc0}
end.
diff --git a/src/rexi/test/rexi_tests.erl b/src/rexi/test/rexi_tests.erl
index 18b05b545..6a388d163 100644
--- a/src/rexi/test/rexi_tests.erl
+++ b/src/rexi/test/rexi_tests.erl
@@ -75,6 +75,7 @@ t_cast(_) ->
Ref = rexi:cast(node(), {?MODULE, rpc_test_fun, [potato]}),
{Res, Dict} =
receive
+ {Ref, {R, D}, {delta, _}} -> {R, maps:from_list(D)};
{Ref, {R, D}} -> {R, maps:from_list(D)}
end,
?assertEqual(potato, Res),
@@ -99,7 +100,12 @@ t_cast_explicit_caller(_) ->
receive
{'DOWN', CallerRef, _, _, Exit} -> Exit
end,
- ?assertMatch({Ref, {potato, [_ | _]}}, Result).
+ case csrt:is_enabled() of
+ true ->
+ ?assertMatch({Ref, {potato, [_ | _]}, {delta, _}}, Result);
+ false ->
+ ?assertMatch({Ref, {potato, [_ | _]}}, Result)
+ end.
t_cast_ref(_) ->
put(nonce, yesh),
@@ -180,6 +186,7 @@ t_cast_error(_) ->
Ref = rexi:cast(node(), self(), {?MODULE, rpc_test_fun, [{error, tomato}]}, []),
Res =
receive
+ {Ref, RexiExit, {delta, _}} -> RexiExit;
{Ref, RexiExit} -> RexiExit
end,
?assertMatch({rexi_EXIT, {tomato, [{?MODULE, rpc_test_fun, 1, _} | _]}}, Res).
@@ -188,6 +195,7 @@ t_kill(_) ->
Ref = rexi:cast(node(), {?MODULE, rpc_test_fun, [{sleep, 10000}]}),
WorkerPid =
receive
+ {Ref, {sleeping, Pid}, {delta, _}} -> Pid;
{Ref, {sleeping, Pid}} -> Pid
end,
?assert(is_process_alive(WorkerPid)),
@@ -207,18 +215,23 @@ t_ping(_) ->
rexi:cast(node(), {?MODULE, rpc_test_fun, [ping]}),
Res =
receive
+ {rexi, Ping, {delta, _}} -> Ping;
{rexi, Ping} -> Ping
end,
?assertEqual('$rexi_ping', Res).
stream_init(Ref) ->
receive
+ {Ref, From, rexi_STREAM_INIT, {delta, _}} ->
+ From;
{Ref, From, rexi_STREAM_INIT} ->
From
end.
recv(Ref) when is_reference(Ref) ->
receive
+ {Ref, _, Msg, {delta, _}} -> Msg;
+ {Ref, Msg, {delta, _}} -> Msg;
{Ref, _, Msg} -> Msg;
{Ref, Msg} -> Msg
after 500 -> timeout