add scanner to upgrade nouveau indexes

Author: rnewson

rel/overlay/etc/default.ini

@@ -1021,6 +1021,7 @@ url = {{nouveau_url}}
 ;couch_scanner_plugin_find = false
 ;couch_scanner_plugin_conflict_finder = false
 ;couch_quickjs_scanner_plugin = false
+;nouveau_index_upgrader = false
 
 ; The following [$plugin*] settings apply to all plugins
 
@@ -1127,6 +1128,12 @@ url = {{nouveau_url}}
 ; Scanner settings to skip dbs and docs would also work:
 ;[couch_quickjs_scanner_plugin.skip_{dbs,ddoc,docs}]
 
+[nouveau_index_upgrader]
+; Common scanner scheduling settings
+;after = restart
+;repeat = restart
+
+
 [chttpd_auth_lockout]
 ; CouchDB can temporarily lock out IP addresses that repeatedly fail authentication
 ; mode can be set to one of three recognised values;

src/nouveau/src/nouveau_fabric_search.erl

@@ -15,7 +15,7 @@
 
 -module(nouveau_fabric_search).
 
--export([go/4]).
+-export([go/3, go/4]).
 
 -include_lib("mem3/include/mem3.hrl").
 -include_lib("couch/include/couch_db.hrl").
@@ -38,12 +38,12 @@ go(DbName, GroupId, IndexName, QueryArgs0) when is_binary(GroupId) ->
 go(DbName, #doc{} = DDoc, IndexName, QueryArgs0) ->
     case nouveau_util:design_doc_to_index(DbName, DDoc, IndexName) of
         {ok, Index} ->
-            go(DbName, DDoc, IndexName, QueryArgs0, Index);
+            go(DbName, QueryArgs0, Index);
         {error, Reason} ->
             {error, Reason}
     end.
 
-go(DbName, #doc{} = _DDoc, _IndexName, QueryArgs0, Index) ->
+go(DbName, QueryArgs0, Index) ->
     Shards = get_shards(DbName, QueryArgs0),
     {PackedBookmark, #{limit := Limit, sort := Sort} = QueryArgs1} =
         maps:take(bookmark, QueryArgs0),

src/nouveau/src/nouveau_index_upgrader.erl

@@ -0,0 +1,138 @@
+% 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(nouveau_index_upgrader).
+-behaviour(couch_scanner_plugin).
+
+-export([
+    start/2,
+    resume/2,
+    complete/1,
+    checkpoint/1,
+    db/2,
+    ddoc/3
+]).
+
+-include("nouveau.hrl").
+-include_lib("couch_scanner/include/couch_scanner_plugin.hrl").
+
+start(ScanId, #{}) ->
+    St = init_config(ScanId),
+    case should_run() of
+        true ->
+            ?INFO("Starting.", [], St),
+            {ok, St};
+        false ->
+            ?INFO("Not starting.", [], St),
+            skip
+    end.
+
+resume(ScanId, #{}) ->
+    St = init_config(ScanId),
+    case should_run() of
+        true ->
+            ?INFO("Resuming.", [], St),
+            {ok, St};
+        false ->
+            ?INFO("Not resuming.", [], St),
+            skip
+    end.
+
+complete(St) ->
+    ?INFO("Completed", [], St),
+    {ok, #{}}.
+
+checkpoint(_St) ->
+    {ok, #{}}.
+
+db(St, _DbName) ->
+    {ok, St}.
+
+ddoc(St, _DbName, #doc{id = <<"_design/_", _/binary>>}) ->
+    {ok, St};
+ddoc(St, DbName, #doc{} = DDoc0) ->
+    case update_ddoc_versions(DDoc0) of
+        DDoc0 ->
+            ok;
+        DDoc1 ->
+            Indexes = nouveau_util:design_doc_to_indexes(DbName, DDoc1),
+            case upgrade_indexes(DbName, Indexes) of
+                true ->
+                    save_ddoc(DbName, DDoc1);
+                false ->
+                    ok
+            end
+    end,
+    {ok, St}.
+
+upgrade_indexes(_DbName, []) ->
+    true;
+upgrade_indexes(DbName, [Index | Rest]) ->
+    case upgrade_index(DbName, Index) of
+        true ->
+            upgrade_indexes(DbName, Rest);
+        false ->
+            false
+    end.
+
+upgrade_index(DbName, #index{} = Index) ->
+    ?INFO("Upgrading ~s/~s to version ~B", [
+        DbName,
+        Index#index.name,
+        ?TARGET_LUCENE_VERSION
+    ]),
+    case
+        nouveau_fabric_search:go(
+            DbName,
+            #{query => <<"*:*">>, bookmark => null, sort => null, limit => 1},
+            Index#index{lucene_version = ?TARGET_LUCENE_VERSION}
+        )
+    of
+        {ok, _SearchResults} ->
+            true;
+        {error, _Reason} ->
+            false
+    end.
+
+update_ddoc_versions(#doc{} = Doc) ->
+    #doc{body = {Fields0}} = Doc,
+    {Indexes0} = couch_util:get_value(<<"nouveau">>, Fields0),
+    Indexes1 = lists:map(fun update_version/1, Indexes0),
+    Fields1 = couch_util:set_value(<<"nouveau">>, Fields0, {Indexes1}),
+    Doc#doc{body = {Fields1}}.
+
+save_ddoc(DbName, #doc{} = DDoc) ->
+    {Pid, Ref} = spawn_monitor(fun() ->
+        case fabric:update_doc(DbName, DDoc, [?ADMIN_CTX]) of
+            {ok, _} ->
+                exit(ok);
+            Else ->
+                exit(Else)
+        end
+    end),
+    receive
+        {'DOWN', Ref, process, Pid, ok} ->
+            ?INFO(
+                "Updated ~s/~s indexes to version ~B", [DbName, DDoc#doc.id, ?TARGET_LUCENE_VERSION]
+            );
+        {'DOWN', Ref, process, Pid, Else} ->
+            ?INFO("Failed to update ~s/~s for reason ~p", [DbName, DDoc#doc.id, Else])
+    end.
+
+update_version({IndexName, {Index}}) ->
+    {IndexName, {couch_util:set_value(<<"lucene_version">>, Index, ?TARGET_LUCENE_VERSION)}}.
+
+init_config(ScanId) ->
+    #{sid => ScanId}.
+
+should_run() ->
+    couch_scanner_util:on_first_node().