iilyak commented on code in PR #4812: URL: https://github.com/apache/couchdb/pull/4812#discussion_r1449149148
########## src/couch_stats/src/couch_stats_resource_tracker.erl: ########## @@ -0,0 +1,1014 @@ +% 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). + +-export([ + start_link/0, + init/1, + handle_call/3, + handle_cast/2, + handle_info/2, + code_change/3, + terminate/2 +]). + +-export([ + inc/1, inc/2, + maybe_inc/2, + get_pid_ref/0, + accumulate_delta/1 +]). + +-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, + set_context_handler_fun/1, + set_context_username/1, + track/1, + should_track/1 +]). + +-export([ + active/0, + active_coordinators/0, + active_workers/0, + find_unmonitored/0 +]). + +-export([ + count_by/1, + group_by/2, + group_by/3, + group_by/4, + sorted/1, + sorted_by/1, + sorted_by/2, + sorted_by/3, + + find_by_pid/1, + + unsafe_foldl/3, + + term_to_flat_json/1 +]). + +-export([ + make_delta/0 +]). + +%% Singular increment operations +-export([ + db_opened/0, + doc_read/0, + row_read/0, + btree_fold/0, + ioq_called/0, + js_evaled/0, + js_filtered/0, + js_filtered_error/0, + js_filtered_doc/0, + mango_match_evaled/0, + get_kv_node/0, + get_kp_node/0 +]). + +%% Plural increment operations +-export([ + js_filtered_docs/1, + io_bytes_read/1, + io_bytes_written/1 +]). + +-export([ + field/2, + curry_field/1 +]). + +-include_lib("couch/include/couch_db.hrl"). + +%% Use these for record upgrades over the wire and in ETS tables +%% TODO: alternatively, just delete these. Currently using a map +%% for shipping deltas over the wire, avoiding much of the +%% problem here. We'll likely still need to handle upgrades to +%% map format over time, so let's decide a course of action here. +-define(RCTX_V1, rctx_v1). +-define(RCTX, ?RCTX_V1). + +-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_FOLDS, btree_folds). +-define(COUCH_BT_GET_KP_NODE, get_kp_node). +-define(COUCH_BT_GET_KV_NODE, get_kv_node). +-define(COUCH_JS_FILTER, js_filter). +-define(COUCH_JS_FILTER_ERROR, js_filter_error). +-define(COUCH_JS_FILTERED_DOCS, js_filtered_docs). +-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(FRPC_CHANGES_ROW, ?ROWS_READ). + +%% Module 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 + + +-record(st, { + eviction_delay = 10 * 1000, %% How many ms dead processes are visible + scan_interval = 2048, %% How regularly to perfom scans + tracking = #{} %% track active processes for eventual eviction +}). + + +%% TODO: switch to: +%% -record(?RCTX, { +-record(rctx, { + %% Metadata + started_at = tnow(), + updated_at = tnow(), + exited_at, %% TODO: do we need a final exit time and additional update times afterwards? + pid_ref, + mon_ref, + mfa, + nonce, + from, + type = unknown, %% unknown/background/system/rpc/coordinator/fabric_rpc/etc_rpc/etc + state = alive, + dbname, + username, + + %% Stats counters + db_open = 0, + docs_read = 0, + rows_read = 0, + btree_folds = 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_filter_error = 0, + js_filtered_docs = 0, + mango_eval_match = 0, + %% TODO: switch record definitions to be macro based, eg: + %% ?COUCH_BT_GET_KP_NODE = 0, + get_kv_node = 0, + get_kp_node = 0 +}). + +%% monotonic time now in millisecionds +tnow() -> + erlang:monotonic_time(millisecond). + +is_enabled() -> + config:get_boolean(?MODULE_STRING, "enabled", true). + +db_opened() -> inc(db_opened). +doc_read() -> inc(docs_read). +row_read() -> inc(rows_read). +btree_fold() -> inc(?COUCH_BT_FOLDS). +%% 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). +js_filtered_doc() -> inc(js_filtered_docs). +mango_match_evaled() -> inc(mango_eval_match). +get_kv_node() -> inc(get_kv_node). +get_kp_node() -> inc(get_kp_node). + +js_filtered_docs(N) -> inc(js_filtered_docs, N). +io_bytes_read(N) -> inc(io_bytes_read, N). +io_bytes_written(N) -> inc(io_bytes_written, N). + +inc(?DB_OPEN) -> + inc(?DB_OPEN, 1); +inc(docs_read) -> + inc(docs_read, 1); +inc(?ROWS_READ) -> + inc(?ROWS_READ, 1); +inc(?FRPC_CHANGES_RETURNED) -> + inc(?FRPC_CHANGES_RETURNED, 1); +inc(?COUCH_BT_FOLDS) -> + inc(?COUCH_BT_FOLDS, 1); +inc(ioq_calls) -> + inc(ioq_calls, 1); +inc(io_bytes_read) -> + inc(io_bytes_read, 1); +inc(io_bytes_written) -> + inc(io_bytes_written, 1); +inc(js_evals) -> + inc(js_evals, 1); +inc(?COUCH_JS_FILTER) -> + inc(?COUCH_JS_FILTER, 1); +inc(?COUCH_JS_FILTER_ERROR) -> + inc(?COUCH_JS_FILTER_ERROR, 1); +inc(?COUCH_JS_FILTERED_DOCS) -> + inc(?COUCH_JS_FILTERED_DOCS, 1); +inc(?MANGO_EVAL_MATCH) -> + inc(?MANGO_EVAL_MATCH, 1); +inc(?COUCH_BT_GET_KP_NODE) -> + inc(?COUCH_BT_GET_KP_NODE, 1); +inc(?COUCH_BT_GET_KV_NODE) -> + inc(?COUCH_BT_GET_KV_NODE, 1); +inc(_) -> + 0. + + +inc(?DB_OPEN, N) -> + update_counter(#rctx.?DB_OPEN, N); +inc(?ROWS_READ, N) -> + update_counter(#rctx.?ROWS_READ, N); +inc(?FRPC_CHANGES_RETURNED, N) -> + update_counter(#rctx.?FRPC_CHANGES_RETURNED, N); +inc(ioq_calls, N) -> + update_counter(#rctx.ioq_calls, N); +inc(io_bytes_read, N) -> + update_counter(#rctx.io_bytes_read, N); +inc(io_bytes_written, N) -> + update_counter(#rctx.io_bytes_written, N); +inc(js_evals, N) -> + update_counter(#rctx.js_evals, N); +inc(?COUCH_JS_FILTER, N) -> + update_counter(#rctx.?COUCH_JS_FILTER, N); +inc(?COUCH_JS_FILTER_ERROR, N) -> + update_counter(#rctx.?COUCH_JS_FILTER_ERROR, N); +inc(?COUCH_JS_FILTERED_DOCS, N) -> + update_counter(#rctx.?COUCH_JS_FILTERED_DOCS, N); +inc(?MANGO_EVAL_MATCH, N) -> + update_counter(#rctx.?MANGO_EVAL_MATCH, N); +inc(?DB_OPEN_DOC, N) -> + update_counter(#rctx.?DB_OPEN_DOC, N); +inc(?FRPC_CHANGES_ROW, N) -> + update_counter(#rctx.?ROWS_READ, N); %% TODO: rework double use of rows_read +inc(?COUCH_BT_GET_KP_NODE, N) -> + update_counter(#rctx.?COUCH_BT_GET_KP_NODE, N); +inc(?COUCH_BT_GET_KV_NODE, N) -> + update_counter(#rctx.?COUCH_BT_GET_KV_NODE, N); +inc(_, _) -> + %% inc needs to allow unknown types to pass for accumulate_update to handle + %% updates from nodes with newer data formats + 0. + +maybe_inc([mango, evaluate_selector], Val) -> + inc(?MANGO_EVAL_MATCH, Val); +maybe_inc([couchdb, database_reads], Val) -> + inc(?DB_OPEN_DOC, Val); +maybe_inc([fabric_rpc, changes, processed], Val) -> + inc(?FRPC_CHANGES_ROW, Val); +maybe_inc([fabric_rpc, changes, returned], Val) -> + inc(?FRPC_CHANGES_RETURNED, Val); +maybe_inc([fabric_rpc, view, rows_read], Val) -> + inc(?ROWS_READ, Val); +maybe_inc([couchdb, couch_server, open], Val) -> + inc(?DB_OPEN, Val); +maybe_inc([couchdb, btree, folds], Val) -> + inc(?COUCH_BT_FOLDS, Val); +maybe_inc([couchdb, btree, kp_node], Val) -> + inc(?COUCH_BT_GET_KP_NODE, Val); +maybe_inc([couchdb, btree, kv_node], Val) -> + inc(?COUCH_BT_GET_KV_NODE, Val); +maybe_inc([couchdb, query_server, js_filter_error], Val) -> + inc(?COUCH_JS_FILTER_ERROR, Val); +maybe_inc([couchdb, query_server, js_filter], Val) -> + inc(?COUCH_JS_FILTER, Val); +maybe_inc([couchdb, query_server, js_filtered_docs], Val) -> + inc(?COUCH_JS_FILTERED_DOCS, Val); +maybe_inc(_Metric, _Val) -> + %%io:format("SKIPPING MAYBE_INC METRIC[~p]: ~p~n", [Val, Metric]), + 0. + + +%% 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. + +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; +accumulate_delta(Other) -> + io:format("CSRT:ACC_DELTA UNKNOWN DELTA: ~p~n", [Other]). + + +update_counter(Field, Count) -> + is_enabled() andalso update_counter(get_pid_ref(), Field, Count). + + +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}). + + +active() -> active_int(all). +active_coordinators() -> active_int(coordinators). +active_workers() -> active_int(workers). + + +active_int(coordinators) -> + select_by_type(coordinators); +active_int(workers) -> + select_by_type(workers); +active_int(all) -> + lists:map(fun to_flat_json/1, ets:tab2list(?MODULE)). + + +select_by_type(coordinators) -> + ets:select(couch_stats_resource_tracker, + [{#rctx{type = {coordinator,'_','_'}, _ = '_'}, [], ['$_']}]); +select_by_type(workers) -> + ets:select(couch_stats_resource_tracker, Review Comment: This would only work with `-include_lib("stdlib/include/ms_transform.hrl")`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
