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/11 16:17:10 UTC
(couchdb) branch couch-stats-resource-tracker-v3 created (now 4764470b0)
This is an automated email from the ASF dual-hosted git repository.
chewbranca pushed a change to branch couch-stats-resource-tracker-v3
in repository https://gitbox.apache.org/repos/asf/couchdb.git
at 4764470b0 Add Dialyzer specs and unit tests
This branch includes the following new commits:
new dfe1baa4d CSRT cleanup and refactor
new 095ba8d4f s/couch_stats_resource_tracker:/csrt:/
new bff7ae880 WIP: Rework modules
new ff6072171 Fix up some broken bits
new d1ed8eef2 Rework time logic and csrt_util
new 4764470b0 Add Dialyzer specs and unit tests
The 6 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) 06/06: Add Dialyzer specs and unit tests
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 4764470b0b5adb96a99af5a90342451a9f5346b6
Author: Russell Branca <ch...@apache.org>
AuthorDate: Tue Mar 11 09:16:06 2025 -0700
Add Dialyzer specs and unit tests
---
src/config/src/config_listener_mon.erl | 2 +
.../src/couch_stats_resource_tracker.hrl | 83 +++-
src/couch_stats/src/csrt.erl | 112 ++++--
src/couch_stats/src/csrt_logger.erl | 4 +
src/couch_stats/src/csrt_query.erl | 13 +-
src/couch_stats/src/csrt_server.erl | 67 +++-
src/couch_stats/test/eunit/csrt_server_tests.erl | 428 +++++++++++++++++++++
7 files changed, 649 insertions(+), 60 deletions(-)
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_stats/src/couch_stats_resource_tracker.hrl b/src/couch_stats/src/couch_stats_resource_tracker.hrl
index abb2b8528..4be2f4fc2 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.hrl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.hrl
@@ -69,10 +69,13 @@
?COUCH_BT_WRITE_KV_NODE => #rctx.?COUCH_BT_WRITE_KV_NODE
}).
+-type pid_ref() :: {pid(), reference()}.
+-type maybe_pid_ref() :: pid_ref() | undefined.
+
-record(rpc_worker, {
mod :: atom() | '_',
func :: atom() | '_',
- from :: {pid(), reference()} | '_'
+ from :: pid_ref() | '_'
}).
-record(coordinator, {
@@ -82,34 +85,74 @@
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,
+ pid_ref :: maybe_pid_ref() | {'_', '_'},
nonce,
- type, %% #coordinator{}/#rpc_worker{}/#replication_worker{}/#compaction_worker
+ type :: rctx_type() | undefined | '_',
dbname,
username,
%% Stats counters
db_open = 0,
- docs_read = 0,
- docs_written = 0,
- rows_read = 0,
- changes_processed = 0,
- changes_returned = 0,
- ioq_calls = 0,
- io_bytes_read = 0,
- io_bytes_written = 0,
- js_evals = 0,
- js_filter = 0,
- js_filtered_docs = 0,
- mango_eval_match = 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,
- get_kv_node = 0,
- get_kp_node = 0,
- write_kv_node = 0,
- write_kp_node = 0
+ %% ?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().
diff --git a/src/couch_stats/src/csrt.erl b/src/couch_stats/src/csrt.erl
index cbf66315f..d6d5e10cc 100644
--- a/src/couch_stats/src/csrt.erl
+++ b/src/couch_stats/src/csrt.erl
@@ -88,24 +88,30 @@
%% PidRef operations
%%
+-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.
+-spec set_pid_ref(PidRef :: pid_ref()) -> pid_ref().
set_pid_ref(PidRef) ->
erlang:put(?PID_REF, PidRef),
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).
@@ -118,6 +124,8 @@ destroy_pid_ref(_PidRef) ->
%% 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 ->
@@ -127,6 +135,8 @@ create_worker_context(From, {M,F,_A}, Nonce) ->
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 ->
@@ -137,6 +147,7 @@ create_coordinator_context(#httpd{method=Verb, nonce=Nonce}, Path0) ->
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
@@ -148,18 +159,22 @@ create_context(Type, Nonce) ->
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) ->
- ok;
+ 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 ->
- ok;
+ false;
true ->
FProps = erlang:fun_info(Fun),
Mod = proplists:get_value(module, FProps),
@@ -167,17 +182,20 @@ set_context_handler_fun(Fun) when is_function(Fun) ->
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 ->
- ok;
+ 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) ->
- ok;
+ false;
update_handler_fun(Mod, Func, PidRef) ->
Rctx = get_resource(PidRef),
%% TODO: #coordinator{} assumption needs to adapt for other types
@@ -186,17 +204,17 @@ update_handler_fun(Mod, Func, PidRef) ->
csrt_server:set_context_type(Coordinator, PidRef),
ok.
-set_context_username(null) ->
- ok;
-set_context_username(undefined) ->
- 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, _) ->
- ok;
+ false;
set_context_username(_, undefined) ->
- ok;
+ false;
set_context_username(#httpd{user_ctx = Ctx}, PidRef) ->
set_context_username(Ctx, PidRef);
set_context_username(#user_ctx{name = Name}, PidRef) ->
@@ -204,9 +222,11 @@ set_context_username(#user_ctx{name = 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) ->
@@ -218,39 +238,43 @@ destroy_context({_, _} = PidRef) ->
%% Public API
%%
+%% @equiv csrt_util:is_enabled().
+-spec is_enabled() -> boolean().
is_enabled() ->
csrt_util:is_enabled().
+-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, PidRef).
+ csrt_logger:do_report(ReportName, get_resource(PidRef)).
%%
%% Stat collection API
%%
+-spec inc(Key :: rctx_field()) -> non_neg_integer().
inc(Key) ->
- inc(Key, 1).
+ is_enabled() andalso csrt_server:inc(get_pid_ref(), Key).
-inc(Key, N) ->
- case maps:is_key(Key, ?KEYS_TO_FIELDS) of
- true ->
- update_counter(maps:get(Key, ?KEYS_TO_FIELDS), N);
- false ->
- 0
- end.
+-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 ->
@@ -260,6 +284,7 @@ maybe_inc(Stat, Val) ->
end.
%% TODO: update stats_descriptions.cfg for relevant apps
+-spec should_track(Stat :: [atom()]) -> boolean().
should_track([fabric_rpc, all_docs, spawned]) ->
is_enabled();
should_track([fabric_rpc, changes, spawned]) ->
@@ -288,19 +313,23 @@ should_track(_Metric) ->
%%io:format("SKIPPING METRIC: ~p~n", [Metric]),
false.
+-spec ioq_called() -> non_neg_integer().
ioq_called() ->
inc(ioq_calls).
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);
+ is_enabled() andalso maps:foreach(fun inc/2, Delta),
+ ok;
accumulate_delta(undefined) ->
ok.
+-spec make_delta() -> map().
make_delta() ->
TA = case get(?DELTA_TA) of
undefined ->
@@ -357,6 +386,7 @@ make_delta() ->
set_delta_a(TB),
Delta.
+-spec make_delta(TA :: Rctx, TB :: Rctx) -> map().
make_delta(#rctx{}=TA, #rctx{}=TB) ->
Delta = #{
docs_read => TB#rctx.docs_read - TA#rctx.docs_read,
@@ -375,35 +405,50 @@ make_delta(#rctx{}=TA, #rctx{}=TB) ->
%% Only return non zero (and also positive) delta fields
maps:filter(fun(_K,V) -> V > 0 end, Delta);
make_delta(_, #rctx{}) ->
- #{error => missing_beg_rctx};
+ %%#{error => missing_beg_rctx};
+ undefined;
make_delta(#rctx{}, _) ->
- #{error => missing_fin_rctx}.
+ %%#{error => missing_fin_rctx}.
+ undefined.
+
+-spec make_delta_base() -> rctx().
+make_delta_base() ->
+ make_delta_base(get_pid_ref()).
-%% TODO: what to do when PidRef=undefined?
+%% TODO: figure this out, maybe not necessary now?
+%% ** TODO: what to do when PidRef=undefined?
+-spec make_delta_base(PidRef :: pid_ref()) -> rctx().
make_delta_base(PidRef) ->
%% TODO: extract user_ctx and db/shard from request
Now = csrt_util:tnow(),
#rctx{
pid_ref = PidRef,
%% TODO: confirm this subtraction works
+ %% TODO: drop this now that make_dt returns 1
started_at = Now - 100, %% give us 100ms rewind time for missing T0
updated_at = Now
}.
-make_delta_base() ->
- make_delta_base(get_pid_ref()).
-
+-spec set_delta_a(TA :: rctx()) -> maybe_rctx().
set_delta_a(TA) ->
erlang:put(?DELTA_TA, TA).
-update_counter(Field, Count) ->
- is_enabled() andalso csrt_server:update_counter(get_pid_ref(), Field, Count).
+
+%% 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).
@@ -411,24 +456,34 @@ 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).
@@ -462,6 +517,7 @@ sorted_by(Key, Val) ->
sorted_by(Key, Val, Agg) ->
csrt_query:sorted_by(Key, Val, Agg).
+%% TODO: encode that this can throw from map:get/2 on missing key
%%
%% Internal Operations assuming is_enabled() == true
%%
diff --git a/src/couch_stats/src/csrt_logger.erl b/src/couch_stats/src/csrt_logger.erl
index 0a431185f..565882d27 100644
--- a/src/couch_stats/src/csrt_logger.erl
+++ b/src/couch_stats/src/csrt_logger.erl
@@ -173,9 +173,13 @@ find_matches(RContexts, Matchers) when is_list(RContexts) andalso is_map(Matcher
get_matchers() ->
persistent_term:get(?MATCHERS_KEY, #{}).
+is_match(undefined) ->
+ false;
is_match(#rctx{}=Rctx) ->
is_match(Rctx, get_matchers()).
+is_match(undefined, _Matchers) ->
+ false;
is_match(#rctx{}=Rctx, Matchers) when is_map(Matchers) ->
maps:size(find_matches([Rctx], Matchers)) > 0.
diff --git a/src/couch_stats/src/csrt_query.erl b/src/couch_stats/src/csrt_query.erl
index 071f2622b..a3580c58a 100644
--- a/src/couch_stats/src/csrt_query.erl
+++ b/src/couch_stats/src/csrt_query.erl
@@ -12,6 +12,7 @@
-module(csrt_query).
+-include_lib("stdlib/include/ms_transform.hrl").
-include_lib("couch_stats_resource_tracker.hrl").
%% aggregate query api
@@ -73,16 +74,20 @@ 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)).
+ %%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 || #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 || #rctx{} = R <- ets:match_object(?MODULE, #rctx{pid_ref=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 || #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
diff --git a/src/couch_stats/src/csrt_server.erl b/src/couch_stats/src/csrt_server.erl
index 210db148e..a3135b97e 100644
--- a/src/couch_stats/src/csrt_server.erl
+++ b/src/couch_stats/src/csrt_server.erl
@@ -27,6 +27,8 @@
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,
@@ -44,6 +46,7 @@
%% Public API
%%
+-spec create_pid_ref() -> pid_ref().
create_pid_ref() ->
{self(), make_ref()}.
@@ -52,6 +55,7 @@ create_pid_ref() ->
%% Context lifecycle API
%%
+-spec new_context(Type :: rctx_type(), Nonce :: nonce()) -> rctx().
new_context(Type, Nonce) ->
#rctx{
nonce = Nonce,
@@ -59,8 +63,10 @@ new_context(Type, Nonce) ->
type = Type
}.
+-spec set_context_dbname(DbName, PidRef) -> boolean() when
+ DbName :: dbname(), PidRef :: maybe_pid_ref().
set_context_dbname(_, undefined) ->
- ok;
+ false;
set_context_dbname(DbName, PidRef) ->
update_element(PidRef, [{#rctx.dbname, DbName}]).
@@ -74,27 +80,33 @@ set_context_dbname(DbName, 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) ->
- ok;
-destroy_resource(#rctx{pid_ref=PidRef}) ->
- destroy_resource(PidRef);
+ 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) ->
@@ -105,11 +117,49 @@ get_resource(PidRef) ->
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) ->
- ok;
-update_counter({_Pid,_Ref}=PidRef, 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?
- catch ets:update_counter(?MODULE, PidRef, {Field, Count}, #rctx{pid_ref=PidRef}).
+ 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
@@ -139,8 +189,9 @@ handle_cast(_Msg, State) ->
%% private functions
%%
+-spec update_element(PidRef :: maybe_pid_ref(), Updates :: [tuple()]) -> boolean().
update_element(undefined, _Update) ->
- ok;
+ 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/test/eunit/csrt_server_tests.erl b/src/couch_stats/test/eunit/csrt_server_tests.erl
new file mode 100644
index 000000000..fc4e6198b
--- /dev/null
+++ b/src/couch_stats/test/eunit/csrt_server_tests.erl
@@ -0,0 +1,428 @@
+% 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(DB_Q, 8).
+
+-define(DEBUG_ENABLED, false).
+
+
+csrt_context_test_() ->
+ {
+ setup,
+ fun setup/0,
+ fun teardown/1,
+ with([
+ ?TDEF(t_context_setting)
+ ])
+ }.
+
+csrt_fabric_test_() ->
+ {
+ setup,
+ fun setup/0,
+ fun teardown/1,
+ with([
+ ?TDEF(t_all_docs_include_false),
+ ?TDEF(t_all_docs_include_true),
+ ?TDEF(t_all_docs_limit_zero),
+ ?TDEF(t_get_doc),
+ ?TDEF(t_put_doc),
+ ?TDEF(t_delete_doc),
+ ?TDEF(t_update_docs),
+ ?TDEF(t_changes),
+ ?TDEF(t_changes_limit_zero),
+ ?TDEF(t_changes_filtered),
+ ?TDEF(t_changes_js_filtered),
+ ?TDEF(t_view_query)
+ ])
+ }.
+
+setup() ->
+ Ctx = test_util:start_couch([fabric, couch_stats]),
+ DbName = ?tempdb(),
+ ok = fabric:create_db(DbName, [{q, ?DB_Q}, {n, 1}]),
+ Docs = [#doc{id = ?l2b("foo_" ++ integer_to_list(I))} || I <- lists:seq(1, ?DOCS_COUNT)],
+ Opts = [],
+ {ok, _} = fabric:update_docs(DbName, Docs, Opts),
+ {Ctx, DbName}.
+
+teardown({Ctx, DbName}) ->
+ ok = fabric:delete_db(DbName, [?ADMIN_CTX]),
+ test_util:stop_couch(Ctx).
+
+t_context_setting({_Ctx, _DbName}) ->
+ false.
+
+t_all_docs_limit_zero({_Ctx, DbName}) ->
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_all_docs"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = csrt_util:to_json(csrt:get_resource(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 = csrt_util:to_json(csrt:get_resource(PidRef)),
+ ?debugFmt("RCTX: ~p~n", [Rctx]),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => 0,
+ docs_read => 0,
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ?assert(maps:get(ioq_calls, Rctx) > 0),
+ ?assert(maps:get(get_kp_node, Rctx) > 0),
+ ?assert(maps:get(get_kv_node, Rctx) > 0),
+ ok = assert_teardown(PidRef).
+
+
+t_all_docs_include_false({_Ctx, DbName}) ->
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_all_docs"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = csrt_util:to_json(csrt:get_resource(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 = csrt_util:to_json(csrt:get_resource(PidRef)),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => ?DOCS_COUNT,
+ docs_read => 0,
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ?assert(maps:get(ioq_calls, Rctx) > 0),
+ ?assert(maps:get(get_kp_node, Rctx) > 0),
+ ?assert(maps:get(get_kv_node, Rctx) > 0),
+ ok = assert_teardown(PidRef).
+
+t_all_docs_include_true({_Ctx, DbName}) ->
+ pdebug(dbname, DbName),
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_all_docs"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = csrt_util:to_json(csrt:get_resource(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 = csrt_util:to_json(csrt:get_resource(PidRef)),
+ ok = rctx_assert(Rctx, #{
+ nonce => Nonce,
+ db_open => ?DB_Q,
+ rows_read => ?DOCS_COUNT,
+ docs_read => ?DOCS_COUNT,
+ docs_written => 0,
+ pid_ref => PidRef
+ }),
+ ok = nonzero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_update_docs({_Ctx, DbName}) ->
+ pdebug(dbname, DbName),
+ Context = #{
+ method => 'POST',
+ path => "/" ++ ?b2l(DbName)
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = csrt_util:to_json(csrt:get_resource(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 = csrt_util:to_json(csrt:get_resource(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 = zero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_get_doc({_Ctx, DbName}) ->
+ pdebug(dbname, DbName),
+ DocId = "foo_17",
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/" ++ DocId
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = csrt_util:to_json(csrt:get_resource(PidRef)),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ _Res = fabric:open_doc(DbName, DocId, [?ADMIN_CTX]),
+ Rctx = csrt_util:to_json(csrt:get_resource(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}) ->
+ pdebug(dbname, DbName),
+ DocId = "bar_put_1919",
+ Context = #{
+ method => 'PUT',
+ path => "/" ++ ?b2l(DbName) ++ "/" ++ DocId
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = csrt_util:to_json(csrt:get_resource(PidRef)),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ Doc = #doc{id = ?l2b(DocId)},
+ _Res = fabric:update_doc(DbName, Doc, [?ADMIN_CTX]),
+ Rctx = csrt_util:to_json(csrt:get_resource(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 = zero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_delete_doc({_Ctx, DbName}) ->
+ 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 = csrt_util:to_json(csrt:get_resource(PidRef)),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ _Res = fabric:update_doc(DbName, Doc, [?ADMIN_CTX]),
+ Rctx = csrt_util:to_json(csrt:get_resource(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 = zero_local_io_assert(Rctx),
+ ok = assert_teardown(PidRef).
+
+t_changes({_Ctx, DbName}) ->
+ pdebug(dbname, DbName),
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_changes"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = csrt_util:to_json(csrt:get_resource(PidRef)),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ _Res = fabric:changes(DbName, fun changes_cb/2, [], #changes_args{limit=0}),
+ Rctx = csrt_util:to_json(csrt:get_resource(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
+ }),
+ %% 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),
+ ?assert(maps:get(ioq_calls, Rctx) > 0),
+ ?assert(maps:get(get_kp_node, Rctx) > 0),
+ ?assert(maps:get(get_kv_node, Rctx) > 0),
+ ok = assert_teardown(PidRef).
+
+
+t_changes_limit_zero({_Ctx, DbName}) ->
+ Context = #{
+ method => 'GET',
+ path => "/" ++ ?b2l(DbName) ++ "/_changes"
+ },
+ {PidRef, Nonce} = coordinator_context(Context),
+ Rctx0 = csrt_util:to_json(csrt:get_resource(PidRef)),
+ ok = fresh_rctx_assert(Rctx0, PidRef, Nonce),
+ _Res = fabric:changes(DbName, fun changes_cb/2, [], #changes_args{limit=0}),
+ Rctx = csrt_util:to_json(csrt:get_resource(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),
+ ?assert(maps:get(ioq_calls, Rctx) > 0),
+ ?assert(maps:get(get_kp_node, Rctx) > 0),
+ ?assert(maps:get(get_kv_node, Rctx) > 0),
+ ok = assert_teardown(PidRef).
+
+t_changes_filtered({_Ctx, DbName}) ->
+ ?assert(false, DbName).
+
+t_changes_js_filtered({_Ctx, DbName}) ->
+ ?assert(false, DbName).
+
+t_view_query({_Ctx, DbName}) ->
+ ?assert(false, DbName).
+
+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, 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.
+
+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).
(couchdb) 01/06: CSRT cleanup and refactor
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit dfe1baa4d0d3a9b935230ee6bff33eddb28503d9
Author: Russell Branca <ch...@apache.org>
AuthorDate: Thu Jan 9 14:35:19 2025 -0800
CSRT cleanup and refactor
---
src/couch_stats/src/couch_stats.erl | 8 +-
.../src/couch_stats_resource_tracker.erl | 2 +
.../src/couch_stats_resource_tracker.hrl | 58 +++-
src/couch_stats/src/csrt.erl | 386 +++++++++++++++++++++
src/couch_stats/src/csrt_logger2.erl | 220 ++++++++++++
src/couch_stats/src/csrt_query.erl | 169 +++++++++
src/couch_stats/src/csrt_server.erl | 139 ++++++++
7 files changed, 974 insertions(+), 8 deletions(-)
diff --git a/src/couch_stats/src/couch_stats.erl b/src/couch_stats/src/couch_stats.erl
index 29190e6b0..a9abfce31 100644
--- a/src/couch_stats/src/couch_stats.erl
+++ b/src/couch_stats/src/couch_stats.erl
@@ -26,7 +26,6 @@
%% couch_stats_resource_tracker API
-export([
- create_context/3,
maybe_track_rexi_init_p/1
]).
@@ -115,17 +114,14 @@ now_sec() ->
-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]),
- couch_stats_resource_tracker:maybe_inc(Name, Val),
+ csrt:maybe_inc(Name, Val),
ok;
maybe_track_local_counter(_, _) ->
ok.
-create_context(From, MFA, Nonce) ->
- couch_stats_resource_tracker:create_context(From, MFA, Nonce).
-
maybe_track_rexi_init_p({M, F, _A}) ->
Metric = [M, F, spawned],
- case couch_stats_resource_tracker:should_track(Metric) of
+ case csrt:should_track(Metric) of
true -> increment_counter(Metric);
false -> ok
end.
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.erl b/src/couch_stats/src/couch_stats_resource_tracker.erl
index afa8cd2c6..70254088f 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.erl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.erl
@@ -104,6 +104,7 @@
-include_lib("couch_stats_resource_tracker.hrl").
%% Module pdict markers
+-ifdef(foo).
-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
@@ -157,6 +158,7 @@
?COUCH_BT_WRITE_KP_NODE => #rctx.?COUCH_BT_WRITE_KP_NODE,
?COUCH_BT_WRITE_KV_NODE => #rctx.?COUCH_BT_WRITE_KV_NODE
}).
+-endif.
-record(st, {}).
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.hrl b/src/couch_stats/src/couch_stats_resource_tracker.hrl
index 6081c9b3d..70266623c 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.hrl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.hrl
@@ -10,6 +10,60 @@
% License for the specific language governing permissions and limitations under
% the License.
+%% 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(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,
+ [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,
+ ?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
+}).
-record(rpc_worker, {
mod :: atom() | '_',
@@ -26,8 +80,8 @@
-record(rctx, {
%% Metadata
- started_at = ?MODULE:tnow(),
- updated_at = ?MODULE:tnow(),
+ started_at = csrt:tnow(),
+ updated_at = csrt:tnow(),
pid_ref,
nonce,
type, %% #coordinator{}/#rpc_worker{}/#replication_worker{}/#compaction_worker
diff --git a/src/couch_stats/src/csrt.erl b/src/couch_stats/src/csrt.erl
new file mode 100644
index 000000000..f7b330179
--- /dev/null
+++ b/src/couch_stats/src/csrt.erl
@@ -0,0 +1,386 @@
+% 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,
+ do_report/2,
+ maybe_report/2,
+ conf_get/1,
+ conf_get/2
+]).
+
+%% stats collection api
+-export([
+ accumulate_delta/1,
+ inc/1,
+ inc/2,
+ ioq_called/0,
+ make_delta/0,
+ maybe_inc/2,
+ should_track/1
+]).
+
+%%
+%% PidRef operations
+%%
+
+get_pid_ref() ->
+ get(?PID_REF).
+
+get_pid_ref(#rctx{pid_ref=PidRef}) ->
+ PidRef.
+
+set_pid_ref(PidRef) ->
+ erlang:put(?PID_REF, PidRef),
+ PidRef.
+
+create_pid_ref() ->
+ csrt_server:create_pid_ref().
+
+destroy_pid_ref() ->
+ destroy_pid_ref(get_pid_ref()).
+
+%%destroy_pid_ref(undefined) ->
+%% undefined;
+destroy_pid_ref(_PidRef) ->
+ erase(?PID_REF).
+
+%%
+%%
+%% Context lifecycle API
+%%
+
+%% TODO: shouldn't need this?
+%% create_resource(#rctx{} = Rctx) ->
+%% csrt_server:create_resource(Rctx).
+
+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.
+
+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.
+
+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),
+ erlang:put(?DELTA_TZ, Rctx),
+ csrt_server:create_resource(Rctx),
+ csrt_logger:track(Rctx),
+ PidRef.
+
+set_context_dbname(DbName) ->
+ set_context_dbname(DbName, get_pid_ref()).
+
+set_context_dbname(_, undefined) ->
+ ok;
+set_context_dbname(DbName, PidRef) ->
+ is_enabled() andalso csrt_server:set_context_dbname(DbName, PidRef).
+
+set_context_handler_fun(Fun) ->
+ set_context_handler_fun(Fun, get_pid_ref()).
+
+set_context_handler_fun(_, undefined) ->
+ ok;
+set_context_handler_fun(Fun, PidRef) when is_function(Fun) ->
+ case is_enabled() of
+ false ->
+ ok;
+ true ->
+ FProps = erlang:fun_info(Fun),
+ Mod = proplists:get_value(module, FProps),
+ Func = proplists:get_value(name, FProps),
+ 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)
+ end.
+
+set_context_username(null) ->
+ ok;
+set_context_username(undefined) ->
+ ok;
+set_context_username(User) ->
+ set_context_username(User, get_pid_ref()).
+
+set_context_username(null, _) ->
+ ok;
+set_context_username(_, undefined) ->
+ ok;
+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).
+
+destroy_context() ->
+ destroy_context(get_pid_ref()).
+
+destroy_context(undefined) ->
+ ok;
+destroy_context({_, _} = PidRef) ->
+ csrt_logger:stop_tracker(),
+ destroy_pid_ref(PidRef),
+ ok.
+
+%%
+%% Public API
+%%
+
+is_enabled() ->
+ config:get_boolean(?MODULE_STRING, "enabled", true).
+
+get_resource() ->
+ get_resource(get_pid_ref()).
+
+get_resource(PidRef) ->
+ csrt_server:get_resource(PidRef).
+
+%% Log a CSRT report if any filters match
+maybe_report(ReportName, PidRef) ->
+ csrt_logger:maybe_report(ReportName, PidRef).
+
+%% Direct report logic skipping should log filters
+do_report(ReportName, PidRef) ->
+ csrt_logger:do_report(ReportName, PidRef).
+
+%%
+%% Stat collection API
+%%
+
+%% monotonic time now in millisecionds
+tnow() ->
+ erlang:monotonic_time(millisecond).
+
+inc(Key) ->
+ inc(Key, 1).
+
+inc(Key, N) ->
+ case maps:is_key(Key, ?KEYS_TO_FIELDS) of
+ true ->
+ update_counter(maps:get(Key, ?KEYS_TO_FIELDS), N);
+ false ->
+ 0
+ end.
+
+
+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.
+
+%% TODO: update stats_descriptions.cfg for relevant apps
+should_track([fabric_rpc, all_docs, spawned]) ->
+ is_enabled();
+should_track([fabric_rpc, changes, spawned]) ->
+ is_enabled();
+should_track([fabric_rpc, changes, processed]) ->
+ is_enabled();
+should_track([fabric_rpc, changes, returned]) ->
+ is_enabled();
+should_track([fabric_rpc, map_view, spawned]) ->
+ is_enabled();
+should_track([fabric_rpc, reduce_view, spawned]) ->
+ is_enabled();
+should_track([fabric_rpc, get_all_security, spawned]) ->
+ is_enabled();
+should_track([fabric_rpc, open_doc, spawned]) ->
+ is_enabled();
+should_track([fabric_rpc, update_docs, spawned]) ->
+ is_enabled();
+should_track([fabric_rpc, open_shard, spawned]) ->
+ is_enabled();
+should_track([mango_cursor, view, all_docs]) ->
+ is_enabled();
+should_track([mango_cursor, view, idx]) ->
+ is_enabled();
+should_track(_Metric) ->
+ %%io:format("SKIPPING METRIC: ~p~n", [Metric]),
+ false.
+
+ioq_called() ->
+ inc(ioq_calls).
+
+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);
+accumulate_delta(undefined) ->
+ ok.
+
+make_delta() ->
+ TA = case get(?DELTA_TA) of
+ undefined ->
+ %% Need to handle this better, can't just make a new T0 at T' as
+ %% the timestamps will be identical causing a divide by zero error.
+ %%
+ %% Realistically need to ensure that all invocations of database
+ %% operations sets T0 appropriately. Perhaps it's possible to do
+ %% this is the couch_db:open chain, and then similarly, in
+ %% couch_server, and uhhhh... couch_file, and...
+ %%
+ %% I think we need some type of approach for establishing a T0 that
+ %% doesn't result in outrageous deltas. For now zero out the
+ %% microseconds field, or subtract a second on the off chance that
+ %% microseconds is zero. I'm not uptodate on the latest Erlang time
+ %% libraries and don't remember how to easily get an
+ %% `os:timestamp()` out of now() - 100ms or some such.
+ %%
+ %% I think it's unavoidable that we'll have some codepaths that do
+ %% not properly instantiate the T0 at spawn resulting in needing to
+ %% do some time of "time warp" or ignoring the timing collection
+ %% entirely. Perhaps if we hoisted out the stats collection into
+ %% the primary flow of the database and funnel that through all the
+ %% function clauses we could then utilize Dialyzer to statically
+ %% analyze and assert all code paths that invoke database
+ %% operations have properly instantinated a T0 at the appropriate
+ %% start time such that we don't have to "fudge" deltas with a
+ %% missing start point, but we're a long ways from that happening
+ %% so I feel it necessary to address the NULL start time.
+
+ %% Track how often we fail to initiate T0 correctly
+ %% Perhaps somewhat naughty we're incrementing stats from within
+ %% couch_stats itself? Might need to handle this differently
+ %% TODO: determine appropriate course of action here
+ %% io:format("~n**********MISSING STARTING DELTA************~n~n", []),
+ couch_stats:increment_counter(
+ [couchdb, csrt, delta_missing_t0]),
+ %%[couch_stats_resource_tracker, delta_missing_t0]),
+
+ case erlang:get(?DELTA_TZ) of
+ undefined ->
+ TA0 = make_delta_base(),
+ %% TODO: handline missing deltas, otherwise divide by zero
+ set_delta_a(TA0),
+ TA0;
+ TA0 ->
+ TA0
+ end;
+ #rctx{} = TA0 ->
+ TA0
+ end,
+ TB = get_resource(),
+ Delta = make_delta(TA, TB),
+ set_delta_a(TB),
+ Delta.
+
+make_delta(#rctx{}=TA, #rctx{}=TB) ->
+ Delta = #{
+ docs_read => TB#rctx.docs_read - TA#rctx.docs_read,
+ 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,
+ 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 => TB#rctx.updated_at - TA#rctx.updated_at
+ },
+ %% TODO: reevaluate this decision
+ %% Only return non zero (and also positive) delta fields
+ maps:filter(fun(_K,V) -> V > 0 end, Delta);
+make_delta(_, #rctx{}) ->
+ #{error => missing_beg_rctx};
+make_delta(#rctx{}, _) ->
+ #{error => missing_fin_rctx}.
+
+%% TODO: what to do when PidRef=undefined?
+make_delta_base(PidRef) ->
+ %% TODO: extract user_ctx and db/shard from request
+ Now = tnow(),
+ #rctx{
+ pid_ref = PidRef,
+ %% TODO: confirm this subtraction works
+ started_at = Now - 100, %% give us 100ms rewind time for missing T0
+ updated_at = Now
+ }.
+
+make_delta_base() ->
+ make_delta_base(get_pid_ref()).
+
+set_delta_a(TA) ->
+ erlang:put(?DELTA_TA, TA).
+
+update_counter(Field, Count) ->
+ is_enabled() andalso update_counter(get_pid_ref(), Field, Count).
+
+update_counter(undefined, _Field, _Count) ->
+ ok;
+update_counter({_Pid,_Ref}=PidRef, Field, Count) ->
+ %% TODO: mem3 crashes without catch, why do we lose the stats table?
+ is_enabled() andalso catch ets:update_counter(?MODULE, PidRef, {Field, Count}, #rctx{pid_ref=PidRef}).
+
+
+conf_get(Key) ->
+ conf_get(Key, undefined).
+
+
+conf_get(Key, Default) ->
+ config:get(?MODULE_STRING, Key, Default).
+
+%%
+%% Internal Operations assuming is_enabled() == true
+%%
diff --git a/src/couch_stats/src/csrt_logger2.erl b/src/couch_stats/src/csrt_logger2.erl
new file mode 100644
index 000000000..57262a00f
--- /dev/null
+++ b/src/couch_stats/src/csrt_logger2.erl
@@ -0,0 +1,220 @@
+% 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_logger2).
+
+-include_lib("stdlib/include/ms_transform.hrl").
+-include_lib("couch_stats_resource_tracker.hrl").
+
+%% Process lifetime logging api
+-export([
+ get_tracker/0,
+ is_logging_enabled/0,
+ log_process_lifetime_report/1,
+ put_tracker/1,
+ stop_tracker/0,
+ stop_tracker/1,
+ track/1,
+ tracker/1
+]).
+
+%% JSON Conversion API
+-export([
+ convert_type/1,
+ convert_pidref/1,
+ convert_pid/1,
+ convert_ref/1,
+ to_json/1
+]).
+
+%% Raw API that bypasses is_enabled checks
+-export([
+ do_lifetime_report/1,
+ do_status_report/1,
+ do_report/2,
+ maybe_report/2
+]).
+
+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.
+
+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_context(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_context(PidRef),
+ ok
+ end.
+
+log_process_lifetime_report(PidRef) ->
+ case csrt:is_enabled() andalso is_logging_enabled() of
+ true ->
+ case csrt:conf_get("logger_type", "csrt_logger") of
+ "csrt_logger" ->
+ csrt_logger:maybe_report(PidRef);
+ _ ->
+ Rctx = csrt_server:get_resource(PidRef),
+ case should_log(Rctx) of
+ true ->
+ do_lifetime_report(Rctx);
+ _ ->
+ ok
+ end
+ end;
+ false ->
+ ok
+ end.
+
+is_logging_enabled() ->
+ logging_enabled() =/= false.
+
+logging_enabled() ->
+ case csrt:conf_get("log_pid_usage_report", "coordinator") of
+ "coordinator" ->
+ coordinator;
+ "true" ->
+ true;
+ _ ->
+ false
+ end.
+
+should_log(undefined) ->
+ false;
+should_log(#rctx{}=Rctx) ->
+ should_log(Rctx, logging_enabled()).
+
+should_log(undefined, _) ->
+ false;
+should_log(#rctx{}, true) ->
+ true;
+should_log(#rctx{}, false) ->
+ false;
+should_log(#rctx{type = #coordinator{}}, coordinator) ->
+ true;
+should_log(#rctx{type = #rpc_worker{mod=fabric_rpc, func=FName}}, _) ->
+ case csrt:conf_get("log_fabric_rpc") of
+ "true" ->
+ true;
+ undefined ->
+ false;
+ Name ->
+ Name =:= atom_to_list(FName)
+ end;
+should_log(#rctx{}, _) ->
+ false.
+
+%% TODO: decide on API
+maybe_report(_ReportName, PidRef) ->
+ log_process_lifetime_report(PidRef).
+
+do_lifetime_report(Rctx) ->
+ do_report("csrt-pid-usage-lifetime", Rctx).
+
+do_status_report(Rctx) ->
+ do_report("csrt-pid-usage-status", Rctx).
+
+do_report(ReportName, #rctx{}=Rctx) ->
+ couch_log:report(ReportName, to_json(Rctx)).
+
+%%
+%% Conversion API for outputting JSON
+%%
+
+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.
+
+convert_pidref({Parent0, ParentRef0}) ->
+ Parent = convert_pid(Parent0),
+ ParentRef = convert_ref(ParentRef0),
+ <<Parent/binary, ":", ParentRef/binary>>;
+convert_pidref(null) ->
+ null;
+convert_pidref(undefined) ->
+ null.
+
+convert_pid(Pid) when is_pid(Pid) ->
+ list_to_binary(pid_to_list(Pid)).
+
+convert_ref(Ref) when is_reference(Ref) ->
+ list_to_binary(ref_to_list(Ref)).
+
+to_json(#rctx{}=Rctx) ->
+ #{
+ updated_at => Rctx#rctx.updated_at,
+ started_at => 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,
+ 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,
+ ioq_calls => Rctx#rctx.ioq_calls
+ }.
+
+%%
+%% Process lifetime logging api
+%%
+
+get_tracker() ->
+ get(?TRACKER_PID).
+
+put_tracker(Pid) when is_pid(Pid) ->
+ put(?TRACKER_PID, Pid).
+
+stop_tracker() ->
+ stop_tracker(get_tracker()).
+
+stop_tracker(undefined) ->
+ ok;
+stop_tracker(Pid) when is_pid(Pid) ->
+ Pid ! stop.
diff --git a/src/couch_stats/src/csrt_query.erl b/src/couch_stats/src/csrt_query.erl
new file mode 100644
index 000000000..1f9f608f7
--- /dev/null
+++ b/src/couch_stats/src/csrt_query.erl
@@ -0,0 +1,169 @@
+% 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("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)).
+
+find_by_pid(Pid) ->
+ [R || #rctx{} = R <- ets:match_object(?MODULE, #rctx{pid_ref={Pid, '_'}, _ = '_'})].
+
+find_by_pidref(PidRef) ->
+ [R || #rctx{} = 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}, _ = '_'})].
+
+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_logger: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_logger: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..2b13a38a8
--- /dev/null
+++ b/src/couch_stats/src/csrt_server.erl
@@ -0,0 +1,139 @@
+% 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,
+ new_context/2,
+ set_context_dbname/2,
+ set_context_username/2,
+ set_context_type/2
+]).
+
+-include_lib("stdlib/include/ms_transform.hrl").
+-include_lib("couch_stats_resource_tracker.hrl").
+
+-record(st, {}).
+
+%%
+%% Public API
+%%
+
+create_pid_ref() ->
+ {self(), make_ref()}.
+
+%%
+%%
+%% Context lifecycle API
+%%
+
+new_context(Type, Nonce) ->
+ #rctx{
+ nonce = Nonce,
+ pid_ref = create_pid_ref(),
+ type = Type
+ }.
+
+set_context_dbname(_, undefined) ->
+ ok;
+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).
+
+set_context_username(_, undefined) ->
+ ok;
+set_context_username(UserName, PidRef) ->
+ update_element(PidRef, [{#rctx.username, UserName}]).
+
+get_context_type(#rctx{type=Type}) ->
+ Type.
+
+set_context_type(Type, PidRef) ->
+ update_element(PidRef, [{#rctx.type, Type}]).
+
+create_resource(#rctx{} = Rctx) ->
+ catch ets:insert(?MODULE, Rctx).
+
+destroy_resource(undefined) ->
+ ok;
+destroy_resource(#rctx{pid_ref=PidRef}) ->
+ destroy_resource(PidRef);
+destroy_resource({_,_}=PidRef) ->
+ catch ets:delete(?MODULE, PidRef).
+
+get_resource(undefined) ->
+ undefined;
+get_resource(PidRef) ->
+ catch case ets:lookup(?MODULE, PidRef) of
+ [#rctx{}=Rctx] ->
+ Rctx;
+ [] ->
+ undefined
+ 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
+%%
+
+update_element(undefined, _Update) ->
+ ok;
+update_element({_Pid,_Ref}=PidRef, Update) ->
+ %% TODO: should we take any action when the update fails?
+ catch ets:update_element(?MODULE, PidRef, Update).
+
(couchdb) 05/06: Rework time logic and csrt_util
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit d1ed8eef2a4b052a45e6f4ea21a170fa61462b0a
Author: Russell Branca <ch...@apache.org>
AuthorDate: Wed Jan 15 16:44:05 2025 -0800
Rework time logic and csrt_util
---
src/couch_stats/src/csrt.erl | 6 ++--
src/couch_stats/src/csrt_logger.erl | 2 +-
src/couch_stats/src/csrt_util.erl | 70 +++++++++++++++++++++++++++++++------
3 files changed, 64 insertions(+), 14 deletions(-)
diff --git a/src/couch_stats/src/csrt.erl b/src/couch_stats/src/csrt.erl
index 00fb59e50..cbf66315f 100644
--- a/src/couch_stats/src/csrt.erl
+++ b/src/couch_stats/src/csrt.erl
@@ -369,7 +369,7 @@ make_delta(#rctx{}=TA, #rctx{}=TB) ->
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 => TB#rctx.updated_at - TA#rctx.updated_at
+ dt => csrt_util:make_dt(TA#rctx.updated_at, TB#rctx.updated_at)
},
%% TODO: reevaluate this decision
%% Only return non zero (and also positive) delta fields
@@ -401,11 +401,11 @@ update_counter(Field, Count) ->
conf_get(Key) ->
- conf_get(Key, undefined).
+ csrt_util:conf_get(Key).
conf_get(Key, Default) ->
- config:get(?MODULE_STRING, Key, Default).
+ csrt_util:conf_get(Key, Default).
%%
%% aggregate query api
diff --git a/src/couch_stats/src/csrt_logger.erl b/src/couch_stats/src/csrt_logger.erl
index 9641d1fd0..0a431185f 100644
--- a/src/couch_stats/src/csrt_logger.erl
+++ b/src/couch_stats/src/csrt_logger.erl
@@ -342,7 +342,7 @@ initialize_matchers() ->
#{},
DefaultMatchers
),
- couch_log:warning("Initialized ~p CSRT Logger matchers", [maps:size(Matchers)]),
+ couch_log:notice("Initialized ~p CSRT Logger matchers", [maps:size(Matchers)]),
persistent_term:put(?MATCHERS_KEY, Matchers),
ok.
diff --git a/src/couch_stats/src/csrt_util.erl b/src/couch_stats/src/csrt_util.erl
index e6a4591fc..5615dd49f 100644
--- a/src/couch_stats/src/csrt_util.erl
+++ b/src/couch_stats/src/csrt_util.erl
@@ -16,7 +16,11 @@
is_enabled/0,
conf_get/1,
conf_get/2,
- tnow/0
+ make_dt/2,
+ make_dt/3,
+ tnow/0,
+ tutc/0,
+ tutc/1
]).
%% JSON Conversion API
@@ -30,25 +34,66 @@
-include_lib("couch_stats_resource_tracker.hrl").
+-spec is_enabled() -> boolean().
is_enabled() ->
config:get_boolean(?CSRT, "enabled", true).
-
+-spec conf_get(Key :: list()) -> list().
conf_get(Key) ->
- conf_get(Key, undefined).
-
+ conf_get(Key, "undefined").
+-spec conf_get(Key :: list(), Default :: list()) -> list().
conf_get(Key, Default) ->
config:get(?MODULE_STRING, Key, Default).
-%% monotonic time now in millisecionds
+%% Monotnonic time now in native format using time forward only event tracking
+-spec tnow() -> integer().
tnow() ->
- erlang:monotonic_time(millisecond).
+ 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),
@@ -62,25 +107,30 @@ convert_type(#rpc_worker{mod=M0, func=F0, from=From0}) ->
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(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 => Rctx#rctx.updated_at,
- started_at => Rctx#rctx.started_at,
+ 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,
(couchdb) 04/06: Fix up some broken bits
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit ff6072171a6378520bfee9dd6bb38fffec26100f
Author: Russell Branca <ch...@apache.org>
AuthorDate: Fri Jan 10 16:53:54 2025 -0800
Fix up some broken bits
---
src/chttpd/src/chttpd_misc.erl | 2 +-
src/couch_stats/src/couch_stats.app.src | 3 +-
.../src/couch_stats_resource_tracker.hrl | 9 +-
src/couch_stats/src/couch_stats_sup.erl | 2 +-
src/couch_stats/src/csrt.erl | 157 ++++++++--
src/couch_stats/src/csrt_logger.erl | 315 ++++++++++++++++-----
src/couch_stats/src/csrt_query.erl | 4 +-
src/couch_stats/src/csrt_server.erl | 10 +-
src/couch_stats/src/csrt_util.erl | 102 +++++++
src/fabric/src/fabric_rpc.erl | 1 +
src/fabric/test/eunit/fabric_rpc_tests.erl | 2 +-
src/rexi/test/rexi_tests.erl | 2 +-
12 files changed, 497 insertions(+), 112 deletions(-)
diff --git a/src/chttpd/src/chttpd_misc.erl b/src/chttpd/src/chttpd_misc.erl
index 73b4fca2b..f39a2849b 100644
--- a/src/chttpd/src/chttpd_misc.erl
+++ b/src/chttpd/src/chttpd_misc.erl
@@ -239,7 +239,7 @@ handle_resource_status_req(#httpd{method = 'POST'} = Req) ->
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:to_json/1,
+ ToJson = fun csrt_util:to_json/1,
JsonKeys = fun(PL) -> [[ToJson(K), V] || {K, V} <- PL] end,
Fun = case {Action, Key, Val} of
diff --git a/src/couch_stats/src/couch_stats.app.src b/src/couch_stats/src/couch_stats.app.src
index de9f00e4e..da100e068 100644
--- a/src/couch_stats/src/couch_stats.app.src
+++ b/src/couch_stats/src/couch_stats.app.src
@@ -16,7 +16,8 @@
{registered, [
couch_stats_aggregator,
couch_stats_process_tracker,
- couch_stats_resource_tracker
+ csrt_server,
+ csrt_logger
]},
{applications, [kernel, stdlib, couch_log]},
{mod, {couch_stats_app, []}},
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.hrl b/src/couch_stats/src/couch_stats_resource_tracker.hrl
index 70266623c..abb2b8528 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.hrl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.hrl
@@ -10,6 +10,8 @@
% License for the specific language governing permissions and limitations under
% the License.
+-define(CSRT, "csrt").
+
%% CSRT pdict markers
-define(DELTA_TA, csrt_delta_ta).
-define(DELTA_TZ, csrt_delta_tz). %% T Zed instead of T0
@@ -27,6 +29,7 @@
-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
@@ -53,6 +56,7 @@
?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,
@@ -80,8 +84,8 @@
-record(rctx, {
%% Metadata
- started_at = csrt:tnow(),
- updated_at = csrt:tnow(),
+ started_at = csrt_util:tnow(),
+ updated_at = csrt_util:tnow(),
pid_ref,
nonce,
type, %% #coordinator{}/#rpc_worker{}/#replication_worker{}/#compaction_worker
@@ -91,6 +95,7 @@
%% Stats counters
db_open = 0,
docs_read = 0,
+ docs_written = 0,
rows_read = 0,
changes_processed = 0,
changes_returned = 0,
diff --git a/src/couch_stats/src/couch_stats_sup.erl b/src/couch_stats/src/couch_stats_sup.erl
index 66b760901..826dfbac4 100644
--- a/src/couch_stats/src/couch_stats_sup.erl
+++ b/src/couch_stats/src/couch_stats_sup.erl
@@ -29,8 +29,8 @@ init([]) ->
{
{one_for_one, 5, 10}, [
?CHILD(couch_stats_server, worker),
+ ?CHILD(csrt_server, worker),
?CHILD(csrt_logger, worker),
- ?CHILD(couch_stats_resource_tracker, worker),
?CHILD(couch_stats_process_tracker, worker)
]
}}.
diff --git a/src/couch_stats/src/csrt.erl b/src/couch_stats/src/csrt.erl
index f7b330179..00fb59e50 100644
--- a/src/couch_stats/src/csrt.erl
+++ b/src/couch_stats/src/csrt.erl
@@ -54,6 +54,7 @@
%% stats collection api
-export([
accumulate_delta/1,
+ docs_written/1,
inc/1,
inc/2,
ioq_called/0,
@@ -62,6 +63,27 @@
should_track/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
%%
@@ -134,12 +156,7 @@ set_context_dbname(_, undefined) ->
set_context_dbname(DbName, PidRef) ->
is_enabled() andalso csrt_server:set_context_dbname(DbName, PidRef).
-set_context_handler_fun(Fun) ->
- set_context_handler_fun(Fun, get_pid_ref()).
-
-set_context_handler_fun(_, undefined) ->
- ok;
-set_context_handler_fun(Fun, PidRef) when is_function(Fun) ->
+set_context_handler_fun(Fun) when is_function(Fun) ->
case is_enabled() of
false ->
ok;
@@ -147,13 +164,28 @@ 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 = 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)
+ update_handler_fun(Mod, Func, get_pid_ref())
+ end.
+
+set_context_handler_fun(Mod, Func)
+ when is_atom(Mod) andalso is_atom(Func) ->
+ case is_enabled() of
+ false ->
+ ok;
+ true ->
+ update_handler_fun(Mod, Func, get_pid_ref())
end.
+update_handler_fun(_, _, undefined) ->
+ ok;
+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.
+
set_context_username(null) ->
ok;
set_context_username(undefined) ->
@@ -187,7 +219,7 @@ destroy_context({_, _} = PidRef) ->
%%
is_enabled() ->
- config:get_boolean(?MODULE_STRING, "enabled", true).
+ csrt_util:is_enabled().
get_resource() ->
get_resource(get_pid_ref()).
@@ -207,10 +239,6 @@ do_report(ReportName, PidRef) ->
%% Stat collection API
%%
-%% monotonic time now in millisecionds
-tnow() ->
- erlang:monotonic_time(millisecond).
-
inc(Key) ->
inc(Key, 1).
@@ -263,6 +291,9 @@ should_track(_Metric) ->
ioq_called() ->
inc(ioq_calls).
+docs_written(N) ->
+ inc(docs_written, N).
+
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
@@ -329,6 +360,7 @@ make_delta() ->
make_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,
@@ -350,7 +382,7 @@ make_delta(#rctx{}, _) ->
%% TODO: what to do when PidRef=undefined?
make_delta_base(PidRef) ->
%% TODO: extract user_ctx and db/shard from request
- Now = tnow(),
+ Now = csrt_util:tnow(),
#rctx{
pid_ref = PidRef,
%% TODO: confirm this subtraction works
@@ -365,13 +397,7 @@ set_delta_a(TA) ->
erlang:put(?DELTA_TA, TA).
update_counter(Field, Count) ->
- is_enabled() andalso update_counter(get_pid_ref(), Field, Count).
-
-update_counter(undefined, _Field, _Count) ->
- ok;
-update_counter({_Pid,_Ref}=PidRef, Field, Count) ->
- %% TODO: mem3 crashes without catch, why do we lose the stats table?
- is_enabled() andalso catch ets:update_counter(?MODULE, PidRef, {Field, Count}, #rctx{pid_ref=PidRef}).
+ is_enabled() andalso csrt_server:update_counter(get_pid_ref(), Field, Count).
conf_get(Key) ->
@@ -381,6 +407,89 @@ conf_get(Key) ->
conf_get(Key, Default) ->
config:get(?MODULE_STRING, Key, Default).
+%%
+%% aggregate query api
+%%
+
+active() ->
+ csrt_query:active().
+
+active(Type) ->
+ csrt_query:active(Type).
+
+active_coordinators() ->
+ csrt_query:active_coordinators().
+
+active_coordinators(Type) ->
+ csrt_query:active_coordinators(Type).
+
+active_workers() ->
+ csrt_query:active_workers().
+
+active_workers(Type) ->
+ csrt_query:active_workers(Type).
+
+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).
+
%%
%% 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)
+ ]
+ }.
+
+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))))).
+
+-endif.
diff --git a/src/couch_stats/src/csrt_logger.erl b/src/couch_stats/src/csrt_logger.erl
index 57262a00f..9641d1fd0 100644
--- a/src/couch_stats/src/csrt_logger.erl
+++ b/src/couch_stats/src/csrt_logger.erl
@@ -10,10 +10,7 @@
% License for the specific language governing permissions and limitations under
% the License.
--module(csrt_logger2).
-
--include_lib("stdlib/include/ms_transform.hrl").
--include_lib("couch_stats_resource_tracker.hrl").
+-module(csrt_logger).
%% Process lifetime logging api
-export([
@@ -27,15 +24,6 @@
tracker/1
]).
-%% JSON Conversion API
--export([
- convert_type/1,
- convert_pidref/1,
- convert_pid/1,
- convert_ref/1,
- to_json/1
-]).
-
%% Raw API that bypasses is_enabled checks
-export([
do_lifetime_report/1,
@@ -44,6 +32,45 @@
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([
+ matcher_on_dbname/1,
+ matcher_on_docs_read/1,
+ matcher_on_docs_written/1,
+ matcher_on_rows_read/1,
+ matcher_on_worker_changes_processed/1,
+ matcher_on_ioq_calls/1,
+ matcher_on_nonce/1,
+ matcher_on_nonce_comp/1,
+ matcher_on_writes/1
+]).
+
+-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").
+
+-record(st, {
+ matchers = #{}
+}).
+
track(#rctx{pid_ref=PidRef}) ->
case get_tracker() of
undefined ->
@@ -60,7 +87,7 @@ tracker({Pid, _Ref}=PidRef) ->
stop ->
%% TODO: do we need cleanup here?
log_process_lifetime_report(PidRef),
- csrt_server:destroy_context(PidRef),
+ csrt_server:destroy_resource(PidRef),
ok;
{'DOWN', MonRef, _Type, _0DPid, _Reason0} ->
%% TODO: should we pass reason to log_process_lifetime_report?
@@ -73,16 +100,17 @@ tracker({Pid, _Ref}=PidRef) ->
%% end,
%% TODO: should we send the induced work delta to the coordinator?
log_process_lifetime_report(PidRef),
- csrt_server:destroy_context(PidRef),
+ csrt_server:destroy_resource(PidRef),
ok
end.
log_process_lifetime_report(PidRef) ->
- case csrt:is_enabled() andalso is_logging_enabled() of
+ case csrt_util:is_enabled() andalso is_logging_enabled() of
true ->
- case csrt:conf_get("logger_type", "csrt_logger") of
+ %%case csrt_util:conf_get("logger_type", "csrt_logger") of
+ case csrt_util:conf_get("logger_type", "csrt_logger") of
"csrt_logger" ->
- csrt_logger:maybe_report(PidRef);
+ maybe_report("csrt-pid-usage-lifetime", PidRef);
_ ->
Rctx = csrt_server:get_resource(PidRef),
case should_log(Rctx) of
@@ -100,7 +128,7 @@ is_logging_enabled() ->
logging_enabled() =/= false.
logging_enabled() ->
- case csrt:conf_get("log_pid_usage_report", "coordinator") of
+ case csrt_util:conf_get("log_pid_usage_report", "coordinator") of
"coordinator" ->
coordinator;
"true" ->
@@ -123,7 +151,7 @@ should_log(#rctx{}, false) ->
should_log(#rctx{type = #coordinator{}}, coordinator) ->
true;
should_log(#rctx{type = #rpc_worker{mod=fabric_rpc, func=FName}}, _) ->
- case csrt:conf_get("log_fabric_rpc") of
+ case csrt_util:conf_get("log_fabric_rpc") of
"true" ->
true;
undefined ->
@@ -134,9 +162,32 @@ should_log(#rctx{type = #rpc_worker{mod=fabric_rpc, func=FName}}, _) ->
should_log(#rctx{}, _) ->
false.
-%% TODO: decide on API
-maybe_report(_ReportName, PidRef) ->
- log_process_lifetime_report(PidRef).
+find_matches(RContexts, Matchers) when is_list(RContexts) andalso is_map(Matchers) ->
+ maps:filter(
+ fun(_Name, {_MSpec, CompMSpec}) ->
+ catch [] =/= ets:match_spec_run(RContexts, CompMSpec)
+ end,
+ Matchers
+ ).
+
+get_matchers() ->
+ persistent_term:get(?MATCHERS_KEY, #{}).
+
+is_match(#rctx{}=Rctx) ->
+ is_match(Rctx, get_matchers()).
+
+is_match(#rctx{}=Rctx, Matchers) when is_map(Matchers) ->
+ maps:size(find_matches([Rctx], Matchers)) > 0.
+
+maybe_report(ReportName, PidRef) ->
+ Rctx = csrt_server:get_resource(PidRef),
+ case is_match(Rctx) of
+ true ->
+ do_report(ReportName, Rctx),
+ ok;
+ false ->
+ ok
+ end.
do_lifetime_report(Rctx) ->
do_report("csrt-pid-usage-lifetime", Rctx).
@@ -145,61 +196,7 @@ do_status_report(Rctx) ->
do_report("csrt-pid-usage-status", Rctx).
do_report(ReportName, #rctx{}=Rctx) ->
- couch_log:report(ReportName, to_json(Rctx)).
-
-%%
-%% Conversion API for outputting JSON
-%%
-
-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.
-
-convert_pidref({Parent0, ParentRef0}) ->
- Parent = convert_pid(Parent0),
- ParentRef = convert_ref(ParentRef0),
- <<Parent/binary, ":", ParentRef/binary>>;
-convert_pidref(null) ->
- null;
-convert_pidref(undefined) ->
- null.
-
-convert_pid(Pid) when is_pid(Pid) ->
- list_to_binary(pid_to_list(Pid)).
-
-convert_ref(Ref) when is_reference(Ref) ->
- list_to_binary(ref_to_list(Ref)).
-
-to_json(#rctx{}=Rctx) ->
- #{
- updated_at => Rctx#rctx.updated_at,
- started_at => 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,
- 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,
- ioq_calls => Rctx#rctx.ioq_calls
- }.
+ couch_log:report(ReportName, csrt_util:to_json(Rctx)).
%%
%% Process lifetime logging api
@@ -218,3 +215,165 @@ stop_tracker(undefined) ->
ok;
stop_tracker(Pid) when is_pid(Pid) ->
Pid ! stop.
+
+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
+%%
+
+matcher_on_dbname(DbName)
+ when is_binary(DbName) ->
+ ets:fun2ms(fun(#rctx{dbname=DbName1} = R) when DbName =:= DbName1 -> R end).
+
+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).
+
+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).
+
+matcher_on_rows_read(Threshold)
+ when is_integer(Threshold) andalso Threshold > 0 ->
+ ets:fun2ms(fun(#rctx{rows_read=DocsRead} = R) when DocsRead >= Threshold -> R end).
+
+matcher_on_writes(Threshold)
+ when is_integer(Threshold) andalso Threshold > 0 ->
+ ets:fun2ms(fun(#rctx{write_kp_node = WKP, write_kv_node = WKV} = R) when WKP + WKV > Threshold-> R end).
+
+matcher_on_nonce(Nonce) ->
+ ets:fun2ms(fun(#rctx{nonce = Nonce1} = R) when Nonce =:= Nonce1 -> R end).
+
+matcher_on_nonce_comp(Nonce) ->
+ ets:match_spec_compile(ets:fun2ms(fun(#rctx{nonce = Nonce1} = R) when Nonce =:= Nonce1 -> R end)).
+
+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
+ ).
+
+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).
+
+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.
+
+set_matchers_term(Matchers) when is_map(Matchers) ->
+ persistent_term:put({?MODULE, all_csrt_matchers}, Matchers).
+
+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},
+ {btree_writes, fun matcher_on_writes/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
+ ),
+ couch_log:warning("Initialized ~p CSRT Logger matchers", [maps:size(Matchers)]),
+ persistent_term:put(?MATCHERS_KEY, Matchers),
+ ok.
+
+
+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).
+
+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
index 1f9f608f7..071f2622b 100644
--- a/src/couch_stats/src/csrt_query.erl
+++ b/src/couch_stats/src/csrt_query.erl
@@ -94,7 +94,7 @@ field(#rctx{pid_ref=Val}, pid_ref) -> Val;
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_logger:convert_type(Val);
+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;
@@ -165,5 +165,5 @@ 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_logger:to_json/1, 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
index 2b13a38a8..210db148e 100644
--- a/src/couch_stats/src/csrt_server.erl
+++ b/src/couch_stats/src/csrt_server.erl
@@ -30,12 +30,14 @@
new_context/2,
set_context_dbname/2,
set_context_username/2,
- set_context_type/2
+ set_context_type/2,
+ update_counter/3
]).
-include_lib("stdlib/include/ms_transform.hrl").
-include_lib("couch_stats_resource_tracker.hrl").
+
-record(st, {}).
%%
@@ -103,6 +105,12 @@ get_resource(PidRef) ->
undefined
end.
+update_counter(undefined, _Field, _Count) ->
+ ok;
+update_counter({_Pid,_Ref}=PidRef, Field, Count) ->
+ %% TODO: mem3 crashes without catch, why do we lose the stats table?
+ catch ets:update_counter(?MODULE, PidRef, {Field, Count}, #rctx{pid_ref=PidRef}).
+
%%
%% gen_server callbacks
%%
diff --git a/src/couch_stats/src/csrt_util.erl b/src/couch_stats/src/csrt_util.erl
new file mode 100644
index 000000000..e6a4591fc
--- /dev/null
+++ b/src/couch_stats/src/csrt_util.erl
@@ -0,0 +1,102 @@
+% 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,
+ conf_get/1,
+ conf_get/2,
+ tnow/0
+]).
+
+%% JSON Conversion API
+-export([
+ convert_type/1,
+ convert_pidref/1,
+ convert_pid/1,
+ convert_ref/1,
+ to_json/1
+]).
+
+-include_lib("couch_stats_resource_tracker.hrl").
+
+is_enabled() ->
+ config:get_boolean(?CSRT, "enabled", true).
+
+
+conf_get(Key) ->
+ conf_get(Key, undefined).
+
+
+conf_get(Key, Default) ->
+ config:get(?MODULE_STRING, Key, Default).
+
+%% monotonic time now in millisecionds
+tnow() ->
+ erlang:monotonic_time(millisecond).
+
+%%
+%% Conversion API for outputting JSON
+%%
+
+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.
+
+convert_pidref({Parent0, ParentRef0}) ->
+ Parent = convert_pid(Parent0),
+ ParentRef = convert_ref(ParentRef0),
+ <<Parent/binary, ":", ParentRef/binary>>;
+convert_pidref(null) ->
+ null;
+convert_pidref(undefined) ->
+ null.
+
+convert_pid(Pid) when is_pid(Pid) ->
+ list_to_binary(pid_to_list(Pid)).
+
+convert_ref(Ref) when is_reference(Ref) ->
+ list_to_binary(ref_to_list(Ref)).
+
+to_json(#rctx{}=Rctx) ->
+ #{
+ updated_at => Rctx#rctx.updated_at,
+ started_at => 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,
+ ioq_calls => Rctx#rctx.ioq_calls
+ }.
+
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 16bea8433..480711237 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) ->
diff --git a/src/fabric/test/eunit/fabric_rpc_tests.erl b/src/fabric/test/eunit/fabric_rpc_tests.erl
index c402affba..f5e4e52f6 100644
--- a/src/fabric/test/eunit/fabric_rpc_tests.erl
+++ b/src/fabric/test/eunit/fabric_rpc_tests.erl
@@ -101,7 +101,7 @@ t_no_config_db_create_fails_for_shard_rpc(DbName) ->
receive
Resp0 -> Resp0
end,
- case couch_stats_resource_tracker:is_enabled() of
+ case csrt:is_enabled() of
true ->
?assertMatch( %% allow for {Ref, {rexi_EXIT, error}, {delta, D}}
{Ref, {'rexi_EXIT', {{error, missing_target}, _}}, _},
diff --git a/src/rexi/test/rexi_tests.erl b/src/rexi/test/rexi_tests.erl
index 7500e57e0..6a388d163 100644
--- a/src/rexi/test/rexi_tests.erl
+++ b/src/rexi/test/rexi_tests.erl
@@ -100,7 +100,7 @@ t_cast_explicit_caller(_) ->
receive
{'DOWN', CallerRef, _, _, Exit} -> Exit
end,
- case couch_stats_resource_tracker:is_enabled() of
+ case csrt:is_enabled() of
true ->
?assertMatch({Ref, {potato, [_ | _]}, {delta, _}}, Result);
false ->
(couchdb) 02/06: s/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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 095ba8d4fa02ae3d303fd4a8cdd0f5a7f5550be3
Author: Russell Branca <ch...@apache.org>
AuthorDate: Thu Jan 9 17:53:47 2025 -0800
s/couch_stats_resource_tracker:/csrt:/
---
src/chttpd/src/chttpd.erl | 14 +++++++-------
src/chttpd/src/chttpd_db.erl | 4 ++--
src/chttpd/src/chttpd_misc.erl | 12 ++++++------
src/couch_stats/src/csrt_logger.erl | 6 +++---
src/fabric/src/fabric_rpc.erl | 6 +++---
src/fabric/src/fabric_util.erl | 2 +-
src/ioq/src/ioq.erl | 2 +-
src/mem3/src/mem3_rpc.erl | 2 +-
src/rexi/src/rexi_server.erl | 8 ++++----
src/rexi/src/rexi_utils.erl | 8 ++++----
10 files changed, 32 insertions(+), 32 deletions(-)
diff --git a/src/chttpd/src/chttpd.erl b/src/chttpd/src/chttpd.erl
index 99220a0b0..a8dec6517 100644
--- a/src/chttpd/src/chttpd.erl
+++ b/src/chttpd/src/chttpd.erl
@@ -324,8 +324,8 @@ handle_request_int(MochiReq) ->
chttpd_util:mochiweb_client_req_set(MochiReq),
%% This is probably better in before_request, but having Path is nice
- couch_stats_resource_tracker:create_coordinator_context(HttpReq0, Path),
- couch_stats_resource_tracker:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
+ csrt:create_coordinator_context(HttpReq0, Path),
+ csrt:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
{HttpReq2, Response} =
case before_request(HttpReq0) of
@@ -357,7 +357,7 @@ handle_request_int(MochiReq) ->
before_request(HttpReq) ->
try
- couch_stats_resource_tracker:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
+ csrt:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
chttpd_stats:init(),
chttpd_plugin:before_request(HttpReq)
catch
@@ -378,7 +378,7 @@ after_request(HttpReq, HttpResp0) ->
chttpd_stats:report(HttpReq, HttpResp2),
maybe_log(HttpReq, HttpResp2),
%% NOTE: do not set_context_handler_fun to preserve the Handler
- couch_stats_resource_tracker:destroy_context(),
+ csrt:destroy_context(),
HttpResp2.
process_request(#httpd{mochi_req = MochiReq} = HttpReq) ->
@@ -391,7 +391,7 @@ process_request(#httpd{mochi_req = MochiReq} = HttpReq) ->
RawUri = MochiReq:get(raw_path),
try
- couch_stats_resource_tracker:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
+ csrt:set_context_handler_fun(?MODULE, ?FUNCTION_NAME),
couch_httpd:validate_host(HttpReq),
check_request_uri_length(RawUri),
check_url_encoding(RawUri),
@@ -417,12 +417,12 @@ handle_req_after_auth(HandlerKey, HttpReq) ->
HandlerKey,
fun chttpd_db:handle_request/1
),
- couch_stats_resource_tracker:set_context_handler_fun(HandlerFun),
+ csrt:set_context_handler_fun(HandlerFun),
AuthorizedReq = chttpd_auth:authorize(
possibly_hack(HttpReq),
fun chttpd_auth_request:authorize_request/1
),
- couch_stats_resource_tracker:set_context_username(AuthorizedReq),
+ 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 8a816abde..4643693e1 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -83,7 +83,7 @@
% Database request handlers
handle_request(#httpd{path_parts = [DbName | RestParts], method = Method} = Req) ->
- couch_stats_resource_tracker:set_context_dbname(DbName),
+ csrt:set_context_dbname(DbName),
case {Method, RestParts} of
{'PUT', []} ->
create_db_req(Req, DbName);
@@ -104,7 +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),
- couch_stats_resource_tracker:set_context_handler_fun(Handler),
+ csrt:set_context_handler_fun(Handler),
do_db_req(Req, Handler)
end.
diff --git a/src/chttpd/src/chttpd_misc.erl b/src/chttpd/src/chttpd_misc.erl
index add5443f9..73b4fca2b 100644
--- a/src/chttpd/src/chttpd_misc.erl
+++ b/src/chttpd/src/chttpd_misc.erl
@@ -233,13 +233,13 @@ handle_resource_status_req(#httpd{method = 'POST'} = Req) ->
Key = proplists:get_value(<<"key">>, Props),
Val = proplists:get_value(<<"val">>, Props),
- CountBy = fun couch_stats_resource_tracker:count_by/1,
- GroupBy = fun couch_stats_resource_tracker:group_by/2,
- SortedBy1 = fun couch_stats_resource_tracker:sorted_by/1,
- SortedBy2 = fun couch_stats_resource_tracker:sorted_by/2,
+ 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 couch_stats_resource_tracker:term_to_flat_json/1,
+ ToJson = fun csrt:to_json/1,
JsonKeys = fun(PL) -> [[ToJson(K), V] || {K, V} <- PL] end,
Fun = case {Action, Key, Val} of
@@ -318,7 +318,7 @@ handle_resource_status_req(#httpd{method = 'GET'} = Req) ->
ok = chttpd:verify_is_server_admin(Req),
{Resp, Bad} = rpc:multicall(erlang, apply, [
fun() ->
- {node(), couch_stats_resource_tracker:active()}
+ {node(), csrt:active()}
end,
[]
]),
diff --git a/src/couch_stats/src/csrt_logger.erl b/src/couch_stats/src/csrt_logger.erl
index c4e064bf3..e4df3d053 100644
--- a/src/couch_stats/src/csrt_logger.erl
+++ b/src/couch_stats/src/csrt_logger.erl
@@ -141,7 +141,7 @@ log_level() ->
maybe_report(PidRef) ->
- Rctx = couch_stats_resource_tracker:get_resource(PidRef),
+ Rctx = csrt:get_resource(PidRef),
case is_match(Rctx) of
true ->
report(Rctx),
@@ -154,7 +154,7 @@ maybe_report_old(PidRef) ->
Level = log_level(),
case Level > ?NO_LOG of
true ->
- Rctx = couch_stats_resource_tracker:get_resource(PidRef),
+ Rctx = csrt:get_resource(PidRef),
RLevel = log_level(Rctx),
%% io:format("CSRT LOGGER TRIGGERING{~p} ~p =<? ~p~n", [Rctx#rctx.type, RLevel, Level]),
case RLevel >= Level of
@@ -257,7 +257,7 @@ maybe_report_int(#rctx{}=Rctx, _RLevel, _Level, #st{matchers=Matchers, should_tr
report(Rctx) ->
couch_log:report(
"csrt-pid-usage-lifetime",
- couch_stats_resource_tracker:to_json(Rctx)
+ csrt:to_json(Rctx)
).
get_server() ->
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 53b34f41a..16bea8433 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -497,7 +497,7 @@ view_cb({row, Props}, #mrargs{extra = Options} = Acc) ->
%% TODO: wire in csrt tracking
%% TODO: distinguish between all_docs vs view call
couch_stats:increment_counter([fabric_rpc, view, rows_read]),
- %%couch_stats_resource_tracker:inc(rows_read),
+ %%csrt:inc(rows_read),
% Adding another row
ViewRow = fabric_view_row:from_props(Props, Options),
ok = rexi:stream2(ViewRow),
@@ -674,13 +674,13 @@ clean_stack(S) ->
).
set_io_priority(DbName, Options) ->
- couch_stats_resource_tracker:set_context_dbname(DbName),
+ 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} ->
- couch_stats_resource_tracker:set_context_username(UserName)
+ csrt:set_context_username(UserName)
end,
case lists:keyfind(io_priority, 1, Options) of
{io_priority, Pri} ->
diff --git a/src/fabric/src/fabric_util.erl b/src/fabric/src/fabric_util.erl
index 52032f0ec..49130228c 100644
--- a/src/fabric/src/fabric_util.erl
+++ b/src/fabric/src/fabric_util.erl
@@ -145,7 +145,7 @@ await_shard_response(Ref, Name, Rest, Opts, Factor, Timeout) ->
receive
Msg0 ->
{Msg, Delta} = rexi_utils:extract_delta(Msg0),
- couch_stats_resource_tracker:accumulate_delta(Delta),
+ csrt:accumulate_delta(Delta),
case Msg of
{Ref, {ok, Db}} ->
{ok, Db};
diff --git a/src/ioq/src/ioq.erl b/src/ioq/src/ioq.erl
index 8a7e68941..1324c07ff 100644
--- a/src/ioq/src/ioq.erl
+++ b/src/ioq/src/ioq.erl
@@ -59,7 +59,7 @@ call_search(Fd, Msg, Metadata) ->
call(Fd, Msg, Metadata).
call(Fd, Msg, Metadata) ->
- couch_stats_resource_tracker:ioq_called(),
+ csrt:ioq_called(),
Priority = io_class(Msg, Metadata),
case bypass(Priority) of
true ->
diff --git a/src/mem3/src/mem3_rpc.erl b/src/mem3/src/mem3_rpc.erl
index 81b4fec27..05b2d3773 100644
--- a/src/mem3/src/mem3_rpc.erl
+++ b/src/mem3/src/mem3_rpc.erl
@@ -393,7 +393,7 @@ wait_message(Node, Ref, Mon, Timeout) ->
process_raw_message(Msg0, Node, Ref, Mon, Timeout) ->
{Msg, Delta} = rexi_utils:extract_delta(Msg0),
- couch_stats_resource_tracker:accumulate_delta(Delta),
+ csrt:accumulate_delta(Delta),
case Msg of
{Ref, {ok, Reply}} ->
Reply;
diff --git a/src/rexi/src/rexi_server.erl b/src/rexi/src/rexi_server.erl
index 20a9f598e..ec35c9dc2 100644
--- a/src/rexi/src/rexi_server.erl
+++ b/src/rexi/src/rexi_server.erl
@@ -139,15 +139,15 @@ init_p(From, {M, F, A}, Nonce) ->
put('$initial_call', MFA),
put(nonce, Nonce),
try
- couch_stats_resource_tracker:create_worker_context(From, MFA, Nonce),
+ csrt:create_worker_context(From, MFA, Nonce),
couch_stats:maybe_track_rexi_init_p(MFA),
apply(M, F, A)
catch
exit:normal ->
- couch_stats_resource_tracker:destroy_context(),
+ csrt:destroy_context(),
ok;
Class:Reason:Stack0 ->
- couch_stats_resource_tracker:destroy_context(),
+ csrt:destroy_context(),
Stack = clean_stack(Stack0),
{ClientPid, _ClientRef} = From,
couch_log:error(
@@ -163,7 +163,7 @@ init_p(From, {M, F, A}, Nonce) ->
]
),
exit(#error{
- delta = couch_stats_resource_tracker:make_delta(),
+ delta = csrt:make_delta(),
timestamp = os:timestamp(),
reason = {Class, Reason},
mfa = {M, F, A},
diff --git a/src/rexi/src/rexi_utils.erl b/src/rexi/src/rexi_utils.erl
index 8b9855169..ea9aec16c 100644
--- a/src/rexi/src/rexi_utils.erl
+++ b/src/rexi/src/rexi_utils.erl
@@ -70,7 +70,7 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
process_raw_message(Payload0, RefList, Keypos, Fun, Acc0, TimeoutRef) ->
{Payload, Delta} = extract_delta(Payload0),
- couch_stats_resource_tracker:accumulate_delta(Delta),
+ csrt:accumulate_delta(Delta),
case Payload of
{timeout, TimeoutRef} ->
{timeout, Acc0};
@@ -128,10 +128,10 @@ extract_delta({A, B, C, D, E, F, G, {delta, Delta}}) -> {{A, B, C, D, E, F, G},
extract_delta(T) -> {T, undefined}.
get_delta() ->
- {delta, couch_stats_resource_tracker:make_delta()}.
+ {delta, csrt:make_delta()}.
maybe_add_delta(T) ->
- case couch_stats_resource_tracker:is_enabled() of
+ case csrt:is_enabled() of
false ->
T;
true ->
@@ -147,7 +147,7 @@ maybe_add_delta(T, undefined) ->
maybe_add_delta(T, Delta) when is_map(Delta) ->
maybe_add_delta(T, {delta, Delta});
maybe_add_delta(T, {delta, _} = Delta) ->
- case couch_stats_resource_tracker:is_enabled() of
+ case csrt:is_enabled() of
false ->
T;
true ->
(couchdb) 03/06: WIP: Rework modules
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit bff7ae8802d0e154428f4ba96720a68a130a60dd
Author: Russell Branca <ch...@apache.org>
AuthorDate: Fri Jan 10 12:12:00 2025 -0800
WIP: Rework modules
---
.../src/couch_stats_resource_tracker.erl | 845 ---------------------
src/couch_stats/src/csrt_logger.erl | 465 +++++-------
src/couch_stats/src/csrt_logger2.erl | 220 ------
3 files changed, 171 insertions(+), 1359 deletions(-)
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.erl b/src/couch_stats/src/couch_stats_resource_tracker.erl
deleted file mode 100644
index 70254088f..000000000
--- a/src/couch_stats/src/couch_stats_resource_tracker.erl
+++ /dev/null
@@ -1,845 +0,0 @@
-% 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(couch_stats_resource_tracker).
-
--behaviour(gen_server).
-
-%% TODO: delete these exports after csrt_logger experiment concluded
--export([
- new/0,
- tnow/0
-]).
-
--export([
- start_link/0,
- init/1,
- handle_call/3,
- handle_cast/2
-]).
-
-%% PidRef API
--export([
- close_pid_ref/0,
- close_pid_ref/1,
- create_pid_ref/0,
- get_pid_ref/0,
- set_pid_ref/1
-]).
-
-%% Context API
--export([
- create_context/2,
- create_coordinator_context/2,
- create_resource/1,
- 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_handler_fun/3,
- set_context_username/1,
- set_context_username/2
-]).
-
-%% stats collection api
--export([
- accumulate_delta/1,
- inc/1,
- inc/2,
- ioq_called/0,
- is_enabled/0,
- make_delta/0,
- maybe_inc/2,
- should_track/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
-]).
-
-%% Process lifetime reporting api
--export([
- is_logging_enabled/0,
- log_process_lifetime_report/1,
- logging_enabled/0,
- should_log/1,
- should_log/2,
- tracker/1,
- to_json/1
-]).
-
--include_lib("couch/include/couch_db.hrl").
--include_lib("stdlib/include/ms_transform.hrl").
--include_lib("couch_stats_resource_tracker.hrl").
-
-%% Module pdict markers
--ifdef(foo).
--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(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,
- [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,
- ?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
-}).
--endif.
-
--record(st, {}).
-
-%%
-%% Public API
-%%
-
-%%
-%% PidRef operations
-%%
-
-new() -> #rctx{}.
-
-get_pid_ref() ->
- get(?PID_REF).
-
-set_pid_ref(PidRef) ->
- erlang:put(?PID_REF, PidRef),
- PidRef.
-
-create_pid_ref() ->
- {self(), make_ref()}.
-
-close_pid_ref() ->
- close_pid_ref(get_pid_ref()).
-
-%%close_pid_ref(undefined) ->
-%% undefined;
-close_pid_ref(_PidRef) ->
- erase(?PID_REF).
-
-get_resource() ->
- get_resource(get_pid_ref()).
-
-get_resource(undefined) ->
- undefined;
-get_resource(PidRef) ->
- catch case ets:lookup(?MODULE, PidRef) of
- [#rctx{}=Rctx] ->
- Rctx;
- [] ->
- undefined
- end.
-
-%% monotonic time now in millisecionds
-tnow() ->
- erlang:monotonic_time(millisecond).
-
-is_enabled() ->
- config:get_boolean(?MODULE_STRING, "enabled", true).
-
-%%
-%% 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)).
-
-find_by_pid(Pid) ->
- [R || #rctx{} = R <- ets:match_object(?MODULE, #rctx{pid_ref={Pid, '_'}, _ = '_'})].
-
-find_by_pidref(PidRef) ->
- [R || #rctx{} = 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}, _ = '_'})].
-
-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;
-field(#rctx{type=Val}, type) -> 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))).
-
-%%
-%% Conversion API for outputting JSON
-%%
-
-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.
-
-convert_pidref({Parent0, ParentRef0}) ->
- Parent = convert_pid(Parent0),
- ParentRef = convert_ref(ParentRef0),
- <<Parent/binary, ":", ParentRef/binary>>;
-convert_pidref(null) ->
- null;
-convert_pidref(undefined) ->
- null.
-
-convert_pid(Pid) when is_pid(Pid) ->
- ?l2b(pid_to_list(Pid)).
-
-convert_ref(Ref) when is_reference(Ref) ->
- ?l2b(ref_to_list(Ref)).
-
-to_json(#rctx{}=Rctx) ->
- #{
- updated_at => Rctx#rctx.updated_at,
- started_at => 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,
- 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,
- ioq_calls => Rctx#rctx.ioq_calls
- }.
-
-%%
-%% Context lifecycle API
-%%
-
-create_resource(#rctx{} = Rctx) ->
- catch ets:insert(?MODULE, Rctx).
-
-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.
-
-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.
-
-create_context(Type, Nonce) ->
- Rctx = new_context(Type, Nonce),
- set_pid_ref(Rctx#rctx.pid_ref),
- erlang:put(?DELTA_TZ, Rctx),
- create_resource(Rctx),
- track(Rctx),
- Rctx#rctx.pid_ref.
-
-%% Might be useful to export this but the internal worker types aren't exported
-new_context(Type, Nonce) ->
- #rctx{
- nonce = Nonce,
- pid_ref = create_pid_ref(),
- type = Type
- }.
-
-set_context_dbname(DbName) ->
- set_context_dbname(DbName, get_pid_ref()).
-
-set_context_dbname(_, undefined) ->
- ok;
-set_context_dbname(DbName, PidRef) ->
- is_enabled() andalso update_element(PidRef, [{#rctx.dbname, DbName}]).
-
-set_context_handler_fun(Fun) when is_function(Fun) ->
- case is_enabled() of
- false ->
- ok;
- true ->
- FProps = erlang:fun_info(Fun),
- Mod = proplists:get_value(module, FProps),
- Func = proplists:get_value(name, FProps),
- set_context_handler_fun(Mod, Func)
- end.
-
-set_context_handler_fun(Mod, Func) when is_atom(Mod) andalso is_atom(Func) ->
- set_context_handler_fun(Mod, Func, get_pid_ref()).
-
-set_context_handler_fun(_, _, undefined) ->
- ok;
-set_context_handler_fun(Mod, Func, PidRef) when is_atom(Mod) andalso is_atom(Func) ->
- case is_enabled() of
- false ->
- ok;
- true ->
- #rctx{type=#coordinator{}=Coordinator} = get_resource(),
- Update = [{#rctx.type, Coordinator#coordinator{mod=Mod, func=Func}}],
- update_element(PidRef, Update)
- end.
-
-set_context_username(null) ->
- ok;
-set_context_username(undefined) ->
- ok;
-set_context_username(User) ->
- set_context_username(User, get_pid_ref()).
-
-set_context_username(null, _) ->
- ok;
-set_context_username(_, undefined) ->
- ok;
-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 update_element(PidRef, [{#rctx.username, UserName}]).
-
-destroy_context() ->
- destroy_context(get_pid_ref()).
-
-destroy_context(undefined) ->
- ok;
-destroy_context({_, _} = PidRef) ->
- stop_tracker(get_tracker()),
- close_pid_ref(PidRef),
- ok.
-
-%% Stat collection API
-
-inc(Key) ->
- inc(Key, 1).
-
-inc(Key, N) ->
- case maps:is_key(Key, ?KEYS_TO_FIELDS) of
- true ->
- update_counter(maps:get(Key, ?KEYS_TO_FIELDS), N);
- false ->
- 0
- end.
-
-
-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.
-
-%% TODO: update stats_descriptions.cfg for relevant apps
-should_track([fabric_rpc, all_docs, spawned]) ->
- is_enabled();
-should_track([fabric_rpc, changes, spawned]) ->
- is_enabled();
-should_track([fabric_rpc, changes, processed]) ->
- is_enabled();
-should_track([fabric_rpc, changes, returned]) ->
- is_enabled();
-should_track([fabric_rpc, map_view, spawned]) ->
- is_enabled();
-should_track([fabric_rpc, reduce_view, spawned]) ->
- is_enabled();
-should_track([fabric_rpc, get_all_security, spawned]) ->
- is_enabled();
-should_track([fabric_rpc, open_doc, spawned]) ->
- is_enabled();
-should_track([fabric_rpc, update_docs, spawned]) ->
- is_enabled();
-should_track([fabric_rpc, open_shard, spawned]) ->
- is_enabled();
-should_track([mango_cursor, view, all_docs]) ->
- is_enabled();
-should_track([mango_cursor, view, idx]) ->
- is_enabled();
-should_track(_Metric) ->
- %%io:format("SKIPPING METRIC: ~p~n", [Metric]),
- false.
-
-ioq_called() ->
- inc(ioq_calls).
-
-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);
-accumulate_delta(undefined) ->
- ok.
-
-make_delta() ->
- TA = case get(?DELTA_TA) of
- undefined ->
- %% Need to handle this better, can't just make a new T0 at T' as
- %% the timestamps will be identical causing a divide by zero error.
- %%
- %% Realistically need to ensure that all invocations of database
- %% operations sets T0 appropriately. Perhaps it's possible to do
- %% this is the couch_db:open chain, and then similarly, in
- %% couch_server, and uhhhh... couch_file, and...
- %%
- %% I think we need some type of approach for establishing a T0 that
- %% doesn't result in outrageous deltas. For now zero out the
- %% microseconds field, or subtract a second on the off chance that
- %% microseconds is zero. I'm not uptodate on the latest Erlang time
- %% libraries and don't remember how to easily get an
- %% `os:timestamp()` out of now() - 100ms or some such.
- %%
- %% I think it's unavoidable that we'll have some codepaths that do
- %% not properly instantiate the T0 at spawn resulting in needing to
- %% do some time of "time warp" or ignoring the timing collection
- %% entirely. Perhaps if we hoisted out the stats collection into
- %% the primary flow of the database and funnel that through all the
- %% function clauses we could then utilize Dialyzer to statically
- %% analyze and assert all code paths that invoke database
- %% operations have properly instantinated a T0 at the appropriate
- %% start time such that we don't have to "fudge" deltas with a
- %% missing start point, but we're a long ways from that happening
- %% so I feel it necessary to address the NULL start time.
-
- %% Track how often we fail to initiate T0 correctly
- %% Perhaps somewhat naughty we're incrementing stats from within
- %% couch_stats itself? Might need to handle this differently
- %% TODO: determine appropriate course of action here
- %% io:format("~n**********MISSING STARTING DELTA************~n~n", []),
- couch_stats:increment_counter(
- [couchdb, csrt, delta_missing_t0]),
- %%[couch_stats_resource_tracker, delta_missing_t0]),
-
- case erlang:get(?DELTA_TZ) of
- undefined ->
- TA0 = make_delta_base(),
- %% TODO: handline missing deltas, otherwise divide by zero
- set_delta_a(TA0),
- TA0;
- TA0 ->
- TA0
- end;
- #rctx{} = TA0 ->
- TA0
- end,
- TB = get_resource(),
- Delta = make_delta(TA, TB),
- set_delta_a(TB),
- Delta.
-
-make_delta(#rctx{}=TA, #rctx{}=TB) ->
- Delta = #{
- docs_read => TB#rctx.docs_read - TA#rctx.docs_read,
- 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,
- 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 => TB#rctx.updated_at - TA#rctx.updated_at
- },
- %% TODO: reevaluate this decision
- %% Only return non zero (and also positive) delta fields
- maps:filter(fun(_K,V) -> V > 0 end, Delta);
-make_delta(_, #rctx{}) ->
- #{error => missing_beg_rctx};
-make_delta(#rctx{}, _) ->
- #{error => missing_fin_rctx}.
-
-%% TODO: what to do when PidRef=undefined?
-make_delta_base(PidRef) ->
- %% TODO: extract user_ctx and db/shard from request
- Now = tnow(),
- #rctx{
- pid_ref = PidRef,
- %% TODO: confirm this subtraction works
- started_at = Now - 100, %% give us 100ms rewind time for missing T0
- updated_at = Now
- }.
-
-make_delta_base() ->
- make_delta_base(get_pid_ref()).
-
-set_delta_a(TA) ->
- erlang:put(?DELTA_TA, TA).
-
-update_counter(Field, Count) ->
- is_enabled() andalso update_counter(get_pid_ref(), Field, Count).
-
-update_counter(undefined, _Field, _Count) ->
- ok;
-update_counter({_Pid,_Ref}=PidRef, Field, Count) ->
- %% TODO: mem3 crashes without catch, why do we lose the stats table?
- is_enabled() andalso catch ets:update_counter(?MODULE, PidRef, {Field, Count}, #rctx{pid_ref=PidRef}).
-
-update_element(undefined, _Update) ->
- ok;
-update_element({_Pid,_Ref}=PidRef, Update) ->
- %% TODO: should we take any action when the update fails?
- is_enabled() andalso catch ets:update_element(?MODULE, PidRef, Update).
-
-%% Process lifetime logging api
-
-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.
-
-tracker({Pid, _Ref}=PidRef) ->
- MonRef = erlang:monitor(process, Pid),
- receive
- stop ->
- %% TODO: do we need cleanup here?
- log_process_lifetime_report(PidRef),
- catch evict(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),
- catch evict(PidRef)
- end.
-
-log_process_lifetime_report(PidRef) ->
- case is_enabled() andalso is_logging_enabled() of
- true ->
- case conf_get("logger_type", "csrt_logger") of
- "csrt_logger" ->
- csrt_logger:maybe_report(PidRef);
- _ ->
- Rctx = get_resource(PidRef),
- case should_log(Rctx) of
- true ->
- couch_log:report("csrt-pid-usage-lifetime", to_json(Rctx));
- _ ->
- ok
- end
- end;
- false ->
- ok
- end.
-
-is_logging_enabled() ->
- logging_enabled() =/= false.
-
-logging_enabled() ->
- case conf_get("log_pid_usage_report", "coordinator") of
- "coordinator" ->
- coordinator;
- "true" ->
- true;
- _ ->
- false
- end.
-
-should_log(undefined) ->
- false;
-should_log(#rctx{}=Rctx) ->
- should_log(Rctx, logging_enabled()).
-
-should_log(undefined, _) ->
- false;
-should_log(#rctx{}, true) ->
- true;
-should_log(#rctx{}, false) ->
- false;
-should_log(#rctx{type = #coordinator{}}, coordinator) ->
- true;
-should_log(#rctx{type = #rpc_worker{mod=fabric_rpc, func=FName}}, _) ->
- case conf_get("log_fabric_rpc") of
- "true" ->
- true;
- undefined ->
- false;
- Name ->
- Name =:= atom_to_list(FName)
- end;
-should_log(#rctx{}, _) ->
- false.
-
-%%
-%% 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({call_search, _}, _From, St) ->
- %% TODO: provide isolated search queries here
- {reply, ok, St};
-handle_call(_, _From, State) ->
- {reply, ok, State}.
-
-handle_cast(_Msg, State) ->
- {noreply, State, 0}.
-
-%%
-%% private functions
-%%
-
-conf_get(Key) ->
- conf_get(Key, undefined).
-
-
-conf_get(Key, Default) ->
- config:get(?MODULE_STRING, Key, Default).
-
-to_json_list(List) when is_list(List) ->
- lists:map(fun to_json/1, List).
-
-%%
-%% Process lifetime logging api
-%%
-
-get_tracker() ->
- get(?TRACKER_PID).
-
-put_tracker(Pid) when is_pid(Pid) ->
- put(?TRACKER_PID, Pid).
-
-evict(PidRef) ->
- ets:delete(?MODULE, PidRef).
-
-stop_tracker(undefined) ->
- ok;
-stop_tracker(Pid) when is_pid(Pid) ->
- Pid ! stop.
-
-
--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)
- ]
- }.
-
-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(ioq_calls, lists:sort(maps:keys(?KEYS_TO_FIELDS)))).
-
--endif.
diff --git a/src/couch_stats/src/csrt_logger.erl b/src/couch_stats/src/csrt_logger.erl
index e4df3d053..57262a00f 100644
--- a/src/couch_stats/src/csrt_logger.erl
+++ b/src/couch_stats/src/csrt_logger.erl
@@ -10,334 +10,211 @@
% License for the specific language governing permissions and limitations under
% the License.
--module(csrt_logger).
+-module(csrt_logger2).
--behaviour(gen_server).
--behaviour(config_listener).
-
--export([
- start_link/0,
- init/1,
- handle_call/3,
- handle_cast/2,
- handle_info/2
-]).
+-include_lib("stdlib/include/ms_transform.hrl").
+-include_lib("couch_stats_resource_tracker.hrl").
+%% Process lifetime logging api
-export([
- log_level/0,
- log_level/1,
- maybe_report/1,
- maybe_report_old/1,
- get_matchers/0
+ get_tracker/0,
+ is_logging_enabled/0,
+ log_process_lifetime_report/1,
+ put_tracker/1,
+ stop_tracker/0,
+ stop_tracker/1,
+ track/1,
+ tracker/1
]).
-%% Config update subscription API
+%% JSON Conversion API
-export([
- subscribe_changes/0,
- handle_config_change/5,
- handle_config_terminate/3
+ convert_type/1,
+ convert_pidref/1,
+ convert_pid/1,
+ convert_ref/1,
+ to_json/1
]).
-%% Matchers
+%% Raw API that bypasses is_enabled checks
-export([
- matcher_on_docs_read/1,
- matcher_on_worker_changes_processed/1,
- matcher_on_ioq_calls/1,
- matcher_on_nonce/1,
- matcher_on_nonce_comp/1
-]).
-
--include_lib("stdlib/include/ms_transform.hrl").
--include_lib("couch_stats_resource_tracker.hrl").
-
--record(st, {
- should_track,
- matchers = #{}
-}).
-
--define(CSRT_CONTEXTS, [
- #rpc_worker{mod=fabric_rpc, func=all_docs},
- #rpc_worker{mod=fabric_rpc, func=changes},
- #rpc_worker{mod=fabric_rpc, func=map_view},
- #rpc_worker{mod=fabric_rpc, func=reduce_view},
- #rpc_worker{mod=fabric_rpc, func=get_all_security},
- #rpc_worker{mod=fabric_rpc, func=open_doc},
- #rpc_worker{mod=fabric_rpc, func=update_docs},
- #rpc_worker{mod=fabric_rpc, func=open_shard},
- #rpc_worker{mod=mango_cursor, func=view}
- %% enable all: #worker{}
- %% TODO: add coordinator/etc
- %% #coordinator{mod=chttpd_db:}
-]).
-
--define(CSRT_LOG_LEVELS, [
- {all, 9},
- {workers_too, 4},
- {custom, 2},
- {coordinators, 1},
- {none, 0}
+ do_lifetime_report/1,
+ do_status_report/1,
+ do_report/2,
+ maybe_report/2
]).
--define(NO_LOG, 0).
--define(DEFAULT_LEVEL, coordinators).
-
--define(MATCHERS_KEY, {?MODULE, all_csrt_matchers}).
--define(CONF_MATCHERS_ENABLED, "csrt_logger.matchers_enabled").
--define(CONF_MATCHERS_THRESHOLD, "csrt_logger.matchers_threshold").
-
-%%
-%% Matchers
-%%
-
-matcher_on_docs_read(Threshold)
- when is_integer(Threshold) andalso Threshold > 0 ->
- ets:fun2ms(fun(#rctx{docs_read=DocsRead} = R) when DocsRead >= Threshold -> R end).
-
-matcher_on_nonce(Nonce) ->
- ets:fun2ms(fun(#rctx{nonce = Nonce1} = R) when Nonce =:= Nonce1 -> R end).
-
-matcher_on_nonce_comp(Nonce) ->
- ets:match_spec_compile(ets:fun2ms(fun(#rctx{nonce = Nonce1} = R) when Nonce =:= Nonce1 -> R end)).
-
-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
- ).
-
-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).
-
-%%
-%% Public API
-%%
-
-log_level(all) -> 9;
-log_level(all_workers) -> 4;
-log_level(rpc_workers) -> 3;
-log_level(#rctx{type=#rpc_worker{}}) -> 3;
-log_level(custom) -> 2;
-log_level(coordinators) -> 1;
-log_level(#rctx{type=#coordinator{}}) -> 1;
-log_level(none) -> ?NO_LOG;
-log_level(_) -> ?NO_LOG.
-
-log_level() ->
- Level = config:get("couch_stats_resource_tracker", "log_level", "coordinators"),
- try
- log_level(list_to_existing_atom(Level))
- catch
- _:_ ->
- log_level(?DEFAULT_LEVEL)
+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.
-
-maybe_report(PidRef) ->
- Rctx = csrt:get_resource(PidRef),
- case is_match(Rctx) of
- true ->
- report(Rctx),
+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_context(PidRef),
ok;
- false ->
+ {'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_context(PidRef),
ok
end.
-maybe_report_old(PidRef) ->
- Level = log_level(),
- case Level > ?NO_LOG of
+log_process_lifetime_report(PidRef) ->
+ case csrt:is_enabled() andalso is_logging_enabled() of
true ->
- Rctx = csrt:get_resource(PidRef),
- RLevel = log_level(Rctx),
- %% io:format("CSRT LOGGER TRIGGERING{~p} ~p =<? ~p~n", [Rctx#rctx.type, RLevel, Level]),
- case RLevel >= Level of
- true ->
- %% NOTE: we drop the PidRef here, it's now safe to delete the ets entry
- Msg = {maybe_report, Rctx, RLevel, Level},
- gen_server:cast(get_server(), Msg);
- false ->
- ok
+ case csrt:conf_get("logger_type", "csrt_logger") of
+ "csrt_logger" ->
+ csrt_logger:maybe_report(PidRef);
+ _ ->
+ Rctx = csrt_server:get_resource(PidRef),
+ case should_log(Rctx) of
+ true ->
+ do_lifetime_report(Rctx);
+ _ ->
+ ok
+ end
end;
false ->
ok
end.
-find_matches(RContexts, Matchers) when is_list(RContexts) andalso is_map(Matchers) ->
- maps:filter(
- fun(_Name, {_MSpec, CompMSpec}) ->
- catch [] =/= ets:match_spec_run(RContexts, CompMSpec)
- end,
- Matchers
- ).
-
-get_matchers() ->
- persistent_term:get(?MATCHERS_KEY, #{}).
-
-is_match(#rctx{}=Rctx) ->
- is_match(Rctx, get_matchers()).
-
-is_match(#rctx{}=Rctx, Matchers) when is_map(Matchers) ->
- maps:size(find_matches([Rctx], Matchers)) > 0.
-
-%% default.ini:
-%% [csrt]
-%% enabled = false
-%%
-%% [csrt.should_log]
-%% worker.fabric_rpc.all_docs = true
-%% worker.fabric_rpc.map_view = true
-%% worker.fabric_rpc.reduce_view = true
-create_tracking_map() ->
- lists:foldl(fun(Ele, Acc) ->
- Key = case Ele of
- #rpc_worker{mod=M0, func=F0} ->
- M = atom_to_list(M0),
- F = atom_to_list(F0),
- "worker." ++ M ++ "." ++ F
- %% TODO: #coordinator{} ->
- end,
- Enabled = config:get_boolean("csrt.should_log", Key, true),
- maps:put(Ele, Enabled, Acc)
- end, #{}, ?CSRT_CONTEXTS).
-
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
-init([]) ->
- ok = initialize_matchers(),
- ok = subscribe_changes(),
- {ok, #st{should_track=create_tracking_map()}}.
-
-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({maybe_report, #rctx{}=Rctx, RLevel, Level}, _From, St) ->
- ok = maybe_report_int(Rctx, RLevel, Level, St),
- {reply, ok, St};
-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}.
-
-%% TODO: should we use originally defined log level? or St#st.level?
-%% ShouldLog = log_level(Rctx) > St#st.log_level
-maybe_report_int(#rctx{}=Rctx, RLevel, Level, #st{}) when RLevel >= Level ->
- report(Rctx);
-maybe_report_int(#rctx{}=Rctx, _RLevel, _Level, #st{matchers=Matchers, should_track=Trackers}) ->
- case is_tracked(Rctx, Trackers) orelse is_match(Rctx, Matchers) of
- true ->
- report(Rctx);
- false ->
- ok
+is_logging_enabled() ->
+ logging_enabled() =/= false.
+
+logging_enabled() ->
+ case csrt:conf_get("log_pid_usage_report", "coordinator") of
+ "coordinator" ->
+ coordinator;
+ "true" ->
+ true;
+ _ ->
+ false
end.
-report(Rctx) ->
- couch_log:report(
- "csrt-pid-usage-lifetime",
- csrt:to_json(Rctx)
- ).
+should_log(undefined) ->
+ false;
+should_log(#rctx{}=Rctx) ->
+ should_log(Rctx, logging_enabled()).
+
+should_log(undefined, _) ->
+ false;
+should_log(#rctx{}, true) ->
+ true;
+should_log(#rctx{}, false) ->
+ false;
+should_log(#rctx{type = #coordinator{}}, coordinator) ->
+ true;
+should_log(#rctx{type = #rpc_worker{mod=fabric_rpc, func=FName}}, _) ->
+ case csrt:conf_get("log_fabric_rpc") of
+ "true" ->
+ true;
+ undefined ->
+ false;
+ Name ->
+ Name =:= atom_to_list(FName)
+ end;
+should_log(#rctx{}, _) ->
+ false.
-get_server() ->
- %% TODO: shard servers like couch_server/etc
- ?MODULE.
+%% TODO: decide on API
+maybe_report(_ReportName, PidRef) ->
+ log_process_lifetime_report(PidRef).
-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.
+do_lifetime_report(Rctx) ->
+ do_report("csrt-pid-usage-lifetime", Rctx).
-is_tracked(#rctx{type=Type}, Trackers) ->
- Res = maps:is_key(Type, Trackers) andalso true =:= maps:get(Type, Trackers),
- Res.
+do_status_report(Rctx) ->
+ do_report("csrt-pid-usage-status", Rctx).
-set_matchers_term(Matchers) when is_map(Matchers) ->
- persistent_term:put({?MODULE, all_csrt_matchers}, Matchers).
+do_report(ReportName, #rctx{}=Rctx) ->
+ couch_log:report(ReportName, to_json(Rctx)).
-initialize_matchers() ->
- DefaultMatchers = [
- {docs_read, fun matcher_on_docs_read/1, 1000},
- %%{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
- ),
- couch_log:warning("Initialized ~p CSRT Logger matchers", [maps:size(Matchers)]),
- persistent_term:put(?MATCHERS_KEY, Matchers),
- ok.
+%%
+%% Conversion API for outputting JSON
+%%
+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.
+
+convert_pidref({Parent0, ParentRef0}) ->
+ Parent = convert_pid(Parent0),
+ ParentRef = convert_ref(ParentRef0),
+ <<Parent/binary, ":", ParentRef/binary>>;
+convert_pidref(null) ->
+ null;
+convert_pidref(undefined) ->
+ null.
+
+convert_pid(Pid) when is_pid(Pid) ->
+ list_to_binary(pid_to_list(Pid)).
+
+convert_ref(Ref) when is_reference(Ref) ->
+ list_to_binary(ref_to_list(Ref)).
+
+to_json(#rctx{}=Rctx) ->
+ #{
+ updated_at => Rctx#rctx.updated_at,
+ started_at => 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,
+ 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,
+ ioq_calls => Rctx#rctx.ioq_calls
+ }.
-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).
+%%
+%% Process lifetime logging api
+%%
-matcher_threshold(Name, Default)
- when is_list(Name) andalso is_integer(Default) andalso Default > 0 ->
- config:get_integer(?CONF_MATCHERS_THRESHOLD, Name, Default).
+get_tracker() ->
+ get(?TRACKER_PID).
-subscribe_changes() ->
- config:listen_for_changes(?MODULE, nil).
+put_tracker(Pid) when is_pid(Pid) ->
+ put(?TRACKER_PID, Pid).
-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}.
+stop_tracker() ->
+ stop_tracker(get_tracker()).
-handle_config_terminate(_, stop, _) ->
+stop_tracker(undefined) ->
ok;
-handle_config_terminate(_, _, _) ->
- erlang:send_after(5000, whereis(?MODULE), restart_config_listener).
+stop_tracker(Pid) when is_pid(Pid) ->
+ Pid ! stop.
diff --git a/src/couch_stats/src/csrt_logger2.erl b/src/couch_stats/src/csrt_logger2.erl
deleted file mode 100644
index 57262a00f..000000000
--- a/src/couch_stats/src/csrt_logger2.erl
+++ /dev/null
@@ -1,220 +0,0 @@
-% 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_logger2).
-
--include_lib("stdlib/include/ms_transform.hrl").
--include_lib("couch_stats_resource_tracker.hrl").
-
-%% Process lifetime logging api
--export([
- get_tracker/0,
- is_logging_enabled/0,
- log_process_lifetime_report/1,
- put_tracker/1,
- stop_tracker/0,
- stop_tracker/1,
- track/1,
- tracker/1
-]).
-
-%% JSON Conversion API
--export([
- convert_type/1,
- convert_pidref/1,
- convert_pid/1,
- convert_ref/1,
- to_json/1
-]).
-
-%% Raw API that bypasses is_enabled checks
--export([
- do_lifetime_report/1,
- do_status_report/1,
- do_report/2,
- maybe_report/2
-]).
-
-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.
-
-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_context(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_context(PidRef),
- ok
- end.
-
-log_process_lifetime_report(PidRef) ->
- case csrt:is_enabled() andalso is_logging_enabled() of
- true ->
- case csrt:conf_get("logger_type", "csrt_logger") of
- "csrt_logger" ->
- csrt_logger:maybe_report(PidRef);
- _ ->
- Rctx = csrt_server:get_resource(PidRef),
- case should_log(Rctx) of
- true ->
- do_lifetime_report(Rctx);
- _ ->
- ok
- end
- end;
- false ->
- ok
- end.
-
-is_logging_enabled() ->
- logging_enabled() =/= false.
-
-logging_enabled() ->
- case csrt:conf_get("log_pid_usage_report", "coordinator") of
- "coordinator" ->
- coordinator;
- "true" ->
- true;
- _ ->
- false
- end.
-
-should_log(undefined) ->
- false;
-should_log(#rctx{}=Rctx) ->
- should_log(Rctx, logging_enabled()).
-
-should_log(undefined, _) ->
- false;
-should_log(#rctx{}, true) ->
- true;
-should_log(#rctx{}, false) ->
- false;
-should_log(#rctx{type = #coordinator{}}, coordinator) ->
- true;
-should_log(#rctx{type = #rpc_worker{mod=fabric_rpc, func=FName}}, _) ->
- case csrt:conf_get("log_fabric_rpc") of
- "true" ->
- true;
- undefined ->
- false;
- Name ->
- Name =:= atom_to_list(FName)
- end;
-should_log(#rctx{}, _) ->
- false.
-
-%% TODO: decide on API
-maybe_report(_ReportName, PidRef) ->
- log_process_lifetime_report(PidRef).
-
-do_lifetime_report(Rctx) ->
- do_report("csrt-pid-usage-lifetime", Rctx).
-
-do_status_report(Rctx) ->
- do_report("csrt-pid-usage-status", Rctx).
-
-do_report(ReportName, #rctx{}=Rctx) ->
- couch_log:report(ReportName, to_json(Rctx)).
-
-%%
-%% Conversion API for outputting JSON
-%%
-
-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.
-
-convert_pidref({Parent0, ParentRef0}) ->
- Parent = convert_pid(Parent0),
- ParentRef = convert_ref(ParentRef0),
- <<Parent/binary, ":", ParentRef/binary>>;
-convert_pidref(null) ->
- null;
-convert_pidref(undefined) ->
- null.
-
-convert_pid(Pid) when is_pid(Pid) ->
- list_to_binary(pid_to_list(Pid)).
-
-convert_ref(Ref) when is_reference(Ref) ->
- list_to_binary(ref_to_list(Ref)).
-
-to_json(#rctx{}=Rctx) ->
- #{
- updated_at => Rctx#rctx.updated_at,
- started_at => 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,
- 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,
- ioq_calls => Rctx#rctx.ioq_calls
- }.
-
-%%
-%% Process lifetime logging api
-%%
-
-get_tracker() ->
- get(?TRACKER_PID).
-
-put_tracker(Pid) when is_pid(Pid) ->
- put(?TRACKER_PID, Pid).
-
-stop_tracker() ->
- stop_tracker(get_tracker()).
-
-stop_tracker(undefined) ->
- ok;
-stop_tracker(Pid) when is_pid(Pid) ->
- Pid ! stop.