Improve index cleanup

Fix these issues:

  * Index cleanup triggered by smoosh only cleaned view indexes and purge
  checkpoints, make sure to also clean nouveau and search indexes and
  checkpoints.

  * `/_search_cleanup` endpoint only cleaned indexes on the coordinator node,
  despite being a clustered fabric endpoint. Fix it so it cleans indexes on
  other nodes as well as most users would expect.

  * Nouveau cleanup didn't clean purge checkpoints, so make it do so. Use the
  mrview purge checkpoint strategy for all indexes. This improves dreyfus logic
  to avoid traversing internal disk paths from clouseau.

  * Make `_view_cleanup` clean all the index types. This is a bit dirty, but it
  may be better than adding another cleanup `_index_cleanup` API. Left
  `_search_cleanup` and `_nouveau_cleanup` APIs as is for now.

Some optimizations:

  * For each index clean request fetch ddocs once. Calculate signatures and
  then call the remote node cleanup logic with them. This avoids fetching
  design documents multiple times or sending all of them to the worker nodes.
  This is what Nouveau is doing so stick with that nice pattern.

  * Use erpc for remote calls. Our Erlang version is high enough (25+) to use
  the multple requests pattern from erpc. This is more compact than rexi. The
  absolute timeout pattern makes it simpler to have a global timeout for the
  whole request.

Cleanups:

  * Make purge checkpoint fetching and cleanup more uniform. Use common utility
  logic in `couch_index_util` for all indexes.

  * Make index cleanup similar between all three indexes. Use the same erpc
  pattern for all of them.

  * Move index cleanup functions from fabric.erl to its own module --
  fabric_index_cleanup.erl

  * Rename fabric function names more uniform and clearly indicate if it cleans
  indexes on all nodes or just the current node.

  * Add the db name to dreyfus `#index{}` record when it initializes so it
  matches Nouveau.

Author: nickva

src/couch_index/src/couch_index_util.erl

@@ -14,6 +14,7 @@
 
 -export([root_dir/0, index_dir/2, index_file/3]).
 -export([load_doc/3, sort_lib/1, hexsig/1]).
+-export([get_purge_checkpoints/2, cleanup_purges/3]).
 
 -include_lib("couch/include/couch_db.hrl").
 
@@ -72,3 +73,49 @@ sort_lib([{LName, LCode} | Rest], LAcc) ->
 
 hexsig(Sig) ->
     couch_util:to_hex(Sig).
+
+% Helper function for indexes to get their purge checkpoints as signatures.
+%
+get_purge_checkpoints(DbName, Type) when is_binary(DbName), is_binary(Type) ->
+    couch_util:with_db(DbName, fun(Db) -> get_purge_checkpoints(Db, Type) end);
+get_purge_checkpoints(Db, Type) when is_binary(Type) ->
+    Prefix = <<?LOCAL_DOC_PREFIX, "purge-", Type:(byte_size(Type))/binary, "-">>,
+    PrefixSize = byte_size(Prefix),
+    FoldFun = fun(#doc{id = Id}, Acc) ->
+        case Id of
+            <<Prefix:PrefixSize/binary, Sig/binary>> -> {ok, Acc#{Sig => Id}};
+            _ -> {stop, Acc}
+        end
+    end,
+    Opts = [{start_key, Prefix}],
+    {ok, Signatures = #{}} = couch_db:fold_local_docs(Db, FoldFun, #{}, Opts),
+    Signatures.
+
+% Helper functions for indexes to clean their purge checkpoints.
+%
+cleanup_purges(DbName, #{} = Sigs, #{} = Checkpoints) when is_binary(DbName) ->
+    couch_util:with_db(DbName, fun(Db) ->
+        cleanup_purges(Db, Sigs, Checkpoints)
+    end);
+cleanup_purges(Db, #{} = Sigs, #{} = CheckpointsMap) ->
+    InactiveMap = maps:without(maps:keys(Sigs), CheckpointsMap),
+    InactiveCheckpoints = maps:values(InactiveMap),
+    DeleteFun = fun(DocId) -> delete_checkpoint(Db, DocId) end,
+    lists:foreach(DeleteFun, InactiveCheckpoints).
+
+delete_checkpoint(Db, DocId) ->
+    DbName = couch_db:name(Db),
+    LogMsg = "~p : deleting inactive purge checkpoint ~s : ~s",
+    couch_log:debug(LogMsg, [?MODULE, DbName, DocId]),
+    try couch_db:open_doc(Db, DocId, []) of
+        {ok, Doc = #doc{}} ->
+            Deleted = Doc#doc{deleted = true, body = {[]}},
+            couch_db:update_doc(Db, Deleted, [?ADMIN_CTX]);
+        {not_found, _} ->
+            ok
+    catch
+        Tag:Error ->
+            ErrLog = "~p : error deleting checkpoint ~s : ~s error: ~p:~p",
+            couch_log:error(ErrLog, [?MODULE, DbName, DocId, Tag, Error]),
+            ok
+    end.

src/couch_mrview/src/couch_mrview_cleanup.erl

@@ -14,29 +14,37 @@
 
 -export([
     run/1,
-    cleanup_purges/3,
-    cleanup_indices/2
+    cleanup/2
 ]).
 
--include_lib("couch/include/couch_db.hrl").
-
 run(Db) ->
     Indices = couch_mrview_util:get_index_files(Db),
     Checkpoints = couch_mrview_util:get_purge_checkpoints(Db),
     {ok, Db1} = couch_db:reopen(Db),
     Sigs = couch_mrview_util:get_signatures(Db1),
-    ok = cleanup_purges(Db1, Sigs, Checkpoints),
+    ok = cleanup_purges(Db, Sigs, Checkpoints),
     ok = cleanup_indices(Sigs, Indices).
 
-cleanup_purges(DbName, Sigs, Checkpoints) when is_binary(DbName) ->
-    couch_util:with_db(DbName, fun(Db) ->
-        cleanup_purges(Db, Sigs, Checkpoints)
-    end);
-cleanup_purges(Db, #{} = Sigs, #{} = CheckpointsMap) ->
-    InactiveMap = maps:without(maps:keys(Sigs), CheckpointsMap),
-    InactiveCheckpoints = maps:values(InactiveMap),
-    DeleteFun = fun(DocId) -> delete_checkpoint(Db, DocId) end,
-    lists:foreach(DeleteFun, InactiveCheckpoints).
+% erpc endpoint for fabric_index_cleanup:cleanup_indexes/2
+%
+cleanup(Dbs, #{} = Sigs) ->
+    try
+        lists:foreach(
+            fun(Db) ->
+                Indices = couch_mrview_util:get_index_files(Db),
+                Checkpoints = couch_mrview_util:get_purge_checkpoints(Db),
+                ok = cleanup_purges(Db, Sigs, Checkpoints),
+                ok = cleanup_indices(Sigs, Indices)
+            end,
+            Dbs
+        )
+    catch
+        error:database_does_not_exist ->
+            ok
+    end.
+
+cleanup_purges(Db, Sigs, Checkpoints) ->
+    couch_index_util:cleanup_purges(Db, Sigs, Checkpoints).
 
 cleanup_indices(#{} = Sigs, #{} = IndexMap) ->
     Fun = fun(_, Files) -> lists:foreach(fun delete_file/1, Files) end,
@@ -54,20 +62,3 @@ delete_file(File) ->
             couch_log:error(ErrLog, [?MODULE, File, Tag, Error]),
             ok
     end.
-
-delete_checkpoint(Db, DocId) ->
-    DbName = couch_db:name(Db),
-    LogMsg = "~p : deleting inactive purge checkpoint ~s : ~s",
-    couch_log:debug(LogMsg, [?MODULE, DbName, DocId]),
-    try couch_db:open_doc(Db, DocId, []) of
-        {ok, Doc = #doc{}} ->
-            Deleted = Doc#doc{deleted = true, body = {[]}},
-            couch_db:update_doc(Db, Deleted, [?ADMIN_CTX]);
-        {not_found, _} ->
-            ok
-    catch
-        Tag:Error ->
-            ErrLog = "~p : error deleting checkpoint ~s : ~s error: ~p:~p",
-            couch_log:error(ErrLog, [?MODULE, DbName, DocId, Tag, Error]),
-            ok
-    end.

src/couch_mrview/src/couch_mrview_util.erl

@@ -16,6 +16,7 @@
 -export([get_local_purge_doc_id/1, get_value_from_options/2]).
 -export([verify_view_filename/1, get_signature_from_filename/1]).
 -export([get_signatures/1, get_purge_checkpoints/1, get_index_files/1]).
+-export([get_signatures_from_ddocs/2]).
 -export([ddoc_to_mrst/2, init_state/4, reset_index/3]).
 -export([make_header/1]).
 -export([index_file/2, compaction_file/2, open_file/1]).
@@ -94,40 +95,35 @@ get_signatures(DbName) when is_binary(DbName) ->
     couch_util:with_db(DbName, fun get_signatures/1);
 get_signatures(Db) ->
     DbName = couch_db:name(Db),
-    % get_design_docs/1 returns ejson for clustered shards, and
-    % #full_doc_info{}'s for other cases.
     {ok, DDocs} = couch_db:get_design_docs(Db),
+    % get_design_docs/1 returns ejson for clustered shards, and
+    % #full_doc_info{}'s for other cases. Both are transformed to #doc{} records
     FoldFun = fun
         ({[_ | _]} = EJsonDoc, Acc) ->
             Doc = couch_doc:from_json_obj(EJsonDoc),
-            {ok, Mrst} = ddoc_to_mrst(DbName, Doc),
-            Sig = couch_util:to_hex_bin(Mrst#mrst.sig),
-            Acc#{Sig => true};
+            [Doc | Acc];
         (#full_doc_info{} = FDI, Acc) ->
             {ok, Doc} = couch_db:open_doc_int(Db, FDI, [ejson_body]),
-            {ok, Mrst} = ddoc_to_mrst(DbName, Doc),
-            Sig = couch_util:to_hex_bin(Mrst#mrst.sig),
-            Acc#{Sig => true}
+            [Doc | Acc]
+    end,
+    DDocs1 = lists:foldl(FoldFun, [], DDocs),
+    get_signatures_from_ddocs(DbName, DDocs1).
+
+% From a list of design #doc{} records returns signatures map: #{Sig => true}
+%
+get_signatures_from_ddocs(DbName, DDocs) when is_list(DDocs) ->
+    FoldFun = fun(#doc{} = Doc, Acc) ->
+        {ok, Mrst} = ddoc_to_mrst(DbName, Doc),
+        Sig = couch_util:to_hex_bin(Mrst#mrst.sig),
+        Acc#{Sig => true}
     end,
     lists:foldl(FoldFun, #{}, DDocs).
 
 % Returns a map of `Sig => DocId` elements for all the purge view
 % checkpoint docs. Sig is a hex-encoded binary.
 %
-get_purge_checkpoints(DbName) when is_binary(DbName) ->
-    couch_util:with_db(DbName, fun get_purge_checkpoints/1);
 get_purge_checkpoints(Db) ->
-    FoldFun = fun(#doc{id = Id}, Acc) ->
-        case Id of
-            <<?LOCAL_DOC_PREFIX, "purge-mrview-", Sig/binary>> ->
-                {ok, Acc#{Sig => Id}};
-            _ ->
-                {stop, Acc}
-        end
-    end,
-    Opts = [{start_key, <<?LOCAL_DOC_PREFIX, "purge-mrview-">>}],
-    {ok, Signatures = #{}} = couch_db:fold_local_docs(Db, FoldFun, #{}, Opts),
-    Signatures.
+    couch_index_util:get_purge_checkpoints(Db, <<"mrview">>).
 
 % Returns a map of `Sig => [FilePath, ...]` elements. Sig is a hex-encoded
 % binary and FilePaths are lists as they intended to be passed to couch_file

src/dreyfus/src/clouseau_rpc.erl

@@ -262,10 +262,11 @@ rename(DbName) ->
 %% and an analyzer represented in a Javascript function in a design document.
 %% `Sig` is used to check if an index description is changed,
 %% and the index needs to be reconstructed.
--spec cleanup(DbName :: string_as_binary(_), ActiveSigs :: [sig()]) ->
+-spec cleanup(DbName :: string_as_binary(_), SigMap :: #{sig() => true}) ->
     ok.
 
-cleanup(DbName, ActiveSigs) ->
+cleanup(DbName, #{} = SigMap) ->
+    ActiveSigs = maps:keys(SigMap),
     gen_server:cast({cleanup, clouseau()}, {cleanup, DbName, ActiveSigs}).
 
 %% a binary with value <<"tokens">>

src/dreyfus/src/dreyfus_fabric_cleanup.erl

@@ -14,99 +14,50 @@
 
 -module(dreyfus_fabric_cleanup).
 
--include("dreyfus.hrl").
--include_lib("mem3/include/mem3.hrl").
--include_lib("couch/include/couch_db.hrl").
-
--export([go/1]).
+-export([go/1, go_local/3]).
 
 go(DbName) ->
-    DesignDocs =
-        case fabric:design_docs(DbName) of
-            {ok, DDocs} when is_list(DDocs) ->
-                DDocs;
-            Else ->
-                couch_log:debug("Invalid design docs: ~p~n", [Else]),
-                []
-        end,
-    ActiveSigs = lists:usort(
-        lists:flatmap(
-            fun active_sigs/1,
-            [couch_doc:from_json_obj(DD) || DD <- DesignDocs]
-        )
-    ),
-    cleanup_local_purge_doc(DbName, ActiveSigs),
-    clouseau_rpc:cleanup(DbName, ActiveSigs),
-    ok.
+    case fabric_util:get_design_doc_records(DbName) of
+        {ok, DDocs} when is_list(DDocs) ->
+            Sigs = dreyfus_util:get_signatures_from_ddocs(DbName, DDocs),
+            Shards = mem3:shards(DbName),
+            ByNode = maps:groups_from_list(fun mem3:node/1, fun mem3:name/1, Shards),
+            Fun = fun(Node, Dbs, Acc) ->
+                erpc:send_request(Node, ?MODULE, go_local, [DbName, Dbs, Sigs], Node, Acc)
+            end,
+            Reqs = maps:fold(Fun, erpc:reqids_new(), ByNode),
+            recv(DbName, Reqs, fabric_util:abs_request_timeout());
+        Error ->
+            couch_log:error("~p : error fetching ddocs db:~p ~p", [?MODULE, DbName, Error]),
+            Error
+    end.
 
-active_sigs(#doc{body = {Fields}} = Doc) ->
+% erpc endpoint for go/1 and fabric_index_cleanup:cleanup_indexes/2
+%
+go_local(DbName, Dbs, #{} = Sigs) ->
     try
-        {RawIndexes} = couch_util:get_value(<<"indexes">>, Fields, {[]}),
-        {IndexNames, _} = lists:unzip(RawIndexes),
-        [
-            begin
-                {ok, Index} = dreyfus_index:design_doc_to_index(Doc, IndexName),
-                Index#index.sig
-            end
-         || IndexName <- IndexNames
-        ]
+        lists:foreach(
+            fun(Db) ->
+                Checkpoints = dreyfus_util:get_purge_checkpoints(Db),
+                ok = couch_index_util:cleanup_purges(Db, Sigs, Checkpoints)
+            end,
+            Dbs
+        ),
+        clouseau_rpc:cleanup(DbName, Sigs),
+        ok
     catch
-        error:{badmatch, _Error} ->
-            []
+        error:database_does_not_exist ->
+            ok
     end.
 
-cleanup_local_purge_doc(DbName, ActiveSigs) ->
-    {ok, BaseDir} = clouseau_rpc:get_root_dir(),
-    DbNamePattern = <<DbName/binary, ".*">>,
-    Pattern0 = filename:join([BaseDir, "shards", "*", DbNamePattern, "*"]),
-    Pattern = binary_to_list(iolist_to_binary(Pattern0)),
-    DirListStrs = filelib:wildcard(Pattern),
-    DirList = [iolist_to_binary(DL) || DL <- DirListStrs],
-    LocalShards = mem3:local_shards(DbName),
-    ActiveDirs = lists:foldl(
-        fun(LS, AccOuter) ->
-            lists:foldl(
-                fun(Sig, AccInner) ->
-                    DirName = filename:join([BaseDir, LS#shard.name, Sig]),
-                    [DirName | AccInner]
-                end,
-                AccOuter,
-                ActiveSigs
-            )
-        end,
-        [],
-        LocalShards
-    ),
-
-    DeadDirs = DirList -- ActiveDirs,
-    lists:foreach(
-        fun(IdxDir) ->
-            Sig = dreyfus_util:get_signature_from_idxdir(IdxDir),
-            case Sig of
-                undefined ->
-                    ok;
-                _ ->
-                    DocId = dreyfus_util:get_local_purge_doc_id(Sig),
-                    LocalShards = mem3:local_shards(DbName),
-                    lists:foreach(
-                        fun(LS) ->
-                            ShardDbName = LS#shard.name,
-                            {ok, ShardDb} = couch_db:open_int(ShardDbName, []),
-                            case couch_db:open_doc(ShardDb, DocId, []) of
-                                {ok, LocalPurgeDoc} ->
-                                    couch_db:update_doc(
-                                        ShardDb,
-                                        LocalPurgeDoc#doc{deleted = true},
-                                        [?ADMIN_CTX]
-                                    );
-                                {not_found, _} ->
-                                    ok
-                            end,
-                            couch_db:close(ShardDb)
-                        end,
-                        LocalShards
-                    )
-            end
-        end,
-        DeadDirs
-    ).
+recv(DbName, Reqs, Timeout) ->
+    case erpc:receive_response(Reqs, Timeout, true) of
+        {ok, _Lable, Reqs1} ->
+            recv(DbName, Reqs1, Timeout);
+        {Error, Label, Reqs1} ->
+            ErrMsg = "~p : error cleaning dreyfus indexes db:~p req:~p error:~p",
+            couch_log:error(ErrMsg, [?MODULE, DbName, Label, Error]),
+            recv(DbName, Reqs1, Timeout);
+        no_request ->
+            ok
+    end.

src/dreyfus/src/dreyfus_index.erl

@@ -22,13 +22,13 @@
 % public api.
 -export([
     start_link/2,
-    design_doc_to_index/2,
+    design_doc_to_index/3,
     await/2,
     search/2,
     info/1,
     group1/2,
     group2/2,
-    design_doc_to_indexes/1
+    design_doc_to_indexes/2
 ]).
 
 % gen_server api.
@@ -87,14 +87,14 @@ to_index_pid(Pid) ->
         false -> Pid
     end.
 
-design_doc_to_indexes(#doc{body = {Fields}} = Doc) ->
+design_doc_to_indexes(DbName, #doc{body = {Fields}} = Doc) ->
     RawIndexes = couch_util:get_value(<<"indexes">>, Fields, {[]}),
     case RawIndexes of
         {IndexList} when is_list(IndexList) ->
             {IndexNames, _} = lists:unzip(IndexList),
             lists:flatmap(
                 fun(IndexName) ->
-                    case (catch design_doc_to_index(Doc, IndexName)) of
+                    case (catch design_doc_to_index(DbName, Doc, IndexName)) of
                         {ok, #index{} = Index} -> [Index];
                         _ -> []
                     end
@@ -301,7 +301,7 @@ open_index(DbName, #index{analyzer = Analyzer, sig = Sig}) ->
             Error
     end.
 
-design_doc_to_index(#doc{id = Id, body = {Fields}}, IndexName) ->
+design_doc_to_index(DbName, #doc{id = Id, body = {Fields}}, IndexName) ->
     Language = couch_util:get_value(<<"language">>, Fields, <<"javascript">>),
     {RawIndexes} = couch_util:get_value(<<"indexes">>, Fields, {[]}),
     InvalidDDocError =
@@ -323,6 +323,7 @@ design_doc_to_index(#doc{id = Id, body = {Fields}}, IndexName) ->
                         )
                     ),
                     {ok, #index{
+                        dbname = DbName,
                         analyzer = Analyzer,
                         ddoc_id = Id,
                         def = Def,