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 2023/11/20 22:25:45 UTC
(couchdb) branch couch-stats-resource-tracker updated (cafa5123b -> 4c05a0f39)
This is an automated email from the ASF dual-hosted git repository.
chewbranca pushed a change to branch couch-stats-resource-tracker
in repository https://gitbox.apache.org/repos/asf/couchdb.git
from cafa5123b Add watchdog scanning for unmonitored processes
new 996c38551 Set dbname context properly
new 1bbf7a90e Add unsafe_foldl for perf testing
new e70ba4d5f Rework csrt context setting
new 715b52419 WIP: ugly but working HTTP API around groupings
new be8ab045d Cleanup debug info
new 4c05a0f39 Make CSRT toggle-able and rework delta sending
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.
Summary of changes:
src/chttpd/src/chttpd.erl | 13 +-
src/chttpd/src/chttpd_misc.erl | 92 +++++++-
src/couch/priv/stats_descriptions.cfg | 2 +-
.../src/couch_stats_resource_tracker.erl | 251 ++++++++++++---------
src/fabric/src/fabric_rpc.erl | 17 +-
src/fabric/src/fabric_util.erl | 2 -
src/rexi/src/rexi.erl | 27 ++-
src/rexi/src/rexi_utils.erl | 33 ++-
8 files changed, 288 insertions(+), 149 deletions(-)
(couchdb) 02/06: Add unsafe_foldl for perf testing
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 1bbf7a90e82ccb224c0ebf1558fabbf2ee4bd795
Author: Russell Branca <ch...@apache.org>
AuthorDate: Wed Nov 15 16:03:20 2023 -0800
Add unsafe_foldl for perf testing
---
.../src/couch_stats_resource_tracker.erl | 43 +++++++++++++++++-----
src/fabric/src/fabric_rpc.erl | 2 +-
2 files changed, 35 insertions(+), 10 deletions(-)
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.erl b/src/couch_stats/src/couch_stats_resource_tracker.erl
index 0a5b2e335..d2dd4406c 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.erl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.erl
@@ -53,10 +53,13 @@
count_by/1,
group_by/2,
group_by/3,
+ group_by/4,
sorted/1,
sorted_by/1,
sorted_by/2,
- sorted_by/3
+ sorted_by/3,
+
+ unsafe_foldl/3
]).
-export([
@@ -393,20 +396,24 @@ group_by(KeyFun, ValFun) ->
group_by(KeyFun, ValFun, fun erlang:'+'/2).
+group_by(KeyFun, ValFun, AggFun) ->
+ group_by(KeyFun, ValFun, AggFun, fun ets:foldl/3).
+
+
%% 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) ->
+group_by(KeyL, ValFun, AggFun, Fold) 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) ->
+ group_by(KeyFun, ValFun, AggFun, Fold);
+group_by(Key, ValFun, AggFun, Fold) when is_atom(Key) ->
+ group_by(curry_field(Key), ValFun, AggFun, Fold);
+group_by(KeyFun, Val, AggFun, Fold) when is_atom(Val) ->
+ group_by(KeyFun, curry_field(Val), AggFun, Fold);
+group_by(KeyFun, ValFun, AggFun, Fold) ->
FoldFun = fun(Ele, Acc) ->
Key = KeyFun(Ele),
Val = ValFun(Ele),
@@ -414,7 +421,8 @@ group_by(KeyFun, ValFun, AggFun) ->
NewVal = AggFun(CurrVal, Val),
maps:put(Key, NewVal, Acc)
end,
- ets:foldl(FoldFun, #{}, ?MODULE).
+ Fold(FoldFun, #{}, ?MODULE).
+
sorted(Map) when is_map(Map) ->
lists:sort(fun({_K1, A}, {_K2, B}) -> B < A end, maps:to_list(Map)).
@@ -924,3 +932,20 @@ log_process_lifetime_report(PidRef) ->
%% TODO: catch error out of here, report crashes on depth>1 json
io:format("CSRT RCTX: ~p~n", [to_flat_json(Rctx)]),
couch_log:report("csrt-pid-usage-lifetime", to_flat_json(Rctx)).
+
+
+%% Reimplementation of: https://github.com/erlang/otp/blob/b2ee4fc9a0b81a139dad2033e9b2bfc178146886/lib/stdlib/src/ets.erl#L633-L658
+%% with wrapping of ets:safe_fixtable/2 removed
+unsafe_foldl(F, Accu, T) ->
+ First = ets:first(T),
+ do_foldl(F, Accu, First, T).
+
+do_foldl(F, Accu0, Key, T) ->
+ case Key of
+ '$end_of_table' ->
+ Accu0;
+ _ ->
+ do_foldl(F,
+ lists:foldl(F, Accu0, ets:lookup(T, Key)),
+ ets:next(T, Key), T)
+ end.
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index bfd039aad..2c8ba1654 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -331,7 +331,7 @@ reset_validation_funs(DbName) ->
open_shard(Name, Opts) ->
set_io_priority(Name, Opts),
- couch_stats_resource_tracker:set_context_dbname(DbName),
+ couch_stats_resource_tracker:set_context_dbname(Name),
try
rexi:reply(mem3_util:get_or_create_db(Name, Opts))
catch
(couchdb) 03/06: Rework csrt context setting
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit e70ba4d5f261b2ec9855aa430810ee25ee83e03d
Author: Russell Branca <ch...@apache.org>
AuthorDate: Fri Nov 17 16:01:21 2023 -0800
Rework csrt context setting
---
src/fabric/src/fabric_rpc.erl | 22 ++++++++--------------
1 file changed, 8 insertions(+), 14 deletions(-)
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 2c8ba1654..6a8b069ea 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -70,7 +70,6 @@ changes(DbName, #changes_args{} = Args, StartSeq, DbOptions) ->
changes(DbName, [Args], StartSeq, DbOptions);
changes(DbName, Options, StartVector, DbOptions) ->
set_io_priority(DbName, DbOptions),
- couch_stats_resource_tracker:set_context_dbname(DbName),
Args0 = lists:keyfind(changes_args, 1, Options),
#changes_args{dir = Dir, filter_fun = Filter} = Args0,
Args =
@@ -147,7 +146,6 @@ all_docs(DbName, Options, Args0) ->
case fabric_util:upgrade_mrargs(Args0) of
#mrargs{keys = undefined} = Args ->
set_io_priority(DbName, Options),
- couch_stats_resource_tracker:set_context_dbname(DbName),
{ok, Db} = get_or_create_db(DbName, Options),
CB = get_view_cb(Args),
couch_mrview:query_all_docs(Db, Args, CB, Args)
@@ -172,7 +170,6 @@ map_view(DbName, {DDocId, Rev}, ViewName, Args0, DbOptions) ->
map_view(DbName, DDoc, ViewName, Args0, DbOptions);
map_view(DbName, DDoc, ViewName, Args0, DbOptions) ->
set_io_priority(DbName, DbOptions),
- couch_stats_resource_tracker:set_context_dbname(DbName),
Args = fabric_util:upgrade_mrargs(Args0),
{ok, Db} = get_or_create_db(DbName, DbOptions),
CB = get_view_cb(Args),
@@ -187,7 +184,6 @@ reduce_view(DbName, {DDocId, Rev}, ViewName, Args0, DbOptions) ->
reduce_view(DbName, DDoc, ViewName, Args0, DbOptions);
reduce_view(DbName, DDoc, ViewName, Args0, DbOptions) ->
set_io_priority(DbName, DbOptions),
- couch_stats_resource_tracker:set_context_dbname(DbName),
Args = fabric_util:upgrade_mrargs(Args0),
{ok, Db} = get_or_create_db(DbName, DbOptions),
VAcc0 = #vacc{db = Db},
@@ -331,7 +327,6 @@ reset_validation_funs(DbName) ->
open_shard(Name, Opts) ->
set_io_priority(Name, Opts),
- couch_stats_resource_tracker:set_context_dbname(Name),
try
rexi:reply(mem3_util:get_or_create_db(Name, Opts))
catch
@@ -358,14 +353,6 @@ get_uuid(DbName) ->
with_db(DbName, Options, {M, F, A}) ->
set_io_priority(DbName, Options),
- couch_stats_resource_tracker: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)
- end,
case get_or_create_db(DbName, Options) of
{ok, Db} ->
rexi:reply(
@@ -391,7 +378,6 @@ with_db(DbName, Options, {M, F, A}) ->
read_repair_filter(DbName, Docs, NodeRevs, Options) ->
set_io_priority(DbName, Options),
- couch_stats_resource_tracker:set_context_dbname(DbName),
case get_or_create_db(DbName, Options) of
{ok, Db} ->
try
@@ -682,6 +668,14 @@ clean_stack(S) ->
).
set_io_priority(DbName, Options) ->
+ couch_stats_resource_tracker: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)
+ end,
case lists:keyfind(io_priority, 1, Options) of
{io_priority, Pri} ->
erlang:put(io_priority, Pri);
(couchdb) 06/06: Make CSRT toggle-able and rework delta sending
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 4c05a0f39a2f4d7e372c80f8701678c006b0df3e
Author: Russell Branca <ch...@apache.org>
AuthorDate: Mon Nov 20 14:25:32 2023 -0800
Make CSRT toggle-able and rework delta sending
---
.../src/couch_stats_resource_tracker.erl | 163 +++++++++++++--------
src/rexi/src/rexi.erl | 27 +++-
src/rexi/src/rexi_utils.erl | 27 +++-
3 files changed, 143 insertions(+), 74 deletions(-)
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.erl b/src/couch_stats/src/couch_stats_resource_tracker.erl
index 5097f2866..fb0d39cd3 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.erl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.erl
@@ -34,6 +34,7 @@
-export([
create_context/0, create_context/1, create_context/3,
create_coordinator_context/1, create_coordinator_context/2,
+ is_enabled/0,
get_resource/0,
get_resource/1,
set_context_dbname/1,
@@ -175,11 +176,15 @@
get_kp_node = 0
}).
+is_enabled() ->
+ config:get_boolean("couch_stats_resource_tracker", "enabled", true).
+
db_opened() -> inc(db_opened).
doc_read() -> inc(docs_read).
row_read() -> inc(rows_read).
btree_fold() -> inc(?COUCH_BT_FOLDS).
-ioq_called() -> inc(ioq_calls).
+%% TODO: do we need ioq_called and this access pattern?
+ioq_called() -> is_enabled() andalso inc(ioq_calls).
js_evaled() -> inc(js_evals).
js_filtered() -> inc(js_filter).
js_filtered_error() -> inc(js_filter_error).
@@ -290,29 +295,29 @@ maybe_inc(_Metric, _Val) ->
%% TODO: update stats_descriptions.cfg for relevant apps
should_track([fabric_rpc, all_docs, spawned]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, changes, spawned]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, changes, processed]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, changes, returned]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, map_view, spawned]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, reduce_view, spawned]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, get_all_security, spawned]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, open_doc, spawned]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, update_docs, spawned]) ->
- true;
+ is_enabled();
should_track([fabric_rpc, open_shard, spawned]) ->
- true;
+ is_enabled();
should_track([mango_cursor, view, all_docs]) ->
- true;
+ is_enabled();
should_track([mango_cursor, view, idx]) ->
- true;
+ is_enabled();
should_track(_Metric) ->
%%io:format("SKIPPING METRIC: ~p~n", [Metric]),
false.
@@ -320,14 +325,14 @@ should_track(_Metric) ->
accumulate_delta(Delta) ->
%% TODO: switch to creating a batch of updates to invoke a single
%% update_counter rather than sequentially invoking it for each field
- maps:foreach(fun inc/2, Delta).
+ is_enabled() andalso maps:foreach(fun inc/2, Delta).
update_counter(Field, Count) ->
- update_counter(get_pid_ref(), Field, Count).
+ is_enabled() andalso update_counter(get_pid_ref(), Field, Count).
update_counter({_Pid,_Ref}=PidRef, Field, Count) ->
- ets:update_counter(?MODULE, PidRef, {Field, Count}, #rctx{pid_ref=PidRef}).
+ is_enabled() andalso ets:update_counter(?MODULE, PidRef, {Field, Count}, #rctx{pid_ref=PidRef}).
active() -> active_int(all).
@@ -622,73 +627,98 @@ get_pid_ref() ->
create_context() ->
- create_context(self()).
+ is_enabled() andalso create_context(self()).
create_context(Pid) ->
- Ref = make_ref(),
- Rctx = make_record(Pid, Ref),
- track(Rctx),
- ets:insert(?MODULE, Rctx),
- Rctx.
+ case is_enabled() of
+ false ->
+ ok;
+ true ->
+ Ref = make_ref(),
+ Rctx = make_record(Pid, Ref),
+ track(Rctx),
+ ets:insert(?MODULE, Rctx),
+ Rctx
+ end.
%% add type to disnguish coordinator vs rpc_worker
create_context(From, {M,F,_A} = MFA, Nonce) ->
- PidRef = get_pid_ref(), %% this will instantiate a new PidRef
- %% TODO: extract user_ctx and db/shard from
- Rctx = #rctx{
- pid_ref = PidRef,
- from = From,
- mfa = MFA,
- type = {worker, M, F},
- nonce = Nonce
- },
- track(Rctx),
- erlang:put(?DELTA_TZ, Rctx),
- true = ets:insert(?MODULE, Rctx),
- Rctx.
+ case is_enabled() of
+ false ->
+ ok;
+ true ->
+ PidRef = get_pid_ref(), %% this will instantiate a new PidRef
+ %% TODO: extract user_ctx and db/shard from
+ Rctx = #rctx{
+ pid_ref = PidRef,
+ from = From,
+ mfa = MFA,
+ type = {worker, M, F},
+ nonce = Nonce
+ },
+ track(Rctx),
+ erlang:put(?DELTA_TZ, Rctx),
+ true = ets:insert(?MODULE, Rctx),
+ Rctx
+ end.
create_coordinator_context(#httpd{path_parts=Parts} = Req) ->
- create_coordinator_context(Req, io_lib:format("~p", [Parts])).
+ is_enabled() andalso create_coordinator_context(Req, io_lib:format("~p", [Parts])).
create_coordinator_context(#httpd{} = Req, Path) ->
- #httpd{
- method = Verb,
- nonce = Nonce
- } = Req,
- PidRef = get_pid_ref(), %% this will instantiate a new PidRef
- Rctx = #rctx{
- pid_ref = PidRef,
- type = {coordinator, Verb, [$/ | Path]},
- nonce = Nonce
- },
- track(Rctx),
- erlang:put(?DELTA_TZ, Rctx),
- true = ets:insert(?MODULE, Rctx),
- Rctx.
+ case is_enabled() of
+ false ->
+ ok;
+ true ->
+ #httpd{
+ method = Verb,
+ nonce = Nonce
+ } = Req,
+ PidRef = get_pid_ref(), %% this will instantiate a new PidRef
+ Rctx = #rctx{
+ pid_ref = PidRef,
+ type = {coordinator, Verb, [$/ | Path]},
+ nonce = Nonce
+ },
+ track(Rctx),
+ erlang:put(?DELTA_TZ, Rctx),
+ true = ets:insert(?MODULE, Rctx),
+ Rctx
+ end.
set_context_dbname(DbName) ->
- case ets:update_element(?MODULE, get_pid_ref(), [{#rctx.dbname, DbName}]) of
+ case is_enabled() of
false ->
- Stk = try throw(42) catch _:_:Stk0 -> Stk0 end,
- io:format("UPDATING DBNAME[~p] FAILURE WITH CONTEXT: ~p AND STACK:~n~pFOO:: ~p~n~n", [DbName, get_resource(), Stk, process_info(self(), current_stacktrace)]),
- timer:sleep(1000),
- erlang:halt(kaboomz);
+ ok;
true ->
- true
+ case ets:update_element(?MODULE, get_pid_ref(), [{#rctx.dbname, DbName}]) of
+ false ->
+ Stk = try throw(42) catch _:_:Stk0 -> Stk0 end,
+ io:format("UPDATING DBNAME[~p] FAILURE WITH CONTEXT: ~p AND STACK:~n~pFOO:: ~p~n~n", [DbName, get_resource(), Stk, process_info(self(), current_stacktrace)]),
+ timer:sleep(1000),
+ erlang:halt(kaboomz);
+ true ->
+ true
+ end
end.
set_context_username(null) ->
ok;
set_context_username(UserName) ->
- case ets:update_element(?MODULE, get_pid_ref(), [{#rctx.username, UserName}]) of
+ case is_enabled() of
false ->
- Stk = try throw(42) catch _:_:Stk0 -> Stk0 end,
- io:format("UPDATING USERNAME[~p] FAILURE WITH CONTEXT: ~p AND STACK:~n~pFOO:: ~p~n~n", [UserName, get_resource(), Stk, process_info(self(), current_stacktrace)]),
- timer:sleep(1000),
- erlang:halt(kaboomz);
+ ok;
true ->
- true
+ case ets:update_element(?MODULE, get_pid_ref(), [{#rctx.username, UserName}]) of
+ false ->
+ Stk = try throw(42) catch _:_:Stk0 -> Stk0 end,
+ io:format("UPDATING USERNAME[~p] FAILURE WITH CONTEXT: ~p AND STACK:~n~pFOO:: ~p~n~n", [UserName, get_resource(), Stk, process_info(self(), current_stacktrace)]),
+ timer:sleep(1000),
+ erlang:halt(kaboomz);
+ true ->
+ true
+ end
end.
track(#rctx{}=Rctx) ->
@@ -830,7 +860,12 @@ init([]) ->
{keypos, #rctx.pid_ref}
]),
St = #st{},
- _TimerRef = erlang:send_after(St#st.scan_interval, self(), scan),
+ case is_enabled() of
+ false ->
+ ok;
+ true ->
+ _TimerRef = erlang:send_after(St#st.scan_interval, self(), scan)
+ end,
{ok, St}.
handle_call(fetch, _from, #st{} = St) ->
@@ -852,7 +887,7 @@ handle_info(scan, #st{tracking=AT0} = St0) ->
AT = maybe_track(Unmonitored, AT0),
_TimerRef = erlang:send_after(St0#st.scan_interval, self(), scan),
{noreply, St0#st{tracking=AT}};
-handle_info({'DOWN', MonRef, Type, DPid, Reason}, #st{tracking=AT0} = St0) ->
+handle_info({'DOWN', MonRef, _Type, DPid, Reason}, #st{tracking=AT0} = St0) ->
%% io:format("CSRT:HI(~p)~n", [{'DOWN', MonRef, Type, DPid, Reason}]),
St = case maps:get(MonRef, AT0, undefined) of
undefined ->
diff --git a/src/rexi/src/rexi.erl b/src/rexi/src/rexi.erl
index 0b005dace..4bb5e9f69 100644
--- a/src/rexi/src/rexi.erl
+++ b/src/rexi/src/rexi.erl
@@ -129,7 +129,8 @@ async_server_call(Server, Caller, Request) ->
-spec reply(any()) -> any().
reply(Reply) ->
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {Ref, Reply, get_delta()}).
+ %%erlang:send(Caller, {Ref, Reply, get_delta()}).
+ erlang:send(Caller, maybe_add_delta({Ref, Reply})).
%% @equiv sync_reply(Reply, 300000)
sync_reply(Reply) ->
@@ -214,7 +215,8 @@ stream(Msg, Limit, Timeout) ->
{ok, Count} ->
put(rexi_unacked, Count + 1),
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {Ref, self(), Msg, get_delta()}),
+ %%erlang:send(Caller, {Ref, self(), Msg, get_delta()}),
+ erlang:send(Caller, maybe_add_delta({Ref, self(), Msg})),
ok
catch
throw:timeout ->
@@ -243,7 +245,7 @@ stream2(Msg, Limit, Timeout) ->
{ok, Count} ->
put(rexi_unacked, Count + 1),
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {Ref, self(), Msg, get_delta()}),
+ erlang:send(Caller, maybe_add_delta({Ref, self(), Msg})),
ok
catch
throw:timeout ->
@@ -281,7 +283,7 @@ ping() ->
%% filtered queries will be silent on usage until they finally return
%% a row or no results. This delay is proportional to the database size,
%% so instead we make sure ping/0 keeps live stats flowing.
- erlang:send(Caller, {rexi, '$rexi_ping'}, get_delta()).
+ erlang:send(Caller, maybe_add_delta({rexi, '$rexi_ping'})).
%% internal functions %%
@@ -337,3 +339,20 @@ drain_acks(Count) ->
get_delta() ->
{delta, couch_stats_resource_tracker:make_delta()}.
+
+maybe_add_delta(T) ->
+ case couch_stats_resource_tracker:is_enabled() of
+ false ->
+ T;
+ true ->
+ add_delta(T, get_delta())
+ end.
+
+add_delta({A}, Delta) -> {A, Delta};
+add_delta({A, B}, Delta) -> {A, B, Delta};
+add_delta({A, B, C}, Delta) -> {A, B, C, Delta};
+add_delta({A, B, C, D}, Delta) -> {A, B, C, D, Delta};
+add_delta({A, B, C, D, E}, Delta) -> {A, B, C, D, E, Delta};
+add_delta({A, B, C, D, E, F}, Delta) -> {A, B, C, D, E, F, Delta};
+add_delta({A, B, C, D, E, F, G}, Delta) -> {A, B, C, D, E, F, G, Delta};
+add_delta(T, _Delta) -> T.
diff --git a/src/rexi/src/rexi_utils.erl b/src/rexi/src/rexi_utils.erl
index f18ca179f..4924bd11e 100644
--- a/src/rexi/src/rexi_utils.erl
+++ b/src/rexi/src/rexi_utils.erl
@@ -67,6 +67,8 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
receive
{timeout, TimeoutRef} ->
{timeout, Acc0};
+ {rexi, '$rexi_ping'} ->
+ {ok, Acc0};
{rexi, '$rexi_ping', {delta, Delta}} ->
couch_stats_resource_tracker:accumulate_delta(Delta),
{ok, Acc0};
@@ -120,9 +122,16 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
%% {#Ref<0.1961702128.3491758082.26965>, {rexi_EXIT,{{query_parse_error,<<78,111,32,114,111,119,115,32,99,97,110,32,109,97
{Ref, Msg} ->
%% TODO: add stack trace to log entry
- couch_log:debug("rexi_utils:process_message no delta: {Ref, Msg} => {~p, ~p}~n", [Ref, Msg]),
- timer:sleep(100),
- erlang:halt(binary_to_list(iolist_to_binary(io_lib:format("{enodelta} rexi_utils:process_message no delta: {Ref, Msg} => {~w, ~w}~n", [Ref, Msg]))));
+ %% couch_log:debug("rexi_utils:process_message no delta: {Ref, Msg} => {~p, ~p}~n", [Ref, Msg]),
+ %% timer:sleep(100),
+ %% erlang:halt(binary_to_list(iolist_to_binary(io_lib:format("{enodelta} rexi_utils:process_message no delta: {Ref, Msg} => {~w, ~w}~n", [Ref, Msg]))));
+ case lists:keyfind(Ref, Keypos, RefList) of
+ false ->
+ % this was some non-matching message which we will ignore
+ {ok, Acc0};
+ Worker ->
+ Fun(Msg, Worker, Acc0)
+ end;
{Ref, From, rexi_STREAM_INIT = Msg} ->
case lists:keyfind(Ref, Keypos, RefList) of
false ->
@@ -132,9 +141,15 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
end;
{Ref, From, Msg} ->
%% TODO: add stack trace to log entry
- couch_log:debug("rexi_utils:process_message no delta: {Ref, From, Msg} => {~p, ~p, ~p}~n", [Ref, From, Msg]),
- timer:sleep(100),
- erlang:halt(binary_to_list(iolist_to_binary(io_lib:format("{enodelta} rexi_utils:process_message no delta: {Ref, From, Msg} => {~w, ~w, ~w}~n", [Ref, From, Msg]))));
+ %% couch_log:debug("rexi_utils:process_message no delta: {Ref, From, Msg} => {~p, ~p, ~p}~n", [Ref, From, Msg]),
+ %% timer:sleep(100),
+ %% erlang:halt(binary_to_list(iolist_to_binary(io_lib:format("{enodelta} rexi_utils:process_message no delta: {Ref, From, Msg} => {~w, ~w, ~w}~n", [Ref, From, Msg]))));
+ case lists:keyfind(Ref, Keypos, RefList) of
+ false ->
+ {ok, Acc0};
+ Worker ->
+ Fun(Msg, {Worker, From}, Acc0)
+ end;
{rexi_DOWN, _, _, _} = Msg ->
Fun(Msg, nil, Acc0)
after PerMsgTO ->
(couchdb) 01/06: Set dbname context properly
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 996c38551ae2a10ee16371bab30f5a06128aa102
Author: Russell Branca <ch...@apache.org>
AuthorDate: Wed Nov 15 15:08:35 2023 -0800
Set dbname context properly
---
src/fabric/src/fabric_rpc.erl | 6 ++++++
1 file changed, 6 insertions(+)
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index b5f41782a..bfd039aad 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -70,6 +70,7 @@ changes(DbName, #changes_args{} = Args, StartSeq, DbOptions) ->
changes(DbName, [Args], StartSeq, DbOptions);
changes(DbName, Options, StartVector, DbOptions) ->
set_io_priority(DbName, DbOptions),
+ couch_stats_resource_tracker:set_context_dbname(DbName),
Args0 = lists:keyfind(changes_args, 1, Options),
#changes_args{dir = Dir, filter_fun = Filter} = Args0,
Args =
@@ -146,6 +147,7 @@ all_docs(DbName, Options, Args0) ->
case fabric_util:upgrade_mrargs(Args0) of
#mrargs{keys = undefined} = Args ->
set_io_priority(DbName, Options),
+ couch_stats_resource_tracker:set_context_dbname(DbName),
{ok, Db} = get_or_create_db(DbName, Options),
CB = get_view_cb(Args),
couch_mrview:query_all_docs(Db, Args, CB, Args)
@@ -170,6 +172,7 @@ map_view(DbName, {DDocId, Rev}, ViewName, Args0, DbOptions) ->
map_view(DbName, DDoc, ViewName, Args0, DbOptions);
map_view(DbName, DDoc, ViewName, Args0, DbOptions) ->
set_io_priority(DbName, DbOptions),
+ couch_stats_resource_tracker:set_context_dbname(DbName),
Args = fabric_util:upgrade_mrargs(Args0),
{ok, Db} = get_or_create_db(DbName, DbOptions),
CB = get_view_cb(Args),
@@ -184,6 +187,7 @@ reduce_view(DbName, {DDocId, Rev}, ViewName, Args0, DbOptions) ->
reduce_view(DbName, DDoc, ViewName, Args0, DbOptions);
reduce_view(DbName, DDoc, ViewName, Args0, DbOptions) ->
set_io_priority(DbName, DbOptions),
+ couch_stats_resource_tracker:set_context_dbname(DbName),
Args = fabric_util:upgrade_mrargs(Args0),
{ok, Db} = get_or_create_db(DbName, DbOptions),
VAcc0 = #vacc{db = Db},
@@ -327,6 +331,7 @@ reset_validation_funs(DbName) ->
open_shard(Name, Opts) ->
set_io_priority(Name, Opts),
+ couch_stats_resource_tracker:set_context_dbname(DbName),
try
rexi:reply(mem3_util:get_or_create_db(Name, Opts))
catch
@@ -386,6 +391,7 @@ with_db(DbName, Options, {M, F, A}) ->
read_repair_filter(DbName, Docs, NodeRevs, Options) ->
set_io_priority(DbName, Options),
+ couch_stats_resource_tracker:set_context_dbname(DbName),
case get_or_create_db(DbName, Options) of
{ok, Db} ->
try
(couchdb) 05/06: Cleanup debug info
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit be8ab045d93b72aa1da4cbcb84d71054cf5d03ee
Author: Russell Branca <ch...@apache.org>
AuthorDate: Mon Nov 20 13:45:29 2023 -0800
Cleanup debug info
---
src/chttpd/src/chttpd.erl | 13 +-------
src/chttpd/src/chttpd_misc.erl | 2 +-
src/couch/priv/stats_descriptions.cfg | 2 +-
.../src/couch_stats_resource_tracker.erl | 36 +++-------------------
src/fabric/src/fabric_rpc.erl | 1 -
src/fabric/src/fabric_util.erl | 2 --
src/rexi/src/rexi_utils.erl | 6 ----
7 files changed, 7 insertions(+), 55 deletions(-)
diff --git a/src/chttpd/src/chttpd.erl b/src/chttpd/src/chttpd.erl
index 688541ae8..78b41a710 100644
--- a/src/chttpd/src/chttpd.erl
+++ b/src/chttpd/src/chttpd.erl
@@ -313,8 +313,6 @@ handle_request_int(MochiReq) ->
]
},
- io:format("CSRTZ PROCESSING: ~w~n", [RequestedPath]),
-
% put small token on heap to keep requests synced to backend calls
erlang:put(nonce, Nonce),
@@ -326,22 +324,15 @@ handle_request_int(MochiReq) ->
chttpd_util:mochiweb_client_req_set(MochiReq),
%% This is probably better in before_request, but having Path is nice
- io:format("CSRTZ INITIATING CONTEXT: ~w~n", [Nonce]),
couch_stats_resource_tracker:create_coordinator_context(HttpReq0, Path),
- io:format("CSRTZ INITIAL CONTEXT: ~w~n", [couch_stats_resource_tracker:get_resource()]),
- Coord = self(),
- PidRef = couch_stats_resource_tracker:get_pid_ref(),
- spawn(fun() -> monitor(process, Coord), receive M -> io:format("CSRTZ PROCESS DOWN[~w]{~w}: ~w~n", [ Coord, M, couch_stats_resource_tracker:get_resource(PidRef)]) end end),
+
{HttpReq2, Response} =
case before_request(HttpReq0) of
{ok, HttpReq1} ->
- io:format("CSRTZ BEFORE_REQUEST OK: ~w~n", [couch_stats_resource_tracker:get_resource()]),
process_request(HttpReq1);
{error, Response0} ->
- io:format("CSRTZ BEFORE_REQUEST ERROR: ~w~n", [couch_stats_resource_tracker:get_resource()]),
{HttpReq0, Response0}
end,
- io:format("CSRTZ AFTER PROCESS_REQUEST: ~w~n", [couch_stats_resource_tracker:get_resource()]),
chttpd_util:mochiweb_client_req_clean(),
@@ -357,10 +348,8 @@ handle_request_int(MochiReq) ->
case after_request(HttpReq2, HttpResp) of
#httpd_resp{status = ok, response = Resp} ->
- io:format("CSRTZ AFTER REQUEST OK: ~w~n", [couch_stats_resource_tracker:get_resource()]),
{ok, Resp};
#httpd_resp{status = aborted, reason = Reason} ->
- io:format("CSRTZ AFTER REQUEST ERROR: ~w~n", [couch_stats_resource_tracker:get_resource()]),
couch_log:error("Response abnormally terminated: ~w", [Reason]),
exit({shutdown, Reason})
end.
diff --git a/src/chttpd/src/chttpd_misc.erl b/src/chttpd/src/chttpd_misc.erl
index 28bbcbee0..eff4c93c1 100644
--- a/src/chttpd/src/chttpd_misc.erl
+++ b/src/chttpd/src/chttpd_misc.erl
@@ -314,7 +314,7 @@ handle_resource_status_req(#httpd{method = 'POST'} = Req) ->
end,
[]
]),
- io:format("{CSRT}***** GOT RESP: ~p~n", [Resp]),
+ %%io:format("{CSRT}***** GOT RESP: ~p~n", [Resp]),
send_json(Req, {Resp});
handle_resource_status_req(#httpd{method = 'GET'} = Req) ->
ok = chttpd:verify_is_server_admin(Req),
diff --git a/src/couch/priv/stats_descriptions.cfg b/src/couch/priv/stats_descriptions.cfg
index 02740e4d8..43794bff6 100644
--- a/src/couch/priv/stats_descriptions.cfg
+++ b/src/couch/priv/stats_descriptions.cfg
@@ -418,7 +418,7 @@
{type, counter},
{desc, <<"number of JS filter invocations">>}
]}.
-{[couchdb, query_server, js_filterered], [
+{[couchdb, query_server, js_filtered_docs], [
{type, counter},
{desc, <<"number of docs filtered through JS invocations">>}
]}.
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.erl b/src/couch_stats/src/couch_stats_resource_tracker.erl
index d5fcd92ae..5097f2866 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.erl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.erl
@@ -317,14 +317,7 @@ should_track(_Metric) ->
%%io:format("SKIPPING METRIC: ~p~n", [Metric]),
false.
-%% TODO: update coordinator stats from worker deltas
accumulate_delta(Delta) ->
- case get_resource() of
- #rctx{type={coordinator,_,_}} = Rctx ->
- io:format("Accumulating delta: ~w || ~w~n", [Delta, Rctx]);
- _ ->
- ok
- end,
%% TODO: switch to creating a batch of updates to invoke a single
%% update_counter rather than sequentially invoking it for each field
maps:foreach(fun inc/2, Delta).
@@ -463,7 +456,6 @@ to_json(#rctx{}=Rctx) ->
changes_processed = ChangesProcessed,
changes_returned = ChangesReturned
} = Rctx,
- %%io:format("TO_JSON_MFA: ~p~n", [MFA0]),
MFA = case MFA0 of
{M, F, A} ->
[M, F, A];
@@ -495,11 +487,9 @@ to_json(#rctx{}=Rctx) ->
#{
updated_at => term_to_json(TP),
started_at => term_to_json(TInit),
- %%pid_ref => [pid_to_list(Pid), ref_to_list(Ref)],
pid_ref => term_to_json(PidRef),
mfa => term_to_json(MFA),
nonce => term_to_json(Nonce),
- %%from => From,
from => term_to_json(From),
dbname => DbName,
username => UserName,
@@ -522,7 +512,6 @@ to_json(#rctx{}=Rctx) ->
term_to_json({Pid, Ref}) when is_pid(Pid), is_reference(Ref) ->
[?l2b(pid_to_list(Pid)), ?l2b(ref_to_list(Ref))];
term_to_json({type, {coordinator, _, _} = Type}) ->
- %%io:format("SETTING JSON TYPE: ~p~n", [Type]),
?l2b(io_lib:format("~p", [Type]));
term_to_json({A, B, C}) ->
[A, B, C];
@@ -536,7 +525,6 @@ term_to_json(T) ->
T.
term_to_flat_json({type, {coordinator, _, _}}=Type) ->
- %%io:format("SETTING FLAT JSON TYPE: ~p~n", [Type]),
?l2b(io_lib:format("~p", [Type]));
term_to_flat_json({coordinator, _, _}=Type) ->
?l2b(io_lib:format("~p", [Type]));
@@ -573,7 +561,6 @@ to_flat_json(#rctx{}=Rctx) ->
changes_returned = ChangesReturned,
ioq_calls = IoqCalls
} = Rctx,
- %%io:format("TO_JSON_MFA: ~p~n", [MFA0]),
MFA = case MFA0 of
{_M, _F, _A} ->
?l2b(io_lib:format("~w", [MFA0]));
@@ -600,12 +587,9 @@ to_flat_json(#rctx{}=Rctx) ->
Nonce0 ->
list_to_binary(Nonce0)
end,
- io:format("NONCE IS: ~p||~p~n", [Nonce0, Nonce]),
#{
- %%updated_at => ?l2b(io_lib:format("~w", [TP])),
updated_at => term_to_flat_json(TP),
started_at => term_to_flat_json(TInit),
- %%pid_ref => [pid_to_list(Pid), ref_to_list(Ref)],
pid_ref => ?l2b(io_lib:format("~w", [PidRef])),
mfa => MFA,
nonce => Nonce,
@@ -650,10 +634,8 @@ create_context(Pid) ->
%% add type to disnguish coordinator vs rpc_worker
create_context(From, {M,F,_A} = MFA, Nonce) ->
- %%io:format("[~p] CREAT_CONTEXT MFA[~p]: {~p}: ~p~n", [self(), From, MFA, Nonce]),
PidRef = get_pid_ref(), %% this will instantiate a new PidRef
- %%Rctx = make_record(self(), Ref),
- %% TODO: extract user_ctx and db/shard from
+ %% TODO: extract user_ctx and db/shard from
Rctx = #rctx{
pid_ref = PidRef,
from = From,
@@ -670,18 +652,13 @@ create_coordinator_context(#httpd{path_parts=Parts} = Req) ->
create_coordinator_context(Req, io_lib:format("~p", [Parts])).
create_coordinator_context(#httpd{} = Req, Path) ->
- %%io:format("CREATING COORDINATOR CONTEXT ON {~p}~n", [Path]),
#httpd{
method = Verb,
- %%path_parts = Parts,
nonce = Nonce
} = Req,
PidRef = get_pid_ref(), %% this will instantiate a new PidRef
- %%Rctx = make_record(self(), Ref),
- %% TODO: extract user_ctx and db/shard from Req
Rctx = #rctx{
pid_ref = PidRef,
- %%type = {cooridantor, Verb, Parts},
type = {coordinator, Verb, [$/ | Path]},
nonce = Nonce
},
@@ -704,7 +681,6 @@ set_context_dbname(DbName) ->
set_context_username(null) ->
ok;
set_context_username(UserName) ->
- %%io:format("CSRT SETTING USERNAME CONTEXT: ~p~n", [UserName]),
case ets:update_element(?MODULE, get_pid_ref(), [{#rctx.username, UserName}]) of
false ->
Stk = try throw(42) catch _:_:Stk0 -> Stk0 end,
@@ -755,7 +731,7 @@ make_delta() ->
%% 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", []),
+ %% io:format("~n**********MISSING STARTING DELTA************~n~n", []),
couch_stats:increment_counter(
[couchdb, csrt, delta_missing_t0]),
%%[couch_stats_resource_tracker, delta_missing_t0]),
@@ -769,7 +745,6 @@ make_delta() ->
TA0 ->
TA0
end;
- %%?RCTX{} = TA0 ->
#rctx{} = TA0 ->
TA0
end,
@@ -804,7 +779,6 @@ make_delta(#rctx{}, _) ->
make_delta_base() ->
Ref = make_ref(),
- %%Rctx = make_record(self(), Ref),
%% TODO: extract user_ctx and db/shard from request
TA0 = #rctx{
pid_ref = {self(), Ref}
@@ -874,14 +848,12 @@ handle_cast(Msg, St) ->
{stop, {unknown_cast, Msg}, St}.
handle_info(scan, #st{tracking=AT0} = St0) ->
- io:format("{CSRT} TRIGGERING SCAN: ~n", []),
Unmonitored = find_unmonitored(),
- io:format("{CSRT} SCAN FOUND ~p UNMONITORED PIDS: ~n", [length(Unmonitored)]),
AT = maybe_track(Unmonitored, AT0),
_TimerRef = erlang:send_after(St0#st.scan_interval, self(), scan),
{noreply, St0#st{tracking=AT}};
handle_info({'DOWN', MonRef, Type, DPid, Reason}, #st{tracking=AT0} = St0) ->
- io:format("CSRT:HI(~p)~n", [{'DOWN', MonRef, Type, DPid, Reason}]),
+ %% io:format("CSRT:HI(~p)~n", [{'DOWN', MonRef, Type, DPid, Reason}]),
St = case maps:get(MonRef, AT0, undefined) of
undefined ->
io:format("ERROR: UNEXPECTED MISSING MONITOR IN TRACKING TABLE: {~p, ~p}~n", [MonRef, DPid]),
@@ -937,7 +909,7 @@ log_process_lifetime_report(PidRef) ->
%% More safely assert this can't ever be undefined
#rctx{} = Rctx = get_resource(PidRef),
%% TODO: catch error out of here, report crashes on depth>1 json
- io:format("CSRT RCTX: ~p~n", [to_flat_json(Rctx)]),
+ %%io:format("CSRT RCTX: ~p~n", [to_flat_json(Rctx)]),
couch_log:report("csrt-pid-usage-lifetime", to_flat_json(Rctx)).
diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl
index 6a8b069ea..6fa990e71 100644
--- a/src/fabric/src/fabric_rpc.erl
+++ b/src/fabric/src/fabric_rpc.erl
@@ -262,7 +262,6 @@ set_purge_infos_limit(DbName, Limit, Options) ->
with_db(DbName, Options, {couch_db, set_purge_infos_limit, [Limit]}).
open_doc(DbName, DocId, Options) ->
- io:format("frpc:open_doc(~p, ~p, ~p)~n", [DbName, DocId, Options]),
with_db(DbName, Options, {couch_db, open_doc, [DocId, Options]}).
open_revs(DbName, IdRevsOpts, Options) ->
diff --git a/src/fabric/src/fabric_util.erl b/src/fabric/src/fabric_util.erl
index 00d6b5e0e..2bde50a29 100644
--- a/src/fabric/src/fabric_util.erl
+++ b/src/fabric/src/fabric_util.erl
@@ -138,7 +138,6 @@ get_shard([#shard{node = Node, name = Name} | Rest], Opts, Timeout, Factor) ->
try
receive
{Ref, {ok, Db}, {delta, Delta}} ->
- io:format("[~p]GET SHARD GOT DELTA: ~p~n", [self(), Delta]),
couch_stats_resource_tracker:accumulate_delta(Delta),
{ok, Db};
{Ref, {'rexi_EXIT', {{unauthorized, _} = Error, _}}} ->
@@ -149,7 +148,6 @@ get_shard([#shard{node = Node, name = Name} | Rest], Opts, Timeout, Factor) ->
couch_log:debug("Failed to open shard ~p because: ~p", [Name, Reason]),
get_shard(Rest, Opts, Timeout, Factor);
Other ->
- io:format("GOT UNEXPECTED MESSAGE FORMAT: ~p~n", [Other]),
erlang:error(Other)
after Timeout ->
couch_log:debug("Failed to open shard ~p after: ~p", [Name, Timeout]),
diff --git a/src/rexi/src/rexi_utils.erl b/src/rexi/src/rexi_utils.erl
index 43f3346d9..f18ca179f 100644
--- a/src/rexi/src/rexi_utils.erl
+++ b/src/rexi/src/rexi_utils.erl
@@ -68,7 +68,6 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
{timeout, TimeoutRef} ->
{timeout, Acc0};
{rexi, '$rexi_ping', {delta, Delta}} ->
- io:format("[~p]GOT PING DELTA: ~p~n", [self(), Delta]),
couch_stats_resource_tracker:accumulate_delta(Delta),
{ok, Acc0};
{rexi, Ref, Msg} ->
@@ -86,7 +85,6 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
Fun(Msg, {Worker, From}, Acc0)
end;
{Ref, Msg, {delta, Delta}} ->
- io:format("[~p]GOT DELTA: ~p -- ~p~n", [self(), Delta, Msg]),
couch_stats_resource_tracker:accumulate_delta(Delta),
case lists:keyfind(Ref, Keypos, RefList) of
false ->
@@ -96,7 +94,6 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
Fun(Msg, Worker, Acc0)
end;
{Ref, From, Msg, {delta, Delta}} ->
- io:format("[~p]GOT DELTA: ~p -- ~p~n", [self(), Delta, Msg]),
couch_stats_resource_tracker:accumulate_delta(Delta),
case lists:keyfind(Ref, Keypos, RefList) of
false ->
@@ -125,7 +122,6 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
%% TODO: add stack trace to log entry
couch_log:debug("rexi_utils:process_message no delta: {Ref, Msg} => {~p, ~p}~n", [Ref, Msg]),
timer:sleep(100),
- %%erlang:halt(enodelta),
erlang:halt(binary_to_list(iolist_to_binary(io_lib:format("{enodelta} rexi_utils:process_message no delta: {Ref, Msg} => {~w, ~w}~n", [Ref, Msg]))));
{Ref, From, rexi_STREAM_INIT = Msg} ->
case lists:keyfind(Ref, Keypos, RefList) of
@@ -135,11 +131,9 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
Fun(Msg, {Worker, From}, Acc0)
end;
{Ref, From, Msg} ->
- %%io:format("GOT NON DELTA MSG: ~p~n", [Msg]),
%% TODO: add stack trace to log entry
couch_log:debug("rexi_utils:process_message no delta: {Ref, From, Msg} => {~p, ~p, ~p}~n", [Ref, From, Msg]),
timer:sleep(100),
- %%erlang:halt(enodelta),
erlang:halt(binary_to_list(iolist_to_binary(io_lib:format("{enodelta} rexi_utils:process_message no delta: {Ref, From, Msg} => {~w, ~w, ~w}~n", [Ref, From, Msg]))));
{rexi_DOWN, _, _, _} = Msg ->
Fun(Msg, nil, Acc0)
(couchdb) 04/06: WIP: ugly but working HTTP API around groupings
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
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 715b524199cd2b1a4c35e26152c8d44d857db793
Author: Russell Branca <ch...@apache.org>
AuthorDate: Fri Nov 17 16:03:12 2023 -0800
WIP: ugly but working HTTP API around groupings
---
src/chttpd/src/chttpd_misc.erl | 92 +++++++++++++++++++++-
.../src/couch_stats_resource_tracker.erl | 13 ++-
2 files changed, 99 insertions(+), 6 deletions(-)
diff --git a/src/chttpd/src/chttpd_misc.erl b/src/chttpd/src/chttpd_misc.erl
index 195c8d828..28bbcbee0 100644
--- a/src/chttpd/src/chttpd_misc.erl
+++ b/src/chttpd/src/chttpd_misc.erl
@@ -37,8 +37,7 @@
[
send_json/2, send_json/3,
send_method_not_allowed/2,
- send_chunk/2,
- start_chunked_response/3
+ send_chunk/2
]
).
@@ -231,6 +230,92 @@ handle_task_status_req(#httpd{method = 'GET'} = Req) ->
handle_task_status_req(Req) ->
send_method_not_allowed(Req, "GET,HEAD").
+handle_resource_status_req(#httpd{method = 'POST'} = Req) ->
+ ok = chttpd:verify_is_server_admin(Req),
+ chttpd:validate_ctype(Req, "application/json"),
+ {Props} = chttpd:json_body_obj(Req),
+ Action = proplists:get_value(<<"action">>, Props),
+ Key = proplists:get_value(<<"key">>, Props),
+ Val = proplists:get_value(<<"val">>, Props),
+
+ CountBy = fun 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,
+ ConvertEle = fun(K) -> list_to_existing_atom(binary_to_list(K)) end,
+ ConvertList = fun(L) -> [ConvertEle(E) || E <- L] end,
+ ToJson = fun couch_stats_resource_tracker:term_to_flat_json/1,
+ JsonKeys = fun(PL) -> [[ToJson(K), V] || {K, V} <- PL] end,
+
+ Fun = case {Action, Key, Val} of
+ {<<"count_by">>, Keys, undefined} when is_list(Keys) ->
+ Keys1 = [ConvertEle(K) || K <- Keys],
+ fun() -> CountBy(Keys1) end;
+ {<<"count_by">>, Key, undefined} ->
+ Key1 = ConvertEle(Key),
+ fun() -> CountBy(Key1) end;
+ {<<"group_by">>, Keys, Vals} when is_list(Keys) andalso is_list(Vals) ->
+ Keys1 = ConvertList(Keys),
+ Vals1 = ConvertList(Vals),
+ fun() -> GroupBy(Keys1, Vals1) end;
+ {<<"group_by">>, Key, Vals} when is_list(Vals) ->
+ Key1 = ConvertEle(Key),
+ Vals1 = ConvertList(Vals),
+ fun() -> GroupBy(Key1, Vals1) end;
+ {<<"group_by">>, Keys, Val} when is_list(Keys) ->
+ Keys1 = ConvertList(Keys),
+ Val1 = ConvertEle(Val),
+ fun() -> GroupBy(Keys1, Val1) end;
+ {<<"group_by">>, Key, Val} ->
+ Key1 = ConvertEle(Key),
+ Val1 = ConvertList(Val),
+ fun() -> GroupBy(Key1, Val1) end;
+
+ {<<"sorted_by">>, Key, undefined} ->
+ Key1 = ConvertEle(Key),
+ fun() -> JsonKeys(SortedBy1(Key1)) end;
+ {<<"sorted_by">>, Keys, undefined} when is_list(Keys) ->
+ Keys1 = [ConvertEle(K) || K <- Keys],
+ fun() -> JsonKeys(SortedBy1(Keys1)) end;
+ {<<"sorted_by">>, Keys, Vals} when is_list(Keys) andalso is_list(Vals) ->
+ Keys1 = ConvertList(Keys),
+ Vals1 = ConvertList(Vals),
+ fun() -> JsonKeys(SortedBy2(Keys1, Vals1)) end;
+ {<<"sorted_by">>, Key, Vals} when is_list(Vals) ->
+ Key1 = ConvertEle(Key),
+ Vals1 = ConvertList(Vals),
+ fun() -> JsonKeys(SortedBy2(Key1, Vals1)) end;
+ {<<"sorted_by">>, Keys, Val} when is_list(Keys) ->
+ Keys1 = ConvertList(Keys),
+ Val1 = ConvertEle(Val),
+ fun() -> JsonKeys(SortedBy2(Keys1, Val1)) end;
+ {<<"sorted_by">>, Key, Val} ->
+ Key1 = ConvertEle(Key),
+ Val1 = ConvertList(Val),
+ fun() -> JsonKeys(SortedBy2(Key1, Val1)) end;
+ _ ->
+ throw({badrequest, invalid_resource_request})
+ end,
+
+ Fun1 = fun() ->
+ case Fun() of
+ Map when is_map(Map) ->
+ {maps:fold(
+ fun(K,V,A) -> [{ToJson(K), V} | A] end,
+ [], Map)};
+ List when is_list(List) ->
+ List
+ end
+ end,
+
+ {Resp, _Bad} = rpc:multicall(erlang, apply, [
+ fun() ->
+ {node(), Fun1()}
+ end,
+ []
+ ]),
+ io:format("{CSRT}***** GOT RESP: ~p~n", [Resp]),
+ send_json(Req, {Resp});
handle_resource_status_req(#httpd{method = 'GET'} = Req) ->
ok = chttpd:verify_is_server_admin(Req),
{Resp, Bad} = rpc:multicall(erlang, apply, [
@@ -243,7 +328,8 @@ handle_resource_status_req(#httpd{method = 'GET'} = Req) ->
io:format("ACTIVE RESP: ~p~nBAD RESP: ~p~n", [Resp, Bad]),
send_json(Req, {Resp});
handle_resource_status_req(Req) ->
- send_method_not_allowed(Req, "GET,HEAD").
+ ok = chttpd:verify_is_server_admin(Req),
+ send_method_not_allowed(Req, "GET,HEAD,POST").
handle_replicate_req(#httpd{method = 'POST', user_ctx = Ctx, req_body = PostBody} = Req) ->
chttpd:validate_ctype(Req, "application/json"),
diff --git a/src/couch_stats/src/couch_stats_resource_tracker.erl b/src/couch_stats/src/couch_stats_resource_tracker.erl
index d2dd4406c..d5fcd92ae 100644
--- a/src/couch_stats/src/couch_stats_resource_tracker.erl
+++ b/src/couch_stats/src/couch_stats_resource_tracker.erl
@@ -59,7 +59,9 @@
sorted_by/2,
sorted_by/3,
- unsafe_foldl/3
+ unsafe_foldl/3,
+
+ term_to_flat_json/1
]).
-export([
@@ -424,6 +426,7 @@ group_by(KeyFun, ValFun, AggFun, Fold) ->
Fold(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)).
@@ -523,6 +526,8 @@ term_to_json({type, {coordinator, _, _} = Type}) ->
?l2b(io_lib:format("~p", [Type]));
term_to_json({A, B, C}) ->
[A, B, C];
+term_to_json(Tuple) when is_tuple(Tuple) ->
+ ?l2b(io_lib:format("~w", [Tuple]));
term_to_json(undefined) ->
null;
term_to_json(null) ->
@@ -530,11 +535,13 @@ term_to_json(null) ->
term_to_json(T) ->
T.
-term_to_flat_json({type, {coordinator, _, _} = Type}) ->
+term_to_flat_json({type, {coordinator, _, _}}=Type) ->
%%io:format("SETTING FLAT JSON TYPE: ~p~n", [Type]),
?l2b(io_lib:format("~p", [Type]));
+term_to_flat_json({coordinator, _, _}=Type) ->
+ ?l2b(io_lib:format("~p", [Type]));
term_to_flat_json(Tuple) when is_tuple(Tuple) ->
- ?l2b(io_lib:format("~w", [Tuple]));
+ ?l2b(io_lib:format("~p", [Tuple]));
term_to_flat_json(undefined) ->
null;
term_to_flat_json(null) ->