PoC Couch Stats Resource Tracker for tracking process local resource usage

src/chttpd/src/chttpd.erl

@@ -323,6 +323,9 @@ handle_request_int(MochiReq) ->
     % Save client socket so that it can be monitored for disconnects
     chttpd_util:mochiweb_client_req_set(MochiReq),
 
+    %% This is probably better in before_request, but having Path is nice
+    couch_stats_resource_tracker:create_coordinator_context(HttpReq0, Path),
+
     {HttpReq2, Response} =
         case before_request(HttpReq0) of
             {ok, HttpReq1} ->
@@ -347,12 +350,14 @@ handle_request_int(MochiReq) ->
         #httpd_resp{status = ok, response = Resp} ->
             {ok, Resp};
         #httpd_resp{status = aborted, reason = Reason} ->
-            couch_log:error("Response abnormally terminated: ~p", [Reason]),
+            couch_log:error("Response abnormally terminated: ~w", [Reason]),
             exit({shutdown, Reason})
     end.
 
 before_request(HttpReq) ->
     try
+        %% TODO: re-enable this here once we have Path
+        %% couch_stats_resource_tracker:create_coordinator_context(HttpReq),
         chttpd_stats:init(),
         chttpd_plugin:before_request(HttpReq)
     catch

src/chttpd/src/chttpd_db.erl

@@ -83,6 +83,7 @@
 
 % Database request handlers
 handle_request(#httpd{path_parts = [DbName | RestParts], method = Method} = Req) ->
+    couch_stats_resource_tracker:set_context_dbname(DbName),
     case {Method, RestParts} of
         {'PUT', []} ->
             create_db_req(Req, DbName);

src/chttpd/src/chttpd_httpd_handlers.erl

@@ -20,6 +20,7 @@ url_handler(<<"_utils">>) -> fun chttpd_misc:handle_utils_dir_req/1;
 url_handler(<<"_all_dbs">>) -> fun chttpd_misc:handle_all_dbs_req/1;
 url_handler(<<"_dbs_info">>) -> fun chttpd_misc:handle_dbs_info_req/1;
 url_handler(<<"_active_tasks">>) -> fun chttpd_misc:handle_task_status_req/1;
+url_handler(<<"_active_resources">>) -> fun chttpd_misc:handle_resource_status_req/1;
 url_handler(<<"_scheduler">>) -> fun couch_replicator_httpd:handle_scheduler_req/1;
 url_handler(<<"_node">>) -> fun chttpd_node:handle_node_req/1;
 url_handler(<<"_reload_query_servers">>) -> fun chttpd_misc:handle_reload_query_servers_req/1;

src/chttpd/src/chttpd_misc.erl

@@ -20,6 +20,7 @@
     handle_replicate_req/1,
     handle_reload_query_servers_req/1,
     handle_task_status_req/1,
+    handle_resource_status_req/1,
     handle_up_req/1,
     handle_utils_dir_req/1,
     handle_utils_dir_req/2,
@@ -36,8 +37,7 @@
     [
         send_json/2, send_json/3,
         send_method_not_allowed/2,
-        send_chunk/2,
-        start_chunked_response/3
+        send_chunk/2
     ]
 ).
 
@@ -230,6 +230,107 @@ 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, [
+        fun() ->
+            {node(), couch_stats_resource_tracker:active()}
+        end,
+        []
+    ]),
+    %% TODO: incorporate Bad responses
+    io:format("ACTIVE RESP: ~p~nBAD RESP: ~p~n", [Resp, Bad]),
+    send_json(Req, {Resp});
+handle_resource_status_req(Req) ->
+    ok = chttpd:verify_is_server_admin(Req),
+    send_method_not_allowed(Req, "GET,HEAD,POST").
+
 handle_replicate_req(#httpd{method = 'POST', user_ctx = Ctx, req_body = PostBody} = Req) ->
     chttpd:validate_ctype(Req, "application/json"),
     %% see HACK in chttpd.erl about replication

src/config/src/config.erl

@@ -158,15 +158,20 @@ get(Section, Key) ->
 get(Section, Key, Default) when is_binary(Section) and is_binary(Key) ->
     ?MODULE:get(binary_to_list(Section), binary_to_list(Key), Default);
 get(Section, Key, Default) when is_list(Section), is_list(Key) ->
-    case ets:lookup(?MODULE, {Section, Key}) of
-        [] when Default == undefined -> Default;
-        [] when is_boolean(Default) -> Default;
-        [] when is_float(Default) -> Default;
-        [] when is_integer(Default) -> Default;
-        [] when is_list(Default) -> Default;
-        [] when is_atom(Default) -> Default;
-        [] -> error(badarg);
-        [{_, Match}] -> Match
+    try
+        case ets:lookup(?MODULE, {Section, Key}) of
+            [] when Default == undefined -> Default;
+            [] when is_boolean(Default) -> Default;
+            [] when is_float(Default) -> Default;
+            [] when is_integer(Default) -> Default;
+            [] when is_list(Default) -> Default;
+            [] when is_atom(Default) -> Default;
+            [] -> error(badarg);
+            [{_, Match}] -> Match
+        end
+    catch
+        error:badarg ->
+            Default
     end.
 
 set(Section, Key, Value) ->

src/couch/priv/stats_descriptions.cfg

@@ -298,6 +298,10 @@
     {type, counter},
     {desc, <<"number of couch_server LRU operations skipped">>}
 ]}.
+{[couchdb, couch_server, open], [
+    {type, counter},
+    {desc, <<"number of couch_server open operations invoked">>}
+]}.
 {[couchdb, query_server, vdu_rejects], [
     {type, counter},
     {desc, <<"number of rejections by validate_doc_update function">>}
@@ -410,10 +414,39 @@
     {type, counter},
     {desc, <<"number of other requests">>}
 ]}.
+{[couchdb, query_server, js_filter], [
+    {type, counter},
+    {desc, <<"number of JS filter invocations">>}
+]}.
+{[couchdb, query_server, js_filtered_docs], [
+    {type, counter},
+    {desc, <<"number of docs filtered through JS invocations">>}
+]}.
+{[couchdb, query_server, js_filter_error], [
+    {type, counter},
+    {desc, <<"number of JS filter invocation errors">>}
+]}.
 {[couchdb, legacy_checksums], [
     {type, counter},
     {desc, <<"number of legacy checksums found in couch_file instances">>}
 ]}.
+{[couchdb, btree, folds], [
+    {type, counter},
+    {desc, <<"number of couch btree kv fold callback invocations">>}
+]}.
+{[couchdb, btree, kp_node], [
+    {type, counter},
+    {desc, <<"number of couch btree kp_nodes read">>}
+]}.
+{[couchdb, btree, kv_node], [
+    {type, counter},
+    {desc, <<"number of couch btree kv_nodes read">>}
+]}.
+%% CSRT (couch_stats_resource_tracker) stats
+{[couchdb, csrt, delta_missing_t0], [
+    {type, counter},
+    {desc, <<"number of csrt contexts without a proper startime">>}
+]}.
 {[pread, exceed_eof], [
     {type, counter},
     {desc, <<"number of the attempts to read beyond end of db file">>}

src/couch/src/couch_btree.erl

@@ -472,6 +472,8 @@ reduce_tree_size(kp_node, NodeSize, [{_K, {_P, _Red, Sz}} | NodeList]) ->
 
 get_node(#btree{fd = Fd}, NodePos) ->
     {ok, {NodeType, NodeList}} = couch_file:pread_term(Fd, NodePos),
+    %% TODO: wire in csrt tracking
+    couch_stats:increment_counter([couchdb, btree, NodeType]),
     {NodeType, NodeList}.
 
 write_node(#btree{fd = Fd, compression = Comp} = Bt, NodeType, NodeList) ->
@@ -1163,6 +1165,7 @@ stream_kv_node2(Bt, Reds, PrevKVs, [{K, V} | RestKVs], InRange, Dir, Fun, Acc) -
         false ->
             {stop, {PrevKVs, Reds}, Acc};
         true ->
+            couch_stats:increment_counter([couchdb, btree, folds]),
             AssembledKV = assemble(Bt, K, V),
             case Fun(visit, AssembledKV, {PrevKVs, Reds}, Acc) of
                 {ok, Acc2} ->

src/couch/src/couch_db.erl

@@ -297,6 +297,7 @@ open_doc(Db, IdOrDocInfo) ->
     open_doc(Db, IdOrDocInfo, []).
 
 open_doc(Db, Id, Options) ->
+    %% TODO: wire in csrt tracking
     increment_stat(Db, [couchdb, database_reads]),
     case open_doc_int(Db, Id, Options) of
         {ok, #doc{deleted = true} = Doc} ->
@@ -1982,6 +1983,7 @@ increment_stat(#db{options = Options}, Stat, Count) when
 ->
     case lists:member(sys_db, Options) of
         true ->
+            %% TODO: we shouldn't leak resource usage just because it's a sys_db
             ok;
         false ->
             couch_stats:increment_counter(Stat, Count)

src/couch/src/couch_query_servers.erl

@@ -542,6 +542,8 @@ filter_docs(Req, Db, DDoc, FName, Docs) ->
         {ok, filter_docs_int(Db, DDoc, FName, JsonReq, JsonDocs)}
     catch
         throw:{os_process_error, {exit_status, 1}} ->
+            %% TODO: wire in csrt tracking
+            couch_stats:increment_counter([couchdb, query_server, js_filter_error]),
             %% batch used too much memory, retry sequentially.
             Fun = fun(JsonDoc) ->
                 filter_docs_int(Db, DDoc, FName, JsonReq, [JsonDoc])
@@ -550,6 +552,12 @@ filter_docs(Req, Db, DDoc, FName, Docs) ->
     end.
 
 filter_docs_int(Db, DDoc, FName, JsonReq, JsonDocs) ->
+    %% Count usage in _int version as this can be repeated for OS error
+    %% Pros & cons... might not have actually processed `length(JsonDocs)` docs
+    %% but it certainly undercounts if we count in `filter_docs/5` above
+    %% TODO: wire in csrt tracking
+    couch_stats:increment_counter([couchdb, query_server, js_filter]),
+    couch_stats:increment_counter([couchdb, query_server, js_filtered_docs], length(JsonDocs)),
     [true, Passes] = ddoc_prompt(
         Db,
         DDoc,

src/couch/src/couch_server.erl

@@ -97,6 +97,8 @@ sup_start_link(N) ->
     gen_server:start_link({local, couch_server(N)}, couch_server, [N], []).
 
 open(DbName, Options) ->
+    %% TODO: wire in csrt tracking
+    couch_stats:increment_counter([couchdb, couch_server, open]),
     try
         validate_open_or_create(DbName, Options),
         open_int(DbName, Options)

src/couch_stats/src/couch_stats.erl

@@ -24,6 +24,12 @@
     update_gauge/2
 ]).
 
+%% couch_stats_resource_tracker API
+-export([
+    create_context/3,
+    maybe_track_rexi_init_p/1
+]).
+
 -type response() :: ok | {error, unknown_metric} | {error, invalid_metric}.
 -type stat() :: {any(), [{atom(), any()}]}.
 
@@ -49,6 +55,11 @@ increment_counter(Name) ->
 
 -spec increment_counter(any(), pos_integer()) -> response().
 increment_counter(Name, Value) ->
+    %% Should maybe_track_local happen before or after notify?
+    %% If after, only currently tracked metrics declared in the app's
+    %% stats_description.cfg will be trackable locally. Pros/cons.
+    %io:format("NOTIFY_EXISTING_METRIC: ~p || ~p || ~p~n", [Name, Op, Type]),
+    ok = maybe_track_local_counter(Name, Value),
     case couch_stats_util:get_counter(Name, stats()) of
         {ok, Ctx} -> couch_stats_counter:increment(Ctx, Value);
         {error, Error} -> {error, Error}
@@ -100,6 +111,25 @@ stats() ->
 now_sec() ->
     erlang:monotonic_time(second).
 
+%% Only potentially track positive increments to counters
+-spec maybe_track_local_counter(any(), any()) -> ok.
+maybe_track_local_counter(Name, Val) when is_integer(Val) andalso Val > 0 ->
+    %%io:format("maybe_track_local[~p]: ~p~n", [Val, Name]),
+    couch_stats_resource_tracker: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
+        true -> increment_counter(Metric);
+        false -> ok
+    end.
+
 -ifdef(TEST).
 
 -include_lib("couch/include/couch_eunit.hrl").