You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by va...@apache.org on 2024/03/24 03:31:55 UTC
(couchdb) 01/01: Implement a background db / ddoc / doc scanner
This is an automated email from the ASF dual-hosted git repository.
vatamane pushed a commit to branch scanner
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 5bdf4eec43fa750ad026ad7b739d5acb2c19a953
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Wed Aug 9 17:04:15 2023 -0400
Implement a background db / ddoc / doc scanner
[WIP] Everything is place except docs and tests
The app scans all the dbs and docs. It has a plugin system to allow gathering
various things from a cluster. The first use is to scan all the javascript
design docs and run them through the new QuickJS javascript engine.
Other possible uses:
- Gather total db and view sizes
- Scan for document features (docs of certain sizes, contained certain fields
and values).
The plugins are managed as individual process by the couch_scanner_server with
the start_link/1 and stop/1 functions. After a plugin runner is spawned, the
only thing couch_scanner_server does is wait for it to exit.
The plugin runner process may exit normally, crash, or exit with {shutdown,
{reschedule, TSec}} if they want to reschedule to run again at some point the
future (next day, a week later, etc).
After the process starts, it will load and validate the plugin module. Then, it
will start scanning all the dbs and docs on the local node. Shard ranges will
be scanned only on one of the cluster nodes to avoid duplicating work. For
instance, if there are 2 shard ranges, 0-7, 8-f, with copies on nodes n1, n2,
n3. Then, 0-7 might be scanned on n1 only, and 8-f on n3.
The plugin API is the following (as OTP callback definitions):
```erlang
-callback start(ScanId :: binary(), EJson :: #{}) ->
{ok, St :: term()} | skip.
-callback resume(ScanId :: binary(), EJson :: #{}) ->
{ok, St :: term()} | skip.
-callback stop(St :: term()) ->
{ok, EJson :: #{}}.
-callback checkpoint(St :: term()) ->
{ok, EJson :: #{}}.
-callback db(St :: term(), DbName :: binary()) ->
{ok | skip | stop, St1 :: term()}.
-callback ddoc(St :: term(), DbName :: binary(), #doc{}) ->
{ok | stop, St1 :: term()}.
-callback shards(St :: term(), [#shard{}]) ->
{[#shard{}], St1 :: term()}.
-callback db_opened(St :: term(), Db :: term()) ->
{ok, St :: term()}.
-callback doc_id(St :: term(), DocId :: binary(), Db :: term()) ->
{ok | skip | stop, St1 :: term()}.
-callback doc(St :: term(), Db :: term(), #doc{}) ->
{ok | stop, St1 :: term()}.
-callback db_closing(St :: term(), Db :: term()) ->
{ok, St1 :: term()}.
```
A simple plugin `couch_scanner_plugin_ddoc_features` is included as first
example implementation. It traverses the design docs on a cluster and reports
when it finds Apache CouchDB 4.x deprecated featuers (lists, shows, etc).
Plugin module are enabled by `$plugin_mod = true` entries in the
`[couch_scanner_plugins]` section. For example, to enable
`couch_scanner_plugin_ddoc_features`:
```
[couch_scanner_plugins]
couch_scanner_plugin_ddoc_features = true
```
Plugins may configure thier scheduling using `after` and `repeat` config
values. For example, to start after Unix time stamp 1711249693 and then run
every 3 days:
```
[couch_scanner_plugin_ddoc_features]
after = 1711249693
repeat = 3_days
```
The default values for `after` and `repeat` is ` = restart`, meaning to run
once after the node starts up.
---
rebar.config.script | 1 +
rel/reltool.config | 2 +
src/couch_scanner/README.md | 4 +
src/couch_scanner/include/couch_scanner_plugin.hrl | 22 +
src/couch_scanner/src/couch_scanner.app.src | 29 +
src/couch_scanner/src/couch_scanner.erl | 36 +
src/couch_scanner/src/couch_scanner_app.erl | 23 +
src/couch_scanner/src/couch_scanner_checkpoint.erl | 136 ++++
src/couch_scanner/src/couch_scanner_plugin.erl | 731 +++++++++++++++++++++
.../src/couch_scanner_plugin_ddoc_features.erl | 252 +++++++
src/couch_scanner/src/couch_scanner_server.erl | 242 +++++++
src/couch_scanner/src/couch_scanner_sup.erl | 34 +
.../test/eunit/couch_scanner_test.erl | 34 +
13 files changed, 1546 insertions(+)
diff --git a/rebar.config.script b/rebar.config.script
index 98e013e06..4e94603df 100644
--- a/rebar.config.script
+++ b/rebar.config.script
@@ -141,6 +141,7 @@ SubDirs = [
"src/smoosh",
"src/weatherreport",
"src/couch_prometheus",
+ "src/couch_scanner",
"rel"
].
diff --git a/rel/reltool.config b/rel/reltool.config
index ab92a374d..abd758ada 100644
--- a/rel/reltool.config
+++ b/rel/reltool.config
@@ -62,6 +62,7 @@
snappy,
weatherreport,
couch_prometheus,
+ couch_scanner,
%% extra
nouveau,
@@ -125,6 +126,7 @@
{app, snappy, [{incl_cond, include}]},
{app, weatherreport, [{incl_cond, include}]},
{app, couch_prometheus, [{incl_cond, include}]},
+ {app, couch_scanner, [{incl_cond, include}]},
%% extra
{app, nouveau, [{incl_cond, include}]},
diff --git a/src/couch_scanner/README.md b/src/couch_scanner/README.md
new file mode 100644
index 000000000..9b8bf66b7
--- /dev/null
+++ b/src/couch_scanner/README.md
@@ -0,0 +1,4 @@
+Couch Scanner
+================
+
+Traverse all dbs periodically and emit various reports
diff --git a/src/couch_scanner/include/couch_scanner_plugin.hrl b/src/couch_scanner/include/couch_scanner_plugin.hrl
new file mode 100644
index 000000000..0fa2e8f5b
--- /dev/null
+++ b/src/couch_scanner/include/couch_scanner_plugin.hrl
@@ -0,0 +1,22 @@
+% 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.
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("mem3/include/mem3.hrl").
+
+-define(LOG(LVL, FMT, ARGS, META), couch_scanner_plugin:log(LVL, ?MODULE, FMT, ARGS, maps:merge(META, #{fn => ?FUNCTION_NAME}))).
+-define(INFO(FMT, ARGS, META), ?LOG(info, FMT, ARGS, META)).
+-define(INFO(FMT, ARGS), ?LOG(info, FMT, ARGS, #{})).
+-define(INFO(META), ?LOG(info, "", [], META)).
+-define(INFO(), ?LOG(info, "", [], #{})).
+-define(WARN(FMT, ARGS), ?LOG(warning, FMT, ARGS, #{})).
+-define(WARN(FMT, ARGS, META), ?LOG(warning, FMT, ARGS, META)).
diff --git a/src/couch_scanner/src/couch_scanner.app.src b/src/couch_scanner/src/couch_scanner.app.src
new file mode 100644
index 000000000..961e9e80e
--- /dev/null
+++ b/src/couch_scanner/src/couch_scanner.app.src
@@ -0,0 +1,29 @@
+% 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.
+
+{application, couch_scanner, [
+ {description, "CouchDB Scanner"},
+ {vsn, git},
+ {registered, [
+ couch_scanner_server
+ ]},
+ {applications, [
+ kernel,
+ stdlib,
+ crypto,
+ config,
+ couch_log,
+ couch_stats,
+ fabric
+ ]},
+ {mod, {couch_scanner_app, []}}
+]}.
diff --git a/src/couch_scanner/src/couch_scanner.erl b/src/couch_scanner/src/couch_scanner.erl
new file mode 100644
index 000000000..96b8e48a0
--- /dev/null
+++ b/src/couch_scanner/src/couch_scanner.erl
@@ -0,0 +1,36 @@
+% 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_scanner).
+
+-export([
+ status/0,
+ stop/0,
+ resume/0,
+ checkpoints/0,
+ reset_checkpoints/0
+]).
+
+status() ->
+ couch_scanner_server:status().
+
+stop() ->
+ couch_scanner_server:stop().
+
+resume() ->
+ couch_scanner_server:resume().
+
+checkpoints() ->
+ couch_scanner_server:checkpoints().
+
+reset_checkpoints() ->
+ couch_scanner_server:reset_checkpoints().
diff --git a/src/couch_scanner/src/couch_scanner_app.erl b/src/couch_scanner/src/couch_scanner_app.erl
new file mode 100644
index 000000000..23c4093dc
--- /dev/null
+++ b/src/couch_scanner/src/couch_scanner_app.erl
@@ -0,0 +1,23 @@
+% 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_scanner_app).
+
+-behaviour(application).
+
+-export([start/2, stop/1]).
+
+start(_StartType, _StartArgs) ->
+ couch_scanner_sup:start_link().
+
+stop(_State) ->
+ ok.
diff --git a/src/couch_scanner/src/couch_scanner_checkpoint.erl b/src/couch_scanner/src/couch_scanner_checkpoint.erl
new file mode 100644
index 000000000..6907ee30e
--- /dev/null
+++ b/src/couch_scanner/src/couch_scanner_checkpoint.erl
@@ -0,0 +1,136 @@
+% 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_scanner_checkpoint).
+
+-export([
+ write/2,
+ read/1,
+ reset/1,
+ list_all/1
+]).
+
+-include_lib("couch/include/couch_db.hrl").
+
+write(<<Plugin/binary>>, #{} = State) ->
+ with_db(fun(Db) -> update_doc(Db, doc_id(Plugin), State) end).
+
+read(<<Plugin/binary>>) ->
+ with_db(fun(Db) -> load_doc(Db, doc_id(Plugin)) end).
+
+reset(<<Plugin/binary>>) ->
+ with_db(fun(Db) -> delete_doc(Db, doc_id(Plugin)) end).
+
+list_all(<<Prefix/binary>>) ->
+ with_db(fun(Db) -> list_all(Db, Prefix) end).
+
+% Private functions
+
+doc_id(Plugin) ->
+ % Dashes are more conventional for doc ids
+ Plugin1 = binary:replace(Plugin, <<"_">>, <<"-">>, [global]),
+ <<?LOCAL_DOC_PREFIX, Plugin1/binary>>.
+
+plugin_id(<<?LOCAL_DOC_PREFIX, DocId/binary>>) ->
+ binary:replace(DocId, <<"-">>, <<"_">>, [global]).
+
+delete_doc(Db, DocId) ->
+ case couch_db:open_doc(Db, DocId, []) of
+ {ok, #doc{revs = {_, RevList}}} ->
+ {ok, _} = couch_db:delete_doc(Db, DocId, RevList),
+ ok;
+ {not_found, _} ->
+ ok
+ end.
+
+update_doc(Db, DocId, #{} = Body) ->
+ EJsonBody = ejson_props(Body#{<<"_id">> => DocId}),
+ Doc = couch_doc:from_json_obj(EJsonBody),
+ case couch_db:open_doc(Db, DocId, []) of
+ {ok, #doc{revs = Revs}} ->
+ {ok, _} = couch_db:update_doc(Db, Doc#doc{revs = Revs}, []);
+ {not_found, _} ->
+ {ok, _} = couch_db:update_doc(Db, Doc, [])
+ end,
+ ok.
+
+load_doc(Db, DocId) ->
+ case couch_db:open_doc(Db, DocId, [ejson_body]) of
+ {ok, #doc{body = EJsonBody}} -> ejson_map(EJsonBody);
+ {not_found, _} -> not_found
+ end.
+
+list_all(Db, Prefix0) ->
+ Prefix = binary:replace(Prefix0, <<"_">>, <<"-">>, [global]),
+ IdPref = <<?LOCAL_DOC_PREFIX, Prefix/binary>>,
+ Len = byte_size(IdPref),
+ FoldFun = fun(#doc{id = Id, body = Body}, #{} = Acc) ->
+ case Id of
+ <<IdPref:Len/binary, _/binary>> ->
+ {ok, Acc#{plugin_id(Id) => ejson_map(Body)}};
+ <<_/binary>> ->
+ {stop, Acc}
+ end
+ end,
+ Opts = [{start_key, IdPref}],
+ {ok, #{} = Docs} = couch_db:fold_local_docs(Db, FoldFun, #{}, Opts),
+ Docs.
+
+with_db(Fun) when is_function(Fun, 1) ->
+ DbName = config:get("mem3", "shards_db", "_dbs"),
+ case mem3_util:ensure_exists(DbName) of
+ {ok, Db} ->
+ try
+ Fun(Db)
+ after
+ catch couch_db:close(Db)
+ end;
+ Else ->
+ throw(Else)
+ end.
+
+ejson_map(Obj) ->
+ couch_scanner_plugin:ejson_map(Obj).
+
+ejson_props(Obj) ->
+ ?JSON_DECODE(?JSON_ENCODE(Obj)).
+
+-ifdef(TEST).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+couch_scanner_checkpoint_test_() ->
+ {
+ foreach,
+ fun test_util:start_couch/0,
+ fun test_util:stop_couch/1,
+ [
+ ?TDEF_FE(t_read_write_reset)
+ ]
+ }.
+
+t_read_write_reset(_) ->
+ Plugin = <<"scanner_plugin_abc">>,
+ ok = reset(Plugin),
+ ?assertEqual(ok, write(Plugin, #{<<"foo">> => 1})),
+ ?assertEqual(#{<<"foo">> => 1}, read(Plugin)),
+ ?assertEqual(ok, write(Plugin, #{<<"bar">> => 2})),
+ ?assertEqual(#{<<"bar">> => 2}, read(Plugin)),
+ ?assertEqual(
+ #{<<"scanner_plugin_abc">> => #{<<"bar">> => 2}},
+ list_all(<<"scanner_plugin_">>)
+ ),
+ ?assertEqual(not_found, read(<<"scanner_plugin_other">>)),
+ ?assertEqual(ok, reset(Plugin)),
+ ?assertEqual(not_found, read(Plugin)).
+
+-endif.
diff --git a/src/couch_scanner/src/couch_scanner_plugin.erl b/src/couch_scanner/src/couch_scanner_plugin.erl
new file mode 100644
index 000000000..79144a961
--- /dev/null
+++ b/src/couch_scanner/src/couch_scanner_plugin.erl
@@ -0,0 +1,731 @@
+% 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 under1
+% the License.
+
+% Scanner plugin runner process
+%
+% This is the process which is spawned and run for each enabled plugin.
+%
+% A number of these processes are managed by the couch_scanner_server via
+% start_link/1 and stop/1 functions. After a plugin runner is spawned, the only
+% thing couch_scanner_server does is wait for it to exit.
+%
+% The plugin runner process may exit normally, crash, or exit with {shutdown,
+% {reschedule, TSec}} if they want to reschedule to run again at some point the
+% future (next day, a week later, etc).
+%
+% After the process starts, it will load and validate the plugin module. Then,
+% it will start scanning all the dbs and docs on the local node. Shard ranges
+% will be scanned only on one of the cluster nodes to avoid duplicating work.
+% For instance, if there are 2 shard ranges, 0-7, 8-f, with copies on nodes n1,
+% n2, n3. Then, 0-7 might be scanned on n1 only, and 8-f on n3.
+%
+% The plugin API defined in the behavior definition section.
+%
+% The start/1 function is called when the plugin starts running. It returns
+% some context (St), which can be any Erlang term. All subsequent function
+% calls will be called with the same St object, and may return an updated
+% version of it.
+%
+% If the plugin hasn't finished runing and has resumed running after the node
+% was restarted or an error happened, the resume/1 function will be called.
+% That's the difference between start and resume: start/1 is called when the
+% scan starts from the beginning (first db, first shard, ...), and resume/1 is
+% called when the scanning hasn't finished and has to continue.
+%
+% The checkpoint/1 callback is periodically called to checkpoint the scanning
+% progress. start/1 and resume/1 function will be called with the last saved
+% checkpoint map value.
+%
+% The stop/1 callback is called when the scan has finished. The stop callback
+% should a final checkpoint map object. The last checkoint will be written and
+% then.
+%
+% As the cluster dbs, shards, ddocs and individual docs are discovered during
+% scanning, the appropriate callbacks will be called. Most callbacks, besides
+% the updated St object, can reply with ok, skip or stop tags. The meaning of
+% those are:
+%
+% * ok - continue to the next object
+%
+% * skip - skip the current object and don't scan its internal (ex: skip a db and
+% don't scan its ddocs, but continue with the next db)
+%
+% * stop - stop scanning any remaining objects of that type (ex: don't scan any more dbs)
+%
+% Plugins may use a few helper functions:
+%
+% * log/5 - Used with macros from couch_scanner_plugin.hrl for simple logging
+% with a bit of context from running plugin.
+%
+% * ejson_map/1 - Normalize ejson objects as maps
+%
+% * schedule_time/4 - Given a current time, last start and configured
+% scheduled times, calculate when the next run should happen as an absolute
+% Posix/Unix time in seconds.
+
+-module(couch_scanner_plugin).
+
+-export([
+ % Main plugin process API
+ spawn_link/1,
+ stop/1,
+ % Utility functions
+ log/5,
+ ejson_map/1,
+ % Internal export
+ run/1
+]).
+
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("mem3/include/mem3.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
+
+% Behaviour callback definitions
+
+-callback start(ScanId :: binary(), EJson :: #{}) ->
+ {ok, St :: term()} | skip.
+-callback resume(ScanId :: binary(), EJson :: #{}) ->
+ {ok, St :: term()} | skip.
+-callback stop(St :: term()) ->
+ {ok, EJson :: #{}}.
+-callback checkpoint(St :: term()) ->
+ {ok, EJson :: #{}}.
+-callback db(St :: term(), DbName :: binary()) ->
+ {ok | skip | stop, St1 :: term()}.
+-callback ddoc(St :: term(), DbName :: binary(), #doc{}) ->
+ {ok | stop, St1 :: term()}.
+-callback shards(St :: term(), [#shard{}]) ->
+ {[#shard{}], St1 :: term()}.
+-callback db_opened(St :: term(), Db :: term()) ->
+ {ok, St :: term()}.
+-callback doc_id(St :: term(), DocId :: binary(), Db :: term()) ->
+ {ok | skip | stop, St1 :: term()}.
+-callback doc(St :: term(), Db :: term(), #doc{}) ->
+ {ok | stop, St1 :: term()}.
+-callback db_closing(St :: term(), Db :: term()) ->
+ {ok, St1 :: term()}.
+
+-define(CHECKPOINT_INTERVAL_SEC, 10).
+-define(STOP_TIMEOUT_MSEC, 5000).
+
+-define(MINUTE, 60).
+-define(HOUR, 3600).
+-define(DAY, 24 * 3600).
+-define(WEEK, 7 * 24 * 3600).
+-define(MONTH, 30 * 24 * 3600).
+
+-record(st, {
+ id,
+ scan_id,
+ mod,
+ pst,
+ cursor,
+ shards_db,
+ db,
+ checkpoint_sec = 0,
+ start_sec = 0,
+ skip_dbs,
+ skip_ddocs,
+ skip_docs
+}).
+
+spawn_link(<<Id/binary>>) ->
+ proc_lib:spawn_link(?MODULE, run, [Id]).
+
+stop(Pid) when is_pid(Pid) ->
+ unlink(Pid),
+ Ref = erlang:monitor(process, Pid),
+ Pid ! stop,
+ receive
+ {'DOWN', Ref, _, _, _} -> ok
+ after ?STOP_TIMEOUT_MSEC ->
+ exit(Pid, kill),
+ receive
+ {'DOWN', Ref, _, _, _} -> ok
+ end
+ end,
+ ok.
+
+% Utilities used by plugins
+
+log(Level, Mod, Fmt, Args, #{} = Meta) when
+ is_atom(Level), is_atom(Mod), is_list(Fmt), is_list(Args)
+->
+ {MetaFmt, MetaArgs} = log_format_meta(Mod, Meta),
+ couch_log:Level(lists:flatten([MetaFmt, Fmt]), MetaArgs ++ Args).
+
+ejson_map(Obj) ->
+ jiffy:decode(jiffy:encode(Obj), [return_maps]).
+
+% Main run function
+
+run(<<Id/binary>>) ->
+ process_flag(priority, low),
+ Mod = plugin_mod(Id),
+ St = #st{id = Id, mod = Mod},
+ St1 = init_config(St),
+ St2 = init_from_checkpoint(St1),
+ St3 = scan_dbs(St2),
+ finalize(St3).
+
+% Private functions
+
+init_config(#st{mod = Mod} = St) ->
+ St#st{
+ skip_dbs = config_match_patterns(Mod, "skip_dbs"),
+ skip_ddocs = config_match_patterns(Mod, "skip_ddocs"),
+ skip_docs = config_match_patterns(Mod, "skip_docs")
+ }.
+
+init_from_checkpoint(#st{id = Id, mod = Mod} = St) ->
+ case couch_scanner_checkpoint:read(Id) of
+ #{
+ <<"state">> := <<"running">>,
+ <<"cursor">> := Cur,
+ <<"scan_id">> := SId,
+ <<"pst">> := EJson,
+ <<"start_sec">> := StartSec
+ } ->
+ Now = tsec(),
+ PSt1 = resume_callback(Mod, SId, EJson),
+ St#st{
+ pst = PSt1,
+ cursor = Cur,
+ checkpoint_sec = Now,
+ start_sec = StartSec,
+ scan_id = SId
+ };
+ not_found ->
+ SId = new_scan_id(),
+ Now = tsec(),
+ LastStartSec = 0,
+ Cur = <<>>,
+ PSt1 = start_callback(Mod, Now, SId, LastStartSec, #{}),
+ ok = start_checkpoint(Id, Now, SId, Mod, Cur, PSt1),
+ St#st{
+ pst = PSt1,
+ cursor = Cur,
+ checkpoint_sec = Now,
+ start_sec = Now,
+ scan_id = SId
+ };
+ #{
+ <<"state">> := <<"finished">>,
+ <<"pst">> := EJson,
+ <<"start_sec">> := LastStartSec
+ } ->
+ SId = new_scan_id(),
+ Now = tsec(),
+ Cur = <<>>,
+ PSt1 = start_callback(Mod, Now, SId, LastStartSec, EJson),
+ ok = start_checkpoint(Id, Now, SId, Mod, Cur, PSt1),
+ St#st{
+ pst = PSt1,
+ cursor = Cur,
+ checkpoint_sec = Now,
+ start_sec = Now,
+ scan_id = SId
+ }
+ end.
+
+new_scan_id() ->
+ TSec = integer_to_binary(erlang:system_time(second)),
+ Rand = string:lowercase(binary:encode_hex(crypto:strong_rand_bytes(6))),
+ <<TSec/binary, "-", Rand/binary>>.
+
+scan_dbs(#st{cursor = Cursor} = St) ->
+ DbsDbName = mem3_sync:shards_db(),
+ ioq:set_io_priority({system, DbsDbName}),
+ {ok, Db} = mem3_util:ensure_exists(DbsDbName),
+ St1 = St#st{shards_db = Db},
+ Opts = [{start_key, Cursor}],
+ try
+ {ok, St2} = couch_db:fold_docs(Db, fun scan_dbs_fold/2, St1, Opts),
+ St2#st{shards_db = undefined}
+ after
+ couch_db:close(Db)
+ end.
+
+scan_dbs_fold(#full_doc_info{} = FDI, #st{shards_db = Db} = Acc) ->
+ Acc1 = Acc#st{cursor = FDI#full_doc_info.id},
+ Acc2 = maybe_checkpoint(Acc1),
+ case couch_db:open_doc(Db, FDI, [ejson_body]) of
+ {ok, #doc{id = <<"_design/", _/binary>>}} ->
+ {ok, Acc2};
+ {ok, #doc{id = DbName, body = Body}} ->
+ scan_db(shards(DbName, Body), Acc2)
+ end.
+
+scan_db([], #st{} = St) ->
+ {ok, St};
+scan_db([_ | _] = Shards, #st{} = St) ->
+ #st{cursor = DbName, mod = Mod, pst = PSt, skip_dbs = Skip} = St,
+ case match_skip_pat(DbName, Skip) of
+ false ->
+ {Go, PSt1} = Mod:db(PSt, DbName),
+ St1 = St#st{pst = PSt1},
+ case Go of
+ ok ->
+ St2 = fold_ddocs(fun scan_ddocs_fold/2, St1),
+ {Shards1, St3} = shards_callback(St2, Shards),
+ St4 = scan_shards(Shards1, St3),
+ {ok, St4};
+ skip ->
+ {ok, St1};
+ stop ->
+ {stop, St1}
+ end;
+ true ->
+ {ok, St}
+ end.
+
+scan_ddocs_fold({meta, _}, #st{} = Acc) ->
+ {ok, Acc};
+scan_ddocs_fold({row, RowProps}, #st{} = Acc) ->
+ DDoc = couch_util:get_value(doc, RowProps),
+ scan_ddoc(couch_doc:from_json_obj(DDoc), Acc);
+scan_ddocs_fold(complete, #st{} = Acc) ->
+ {ok, Acc};
+scan_ddocs_fold({error, Error}, _Acc) ->
+ exit({shutdown, {scan_ddocs_fold, Error}}).
+
+scan_shards([], #st{} = St) ->
+ St;
+scan_shards([#shard{} = Shard | Rest], #st{} = St) ->
+ St1 = maybe_checkpoint(St),
+ St2 = scan_docs(St1, Shard),
+ scan_shards(Rest, St2).
+
+scan_ddoc(#doc{id = DDocId} = DDoc, #st{} = St) ->
+ #st{cursor = DbName, mod = Mod, pst = PSt, skip_ddocs = Skip} = St,
+ case match_skip_pat(DDocId, Skip) of
+ false ->
+ {Go, PSt1} = Mod:ddoc(PSt, DbName, DDoc),
+ St1 = St#st{pst = PSt1},
+ case Go of
+ ok -> {ok, St1};
+ skip -> {ok, St1};
+ stop -> {stop, St1}
+ end;
+ true ->
+ {ok, St}
+ end.
+
+scan_docs(#st{} = St, #shard{name = ShardDbName}) ->
+ case couch_db:open_int(ShardDbName, [?ADMIN_CTX]) of
+ {ok, Db} ->
+ St1 = St#st{db = Db},
+ St2 = db_opened_callback(St1),
+ {ok, St3} = couch_db:fold_docs(Db, fun scan_docs_fold/2, St2, []),
+ St4 = db_closing_callback(St3),
+ couch_db:close(Db),
+ St4#st{db = undefined};
+ {not_found, _} ->
+ St
+ end.
+
+scan_docs_fold(#full_doc_info{id = Id} = FDI, #st{} = St) ->
+ #st{db = Db, mod = Mod, pst = PSt, skip_dbs = Skip} = St,
+ case match_skip_pat(Id, Skip) of
+ false ->
+ {Go, PSt1} = Mod:doc_id(PSt, Id, Db),
+ St1 = St#st{pst = PSt1},
+ case Go of
+ ok -> scan_doc(FDI, St1);
+ skip -> {ok, St1};
+ stop -> {stop, St1}
+ end;
+ true ->
+ {ok, St}
+ end.
+
+scan_doc(#full_doc_info{} = FDI, #st{} = St) ->
+ St1 = maybe_checkpoint(St),
+ #st{db = Db, mod = Mod, pst = PSt} = St1,
+ {ok, #doc{} = Doc} = couch_db:open_doc(Db, FDI, [ejson_body]),
+ {Go, PSt1} = Mod:doc(PSt, Db, Doc),
+ case Go of
+ ok -> {ok, St1#st{pst = PSt1}};
+ stop -> {stop, St1#st{pst = PSt1}}
+ end.
+
+maybe_checkpoint(#st{checkpoint_sec = LastCheckpointTSec} = St) ->
+ receive
+ stop ->
+ checkpoint(St),
+ exit({shutdown, stop})
+ after 0 ->
+ ok
+ end,
+ erlang:garbage_collect(),
+ case tsec() - LastCheckpointTSec > ?CHECKPOINT_INTERVAL_SEC of
+ true -> checkpoint(St);
+ false -> St
+ end.
+
+checkpoint(#st{} = St) ->
+ #st{
+ id = Id,
+ mod = Mod,
+ pst = PSt,
+ cursor = Cursor,
+ start_sec = StartSec,
+ scan_id = SId
+ } = St,
+ JsonPSt = checkpoint_callback(Mod, PSt),
+ EJson = #{
+ <<"cursor">> => Cursor,
+ <<"pst">> => JsonPSt,
+ <<"state">> => <<"running">>,
+ <<"scan_id">> => SId,
+ <<"start_sec">> => StartSec
+ },
+ ok = couch_scanner_checkpoint:write(Id, EJson),
+ St#st{checkpoint_sec = tsec()}.
+
+finalize(#st{} = St) ->
+ #st{
+ id = Id,
+ mod = Mod,
+ pst = PSt,
+ start_sec = StartSec,
+ scan_id = SId
+ } = St,
+ {ok, #{} = PStEJson} = Mod:stop(PSt),
+ EJson = #{
+ <<"cursor">> => <<>>,
+ <<"pst">> => ejson_map(PStEJson),
+ <<"state">> => <<"finished">>,
+ <<"scan_id">> => SId,
+ <<"start_sec">> => StartSec
+ },
+ ok = couch_scanner_checkpoint:write(Id, EJson),
+ case schedule_time(Mod, StartSec, tsec()) of
+ infinity -> ok;
+ TSec when is_integer(TSec) -> exit_resched(TSec)
+ end.
+
+exit_resched(TimeSec) when is_integer(TimeSec) orelse TimeSec == infinity ->
+ exit({shutdown, {reschedule, TimeSec}}).
+
+% Call plugin module API functions
+
+start_callback(Mod, Now, ScanId, LastStartSec, #{} = EJson) when
+ is_atom(Mod), is_binary(ScanId), is_integer(LastStartSec)
+->
+ case schedule_time(Mod, LastStartSec, Now) of
+ infinity ->
+ exit_resched(infinity);
+ TSec when is_integer(TSec), TSec =< Now ->
+ case Mod:start(ScanId, EJson) of
+ {ok, PSt} -> PSt;
+ skip -> exit_resched(infinity)
+ end;
+ TSec when is_integer(TSec), TSec > Now ->
+ exit_resched(TSec)
+ end.
+
+resume_callback(Mod, ScanId, #{} = EJson) when
+ is_atom(Mod), is_binary(ScanId)
+->
+ case Mod:resume(ScanId, EJson) of
+ {ok, PSt} -> PSt;
+ skip -> exit_resched(infinity)
+ end.
+
+db_opened_callback(#st{pst = PSt, mod = Mod, db = Db} = St) ->
+ {ok, PSt1} = Mod:db_opened(PSt, Db),
+ St#st{pst = PSt1}.
+
+db_closing_callback(#st{pst = PSt, mod = Mod, db = Db} = St) ->
+ {ok, PSt1} = Mod:db_closing(PSt, Db),
+ St#st{pst = PSt1}.
+
+shards_callback(#st{pst = PSt, mod = Mod} = St, Shards) ->
+ {Shards1, PSt1} = Mod:shards(PSt, Shards),
+ {Shards1, St#st{pst = PSt1}}.
+
+start_checkpoint(Id, StartSec, ScanId, Mod, Cur, PSt1) when
+ is_atom(Mod), is_binary(Id), is_binary(ScanId), is_integer(StartSec)
+->
+ PStEJson = checkpoint_callback(Mod, PSt1),
+ EJson = #{
+ <<"cursor">> => Cur,
+ <<"pst">> => PStEJson,
+ <<"state">> => <<"running">>,
+ <<"scan_id">> => ScanId,
+ <<"start_sec">> => StartSec
+ },
+ ok = couch_scanner_checkpoint:write(Id, EJson).
+
+checkpoint_callback(Mod, PSt) when is_atom(Mod) ->
+ {ok, #{} = PStEJson} = Mod:checkpoint(PSt),
+ ejson_map(PStEJson).
+
+% Plugin discovery, loading and validation
+
+plugin_mod(<<Plugin/binary>>) ->
+ Mod = binary_to_atom(Plugin),
+ case code:ensure_loaded(Mod) of
+ {module, _} ->
+ check_callbacks(Mod),
+ Mod;
+ {error, Error} ->
+ error({?MODULE, {missing_plugin_module, Mod, Error}})
+ end.
+
+check_callbacks(Mod) when is_atom(Mod) ->
+ Cbks = [
+ {start, 2},
+ {resume, 2},
+ {stop, 1},
+ {checkpoint, 1},
+ {db, 2},
+ {ddoc, 3},
+ {shards, 2},
+ {db_opened, 2},
+ {doc_id, 3},
+ {doc, 3},
+ {db_closing, 2}
+ ],
+ Fun = fun({F, A}) ->
+ case erlang:function_exported(Mod, F, A) of
+ true -> ok;
+ false -> error({?MODULE, {undefined_plugin_fun, Mod, F, A}})
+ end
+ end,
+ lists:foreach(Fun, Cbks).
+
+% Shard selection
+
+shards(DbName, {Props = [_ | _]}) ->
+ Shards = lists:sort(mem3_util:build_shards(DbName, Props)),
+ Fun = fun({R, SList}) ->
+ case mem3_util:rotate_list({DbName, R}, SList) of
+ [#shard{node = N} = S | _] when N =:= node() ->
+ {true, S};
+ [_ | _] ->
+ false
+ end
+ end,
+ lists:filtermap(Fun, shards_by_range(lists:sort(Shards))).
+
+shards_by_range(Shards) ->
+ Fun = fun(#shard{range = R} = S, Acc) -> orddict:append(R, S, Acc) end,
+ Dict = lists:foldl(Fun, orddict:new(), Shards),
+ orddict:to_list(Dict).
+
+% Design doc fetching helper
+
+fold_ddocs(Fun, #st{cursor = DbName} = Acc) ->
+ QArgs = #mrargs{
+ include_docs = true,
+ extra = [{namespace, <<"_design">>}]
+ },
+ {ok, Acc1} = fabric:all_docs(DbName, [?ADMIN_CTX], Fun, Acc, QArgs),
+ Acc1.
+
+% Skip patterns
+
+config_match_patterns(Module, Type) ->
+ Section = atom_to_list(Module) ++ "." ++ Type,
+ Fun = fun
+ ({K, "true"}, Acc) -> [list_to_binary(K) | Acc];
+ ({_, _}, Acc) -> Acc
+ end,
+ Items = lists:foldl(Fun, [], config:get(Section)),
+ case Items of
+ [] -> undefined;
+ [<<_/binary>> | _] -> binary:compile_pattern(Items)
+ end.
+
+match_skip_pat(<<_/binary>>, undefined) ->
+ false;
+match_skip_pat(<<_/binary>> = Bin, Pat) ->
+ binary:match(Bin, Pat) /= nomatch.
+
+% Time parsing functions
+
+cfg(Mod, Key, Default) when is_list(Key) ->
+ Section = atom_to_list(Mod),
+ config:get(Section, Key, Default).
+
+schedule_time(Mod, LastSec, NowSec) ->
+ After = cfg(Mod, "after", "restart"),
+ Repeat = cfg(Mod, "repeat", "restart"),
+ schedule_time(NowSec, LastSec, After, Repeat).
+
+schedule_time(Now, Last, AfterCfg, RepeatCfg) when
+ is_integer(Now), is_integer(Last), is_list(AfterCfg), is_list(RepeatCfg)
+->
+ RepeatPeriod = repeat_period(Now, Last, parse_repeat(RepeatCfg)),
+ Restart = restart_tsec(),
+ case {parse_after(AfterCfg), RepeatPeriod} of
+ {undefined, undefined} when Last >= Restart ->
+ % Run after restart, and already ran
+ infinity;
+ {undefined, undefined} when Last < Restart ->
+ % Run after restart, but haven't run yet
+ Now;
+ {After, undefined} when is_integer(After), Last > After ->
+ % Run after an absolute timestamp, and already ran
+ infinity;
+ {After, undefined} when is_integer(After), Last =< After ->
+ % Run once, haven't run yet, schedule to run
+ max(Now, After);
+ {undefined, Period} ->
+ % No after time, just period. Either need to wait
+ % since last time it ran, or is actually ready to run
+ max(Now, Last + Period);
+ {After, Period} ->
+ % Both after time set and a period. Wait for whichever
+ % takes the longest
+ lists:max([Now, After, Last + Period])
+ end.
+
+tsec() ->
+ erlang:system_time(second).
+
+parse_after(Time) when is_list(Time), length(Time) >= 7 ->
+ case string:uppercase(Time) of
+ "RESTART" ->
+ undefined;
+ [_, _, _, _, $-, _, _, $-, _, _] = T ->
+ parse_rfc3339(T ++ "T00:00:00Z");
+ [_, _, _, _, $-, _, _, $-, _, _, $T, _, _] = T ->
+ parse_rfc3339(T ++ ":00:00Z");
+ [_, _, _, _, $-, _, _, $-, _, _, $T, _, _, $:, _, _] = T ->
+ parse_rfc3339(T ++ ":00Z");
+ [_, _, _, _, $-, _, _, $-, _, _, $T, _, _, $:, _, _, $:, _, _] = T ->
+ parse_rfc3339(T ++ "Z");
+ [_, _, _, _, $-, _, _, $-, _, _, $T, _, _, $:, _, _, $:, _, _ | _] = T ->
+ parse_rfc3339(T);
+ [_ | _] = T ->
+ parse_unix(T)
+ end.
+
+parse_rfc3339(Time) ->
+ try calendar:rfc3339_to_system_time(Time) of
+ Sec when is_integer(Sec), Sec > 0 ->
+ Sec;
+ Sec when is_integer(Sec) ->
+ undefined
+ catch
+ error:_ ->
+ undefined
+ end.
+
+parse_unix(Time) ->
+ try list_to_integer(Time) of
+ Sec when is_integer(Sec), Sec > 0 ->
+ Sec;
+ Sec when is_integer(Sec) ->
+ undefined
+ catch
+ error:badarg ->
+ undefined
+ end.
+
+repeat_period(_Now, _Last, undefined) ->
+ undefined;
+repeat_period(Now, Last, {weekday, WeekdayNum}) ->
+ {NowDate, DayTime} = calendar:system_time_to_universal_time(Now, second),
+ case abs(calendar:day_of_the_week(NowDate) - WeekdayNum) rem 7 of
+ 0 ->
+ % It's today. Run only if it hasn't started yet
+ DayStartSec = calendar:datetime_to_gregorian_seconds({{0, 0, 0}, DayTime}),
+ DayStartUnixTSec = Now - DayStartSec,
+ case Last > DayStartUnixTSec of
+ true -> ?WEEK;
+ false -> 0
+ end;
+ Days ->
+ ?DAY * Days
+ end;
+repeat_period(_Now, _Last, Period) when is_integer(Period), Period > 0 ->
+ Period.
+
+parse_repeat(Repeat) when is_list(Repeat) ->
+ % Numbering follows https://www.erlang.org/doc/man/calendar#day_of_the_week-1
+ case string:lowercase(Repeat) of
+ "restart" -> undefined;
+ "mon" ++ _ -> {weekday, 1};
+ "tue" ++ _ -> {weekday, 2};
+ "wed" ++ _ -> {weekday, 3};
+ "thu" ++ _ -> {weekday, 4};
+ "fri" ++ _ -> {weekday, 5};
+ "sat" ++ _ -> {weekday, 6};
+ "sun" ++ _ -> {weekday, 7};
+ Val -> parse_non_weekday_period(Val)
+ end.
+
+parse_non_weekday_period(Period) ->
+ case string:split(Period, "_") of
+ [UnitStr] ->
+ % "hour" -> ?HOUR
+ parse_period_unit(UnitStr);
+ [NumStr, UnitStr] ->
+ Unit = parse_period_unit(UnitStr),
+ % 10_hours -> ["10", "hours"] -> 10 * ?HOUR
+ try list_to_integer(NumStr) of
+ Num when is_integer(Num), Num > 0 ->
+ Num * Unit;
+ Num when is_integer(Num) ->
+ undefined
+ catch
+ error:badarg ->
+ undefined
+ end
+ end.
+
+parse_period_unit(Period) when is_list(Period) ->
+ case Period of
+ "min" ++ _ -> ?MINUTE;
+ "hour" ++ _ -> ?HOUR;
+ "day" ++ _ -> ?DAY;
+ "week" ++ _ -> ?WEEK;
+ "month" ++ _ -> ?MONTH;
+ _ -> 1
+ end.
+
+restart_tsec() ->
+ Native = erlang:system_info(start_time) + erlang:time_offset(),
+ erlang:convert_time_unit(Native, native, second).
+
+% Logging bits
+
+log_format_meta(Mod, #{} = Meta) ->
+ SId = {"s:~s ", maps:get(sid, Meta, undefined)},
+ Fun = {"f:~s ", maps:get(fn, Meta, undefined)},
+ Db = {"db:~s ", format_db(maps:get(db, Meta, undefined))},
+ DDocId = {"ddoc:~s ", maps:get(ddoc, Meta, undefined)},
+ DocId = {"doc:~s ", maps:get(doc, Meta, undefined)},
+ FmtArgs = [{"~s ", Mod}, SId, Fun, Db, DDocId, DocId],
+ lists:unzip([{Fmt, Arg} || {Fmt, Arg} <- FmtArgs, Arg /= undefined]).
+
+format_db(undefined) ->
+ undefined;
+format_db(Db) when is_list(Db) ->
+ format_db(list_to_binary(Db));
+format_db(#shard{dbname = Db, range = [B, E]}) ->
+ {BStr, EStr} = {hex(B), hex(E)},
+ <<Db/binary, "/", BStr/binary, "-", EStr/binary>>;
+format_db(Db) when is_tuple(Db) ->
+ format_db(couch_db:name(Db));
+format_db(<<"shards/", B:8/binary, "-", E:8/binary, "/", Rest/binary>>) ->
+ [Db, _] = binary:split(Rest, <<".">>),
+ <<Db/binary, "/", B/binary, "-", E/binary>>;
+format_db(<<Db/binary>>) ->
+ Db.
+
+hex(Val) when is_integer(Val) ->
+ string:lowercase(erlang:integer_to_binary(Val, 16)).
diff --git a/src/couch_scanner/src/couch_scanner_plugin_ddoc_features.erl b/src/couch_scanner/src/couch_scanner_plugin_ddoc_features.erl
new file mode 100644
index 000000000..a9309343d
--- /dev/null
+++ b/src/couch_scanner/src/couch_scanner_plugin_ddoc_features.erl
@@ -0,0 +1,252 @@
+% 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.
+
+% Scanner plugin to detect various design document features
+%
+% By default when enabled it will scan design documents for features slated to be
+% deprecated in 4.0 such as: rewrites, lists, shows, updates. There are options
+% to enable other feature checks: custom JS reducers, JS libraries, VDU functions,
+% and JS filters.
+%
+% By default the scanning will start and run on the first node of the cluster only.
+% It's possible to make the plugin run on all the and each node will pick a fraction
+% of dbs to scan. That will make the scan go fast but might consume more resources. That options
+% is contorlled via the run_on_first_node boolean setting.
+%
+% When features are detected they are aggregated per database and reported once all the
+% design documents for a particular database have been processed. To get details about
+% the specific design document enable the ddoc_report = true setting.
+%
+
+-module(couch_scanner_plugin_ddoc_features).
+-behaviour(couch_scanner_plugin).
+
+-export([
+ start/2,
+ resume/2,
+ stop/1,
+ checkpoint/1,
+ db/2,
+ ddoc/3,
+ shards/2,
+ db_opened/2,
+ doc_id/3,
+ doc/3,
+ db_closing/2
+]).
+
+-include_lib("couch_scanner/include/couch_scanner_plugin.hrl").
+
+-record(st, {
+ sid,
+ db,
+ report = #{},
+ updates = true,
+ rewrites = true,
+ shows = true,
+ lists = true,
+ filters = false,
+ libs = false,
+ jsreducers = false,
+ vdus = false,
+ run_on_first_node = true,
+ ddoc_report = false
+}).
+
+% DDoc fields
+
+-define(FILTERS, <<"filters">>).
+-define(VIEWS, <<"views">>).
+-define(REDUCE, <<"reduce">>).
+-define(LIB, <<"lib">>).
+-define(VDU, <<"validate_doc_update">>).
+-define(UPDATES, <<"updates">>).
+-define(SHOWS, <<"shows">>).
+-define(LISTS, <<"lists">>).
+-define(REWRITES, <<"rewrites">>).
+
+% Behavior callbacks
+
+start(SId, #{}) ->
+ St = init_config(#st{sid = SId}),
+ case should_run(St) of
+ true ->
+ ?INFO("Starting.", [], #{sid => SId}),
+ {ok, St};
+ false ->
+ ?INFO("Not starting. Not on first node.", [], #{sid => SId}),
+ skip
+ end.
+
+resume(SId, #{}) ->
+ St = init_config(#st{sid = SId}),
+ case should_run(St) of
+ true ->
+ ?INFO("Resuming.", [], #{sid => SId}),
+ {ok, St};
+ false ->
+ ?INFO("Not resuming. Not on first node.", [], #{sid => SId}),
+ skip
+ end.
+
+stop(#st{sid = SId, db = Db, report = Total} = St) ->
+ report_per_db(St, Db, Total),
+ ?INFO("Stopped", [], #{sid => SId}),
+ {ok, #{<<"sid">> => SId}}.
+
+checkpoint(#st{sid = SId}) ->
+ ?INFO(#{sid => SId}),
+ {ok, #{<<"sid">> => SId}}.
+
+db(#st{} = St, DbName) ->
+ % If we run on all nodes spread db checks across nodes
+ case should_pick_db(DbName, St) of
+ true -> {ok, St};
+ false -> {skip, St}
+ end.
+
+ddoc(#st{} = St, _DbName, #doc{id = <<"_design/_", _/binary>>}) ->
+ % These are auto-inserted ddocs _design/_auth, etc.
+ {ok, St};
+ddoc(#st{} = St, DbName, #doc{} = DDoc) ->
+ #doc{body = {Props = [_ | _]}} = DDoc,
+ case couch_util:get_value(<<"language">>, Props, <<"javascript">>) of
+ <<"javascript">> -> {ok, check_ddoc(St, DbName, DDoc)};
+ _ -> {ok, St}
+ end.
+
+shards(#st{} = St, _Shards) ->
+ % Stop scanning here by filtering out all the shards. The rest of the
+ % callbacks are just boilerplate
+ {[], St}.
+
+db_opened(#st{} = St, _Db) ->
+ {ok, St}.
+
+doc_id(#st{} = St, _DocId, _Db) ->
+ {stop, St}.
+
+doc(#st{} = St, _Db, _Doc) ->
+ {ok, St}.
+
+db_closing(#st{} = St, _Db) ->
+ {ok, St}.
+
+% Private
+
+init_config(#st{} = St) ->
+ St#st{
+ updates = cfg_bool("updates", St#st.updates),
+ rewrites = cfg_bool("rewrites", St#st.rewrites),
+ shows = cfg_bool("shows", St#st.shows),
+ lists = cfg_bool("lists", St#st.lists),
+ filters = cfg_bool("filters", St#st.filters),
+ libs = cfg_bool("libs", St#st.libs),
+ jsreducers = cfg_bool("jsreducers", St#st.jsreducers),
+ vdus = cfg_bool("vdus", St#st.vdus),
+ run_on_first_node = cfg_bool("run_first_node", St#st.run_on_first_node),
+ ddoc_report = cfg_bool("ddoc_report", St#st.ddoc_report)
+ }.
+
+cfg_bool(Key, Default) when is_list(Key), is_boolean(Default) ->
+ config:get_boolean(atom_to_list(?MODULE), Key, Default).
+
+should_run(#st{run_on_first_node = true}) ->
+ hd(mem3_util:live_nodes()) =:= node();
+should_run(#st{run_on_first_node = false}) ->
+ true.
+
+should_pick_db(_DbName, #st{run_on_first_node = true}) ->
+ true;
+should_pick_db(DbName, #st{run_on_first_node = false}) ->
+ Nodes = mem3_util:live_nodes(),
+ hd(mem3_util:rotate_list(DbName, Nodes)) =:= node().
+
+check_ddoc(#st{} = St, DbName, #doc{} = DDoc0) ->
+ #doc{id = DDocId, body = Body} = DDoc0,
+ DDoc = couch_scanner_plugin:ejson_map(Body),
+ {St1, Report} = maps:fold(fun check/3, {St, #{}}, DDoc),
+ report(St1, DbName, DDocId, Report).
+
+check(?UPDATES, #{} = Map, {#st{updates = true} = St, Acc}) ->
+ bump(St, updates, map_size(Map), Acc);
+check(?REWRITES, #{} = Map, {#st{rewrites = true} = St, Acc}) ->
+ bump(St, rewrites, map_size(Map), Acc);
+check(?SHOWS, #{} = Map, {#st{shows = true} = St, Acc}) ->
+ bump(St, shows, map_size(Map), Acc);
+check(?LISTS, #{} = Map, {#st{lists = true} = St, Acc}) ->
+ bump(St, lists, map_size(Map), Acc);
+check(?FILTERS, #{} = Map, {#st{filters = true} = St, Acc}) ->
+ bump(St, filters, map_size(Map), Acc);
+check(?VDU, _, {#st{vdus = true} = St, Acc}) ->
+ bump(St, vdus, 1, Acc);
+check(?LIB, _, {#st{libs = true} = St, Acc}) ->
+ bump(St, libs, 1, Acc);
+check(?VIEWS, #{} = Views, {#st{jsreducers = true} = St, Acc}) ->
+ maps:fold(fun check_view/3, {St, Acc}, Views);
+check(<<_/binary>>, _, {#st{} = St, Acc}) ->
+ {St, Acc}.
+
+check_view(_Name, #{?REDUCE := <<"_", _/binary>>}, {#st{} = St, Acc}) ->
+ % Built-in reducers
+ {St, Acc};
+check_view(_Name, #{?REDUCE := <<_/binary>>}, {#st{} = St, Acc}) ->
+ bump(St, jsreducers, 1, Acc);
+check_view(_Name, _, {#st{} = St, Acc}) ->
+ {St, Acc}.
+
+bump(#st{} = St, Field, N, #{} = Map) ->
+ {St, maps:update_with(Field, fun(V) -> V + N end, N, Map)}.
+
+report(#st{} = St, _, _, #{} = Report) when map_size(Report) == 0 ->
+ St;
+report(#st{} = St, Db, DDocId, #{} = Report) ->
+ #st{report = Total, db = PrevDb} = St,
+ report_per_ddoc(#st{} = St, Db, DDocId, Report),
+ case is_binary(PrevDb) andalso Db =/= PrevDb of
+ true ->
+ % We switched dbs, so report stats for old db
+ % and make the new one the current one
+ report_per_db(St, PrevDb, Total),
+ St#st{report = Report, db = Db};
+ false ->
+ % Keep accumulating per-db stats
+ St#st{report = merge_report(Total, Report), db = Db}
+ end.
+
+merge_report(#{} = Total, #{} = Update) ->
+ Fun = fun(_K, V1, V2) -> V1 + V2 end,
+ maps:merge_with(Fun, Total, Update).
+
+report_per_db(#st{sid = SId}, Db, #{} = Report) when
+ map_size(Report) > 0, is_binary(Db)
+->
+ {Fmt, Args} = report_fmt(Report),
+ Meta = #{sid => SId, db => Db},
+ ?WARN(Fmt, Args, Meta);
+report_per_db(#st{}, _, _) ->
+ ok.
+
+report_per_ddoc(#st{ddoc_report = false}, _Db, _DDocId, _Report) ->
+ ok;
+report_per_ddoc(#st{ddoc_report = true, sid = SId}, Db, DDocId, Report) ->
+ {Fmt, Args} = report_fmt(Report),
+ Meta = #{sid => SId, db => Db, ddoc => DDocId},
+ ?WARN(Fmt, Args, Meta).
+
+report_fmt(Report) ->
+ Sorted = lists:sort(maps:to_list(Report)),
+ FmtArgs = [{"~s:~p ", [K, V]} || {K, V} <- Sorted],
+ {Fmt1, Args1} = lists:unzip(FmtArgs),
+ Fmt2 = lists:flatten(Fmt1),
+ Args2 = lists:flatten(Args1),
+ {Fmt2, Args2}.
diff --git a/src/couch_scanner/src/couch_scanner_server.erl b/src/couch_scanner/src/couch_scanner_server.erl
new file mode 100644
index 000000000..f7b294895
--- /dev/null
+++ b/src/couch_scanner/src/couch_scanner_server.erl
@@ -0,0 +1,242 @@
+% 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.
+
+% Scanner plugin server.
+%
+% This gen_server starts, stops and reschedules plugin processes.
+%
+
+-module(couch_scanner_server).
+
+-export([
+ start_link/0,
+ status/0,
+ stop/0,
+ resume/0,
+ checkpoints/0,
+ reset_checkpoints/0
+]).
+
+-export([
+ init/1,
+ terminate/2,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2
+]).
+
+-define(SCHED_INTERVAL_SEC, 5).
+-define(PENALTY_BASE_SEC, 60).
+-define(MAX_PENALTY_SEC, 8 * 3600).
+-define(HEAL_THRESHOLD_SEC, 5 * 60).
+
+-record(sched, {
+ start_time = 0,
+ error_count = 0,
+ reschedule = 0
+}).
+
+-record(st, {
+ stopped = false,
+ pids = #{},
+ scheduling = #{},
+ tref
+}).
+
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+status() ->
+ gen_server:call(?MODULE, status, infinity).
+
+stop() ->
+ gen_server:call(?MODULE, stop, infinity).
+
+resume() ->
+ gen_server:call(?MODULE, resume, infinity).
+
+checkpoints() ->
+ couch_scanner_checkpoint:list_all(<<"couch_scanner_plugin_">>).
+
+reset_checkpoints() ->
+ Fun = fun(Id, #{}) -> couch_scanner_checkpoint:reset(Id) end,
+ maps:map(Fun, checkpoints()).
+
+% Private
+
+init(_Args) ->
+ process_flag(trap_exit, true),
+ {ok, #st{}, interval_msec()}.
+
+terminate(_Reason, #st{pids = Pids} = St) ->
+ ToStop = maps:keys(Pids),
+ lists:foldl(fun stop_plugin/2, St, ToStop),
+ ok.
+
+handle_call(status, _From, #st{} = St) ->
+ #st{pids = Pids, stopped = Stopped} = St,
+ Fun = fun(_, Sched) ->
+ #{
+ start_time => Sched#sched.start_time,
+ error_count => Sched#sched.error_count,
+ reschedule => Sched#sched.reschedule
+ }
+ end,
+ Scheds = maps:map(Fun, St#st.scheduling),
+ {reply, #{pids => Pids, scheduling => Scheds, stopped => Stopped}, St};
+handle_call(stop, _From, #st{} = St) ->
+ St1 = St#st{stopped = true},
+ St2 = stop_in_maintenance(St1),
+ {reply, ok, St2};
+handle_call(resume, _From, #st{} = St) ->
+ St1 = start_stop(St#st{stopped = false}),
+ {reply, ok, St1};
+handle_call(Msg, _From, #st{} = St) ->
+ couch_log:error("~p : unknown call ~p", [?MODULE, Msg]),
+ {reply, {error, {invalid_call, Msg}}, St}.
+
+handle_cast(Msg, #st{} = St) ->
+ couch_log:error("~p : unknown cast ~p", [?MODULE, Msg]),
+ {noreply, St}.
+
+handle_info(timeout, #st{} = St) ->
+ St1 = start_stop(St),
+ {noreply, schedule_timeout(St1)};
+handle_info({'EXIT', Pid, Reason}, #st{pids = Pids} = St) ->
+ case maps:filter(fun(_, P) -> P =:= Pid end, Pids) of
+ Map when map_size(Map) == 1 ->
+ [{Id, Pid}] = maps:to_list(Map),
+ St1 = St#st{pids = maps:remove(Id, Pids)},
+ {noreply, handle_exit(Id, Reason, St1)};
+ Map when map_size(Map) == 0 ->
+ {noreply, St}
+ end;
+handle_info(Msg, St) ->
+ couch_log:error("~p : unknown info message ~p", [?MODULE, Msg]),
+ {noreply, St}.
+
+stop_in_maintenance(#st{pids = Pids} = St) ->
+ case map_size(Pids) > 0 of
+ true ->
+ couch_log:info("~p stopping in maintenance mode", [?MODULE]),
+ lists:foldl(fun stop_plugin/2, St, maps:keys(Pids));
+ false ->
+ St
+ end.
+
+start_stop(#st{stopped = Stopped} = St) ->
+ case in_maintenance() orelse Stopped of
+ true -> stop_in_maintenance(St);
+ false -> start_stop_cfg(St)
+ end.
+
+start_stop_cfg(#st{pids = Pids, scheduling = Scheduling} = St) ->
+ PluginIds = plugins(),
+ RunningIds = maps:keys(Pids),
+ ToStart = PluginIds -- RunningIds,
+ ToStop = RunningIds -- PluginIds,
+ St1 = lists:foldl(fun stop_plugin/2, St, ToStop),
+ lists:foreach(fun couch_scanner_checkpoint:reset/1, ToStop),
+ ToRemove = maps:keys(Scheduling) -- PluginIds,
+ St2 = St1#st{scheduling = maps:without(ToRemove, Scheduling)},
+ lists:foldl(fun start_plugin/2, St2, ToStart).
+
+stop_plugin(Id, #st{} = St) ->
+ #st{pids = Pids, scheduling = Scheduling} = St,
+ {Pid, Pids1} = maps:take(Id, Pids),
+ #{Id := #sched{} = Sched} = Scheduling,
+ couch_log:info("~p : stopping ~s", [?MODULE, Id]),
+ ok = couch_scanner_plugin:stop(Pid),
+ Sched1 = Sched#sched{start_time = 0, reschedule = 0},
+ Scheduling1 = Scheduling#{Id := Sched1},
+ St#st{pids = Pids1, scheduling = Scheduling1}.
+
+start_plugin(Id, #st{pids = Pids, scheduling = Scheduling} = St) ->
+ Sched = maps:get(Id, Scheduling, #sched{}),
+ Now = tsec(),
+ case Now >= Sched#sched.reschedule of
+ true ->
+ couch_log:info("~p : starting ~s", [?MODULE, Id]),
+ Pids1 = Pids#{Id => couch_scanner_plugin:spawn_link(Id)},
+ Sched1 = Sched#sched{
+ start_time = Now,
+ reschedule = 0
+ },
+ Scheduling1 = Scheduling#{Id => Sched1},
+ St#st{pids = Pids1, scheduling = Scheduling1};
+ false ->
+ St
+ end.
+
+plugins() ->
+ Fun = fun
+ ({Module, "true"}, Acc) -> [list_to_binary(Module) | Acc];
+ ({_, _}, Acc) -> Acc
+ end,
+ lists:foldl(Fun, [], config:get("couch_scanner_plugins")).
+
+handle_exit(Id, Reason, #st{} = St) ->
+ #st{scheduling = Scheduling} = St,
+ #{Id := Sched} = Scheduling,
+ Sched1 = sched_exit_update(Id, Sched, Reason),
+ St#st{scheduling = Scheduling#{Id := Sched1}}.
+
+sched_exit_update(Id, #sched{} = Sched, {shutdown, {reschedule, TSec}}) ->
+ couch_log:notice("~p : ~s rescheduled after ~p", [?MODULE, Id, TSec]),
+ Sched#sched{start_time = 0, error_count = 0, reschedule = TSec};
+sched_exit_update(Id, #sched{} = Sched, Norm) when
+ Norm == shutdown; Norm == normal
+->
+ couch_log:notice("~p : ~s finished", [?MODULE, Id]),
+ Sched#sched{start_time = 0, error_count = 0, reschedule = infinity};
+sched_exit_update(Id, #sched{} = Sched, Error) ->
+ couch_log:error("~p : ~s exited with error ~p", [?MODULE, Id, Error]),
+ #sched{start_time = Start, error_count = ErrorCount} = Sched,
+ Sched1 = Sched#sched{start_time = 0},
+ Now = tsec(),
+ % If process has been running successfully for a while without crashing
+ % reset (forgive) its previous errors.
+ case Now - Start =< heal_threshold_sec() of
+ true -> penalize(Now, Sched1#sched{error_count = 1});
+ false -> penalize(Now, Sched1#sched{error_count = ErrorCount + 1})
+ end.
+
+penalize(Now, #sched{error_count = ErrorCount} = Sched) ->
+ Penalty = ?PENALTY_BASE_SEC * (1 bsl (ErrorCount - 1)),
+ Penalty1 = min(Penalty, max_penalty_sec()),
+ Sched#sched{reschedule = Now + Penalty1}.
+
+in_maintenance() ->
+ "false" /= config:get("couchdb", "maintenance_mode", "false").
+
+tsec() ->
+ erlang:system_time(second).
+
+schedule_timeout(#st{tref = TRef} = St) ->
+ case TRef of
+ undefined -> ok;
+ _ when is_reference(TRef) -> erlang:cancel_timer(TRef)
+ end,
+ TRef1 = erlang:send_after(interval_msec(), self(), timeout),
+ St#st{tref = TRef1}.
+
+interval_msec() ->
+ 1000 * cfg_int("sched_interval_sec", ?SCHED_INTERVAL_SEC).
+
+heal_threshold_sec() ->
+ cfg_int("heal_threshold_sec", ?HEAL_THRESHOLD_SEC).
+
+max_penalty_sec() ->
+ cfg_int("max_penalty_sec", ?MAX_PENALTY_SEC).
+
+cfg_int(Key, Default) when is_list(Key), is_integer(Default) ->
+ config:get_integer("couch_scanner", Key, Default).
diff --git a/src/couch_scanner/src/couch_scanner_sup.erl b/src/couch_scanner/src/couch_scanner_sup.erl
new file mode 100644
index 000000000..7baf0037a
--- /dev/null
+++ b/src/couch_scanner/src/couch_scanner_sup.erl
@@ -0,0 +1,34 @@
+% 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_scanner_sup).
+
+-behaviour(supervisor).
+
+-export([
+ start_link/0,
+ init/1
+]).
+
+start_link() ->
+ supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+init([]) ->
+ Children = [
+ #{
+ id => couch_scanner_server,
+ start => {couch_scanner_server, start_link, []},
+ shutdown => 5000
+ }
+ ],
+ SupFlags = #{strategy => rest_for_one, intensity => 25, period => 1},
+ {ok, {SupFlags, Children}}.
diff --git a/src/couch_scanner/test/eunit/couch_scanner_test.erl b/src/couch_scanner/test/eunit/couch_scanner_test.erl
new file mode 100644
index 000000000..559827109
--- /dev/null
+++ b/src/couch_scanner/test/eunit/couch_scanner_test.erl
@@ -0,0 +1,34 @@
+% 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_scanner_test).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+couch_scanner_test_() ->
+ {
+ foreach,
+ fun setup/0,
+ fun teardown/1,
+ [
+ ?TDEF_FE(t_scanner)
+ ]
+ }.
+
+setup() ->
+ test_util:start_couch().
+
+teardown(Ctx) ->
+ test_util:stop_couch(Ctx).
+
+t_scanner(_Ctx) ->
+ ok.