diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl index 20f41e04ac..ee1b9112c5 100644 --- a/src/chttpd/src/chttpd_db.erl +++ b/src/chttpd/src/chttpd_db.erl @@ -26,6 +26,7 @@ handle_changes_req/2, update_doc_result_to_json/1, update_doc_result_to_json/2, handle_design_info_req/3, + handle_index_info_req/2, handle_view_cleanup_req/2, update_doc/4, http_code_from_status/1, @@ -464,6 +465,65 @@ handle_design_info_req(#httpd{method = 'GET'} = Req, Db, #doc{} = DDoc) -> handle_design_info_req(Req, _Db, _DDoc) -> send_method_not_allowed(Req, "GET"). +% $db/_index_info: build status of indexes in a db +handle_index_info_req(#httpd{method = 'GET', path_parts = [_, _]} = Req, Db) -> + Types = parse_types(Req), + DbName = couch_db:name(Db), + {ok, {CopiesExpected, Indexes}} = fabric:get_index_info(DbName, Types), + IndexesEjson = {index_info_to_json(Indexes)}, + Ejson = {[{name, DbName}, {copies_expected, CopiesExpected}, {indexes, IndexesEjson}]}, + send_json(Req, 200, Ejson); +handle_index_info_req(#httpd{method = 'GET'} = Req, _Db) -> + chttpd:send_error(Req, not_found); +handle_index_info_req(Req, _Db) -> + send_method_not_allowed(Req, "GET"). + +% _index_info types: one or more index types. ex: ?type=nouveau,search +parse_types(Req) -> + TypeArgs = [V || {K, V} <- chttpd:qs(Req), K == "type"], + case TypeArgs of + [] -> + [view, search, nouveau]; + [_ | _] -> + Tokens = lists:flatmap(fun(V) -> string:tokens(V, ",") end, TypeArgs), + case lists:usort(lists:map(fun parse_type/1, Tokens)) of + [] -> throw({query_parse_error, <<"`type` must not be empty">>}); + Types -> [T || T <- [view, search, nouveau], lists:member(T, Types)] + end + end. + +parse_type("view") -> + view; +parse_type("search") -> + search; +parse_type("nouveau") -> + nouveau; +parse_type(Other) -> + Msg = io_lib:format("Invalid index type: ~s. Must be view, search or nouveau", [Other]), + throw({query_parse_error, ?l2b(Msg)}). + +index_info_to_json(Indexes) -> + [{DDocId, index_info(Res)} || {DDocId, Res} <- Indexes]. + +index_info({error, Error}) -> + errobj(Error); +index_info({Sections}) -> + {[{Sect, index_section_json(Sect, Val)} || {Sect, Val} <- Sections]}. + +index_section_json(view_index, Res) -> + res_to_json(Res); +index_section_json(_Named, {Idxs}) -> + {[{Name, res_to_json(Res)} || {Name, Res} <- Idxs]}. + +res_to_json({error, Error}) -> + errobj(Error); +res_to_json(Info) -> + Info. + +errobj(Error) -> + {_Code, ErrorStr, ReasonStr} = chttpd:error_info(Error), + {[{error, ErrorStr}, {reason, ReasonStr}]}. + create_db_req(#httpd{} = Req, DbName) -> couch_httpd:verify_is_server_admin(Req), ShardsOpt = parse_shards_opt(Req), @@ -2660,6 +2720,38 @@ monitor_attachments_test_() -> ?_assertEqual([], monitor_attachments(Atts)) end}. +index_info_to_json_test_() -> + % {error, Reason} turn into {"error":..., "reason":...} + Unavailable = {service_unavailable, <<"Search is not available">>}, + Indexes = [ + {<<"_design/err">>, {error, not_found}}, + {<<"_design/ok">>, + {[ + {view_index, {[{update_seq, 1}]}}, + {search_indexes, + {[{<<"i">>, {[{doc_count, 2}]}}, {<<"j">>, {error, Unavailable}}]}}, + {nouveau_indexes, {[{<<"n">>, #{num_docs => 3}}]}} + ]}} + ], + Expected = [ + {<<"_design/err">>, {[{error, <<"not_found">>}, {reason, <<"missing">>}]}}, + {<<"_design/ok">>, + {[ + {view_index, {[{update_seq, 1}]}}, + {search_indexes, + {[ + {<<"i">>, {[{doc_count, 2}]}}, + {<<"j">>, + {[ + {error, <<"service unavailable">>}, + {reason, <<"Search is not available">>} + ]}} + ]}}, + {nouveau_indexes, {[{<<"n">>, #{num_docs => 3}}]}} + ]}} + ], + ?_assertEqual(Expected, index_info_to_json(Indexes)). + parse_partitioned_opt_test_() -> { foreach, diff --git a/src/chttpd/src/chttpd_httpd_handlers.erl b/src/chttpd/src/chttpd_httpd_handlers.erl index 3e499b72d0..4251cbfbce 100644 --- a/src/chttpd/src/chttpd_httpd_handlers.erl +++ b/src/chttpd/src/chttpd_httpd_handlers.erl @@ -30,6 +30,7 @@ url_handler(<<"_up">>) -> fun chttpd_misc:handle_up_req/1; url_handler(_) -> no_match. db_handler(<<"_view_cleanup">>) -> fun chttpd_db:handle_view_cleanup_req/2; +db_handler(<<"_index_info">>) -> fun chttpd_db:handle_index_info_req/2; db_handler(<<"_compact">>) -> fun chttpd_db:handle_compact_req/2; db_handler(<<"_design">>) -> fun chttpd_db:handle_design_req/2; db_handler(<<"_partition">>) -> fun chttpd_db:handle_partition_req/2; diff --git a/src/docs/src/api/database/index.rst b/src/docs/src/api/database/index.rst index 914c0d6ff4..8b5f23c1cd 100644 --- a/src/docs/src/api/database/index.rst +++ b/src/docs/src/api/database/index.rst @@ -44,5 +44,6 @@ Where ``{db}`` is the name of any database. changes compact cleanup + index_info security misc diff --git a/src/docs/src/api/database/index_info.rst b/src/docs/src/api/database/index_info.rst new file mode 100644 index 0000000000..a862f1b738 --- /dev/null +++ b/src/docs/src/api/database/index_info.rst @@ -0,0 +1,224 @@ +.. 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. + +.. _api/db/index_info: + +===================== +``/{db}/_index_info`` +===================== + +.. versionadded:: 3.6 + +.. http:get:: /{db}/_index_info + :synopsis: Returns status all the indexes in the database + + Index info objects have the same fields as the ones returned by + :get:`/{db}/_design/{ddoc}/_info`, + :get:`/{db}/_design/{ddoc}/_search_info/{index}` and + :get:`/{db}/_design/{ddoc}/_nouveau_info/{index}`, with an + ``updates_pending`` object with the index build pending bounds across all + copies of the database. See :ref:`api/db/index_info/structure`. + + This endpoint may be used when one needs to know whether a database's + indexes are fully built, for example before switching traffic to a replica. + + :param db: Database name + :query string type: Filter response to the given index types. Can be one of + these values: ``view``, ``search`` and ``nouveau``, comma separated. + *Optional*, default: all types. + :
header Content-Type: - :mimetype:`application/json` + - :mimetype:`text/plain; charset=utf-8` + :>json string name: Database name + :>json number copies_expected: The number of shard copies of the database + (Q*N) Compare with the ``copies`` count of each ``updates_pending`` + object. + :>json object indexes: Index information keyed by design doc ID. + See :ref:`api/db/index_info/structure`. + :code 200: Request completed successfully + :code 400: Invalid database name or ``type`` value + :code 401: Unauthorized request to a protected API + :code 403: Insufficient permissions / :ref:`Too many requests with invalid credentials` + :code 404: Database doesn't exist + + **Request**: + + .. code-block:: http + + GET /recipes/_index_info HTTP/1.1 + Accept: application/json + Host: localhost:5984 + + **Response**: + + .. code-block:: http + + HTTP/1.1 200 OK + Cache-Control: must-revalidate + Content-Type: application/json + Date: Mon, 01 Sep 2025 15:42:11 GMT + Server: CouchDB (Erlang/OTP) + + { + "name": "recipes", + "copies_expected": 6, + "indexes": { + "_design/cookbook": { + "view_index": { + "collator_versions": [ + "153.136" + ], + "compact_running": false, + "language": "javascript", + "purge_seq": 0, + "signature": "a6d97b0199e54a1eb56e4becb1322587", + "sizes": { + "active": 1825, + "external": 1355, + "file": 16750 + }, + "update_seq": 54, + "updater_running": false, + "waiting_clients": 0, + "waiting_commit": false, + "updates_pending": { + "minimum": 0, + "maximum": 0, + "copies": 6 + } + }, + "search_indexes": { + "ingredients": { + "committed_seq": 54, + "disk_size": 3960, + "doc_count": 50, + "doc_del_count": 0, + "pending_seq": 54, + "signature": "0b4ba635d5eb4fcbb2f6c9c2247460ec", + "updates_pending": { + "minimum": 0, + "maximum": 0, + "copies": 6 + } + } + }, + "nouveau_indexes": { + "ingredients": { + "disk_size": 6324, + "num_docs": 50, + "purge_seq": 0, + "signature": "ea87fe8f9517403691850f51d0a1ce3e5afaf89347204dd79430252a0591e503", + "update_seq": 54, + "updates_pending": { + "minimum": 0, + "maximum": 0, + "copies": 6 + } + } + } + }, + "_design/8c2a4caf8ea1b581ac43a062fd43a876dee1382d": { + "view_index": { + "collator_versions": [ + "153.136" + ], + "compact_running": false, + "language": "query", + "purge_seq": 0, + "signature": "b77547252cb8b19ff12831973b576c0f", + "sizes": { + "active": 0, + "external": 10, + "file": 102 + }, + "update_seq": 0, + "updater_running": false, + "waiting_clients": 0, + "waiting_commit": false, + "updates_pending": { + "minimum": 54, + "maximum": 54, + "copies": 6 + } + } + } + } + } + + In this example the ``cookbook`` design document has a view group, a + search index and a nouveau index, all fully built on all six copies. The + second design document is a Mango index which has not been built on + any copy yet: every copy is 54 updates behind. + +.. _api/db/index_info/structure: + +Index Information +================= + +The ``indexes`` object of the :get:`/{db}/_index_info` response has one entry +per design document with at least one index, keyed by design doc ID. Each entry +may contain: + +* **view_index** (*object*): The design doc's view group, with the fields + described in :ref:`api/ddoc/view_index_info` and ``updates_pending``. A design + document has one view group for all of its views. + +* **search_indexes** (*object*): One entry per search index of the design + document, keyed by index name, with the fields of the ``search_index`` object + returned by :get:`/{db}/_design/{ddoc}/_search_info/{index}` plus + ``updates_pending``. +* **nouveau_indexes** (*object*): One entry per nouveau index of the design + document, keyed by index name, with the fields of the ``search_index`` object + returned by :get:`/{db}/_design/{ddoc}/_nouveau_info/{index}` plus + ``updates_pending``. + +Pending stats +-------------- + +Every index object contains an ``updates_pending`` object which shows the +number of pending changes for that index. + +* **minimum** (*number*): Backlog of the most up to date copy of each + range. ``0`` means at least one fully built copy of the index exists for + every range. +* **maximum** (*number*): Backlog of the least up to date copy of each range. + ``0`` means the index is fully built on all returned shard copies. +* **copies** (*number*): The number of shard copies with returned a response. + Copies which are unreachable, in maintenance mode, or which did not answer + before the request timed out are skipped. When this is much smaller than + ``copies_expected`` the bounds are partial and ``maximum`` may underestimate + the true pending backlog. + +Errors +------ + +If an index cannot be inspected it will return an error for that index only. +For example is Closeau is not available search indexes will return: + +.. code-block:: javascript + + "search_indexes": { + "ingredients": { + "error": "service unavailable", + "reason": "Search is not available" + } + } + +Invalid ``type`` values return a ``400 Bad Request``: + +.. code-block:: javascript + + { + "error": "query_parse_error", + "reason": "Invalid index type: foo. Must be view, search or nouveau" + } diff --git a/src/docs/src/api/ddoc/common.rst b/src/docs/src/api/ddoc/common.rst index 961af135b0..d8193bcef9 100644 --- a/src/docs/src/api/ddoc/common.rst +++ b/src/docs/src/api/ddoc/common.rst @@ -148,6 +148,10 @@ index, index size and current status of the design document and associated index information. + To obtain this information for every design document of a database, + including search and nouveau indexes, in a single request see + :get:`/{db}/_index_info`. + :param db: Database name :param ddoc: Design document name :
[{committed_seq, lists:sum(X)} | Acc]; (pending_seq, X, Acc) -> [{pending_seq, lists:sum(X)} | Acc]; + (pending_updates, X, Acc) -> + [{pending_updates, lists:sum(X)} | Acc]; (signature, [X | _], Acc) -> [{signature, X} | Acc]; (_, _, Acc) -> diff --git a/src/dreyfus/src/dreyfus_index.erl b/src/dreyfus/src/dreyfus_index.erl index d754050015..29afaedb4f 100644 --- a/src/dreyfus/src/dreyfus_index.erl +++ b/src/dreyfus/src/dreyfus_index.erl @@ -26,6 +26,7 @@ await/2, search/2, info/1, + info/3, group1/2, group2/2, design_doc_to_indexes/2 @@ -79,6 +80,39 @@ info(Pid0) -> MFA = {?MODULE, info_int, [Pid]}, dreyfus_util:time([index, info], MFA). +% Get shard local index info based on ddoc and index name. This can be used by +% _index_info worker when getting index info batches. +info(DbName, DDoc, IndexName) -> + case design_doc_to_index(DbName, DDoc, IndexName) of + {ok, Index} -> + case dreyfus_index_manager:get_index(DbName, Index) of + {ok, Pid} -> + case info(Pid) of + {ok, Fields} -> + Info = [ + {signature, Index#index.sig}, + {pending_updates, pending_updates(DbName, Fields)} + | Fields + ], + {ok, Info}; + Else -> + Else + end; + Error -> + Error + end; + Error -> + Error + end. + +% Similar idea as view pending_updates from couch_index. Return a difference +% betweenn committed db seq and index +pending_updates(DbName, Fields) -> + GetCommSeq = fun(Db) -> couch_db:get_committed_update_seq(Db) end, + CommittedSeq = couch_util:with_db(DbName, GetCommSeq), + IndexSeq = couch_util:get_value(pending_seq, Fields, 0), + max(CommittedSeq - IndexSeq, 0). + %% We either have a dreyfus_index gen_server pid or the remote %% clouseau pid. to_index_pid(Pid) -> diff --git a/src/dreyfus/src/dreyfus_rpc.erl b/src/dreyfus/src/dreyfus_rpc.erl index 5fbe194dca..b00af49e6b 100644 --- a/src/dreyfus/src/dreyfus_rpc.erl +++ b/src/dreyfus/src/dreyfus_rpc.erl @@ -80,23 +80,7 @@ info(DbName, DDoc, IndexName) -> info_int(DbName, DDoc, IndexName) -> erlang:put(io_priority, {search, DbName}), check_interactive_mode(), - case dreyfus_index:design_doc_to_index(DbName, DDoc, IndexName) of - {ok, Index} -> - case dreyfus_index_manager:get_index(DbName, Index) of - {ok, Pid} -> - case dreyfus_index:info(Pid) of - {ok, Fields} -> - Info = [{signature, Index#index.sig} | Fields], - rexi:reply({ok, Info}); - Else -> - rexi:reply(Else) - end; - Error -> - rexi:reply(Error) - end; - Error -> - rexi:reply(Error) - end. + rexi:reply(dreyfus_index:info(DbName, DDoc, IndexName)). disk_size(DbName, DDoc, IndexName) -> erlang:put(io_priority, {search, DbName}), diff --git a/src/fabric/src/fabric.erl b/src/fabric/src/fabric.erl index ab575f135e..5099ab519a 100644 --- a/src/fabric/src/fabric.erl +++ b/src/fabric/src/fabric.erl @@ -59,6 +59,7 @@ changes/4, query_view/3, query_view/4, query_view/6, query_view/7, get_view_group_info/2, + get_index_info/2, end_changes/0 ]). @@ -595,6 +596,13 @@ query_view(Db, Options, DDoc, ViewName, Callback, Acc0, QueryArgs0) -> get_view_group_info(DbName, DesignId) -> fabric_group_info:go(dbname(DbName), design_doc(DesignId)). +%% @doc get index info of all indexes: view, search and nouveau etc +%% +-spec get_index_info(dbname(), [view | search | nouveau]) -> + {ok, {non_neg_integer(), [{binary(), any()}]}}. +get_index_info(DbName, Types) -> + fabric_index_info:go(dbname(DbName), Types). + -spec end_changes() -> ok. end_changes() -> fabric_view_changes:increment_changes_epoch(). diff --git a/src/fabric/src/fabric_index_info.erl b/src/fabric/src/fabric_index_info.erl new file mode 100644 index 0000000000..433075ce75 --- /dev/null +++ b/src/fabric/src/fabric_index_info.erl @@ -0,0 +1,372 @@ +% 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. + +% Get status of all the indexes in db. Info gathering is optimized by sending +% the ddocs to the works and they locally gather index info in parallel. +% +% To get an idea about how many shards are built we query every live copy and +% return a summary of expected shards vs actual shards included in the results. +% The result returns the expected number of shards (workers) and each index +% summary will contain the number of copies which actually replied. This gives +% us an idea how many indexes and how far along they are before they are built. +% +% For each index besides their sizes and other info we return: +% +% minimum : best copy of each range. When 0 it means at least one complete +% copy of every range exists. +% +% maximum : the worst copy of each range. When 0 it means every copy of every +% live range is fully bult. +% +% copies : live shard copies which returned results. If this value is too low +% compared to the expected copies the maximum/minimum values may not be +% trusted much. +% +% For other index info fields use the lowest values per range so they are +% more or less stable from call to call. + +-module(fabric_index_info). + +-export([go/2]). + +-include_lib("mem3/include/mem3.hrl"). + +go(DbName, Types) -> + {ok, DDocs} = fabric:design_docs(DbName), + {CopiesExpected, Responses} = index_info(DbName, DDocs, Types), + {ok, {CopiesExpected, merge_responses(Responses)}}. + +index_info(DbName, DDocs, Types) -> + Shards = mem3:shards(DbName), + Workers = fabric_util:submit_jobs(Shards, fabric_rpc, index_info, [DDocs, Types]), + RexiMon = fabric_util:create_monitors(Workers), + Acc0 = {fabric_dict:init(Workers, nil), []}, + try fabric_util:recv(Workers, #shard.ref, fun handle_message/3, Acc0) of + {ok, Responses} -> + {length(Workers), tag_responses(Responses)}; + {timeout, {WorkersDict, Responses}} -> + DefunctWorkers = fabric_util:remove_done_workers(WorkersDict, nil), + fabric_util:log_timeout(DefunctWorkers, "index_info"), + fabric_util:cleanup(DefunctWorkers), + {length(Workers), tag_responses(Responses)}; + {error, Error} -> + fabric_util:cleanup(Workers), + throw(Error) + after + rexi_monitor:stop(RexiMon) + end. + +tag_responses(Responses) -> + [{W#shard.range, W#shard.node, PerDDoc} || {W, PerDDoc} <- Responses]. + +% Failed / unreachables dropped. Otherwise responses are merged +% +handle_message({ok, Result}, Worker, {Counters, Acc}) -> + Counters1 = fabric_dict:erase(Worker, Counters), + maybe_stop(Counters1, [{Worker, Result} | Acc]); +handle_message({rexi_DOWN, _, {_, NRef}, _}, _Worker, {Counters, Acc}) -> + Counters1 = fabric_dict:filter(fun(#shard{node = N}, _) -> N =/= NRef end, Counters), + maybe_stop(Counters1, Acc); +handle_message(_Error, Worker, {Counters, Acc}) -> + Counters1 = fabric_dict:erase(Worker, Counters), + maybe_stop(Counters1, Acc). + +maybe_stop(Counters, Acc) -> + case fabric_dict:size(Counters) of + 0 -> {stop, Acc}; + _ -> {ok, {Counters, Acc}} + end. + +% Each response is {Range, Node, [{DDocId, Sections}]} +% Sections: {error, Err} | [ +% {view_index, Res}, +% {search_indexes, [{Name, Res}]}, +% {nouveau_indexes, [{Name, Res}]} +% ] +% Res: {ok, Info} | {error, Err} +% +% Note: view ddoc is a group so there only one per ddoc +% +merge_responses(Responses) -> + ByDDoc = lists:foldl( + fun({Range, Node, PerDDoc}, Acc0) -> + lists:foldl( + fun({DDocId, Sect}, Acc) -> + orddict:append(DDocId, {Range, Node, Sect}, Acc) + end, + Acc0, + PerDDoc + ) + end, + orddict:new(), + Responses + ), + [{DDocId, merge_ddoc(Entries)} || {DDocId, Entries} <- ByDDoc]. + +merge_ddoc(Entries) -> + case [Error || {_, _, {error, Error}} <- Entries] of + [Error | _] -> + {error, Error}; + [] -> + BySection = lists:foldl( + fun({Range, Node, Sect}, Acc0) -> + lists:foldl( + fun({K, V}, Acc) -> orddict:append(K, {Range, Node, V}, Acc) end, + Acc0, + Sect + ) + end, + orddict:new(), + Entries + ), + {[merge_section(K, Vs) || {K, Vs} <- BySection]} + end. + +merge_section(view_index, Results) -> + {view_index, merge_leaf(view, Results)}; +merge_section(search_indexes, PerCopyIdxLists) -> + {search_indexes, {merge_named(search, PerCopyIdxLists)}}; +merge_section(nouveau_indexes, PerCopyIdxLists) -> + {nouveau_indexes, {merge_named(nouveau, PerCopyIdxLists)}}. + +merge_named(Type, PerCopyIdxLists) -> + ByName = lists:foldl( + fun({Range, Node, IdxList}, Acc0) -> + lists:foldl( + fun({Name, Res}, Acc) -> orddict:append(Name, {Range, Node, Res}, Acc) end, + Acc0, + IdxList + ) + end, + orddict:new(), + PerCopyIdxLists + ), + [{Name, merge_leaf(Type, Results)} || {Name, Results} <- ByName]. + +% Results: {Range, Node, {ok, Info} | {error, Err}} per copy. Errors "win" +merge_leaf(Type, Results) -> + case [Error || {_, _, {error, Error}} <- Results] of + [Error | _] -> {error, Error}; + [] -> merge_oks(Type, Results) + end. + +% Non pending updates are merged by lowest copy. We aggregate pending at the +% end since we're doing a min/max over those. +merge_oks(Type, Results) -> + ByRange = lists:foldl( + fun({Range, Node, {ok, Info}}, Acc) -> orddict:append(Range, {Node, Info}, Acc) end, + orddict:new(), + Results + ), + RangeCopies = [Copies || {_Range, Copies} <- ByRange], + Reps = [rem_pending(Type, first(Copies)) || Copies <- RangeCopies], + PerRange = [[pending(Type, Info) || {_Node, Info} <- Copies] || Copies <- RangeCopies], + add_pending(Type, merge_infos(Type, Reps), agg_pending(PerRange)). + +% For some stable-ish order results don't flip flop with each call +first(Copies) -> + [{_Node, Info} | _] = lists:sort(Copies), + Info. + +% Skip copies which did not report pending +% Ranges with no reporting copies contribute 0 +% Return undefined when no copy of any range sent reports +agg_pending(PerRange) -> + case lists:foldl(fun pending_range/2, undefined, PerRange) of + undefined -> undefined; + {Min, Max, Copies} -> [{minimum, Min}, {maximum, Max}, {copies, Copies}] + end. + +pending_range(Vals, Acc) -> + case lists:sort([P || P <- Vals, is_integer(P)]) of + [] -> + Acc; + [_ | _] = Vs -> + {Min, Max, Copies} = + case Acc of + undefined -> {0, 0, 0}; + _ -> Acc + end, + {Min + hd(Vs), Max + lists:last(Vs), Copies + length(Vs)} + end. + +merge_infos(view, Infos) -> + merge_view_infos(Infos); +merge_infos(search, Infos) -> + merge_search_infos(Infos); +merge_infos(nouveau, Infos) -> + merge_nouveau_infos(Infos). + +pending(nouveau, #{} = Info) -> + maps:get(pending_updates, Info, undefined); +pending(_Type, Info) -> + couch_util:get_value(pending_updates, Info). + +rem_pending(nouveau, #{} = Info) -> + maps:remove(pending_updates, Info); +rem_pending(_Type, Info) -> + lists:keydelete(pending_updates, 1, Info). + +add_pending(_Type, Merged, undefined) -> + Merged; +add_pending(nouveau, #{} = Merged, Bounds) -> + Merged#{updates_pending => maps:from_list(Bounds)}; +add_pending(_Type, {Props}, Bounds) -> + {lists:keystore(updates_pending, 1, Props, {updates_pending, {Bounds}})}. + +% Copied from fabric_group_info:merge_results/1 mostly. Except we remove the pending updates +% then do our own (hopefully better) calculation over them. +merge_view_infos(Infos) -> + Dict = to_orddict(lists:append(Infos)), + Merged = orddict:fold( + fun + (signature, [X | _], Acc) -> + [{signature, X} | Acc]; + (language, [X | _], Acc) -> + [{language, X} | Acc]; + (sizes, X, Acc) -> + [{sizes, {merge_obj(X)}} | Acc]; + (compact_running, X, Acc) -> + [{compact_running, lists:member(true, X)} | Acc]; + (updater_running, X, Acc) -> + [{updater_running, lists:member(true, X)} | Acc]; + (waiting_commit, X, Acc) -> + [{waiting_commit, lists:member(true, X)} | Acc]; + (waiting_clients, X, Acc) -> + [{waiting_clients, lists:sum(X)} | Acc]; + (update_seq, X, Acc) -> + [{update_seq, lists:sum(X)} | Acc]; + (purge_seq, X, Acc) -> + [{purge_seq, lists:sum(X)} | Acc]; + (collator_versions, X, Acc) -> + Vs = lists:usort(lists:flatmap(fun(V) -> V end, X)), + [{collator_versions, Vs} | Acc]; + (_, _, Acc) -> + Acc + end, + [], + Dict + ), + {Merged}. + +% Copied from dreyfus_fabric_info:merge_results/1. We remove pending updates +% and add them later. +merge_search_infos(Infos) -> + Dict = to_orddict(lists:append(Infos)), + Merged = orddict:fold( + fun + (signature, [X | _], Acc) -> [{signature, X} | Acc]; + (disk_size, X, Acc) -> [{disk_size, lists:sum(X)} | Acc]; + (doc_count, X, Acc) -> [{doc_count, lists:sum(X)} | Acc]; + (doc_del_count, X, Acc) -> [{doc_del_count, lists:sum(X)} | Acc]; + (committed_seq, X, Acc) -> [{committed_seq, lists:sum(X)} | Acc]; + (pending_seq, X, Acc) -> [{pending_seq, lists:sum(X)} | Acc]; + (_, _, Acc) -> Acc + end, + [], + Dict + ), + {Merged}. + +merge_nouveau_infos(Maps) -> + lists:foldl(fun(M, Acc) -> maps:merge_with(fun merge_nouveau_val/3, M, Acc) end, #{}, Maps). + +merge_nouveau_val(signature, Val, Val) -> + % Can't sum signatures, but we can sum everything else + Val; +merge_nouveau_val(_Key, Val1, Val2) -> + Val1 + Val2. + +to_orddict(KVs) -> + lists:foldl(fun({K, V}, D) -> orddict:append(K, V, D) end, orddict:new(), KVs). + +% Merge list of {[{K, V}]} objects where V is a number +merge_obj(Objects) -> + Dict = lists:foldl( + fun({Props}, D) -> + lists:foldl(fun({K, V}, D0) -> orddict:append(K, V, D0) end, D, Props) + end, + orddict:new(), + Objects + ), + orddict:fold( + fun(Key, X, Acc) -> + [{Key, lists:sum(X)} | Acc] + end, + [], + Dict + ). + +-ifdef(TEST). +-include_lib("couch/include/couch_eunit.hrl"). + +aggregate_pending_test() -> + ?assertEqual(undefined, agg_pending([])), + ?assertEqual(undefined, agg_pending([[]])), + ?assertEqual(undefined, agg_pending([[undefined]])), + ?assertEqual([{minimum, 5}, {maximum, 5}, {copies, 1}], agg_pending([[5]])), + ?assertEqual( + [{minimum, 3}, {maximum, 11}, {copies, 5}], + agg_pending([[4, 0, 8], [3, 3]]) + ), + ?assertEqual( + [{minimum, 2}, {maximum, 2}, {copies, 1}], + agg_pending([[undefined, 2], [undefined]]) + ). + +viewres(Pending, Seq) -> + {ok, [{signature, <<"sig">>}, {pending_updates, Pending}, {update_seq, Seq}]}. + +merge_responses_test() -> + % Some field are added while for pending we compute our min/max stats + Responses = [ + {[0, 10], n1, [{<<"_design/d">>, [{view_index, viewres(0, 100)}]}]}, + {[0, 10], n2, [{<<"_design/d">>, [{view_index, viewres(6, 94)}]}]}, + {[11, 20], n2, [{<<"_design/d">>, [{view_index, viewres(2, 98)}]}]}, + {[11, 20], n3, [{<<"_design/d">>, [{view_index, viewres(4, 96)}]}]} + ], + [{<<"_design/d">>, {[{view_index, {Props}}]}}] = merge_responses(Responses), + ?assertEqual(198, couch_util:get_value(update_seq, Props)), + ?assertEqual(undefined, couch_util:get_value(pending_updates, Props)), + ?assertEqual( + {[{minimum, 2}, {maximum, 10}, {copies, 4}]}, + couch_util:get_value(updates_pending, Props) + ). + +merge_responses_error_wins_test() -> + Responses = [ + {[0, 10], n1, [{<<"_design/d">>, [{view_index, viewres(0, 100)}]}]}, + {[0, 10], n2, [{<<"_design/d">>, [{view_index, {error, not_found}}]}]} + ], + ?assertEqual( + [{<<"_design/d">>, {[{view_index, {error, not_found}}]}}], + merge_responses(Responses) + ). + +merge_responses_nouveau_test() -> + % Nouveau uses maps for info so add a separate test for it + NRes = fun(Pending, Seq) -> + {ok, #{signature => <<"s">>, pending_updates => Pending, <<"update_seq">> => Seq}} + end, + Responses = [ + {[0, 10], n1, [{<<"_design/d">>, [{nouveau_indexes, [{<<"n">>, NRes(5, 10)}]}]}]}, + {[0, 10], n2, [{<<"_design/d">>, [{nouveau_indexes, [{<<"n">>, NRes(0, 15)}]}]}]} + ], + [{<<"_design/d">>, {[{nouveau_indexes, {[{<<"n">>, Merged}]}}]}}] = + merge_responses(Responses), + ?assertEqual(10, maps:get(<<"update_seq">>, Merged)), + ?assertNot(maps:is_key(pending_updates, Merged)), + ?assertEqual( + #{minimum => 0, maximum => 5, copies => 2}, maps:get(updates_pending, Merged) + ). + +-endif. diff --git a/src/fabric/src/fabric_rpc.erl b/src/fabric/src/fabric_rpc.erl index fd06b391b3..0553f61909 100644 --- a/src/fabric/src/fabric_rpc.erl +++ b/src/fabric/src/fabric_rpc.erl @@ -26,7 +26,7 @@ get_missing_revs/2, get_missing_revs/3, update_docs/3 ]). --export([all_docs/3, changes/3, map_view/4, reduce_view/4, group_info/2]). +-export([all_docs/3, changes/3, map_view/4, reduce_view/4, group_info/2, index_info/3]). -export([ create_db/1, create_db/2, delete_db/1, @@ -66,6 +66,9 @@ -include_lib("couch/include/couch_db.hrl"). -include_lib("couch_mrview/include/couch_mrview.hrl"). +% How many design docs an index_info/3 worker processes concurrently +-define(MAX_CONCURRENCY, 8). + %% rpc endpoints %% call to with_db will supply your M:F with a Db instance %% and then remaining args @@ -337,6 +340,130 @@ purge_docs(DbName, UUIdsIdsRevs, Options) -> group_info(DbName, DDocId) -> group_info(DbName, DDocId, []). +% Shard local worker _index_info. Get info for all ddocs on this shard copy. +index_info(DbName, DDocs, Types) -> + set_io_priority(DbName, []), + Avail = {search_available(), nouveau_enabled()}, + Infos = pmap(fun(DDoc) -> ddoc_info(DbName, DDoc, Types, Avail) end, DDocs), + rexi:reply({ok, lists:append(Infos)}). + +% Return [{DDocId, Sections}] or [] for empty ddocs +ddoc_info(ShardName, {Props} = DDocEJson, Types, Avail) -> + DDocId = couch_util:get_value(<<"_id">>, Props), + try + #doc{body = {Body}} = DDoc = couch_doc:from_json_obj(DDocEJson), + Sections = lists:append([section(T, ShardName, DDoc, Body, Avail) || T <- Types]), + case Sections of + [] -> []; + [_ | _] -> [{DDocId, Sections}] + end + catch + _Tag:Error -> + [{DDocId, {error, Error}}] + end. + +section(view, ShardName, DDoc, Body, _Avail) -> + case couch_util:get_value(<<"views">>, Body) of + {[_ | _]} -> [{view_index, view_info(ShardName, DDoc)}]; + _ -> [] + end; +section(search, ShardName, DDoc, Body, {SearchAvailable, _}) -> + case couch_util:get_value(<<"indexes">>, Body) of + {[_ | _] = Indexes} -> + Infos = [ + {Name, search_info(SearchAvailable, ShardName, DDoc, Name)} + || {Name, _} <- Indexes + ], + [{search_indexes, Infos}]; + _ -> + [] + end; +section(nouveau, ShardName, DDoc, Body, {_, NouveauEnabled}) -> + case couch_util:get_value(<<"nouveau">>, Body) of + {[_ | _] = Indexes} -> + Infos = [ + {Name, nouveau_info(NouveauEnabled, ShardName, DDoc, Name)} + || {Name, _} <- Indexes + ], + [{nouveau_indexes, Infos}]; + _ -> + [] + end. + +% Same node local call as group_info/3 above. An index which cannot be +% opened is reported as {error, Reason} instead of crashing the worker. +view_info(ShardName, DDoc) -> + try + norm_error(couch_mrview:get_info(ShardName, DDoc)) + catch + error:{badmatch, Error} -> norm_error(Error); + _Tag:Error -> {error, Error} + end. + +search_info(false, _ShardName, _DDoc, _IndexName) -> + {error, {service_unavailable, <<"Search is not available">>}}; +search_info(true, ShardName, DDoc, IndexName) -> + try + norm_error(dreyfus_index:info(ShardName, DDoc, IndexName)) + catch + _Tag:Error -> {error, Error} + end. + +nouveau_info(false, _ShardName, _DDoc, _IndexName) -> + {error, {service_unavailable, <<"nouveau is not enabled">>}}; +nouveau_info(true, ShardName, DDoc, IndexName) -> + try + norm_error(nouveau_util:index_info(ShardName, DDoc, IndexName)) + catch + _Tag:Error -> {error, Error} + end. + +norm_error({ok, Info}) -> + {ok, Info}; +norm_error({error, Error}) -> + {error, Error}; +norm_error(Else) -> + {error, Else}. + +search_available() -> + try + dreyfus:available() + catch + _:_ -> false + end. + +nouveau_enabled() -> + try + nouveau:enabled() + catch + _:_ -> false + end. + +% Run fun with bounded concurrency and isolated in separate processes +% +pmap(Fun, Items) -> + lists:append([pmap_chunk(Fun, Chunk) || Chunk <- chunk(Items, ?MAX_CONCURRENCY)]). + +pmap_chunk(Fun, Items) -> + PidRefs = [spawn_monitor(fun() -> exit({pmap_res, Fun(Item)}) end) || Item <- Items], + lists:map( + fun({Pid, Ref}) -> + receive + {'DOWN', Ref, process, Pid, {pmap_res, Result}} -> Result; + {'DOWN', Ref, process, Pid, Error} -> exit(Error) + end + end, + PidRefs + ). + +chunk([], _N) -> + []; +chunk(Items, N) when length(Items) =< N -> + [Items]; +chunk(Items, N) -> + {Chunk, Rest} = lists:split(N, Items), + [Chunk | chunk(Rest, N)]. + group_info(DbName, DDocId, DbOptions) -> with_db(DbName, DbOptions, {couch_mrview, get_info, [DDocId]}). diff --git a/src/nouveau/src/nouveau_rpc.erl b/src/nouveau/src/nouveau_rpc.erl index 2037c7e7ef..08d808a124 100644 --- a/src/nouveau/src/nouveau_rpc.erl +++ b/src/nouveau/src/nouveau_rpc.erl @@ -77,13 +77,5 @@ update_and_retry(DbName, Index, QueryArgs, UpdateLatency) -> rexi:reply(Else) end. -info(DbName, #index{} = Index0) -> - %% Incorporate the shard name into the record. - Index1 = Index0#index{dbname = DbName}, - case nouveau_api:index_info(Index1) of - {ok, Info0} -> - Info1 = Info0#{signature => Index0#index.sig}, - rexi:reply({ok, Info1}); - {error, Reason} -> - rexi:reply({error, Reason}) - end. +info(DbName, #index{} = Index) -> + rexi:reply(nouveau_util:index_info(DbName, Index)). diff --git a/src/nouveau/src/nouveau_util.erl b/src/nouveau/src/nouveau_util.erl index 567b77f492..e0b5700a45 100644 --- a/src/nouveau/src/nouveau_util.erl +++ b/src/nouveau/src/nouveau_util.erl @@ -22,6 +22,8 @@ index_name/1, design_doc_to_indexes/2, design_doc_to_index/3, + index_info/2, + index_info/3, verify_index_exists/2, ensure_local_purge_docs/2, maybe_create_local_purge_doc/2, @@ -40,6 +42,35 @@ index_name(#index{} = Index) -> node_prefix() -> atom_to_binary(node(), utf8). +index_info(DbName, #index{} = Index0) -> + %% Incorporate the shard name into the record. + Index1 = Index0#index{dbname = DbName}, + case nouveau_api:index_info(Index1) of + {ok, Info} -> + {ok, Info#{ + signature => Index1#index.sig, + pending_updates => pending_updates(DbName, Info) + }}; + {error, Reason} -> + {error, Reason} + end. + +index_info(DbName, DDoc, IndexName) -> + case design_doc_to_index(DbName, DDoc, IndexName) of + {ok, Index} -> + index_info(DbName, Index); + {error, Reason} -> + {error, Reason} + end. + +% Similar idea to view's pending_updates in couch_index. Return number of how +% much committed db seq is ahead of the index +pending_updates(DbName, #{} = Info) -> + GetCommSeq = fun(Db) -> couch_db:get_committed_update_seq(Db) end, + CommittedSeq = couch_util:with_db(DbName, GetCommSeq), + IndexSeq = maps:get(<<"update_seq">>, Info, 0), + max(CommittedSeq - IndexSeq, 0). + %% copied from dreyfus_index.erl design_doc_to_indexes(DbName, #doc{body = {Fields}} = Doc) -> RawIndexes = couch_util:get_value(<<"nouveau">>, Fields, {[]}),