You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2014/05/14 16:22:55 UTC
[4/5] couch-replicator commit: updated refs/heads/master to edd5dc4
Merge remote-tracking branch 'origin/import-master' into integration
Conflicts:
Makefile.am
src/couch_replicator.erl
src/couch_replicator_api_wrap.erl
src/couch_replicator_manager.erl
src/couch_replicator_utils.erl
test/04-replication-large-atts.t
Project: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/commit/c509ef21
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/tree/c509ef21
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/diff/c509ef21
Branch: refs/heads/master
Commit: c509ef2145d8a26a7402b5822ad6a9f9c118dc2c
Parents: b38025c 0ebc319
Author: Robert Newson <rn...@apache.org>
Authored: Mon May 12 16:00:27 2014 +0100
Committer: Robert Newson <rn...@apache.org>
Committed: Mon May 12 16:01:23 2014 +0100
----------------------------------------------------------------------
src/couch_replicator.erl | 25 ++--
src/couch_replicator_api_wrap.erl | 245 +++++++++++++++++++------------
src/couch_replicator_httpd.erl | 2 +-
src/couch_replicator_manager.erl | 13 +-
src/couch_replicator_utils.erl | 11 +-
test/04-replication-large-atts.t | 4 +-
test/07-use-checkpoints.t | 256 +++++++++++++++++++++++++++++++++
7 files changed, 445 insertions(+), 111 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/c509ef21/src/couch_replicator.erl
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/c509ef21/src/couch_replicator_api_wrap.erl
----------------------------------------------------------------------
diff --cc src/couch_replicator_api_wrap.erl
index d072187,3cf942d..7410b07
--- a/src/couch_replicator_api_wrap.erl
+++ b/src/couch_replicator_api_wrap.erl
@@@ -242,10 -305,14 +305,14 @@@ update_doc(#httpdb{} = HttpDb, #doc{id
end ++ [{"Content-Type", ?b2l(ContentType)}, {"Content-Length", Len}],
Body = {fun stream_doc/1, {JsonBytes, Doc#doc.atts, Boundary, Len}},
send_req(
- HttpDb,
+ % A crash here bubbles all the way back up to run_user_fun inside
+ % open_doc_revs, which will retry the whole thing. That's the
+ % appropriate course of action, since we've already started streaming
+ % the response body from the GET request.
+ HttpDb#httpdb{retries = 0},
[{method, put}, {path, encode_doc_id(DocId)},
{qs, QArgs}, {headers, Headers}, {body, Body}],
- fun(Code, _, {Props}) when Code =:= 200 orelse Code =:= 201 ->
+ fun(Code, _, {Props}) when Code =:= 200 orelse Code =:= 201 orelse Code =:= 202 ->
{ok, couch_doc:parse_rev(get_value(<<"rev">>, Props))};
(409, _, _) ->
throw(conflict);
@@@ -482,15 -553,12 +553,12 @@@ options_to_query_args([{open_revs, all
options_to_query_args([latest | Rest], Acc) ->
options_to_query_args(Rest, [{"latest", "true"} | Acc]);
options_to_query_args([{open_revs, Revs} | Rest], Acc) ->
- JsonRevs = ?b2l(?JSON_ENCODE(couch_doc:revs_to_strs(Revs))),
+ JsonRevs = ?b2l(iolist_to_binary(?JSON_ENCODE(couch_doc:revs_to_strs(Revs)))),
options_to_query_args(Rest, [{"open_revs", JsonRevs} | Acc]).
-
- -define(MAX_URL_LEN, 7000).
-
- atts_since_arg(_UrlLen, [], Acc) ->
+ atts_since_arg(_UrlLen, [], _MaxLen, Acc) ->
lists:reverse(Acc);
- atts_since_arg(UrlLen, [PA | Rest], Acc) ->
+ atts_since_arg(UrlLen, [PA | Rest], MaxLen, Acc) ->
RevStr = couch_doc:rev_to_str(PA),
NewUrlLen = case Rest of
[] ->
@@@ -534,8 -602,8 +602,8 @@@ receive_docs(Streamer, UserFun, Ref, Us
ContentType,
fun() -> receive_doc_data(Streamer, Ref) end,
Ref) of
- {ok, Doc, Parser} ->
+ {ok, Doc, WaitFun, Parser} ->
- case UserFun({ok, Doc}, UserAcc) of
+ case run_user_fun(UserFun, {ok, Doc}, UserAcc, Ref) of
{ok, UserAcc2} ->
ok;
{skip, UserAcc2} ->
@@@ -631,43 -728,44 +729,6 @@@ receive_doc_data(Streamer, Ref) -
{<<>>, fun() -> receive_doc_data(Streamer, Ref) end}
end.
--doc_from_multi_part_stream(ContentType, DataFun, Ref) ->
-- Self = self(),
-- Parser = spawn_link(fun() ->
-- {<<"--">>, _, _} = couch_httpd:parse_multipart_request(
-- ContentType, DataFun,
- fun(Next) -> couch_replicator_utils:mp_parse_doc(Next, []) end),
- unlink(Self)
- end),
- Parser ! {get_doc_bytes, Ref, self()},
- receive
- {started_open_doc_revs, NewRef} ->
- unlink(Parser),
- exit(Parser, kill),
- restart_remote_open_doc_revs(Ref, NewRef);
- {doc_bytes, Ref, DocBytes} ->
- Doc = couch_doc:from_json_obj(?JSON_DECODE(DocBytes)),
- ReadAttachmentDataFun = fun() ->
- Parser ! {get_bytes, Ref, self()},
- receive
- {started_open_doc_revs, NewRef} ->
- unlink(Parser),
- exit(Parser, kill),
- receive {bytes, Ref, _} -> ok after 0 -> ok end,
- restart_remote_open_doc_revs(Ref, NewRef);
- {bytes, Ref, Bytes} ->
- Bytes
- end
- end,
- Atts2 = lists:map(
- fun(#att{data = follows} = A) ->
- A#att{data = ReadAttachmentDataFun};
- (A) ->
- A
- end, Doc#doc.atts),
- {ok, Doc#doc{atts = Atts2}, Parser}
- end.
-
- fun(Next) -> couch_doc:mp_parse_doc(Next, []) end),
- unlink(Self)
- end),
- Parser ! {get_doc_bytes, Ref, self()},
- receive
- {started_open_doc_revs, NewRef} ->
- unlink(Parser),
- exit(Parser, kill),
- restart_remote_open_doc_revs(Ref, NewRef);
- {doc_bytes, Ref, DocBytes} ->
- Doc = couch_doc:from_json_obj(?JSON_DECODE(DocBytes)),
- ReadAttachmentDataFun = fun() ->
- link(Parser),
- Parser ! {get_bytes, Ref, self()},
- receive
- {started_open_doc_revs, NewRef} ->
- unlink(Parser),
- exit(Parser, kill),
- receive {bytes, Ref, _} -> ok after 0 -> ok end,
- restart_remote_open_doc_revs(Ref, NewRef);
- {bytes, Ref, Bytes} ->
- Bytes
- end
- end,
- Atts2 = lists:map(
- fun(#att{data = follows} = A) ->
- A#att{data = ReadAttachmentDataFun};
- (A) ->
- A
- end, Doc#doc.atts),
- {ok, Doc#doc{atts = Atts2}, Parser}
- end.
-
changes_ev1(object_start, UserFun, UserAcc) ->
fun(Ev) -> changes_ev2(Ev, UserFun, UserAcc) end.
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/c509ef21/src/couch_replicator_httpd.erl
----------------------------------------------------------------------
diff --cc src/couch_replicator_httpd.erl
index f84152c,c0fe004..6b43472
--- a/src/couch_replicator_httpd.erl
+++ b/src/couch_replicator_httpd.erl
@@@ -31,10 -31,11 +31,10 @@@ handle_req(#httpd{method = 'POST', user
couch_httpd:validate_ctype(Req, "application/json"),
RepDoc = {Props} = couch_httpd:json_body_obj(Req),
validate_rep_props(Props),
- {ok, Rep} = couch_replicator_utils:parse_rep_doc(RepDoc, UserCtx),
- case couch_replicator:replicate(Rep) of
+ case couch_replicator:replicate(RepDoc, UserCtx) of
{error, {Error, Reason}} ->
send_json(
- Req, 404,
+ Req, 500,
{[{error, to_binary(Error)}, {reason, to_binary(Reason)}]});
{error, not_found} ->
% Tried to cancel a replication that didn't exist.
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/c509ef21/src/couch_replicator_manager.erl
----------------------------------------------------------------------
diff --cc src/couch_replicator_manager.erl
index 5ec2d33,4891c4c..565f136
--- a/src/couch_replicator_manager.erl
+++ b/src/couch_replicator_manager.erl
@@@ -69,9 -65,10 +69,10 @@@ replication_started(#rep{id = {BaseId,
case rep_state(RepId) of
nil ->
ok;
- #rep_state{rep = #rep{doc_id = DocId}} ->
- update_rep_doc(DocId, [
+ #rep_state{dbname = DbName, rep = #rep{doc_id = DocId}} ->
+ update_rep_doc(DbName, DocId, [
{<<"_replication_state">>, <<"triggered">>},
+ {<<"_replication_state_reason">>, undefined},
{<<"_replication_id">>, ?l2b(BaseId)},
{<<"_replication_stats">>, undefined}]),
ok = gen_server:call(?MODULE, {rep_started, RepId}, infinity),
@@@ -84,12 -81,13 +85,13 @@@ replication_completed(#rep{id = RepId}
case rep_state(RepId) of
nil ->
ok;
- #rep_state{rep = #rep{doc_id = DocId}} ->
- update_rep_doc(DocId, [
+ #rep_state{dbname = DbName, rep = #rep{doc_id = DocId}} ->
+ update_rep_doc(DbName, DocId, [
{<<"_replication_state">>, <<"completed">>},
+ {<<"_replication_state_reason">>, undefined},
{<<"_replication_stats">>, {Stats}}]),
ok = gen_server:call(?MODULE, {rep_complete, RepId}, infinity),
- ?LOG_INFO("Replication `~s` finished (triggered by document `~s`)",
+ couch_log:notice("Replication `~s` finished (triggered by document `~s`)",
[pp_rep_id(RepId), DocId])
end.
@@@ -98,10 -96,10 +100,10 @@@ replication_error(#rep{id = {BaseId, _
case rep_state(RepId) of
nil ->
ok;
- #rep_state{rep = #rep{doc_id = DocId}} ->
- update_rep_doc(DocId, [
+ #rep_state{dbname = DbName, rep = #rep{doc_id = DocId}} ->
- % TODO: maybe add error reason to replication document
+ update_rep_doc(DbName, DocId, [
{<<"_replication_state">>, <<"error">>},
+ {<<"_replication_state_reason">>, to_binary(error_reason(Error))},
{<<"_replication_id">>, ?l2b(BaseId)}]),
ok = gen_server:call(?MODULE, {rep_error, RepId, Error}, infinity)
end.
@@@ -415,9 -344,10 +417,10 @@@ rep_db_update_error(Error, DbName, DocI
_ ->
Reason = to_binary(Error)
end,
- ?LOG_ERROR("Replication manager, error processing document `~s`: ~s",
+ couch_log:error("Replication manager, error processing document `~s`: ~s",
[DocId, Reason]),
- update_rep_doc(DbName, DocId, [{<<"_replication_state">>, <<"error">>}]).
- update_rep_doc(DocId, [{<<"_replication_state">>, <<"error">>},
++ update_rep_doc(DbName, DocId, [{<<"_replication_state">>, <<"error">>},
+ {<<"_replication_state_reason">>, Reason}]).
rep_user_ctx({RepDoc}) ->
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/c509ef21/src/couch_replicator_utils.erl
----------------------------------------------------------------------
diff --cc src/couch_replicator_utils.erl
index 9b9dd74,99ddebf..dff8e50
--- a/src/couch_replicator_utils.erl
+++ b/src/couch_replicator_utils.erl
@@@ -224,14 -223,15 +224,15 @@@ maybe_add_trailing_slash(Url) -
make_options(Props) ->
Options = lists:ukeysort(1, convert_options(Props)),
- DefWorkers = couch_config:get("replicator", "worker_processes", "4"),
- DefBatchSize = couch_config:get("replicator", "worker_batch_size", "500"),
- DefConns = couch_config:get("replicator", "http_connections", "20"),
- DefTimeout = couch_config:get("replicator", "connection_timeout", "30000"),
- DefRetries = couch_config:get("replicator", "retries_per_request", "10"),
- UseCheckpoints = couch_config:get("replicator", "use_checkpoints", "true"),
- DefCheckpointInterval = couch_config:get("replicator", "checkpoint_interval", "5000"),
+ DefWorkers = config:get("replicator", "worker_processes", "4"),
+ DefBatchSize = config:get("replicator", "worker_batch_size", "500"),
+ DefConns = config:get("replicator", "http_connections", "20"),
+ DefTimeout = config:get("replicator", "connection_timeout", "30000"),
+ DefRetries = config:get("replicator", "retries_per_request", "10"),
+ UseCheckpoints = config:get("replicator", "use_checkpoints", "true"),
++ DefCheckpointInterval = config:get("replicator", "checkpoint_interval", "5000"),
{ok, DefSocketOptions} = couch_util:parse_term(
- couch_config:get("replicator", "socket_options",
+ config:get("replicator", "socket_options",
"[{keepalive, true}, {nodelay, false}]")),
lists:ukeymerge(1, Options, lists:keysort(1, [
{connection_timeout, list_to_integer(DefTimeout)},
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/c509ef21/test/04-replication-large-atts.t
----------------------------------------------------------------------
diff --cc test/04-replication-large-atts.t
index 9adaead,a7063c7..adc3101
--- a/test/04-replication-large-atts.t
+++ b/test/04-replication-large-atts.t
@@@ -53,10 -63,10 +53,10 @@@ main(_) -
test() ->
- couch_server_sup:start_link(test_util:config_files()),
+ test_util:start_couch(),
- ibrowse:start(),
- crypto:start(),
+ application:start(ibrowse),
+ application:start(crypto),
- couch_config:set("attachments", "compressible_types", "text/*", false),
+ config:set("attachments", "compressible_types", "text/*", false),
Pairs = [
{source_db_name(), target_db_name()},
http://git-wip-us.apache.org/repos/asf/couchdb-couch-replicator/blob/c509ef21/test/07-use-checkpoints.t
----------------------------------------------------------------------
diff --cc test/07-use-checkpoints.t
index 0000000,cefc1a7..f2ed1de
mode 000000,100755..100755
--- a/test/07-use-checkpoints.t
+++ b/test/07-use-checkpoints.t
@@@ -1,0 -1,256 +1,256 @@@
+ #!/usr/bin/env escript
+ %% -*- erlang -*-
+ % 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.
+
+ % Verify that compacting databases that are being used as the source or
+ % target of a replication doesn't affect the replication and that the
+ % replication doesn't hold their reference counters forever.
+
+ -define(b2l(B), binary_to_list(B)).
+
+ -record(user_ctx, {
+ name = null,
+ roles = [],
+ handler
+ }).
+
+ -record(doc, {
+ id = <<"">>,
+ revs = {0, []},
+ body = {[]},
+ atts = [],
+ deleted = false,
+ meta = []
+ }).
+
+ -record(db, {
+ main_pid = nil,
+ update_pid = nil,
+ compactor_pid = nil,
+ instance_start_time, % number of microsecs since jan 1 1970 as a binary string
+ fd,
+ updater_fd,
+ fd_ref_counter,
+ header = nil,
+ committed_update_seq,
+ fulldocinfo_by_id_btree,
+ docinfo_by_seq_btree,
+ local_docs_btree,
+ update_seq,
+ name,
+ filepath,
+ validate_doc_funs = [],
+ security = [],
+ security_ptr = nil,
+ user_ctx = #user_ctx{},
+ waiting_delayed_commit = nil,
+ revs_limit = 1000,
+ fsync_options = [],
+ options = [],
+ compression,
+ before_doc_update,
+ after_doc_read
+ }).
+
+ -record(rep, {
+ id,
+ source,
+ target,
+ options,
+ user_ctx,
+ doc_id
+ }).
+
+
+ source_db_name() -> <<"couch_test_rep_db_a">>.
+ target_db_name() -> <<"couch_test_rep_db_b">>.
+
+
+ main(_) ->
+ test_util:init_code_path(),
+
+ etap:plan(16),
+ case (catch test()) of
+ ok ->
+ etap:end_tests();
+ Other ->
+ etap:diag(io_lib:format("Test died abnormally: ~p", [Other])),
+ etap:bail(Other)
+ end,
+ ok.
+
+
+ test() ->
+ couch_server_sup:start_link(test_util:config_files()),
+ ibrowse:start(),
+
+ test_use_checkpoints(false),
+ test_use_checkpoints(true),
+
+ couch_server_sup:stop(),
+ ok.
+
+
+ test_use_checkpoints(UseCheckpoints) ->
+ Pairs = [
+ {source_db_name(), target_db_name()},
+ {{remote, source_db_name()}, target_db_name()},
+ {source_db_name(), {remote, target_db_name()}},
+ {{remote, source_db_name()}, {remote, (target_db_name())}}
+ ],
+
+ ListenerFun = case UseCheckpoints of
+ false ->
+ fun({finished, _, {CheckpointHistory}}) ->
+ etap:is(CheckpointHistory,
+ [{<<"use_checkpoints">>,false}],
+ "No checkpoints found");
+ (_) ->
+ ok
+ end;
+ true ->
+ fun({finished, _, {CheckpointHistory}}) ->
+ SessionId = lists:keyfind(
+ <<"session_id">>, 1, CheckpointHistory),
+ etap:isnt(SessionId, false, "There's a checkpoint");
+ (_) ->
+ ok
+ end
+ end,
+ {ok, Listener} = couch_replicator_notifier:start_link(ListenerFun),
+
+ lists:foreach(
+ fun({Source, Target}) ->
+ {ok, SourceDb} = create_db(source_db_name()),
+ etap:diag("Populating source database"),
+ populate_db(SourceDb, 100),
+ ok = couch_db:close(SourceDb),
+
+ etap:diag("Creating target database"),
+ {ok, TargetDb} = create_db(target_db_name()),
+ ok = couch_db:close(TargetDb),
+
+ etap:diag("Setup replicator notifier listener"),
+
+ etap:diag("Triggering replication"),
+ replicate(Source, Target, UseCheckpoints),
+
+ etap:diag("Replication finished, comparing source and target databases"),
+ compare_dbs(SourceDb, TargetDb),
+
+ etap:diag("Deleting databases"),
+ delete_db(TargetDb),
+ delete_db(SourceDb),
+
+ ok = timer:sleep(1000)
+ end,
+ Pairs),
+
+ couch_replicator_notifier:stop(Listener).
+
+
+ populate_db(Db, DocCount) ->
+ Docs = lists:foldl(
+ fun(DocIdCounter, Acc) ->
+ Id = iolist_to_binary(["doc", integer_to_list(DocIdCounter)]),
+ Value = iolist_to_binary(["val", integer_to_list(DocIdCounter)]),
+ Doc = #doc{
+ id = Id,
+ body = {[ {<<"value">>, Value} ]}
+ },
+ [Doc | Acc]
+ end,
+ [], lists:seq(1, DocCount)),
+ {ok, _} = couch_db:update_docs(Db, Docs, []).
+
+
+ compare_dbs(#db{name = SourceName}, #db{name = TargetName}) ->
+ {ok, SourceDb} = couch_db:open_int(SourceName, []),
+ {ok, TargetDb} = couch_db:open_int(TargetName, []),
+ Fun = fun(FullDocInfo, _, Acc) ->
+ {ok, Doc} = couch_db:open_doc(SourceDb, FullDocInfo),
+ {Props} = DocJson = couch_doc:to_json_obj(Doc, [attachments]),
+ DocId = couch_util:get_value(<<"_id">>, Props),
+ DocTarget = case couch_db:open_doc(TargetDb, DocId) of
+ {ok, DocT} ->
+ DocT;
+ Error ->
+ etap:bail("Error opening document '" ++ ?b2l(DocId) ++
+ "' from target: " ++ couch_util:to_list(Error))
+ end,
+ DocTargetJson = couch_doc:to_json_obj(DocTarget, [attachments]),
+ case DocTargetJson of
+ DocJson ->
+ ok;
+ _ ->
+ etap:bail("Content from document '" ++ ?b2l(DocId) ++
+ "' differs in target database")
+ end,
+ {ok, Acc}
+ end,
+ {ok, _, _} = couch_db:enum_docs(SourceDb, Fun, [], []),
+ etap:diag("Target database has the same documents as the source database"),
+ ok = couch_db:close(SourceDb),
+ ok = couch_db:close(TargetDb).
+
+
+ db_url(DbName) ->
+ iolist_to_binary([
- "http://", couch_config:get("httpd", "bind_address", "127.0.0.1"),
++ "http://", config:get("httpd", "bind_address", "127.0.0.1"),
+ ":", integer_to_list(mochiweb_socket_server:get(couch_httpd, port)),
+ "/", DbName
+ ]).
+
+
+ create_db(DbName) ->
+ {ok, Db} = couch_db:create(
+ DbName,
+ [{user_ctx, #user_ctx{roles = [<<"_admin">>]}}, overwrite]),
+ couch_db:close(Db),
+ {ok, Db}.
+
+
+ delete_db(#db{name = DbName, main_pid = Pid}) ->
+ ok = couch_server:delete(
+ DbName, [{user_ctx, #user_ctx{roles = [<<"_admin">>]}}]),
+ MonRef = erlang:monitor(process, Pid),
+ receive
+ {'DOWN', MonRef, process, Pid, _Reason} ->
+ ok
+ after 30000 ->
+ etap:bail("Timeout deleting database")
+ end.
+
+
+ replicate({remote, Db}, Target, UseCheckpoints) ->
+ replicate(db_url(Db), Target, UseCheckpoints);
+
+ replicate(Source, {remote, Db}, UseCheckpoints) ->
+ replicate(Source, db_url(Db), UseCheckpoints);
+
+ replicate(Source, Target, UseCheckpoints) ->
+ RepObject = {[
+ {<<"source">>, Source},
+ {<<"target">>, Target},
+ {<<"use_checkpoints">>, UseCheckpoints}
+ ]},
+ {ok, Rep} = couch_replicator_utils:parse_rep_doc(
+ RepObject, #user_ctx{roles = [<<"_admin">>]}),
+ {ok, Pid} = couch_replicator:async_replicate(Rep),
+ MonRef = erlang:monitor(process, Pid),
+ receive
+ {'DOWN', MonRef, process, Pid, Reason} ->
+ etap:is(Reason, normal, "Replication finished successfully")
+ after 300000 ->
+ etap:bail("Timeout waiting for replication to finish")
+ end.