You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by ja...@apache.org on 2023/07/22 08:38:26 UTC
[couchdb] branch main updated (4df0dc45e -> 480d3a3cb)
This is an automated email from the ASF dual-hosted git repository.
jan pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/couchdb.git
from 4df0dc45e docs: add Centos/RHEL9+, thanks for nothing RedHat
new d2edfd9a3 feat: allow restarting of failed jobs
new 480d3a3cb feat: make rexi timeouts and replication batches for shard split topoff configurable
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
rel/overlay/etc/default.ini | 2 ++
src/mem3/src/mem3_rep.erl | 41 ++++++++++++++++++---------
src/mem3/src/mem3_reshard.erl | 2 +-
src/mem3/src/mem3_reshard_job.erl | 13 +++++++--
src/mem3/src/mem3_rpc.erl | 3 ++
src/mem3/test/eunit/mem3_reshard_api_test.erl | 32 +++++++++++++++++++++
6 files changed, 77 insertions(+), 16 deletions(-)
[couchdb] 01/02: feat: allow restarting of failed jobs
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit d2edfd9a32956b439f96502ae6fa9a46369a8b27
Author: Jan Lehnardt <ja...@apache.org>
AuthorDate: Wed Jul 19 14:07:12 2023 +0200
feat: allow restarting of failed jobs
---
src/mem3/src/mem3_reshard.erl | 2 +-
src/mem3/test/eunit/mem3_reshard_api_test.erl | 32 +++++++++++++++++++++++++++
2 files changed, 33 insertions(+), 1 deletion(-)
diff --git a/src/mem3/src/mem3_reshard.erl b/src/mem3/src/mem3_reshard.erl
index ec08c72cd..b3bed69ff 100644
--- a/src/mem3/src/mem3_reshard.erl
+++ b/src/mem3/src/mem3_reshard.erl
@@ -266,7 +266,7 @@ handle_call({resume_job, _}, _From, #state{state = stopped} = State) ->
handle_call({resume_job, Id}, _From, State) ->
couch_log:notice("~p resume_job call ~p", [?MODULE, Id]),
case job_by_id(Id) of
- #job{job_state = stopped} = Job ->
+ #job{job_state = JobState} = Job when JobState == stopped; JobState == failed ->
case start_job_int(Job, State) of
ok ->
{reply, ok, State};
diff --git a/src/mem3/test/eunit/mem3_reshard_api_test.erl b/src/mem3/test/eunit/mem3_reshard_api_test.erl
index 6cbc4dba8..59ec90553 100644
--- a/src/mem3/test/eunit/mem3_reshard_api_test.erl
+++ b/src/mem3/test/eunit/mem3_reshard_api_test.erl
@@ -50,6 +50,8 @@ teardown({Url, {Db1, Db2, Db3}}) ->
ok = config:delete("reshard", "max_jobs", Persist),
ok = config:delete("reshard", "require_node_param", Persist),
ok = config:delete("reshard", "require_range_param", Persist),
+ ok = config:delete("reshard", "max_retries", Persist),
+ ok = config:delete("reshard", "retry_interval_sec", Persist),
ok = config:delete("admins", ?USER, Persist),
meck:unload().
@@ -79,6 +81,7 @@ mem3_reshard_api_test_() ->
fun test_disabled/1,
fun start_stop_cluster_with_a_job/1,
fun individual_job_start_stop/1,
+ fun individual_job_start_after_failure/1,
fun individual_job_stop_when_cluster_stopped/1,
fun create_job_with_invalid_arguments/1,
fun create_job_with_db/1,
@@ -417,6 +420,35 @@ individual_job_start_stop({Top, {Db1, _, _}}) ->
wait_state(StUrl, <<"completed">>)
end)}.
+individual_job_start_after_failure({Top, {Db1, _, _}}) ->
+ {timeout, ?TIMEOUT,
+ ?_test(begin
+ config:set("reshard", "retry_interval_sec", "0", false),
+ config:set("reshard", "max_retries", "1", false),
+ meck:expect(couch_db_split, split, fun(_, _, _) ->
+ meck:exception(error, kapow)
+ end),
+
+ Body = #{type => split, db => Db1},
+ {201, [#{?ID := Id}]} = req(post, Top ++ ?JOBS, Body),
+
+ JobUrl = Top ++ ?JOBS ++ ?b2l(Id),
+ StUrl = JobUrl ++ "/state",
+
+ wait_state(StUrl, <<"failed">>),
+
+ % Stop/start resharding globally and job should still stay failed
+ ?assertMatch({200, _}, req(put, Top ++ ?STATE, #{state => stopped})),
+ ?assertMatch({200, _}, req(put, Top ++ ?STATE, #{state => running})),
+ ?assertMatch({200, #{<<"state">> := <<"failed">>}}, req(get, StUrl)),
+
+ meck:unload(),
+
+ % Start the job again
+ ?assertMatch({200, _}, req(put, StUrl, #{state => running})),
+ wait_state(StUrl, <<"completed">>)
+ end)}.
+
individual_job_stop_when_cluster_stopped({Top, {Db1, _, _}}) ->
{timeout, ?TIMEOUT,
?_test(begin
[couchdb] 02/02: feat: make rexi timeouts and replication batches for shard split topoff configurable
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit 480d3a3cbca4828657c70871dfa44d2fa7bb7d86
Author: Jan Lehnardt <ja...@apache.org>
AuthorDate: Wed Jul 19 14:32:03 2023 +0200
feat: make rexi timeouts and replication batches for shard split topoff configurable
---
rel/overlay/etc/default.ini | 2 ++
src/mem3/src/mem3_rep.erl | 41 ++++++++++++++++++++++++++-------------
src/mem3/src/mem3_reshard_job.erl | 13 +++++++++++--
src/mem3/src/mem3_rpc.erl | 3 +++
4 files changed, 44 insertions(+), 15 deletions(-)
diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index 2903e7603..bb440e0aa 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -373,6 +373,8 @@ hash_algorithms = sha256, sha
;buffer_count = 2000
;server_per_node = true
;stream_limit = 5
+;shard_split_timeout_msec = 600000
+;shard_split_topoff_batch_size = 500
; Use a single message to kill a group of remote workers. This feature is
; available starting with 3.0. When performing a rolling upgrade from 2.x to
diff --git a/src/mem3/src/mem3_rep.erl b/src/mem3/src/mem3_rep.erl
index b452fd2fa..ed79e69cb 100644
--- a/src/mem3/src/mem3_rep.erl
+++ b/src/mem3/src/mem3_rep.erl
@@ -41,7 +41,8 @@
filter,
db,
hashfun,
- incomplete_ranges
+ incomplete_ranges,
+ rexi_timeout
}).
-record(tgt, {
@@ -54,6 +55,8 @@
remaining = 0
}).
+-define(DEFAULT_REXI_TIMEOUT, 600000).
+
go(Source, Target) ->
go(Source, Target, []).
@@ -90,6 +93,11 @@ go(#shard{} = Source, #{} = Targets0, Opts) when map_size(Targets0) > 0 ->
"incomplete_ranges",
false
),
+ RexiTimeout =
+ case proplists:get_value(rexi_timeout, Opts) of
+ T when is_integer(T), T > 0 -> T;
+ _ -> ?DEFAULT_REXI_TIMEOUT
+ end,
Filter = proplists:get_value(filter, Opts),
Acc = #acc{
batch_size = BatchSize,
@@ -97,7 +105,8 @@ go(#shard{} = Source, #{} = Targets0, Opts) when map_size(Targets0) > 0 ->
source = Source,
targets = Targets,
filter = Filter,
- incomplete_ranges = IncompleteRanges
+ incomplete_ranges = IncompleteRanges,
+ rexi_timeout = RexiTimeout
},
go(Acc);
false ->
@@ -606,16 +615,16 @@ changes_append_fdi(
)
end.
-replicate_batch_multi(#acc{targets = Targets0, seq = Seq, db = Db} = Acc) ->
+replicate_batch_multi(#acc{targets = Targets0, seq = Seq, db = Db, rexi_timeout = Timeout} = Acc) ->
Targets = maps:map(
fun(_, #tgt{} = T) ->
- replicate_batch(T, Db, Seq)
+ replicate_batch(T, Db, Seq, Timeout)
end,
Targets0
),
{ok, Acc#acc{targets = Targets, revcount = 0}}.
-replicate_batch(#tgt{shard = TgtShard, infos = Infos} = Target, Db, Seq) ->
+replicate_batch(#tgt{shard = TgtShard, infos = Infos} = Target, Db, Seq, Timeout) ->
#shard{node = Node, name = Name} = TgtShard,
case find_missing_revs(Target) of
[] ->
@@ -624,7 +633,7 @@ replicate_batch(#tgt{shard = TgtShard, infos = Infos} = Target, Db, Seq) ->
lists:map(
fun(Chunk) ->
Docs = open_docs(Db, Infos, Chunk),
- ok = save_on_target(Node, Name, Docs)
+ ok = save_on_target(Node, Name, Docs, Timeout)
end,
chunk_revs(Missing)
)
@@ -694,13 +703,19 @@ open_docs(Db, Infos, Missing) ->
Missing
).
-save_on_target(Node, Name, Docs) ->
- mem3_rpc:update_docs(Node, Name, Docs, [
- ?REPLICATED_CHANGES,
- full_commit,
- ?ADMIN_CTX,
- {io_priority, {internal_repl, Name}}
- ]),
+save_on_target(Node, Name, Docs, Timeout) ->
+ mem3_rpc:update_docs(
+ Node,
+ Name,
+ Docs,
+ [
+ ?REPLICATED_CHANGES,
+ full_commit,
+ ?ADMIN_CTX,
+ {io_priority, {internal_repl, Name}}
+ ],
+ Timeout
+ ),
ok.
purge_on_target(Node, Name, PurgeInfos) ->
diff --git a/src/mem3/src/mem3_reshard_job.erl b/src/mem3/src/mem3_reshard_job.erl
index 4c93bf98f..aa5b028d1 100644
--- a/src/mem3/src/mem3_reshard_job.erl
+++ b/src/mem3/src/mem3_reshard_job.erl
@@ -49,7 +49,8 @@
-include("mem3_reshard.hrl").
% Batch size for internal replication topoffs
--define(INTERNAL_REP_BATCH_SIZE, 2000).
+-define(INTERNAL_REP_BATCH_SIZE, 500).
+-define(DEFAULT_REXI_TIMEOUT, 600000).
% The list of possible job states. The order of this
% list is important as a job will progress linearly
@@ -406,8 +407,16 @@ topoff_impl(#job{source = #shard{} = Source, target = Targets}) ->
couch_log:notice("~p topoff ~p", [?MODULE, shardsstr(Source, Targets)]),
check_source_exists(Source, topoff),
check_targets_exist(Targets, topoff),
+ Timeout = config:get_integer("rexi", "shard_split_timeout_msec", ?DEFAULT_REXI_TIMEOUT),
+ BatchSize = config:get_integer(
+ "rexi", "shard_split_topoff_batch_size", ?INTERNAL_REP_BATCH_SIZE
+ ),
TMap = maps:from_list([{R, T} || #shard{range = R} = T <- Targets]),
- Opts = [{batch_size, ?INTERNAL_REP_BATCH_SIZE}, {batch_count, all}],
+ Opts = [
+ {batch_size, BatchSize},
+ {batch_count, all},
+ {rexi_timeout, Timeout}
+ ],
case mem3_rep:go(Source, TMap, Opts) of
{ok, Count} ->
Args = [?MODULE, shardsstr(Source, Targets), Count],
diff --git a/src/mem3/src/mem3_rpc.erl b/src/mem3/src/mem3_rpc.erl
index 468bdee21..60c24e1d4 100644
--- a/src/mem3/src/mem3_rpc.erl
+++ b/src/mem3/src/mem3_rpc.erl
@@ -18,6 +18,7 @@
find_common_seq/4,
get_missing_revs/4,
update_docs/4,
+ update_docs/5,
pull_replication/1,
load_checkpoint/4,
load_checkpoint/5,
@@ -61,6 +62,8 @@ get_missing_revs(Node, DbName, IdsRevs, Options) ->
update_docs(Node, DbName, Docs, Options) ->
rexi_call(Node, {fabric_rpc, update_docs, [DbName, Docs, Options]}).
+update_docs(Node, DbName, Docs, Options, Timeout) ->
+ rexi_call(Node, {fabric_rpc, update_docs, [DbName, Docs, Options]}, Timeout).
load_checkpoint(Node, DbName, SourceNode, SourceUUID, <<>>) ->
% Upgrade clause for a mixed cluster for old nodes that don't have