diff options
Diffstat (limited to 'src/couch_replicator/src/couch_replicator_httpc.erl')
| -rw-r--r-- | src/couch_replicator/src/couch_replicator_httpc.erl | 93 |
1 files changed, 73 insertions, 20 deletions
diff --git a/src/couch_replicator/src/couch_replicator_httpc.erl b/src/couch_replicator/src/couch_replicator_httpc.erl index f11d1895d..28b0f3811 100644 --- a/src/couch_replicator/src/couch_replicator_httpc.erl +++ b/src/couch_replicator/src/couch_replicator_httpc.erl @@ -15,6 +15,7 @@ -include_lib("couch/include/couch_db.hrl"). -include_lib("ibrowse/include/ibrowse.hrl"). -include_lib("couch_replicator/include/couch_replicator_api_wrap.hrl"). +-include_lib("kernel/include/logger.hrl"). -export([setup/1]). -export([send_req/3]). @@ -55,6 +56,12 @@ setup(Db) -> {error, Error} -> LogMsg = "~p: auth plugin initialization failed ~p ~p", LogUrl = couch_util:url_strip_password(Url), + ?LOG_ERROR(#{ + what => auth_plugin_initialization_failure, + in => replicator, + db => LogUrl, + details => Error + }), couch_log:error(LogMsg, [?MODULE, LogUrl, Error]), throw({replication_auth_error, Error}) end. @@ -97,8 +104,8 @@ send_req(HttpDb, Params1, Callback) -> send_ibrowse_req(#httpdb{headers = BaseHeaders} = HttpDb0, Params) -> Method = get_value(method, Params, get), - UserHeaders = lists:keysort(1, get_value(headers, Params, [])), - Headers1 = lists:ukeymerge(1, UserHeaders, BaseHeaders), + UserHeaders = get_value(headers, Params, []), + Headers1 = merge_headers(BaseHeaders, UserHeaders), {Headers2, HttpDb} = couch_replicator_auth:update_headers(HttpDb0, Headers1), Url = full_url(HttpDb, Params), Body = get_value(body, Params, []), @@ -155,9 +162,15 @@ process_response({ibrowse_req_id, ReqId}, Worker, HttpDb, Params, Callback) -> process_response({ok, Code, Headers, Body}, Worker, HttpDb, Params, Callback) -> case list_to_integer(Code) of + R when R =:= 301 ; R =:= 302 ; R =:= 303 -> + backoff_success(HttpDb, Params), + do_redirect(Worker, R, Headers, HttpDb, Params, Callback); 429 -> backoff(HttpDb, Params); - Ok when (Ok >= 200 andalso Ok < 300) ; (Ok >= 400 andalso Ok < 500) -> + Error when Error =:= 408 ; Error >= 500 -> + couch_stats:increment_counter([couch_replicator, responses, failure]), + maybe_retry({code, Error}, Worker, HttpDb, Params); + Ok when Ok >= 200 , Ok < 500 -> backoff_success(HttpDb, Params), couch_stats:increment_counter([couch_replicator, responses, success]), EJson = case Body of @@ -168,13 +181,7 @@ process_response({ok, Code, Headers, Body}, Worker, HttpDb, Params, Callback) -> end, process_auth_response(HttpDb, Ok, Headers, Params), if Ok =:= 413 -> put(?STOP_HTTP_WORKER, stop); true -> ok end, - Callback(Ok, Headers, EJson); - R when R =:= 301 ; R =:= 302 ; R =:= 303 -> - backoff_success(HttpDb, Params), - do_redirect(Worker, R, Headers, HttpDb, Params, Callback); - Error -> - couch_stats:increment_counter([couch_replicator, responses, failure]), - maybe_retry({code, Error}, Worker, HttpDb, Params) + Callback(Ok, Headers, EJson) end; process_response(Error, Worker, HttpDb, Params, _Callback) -> @@ -185,10 +192,18 @@ process_stream_response(ReqId, Worker, HttpDb, Params, Callback) -> receive {ibrowse_async_headers, ReqId, Code, Headers} -> case list_to_integer(Code) of + R when R =:= 301 ; R =:= 302 ; R =:= 303 -> + backoff_success(HttpDb, Params), + do_redirect(Worker, R, Headers, HttpDb, Params, Callback); 429 -> Timeout = couch_replicator_rate_limiter:max_interval(), backoff(HttpDb#httpdb{timeout = Timeout}, Params); - Ok when (Ok >= 200 andalso Ok < 300) ; (Ok >= 400 andalso Ok < 500) -> + Error when Error =:= 408 ; Error >= 500 -> + couch_stats:increment_counter( + [couch_replicator, stream_responses, failure] + ), + report_error(Worker, HttpDb, Params, {code, Error}); + Ok when Ok >= 200 , Ok < 500 -> backoff_success(HttpDb, Params), HttpDb1 = process_auth_response(HttpDb, Ok, Headers, Params), StreamDataFun = fun() -> @@ -206,15 +221,7 @@ process_stream_response(ReqId, Worker, HttpDb, Params, Callback) -> Worker, HttpDb1, Params); throw:{maybe_retry_req, Err} -> maybe_retry(Err, Worker, HttpDb1, Params) - end; - R when R =:= 301 ; R =:= 302 ; R =:= 303 -> - backoff_success(HttpDb, Params), - do_redirect(Worker, R, Headers, HttpDb, Params, Callback); - Error -> - couch_stats:increment_counter( - [couch_replicator, stream_responses, failure] - ), - report_error(Worker, HttpDb, Params, {code, Error}) + end end; {ibrowse_async_response, ReqId, {error, _} = Error} -> couch_stats:increment_counter( @@ -348,6 +355,14 @@ update_first_error_timestamp(HttpDb) -> log_retry_error(Params, HttpDb, Wait, Error) -> Method = string:to_upper(atom_to_list(get_value(method, Params, get))), Url = couch_util:url_strip_password(full_url(HttpDb, Params)), + ?LOG_NOTICE(#{ + what => retry_request, + in => replicator, + method => Method, + url => Url, + retry_delay_sec => Wait / 1000, + details => error_cause(Error) + }), couch_log:notice("Retrying ~s request to ~s in ~p seconds due to error ~s", [Method, Url, Wait / 1000, error_cause(Error)]). @@ -360,10 +375,24 @@ report_error(_Worker, HttpDb, Params, Error) -> do_report_error(Url, Method, {code, Code}) -> + ?LOG_ERROR(#{ + what => failed_request, + in => replicator, + method => Method, + url => Url, + status_code => Code + }), couch_log:error("Replicator, request ~s to ~p failed. The received " "HTTP error code is ~p", [Method, Url, Code]); do_report_error(FullUrl, Method, Error) -> + ?LOG_ERROR(#{ + what => failed_request, + in => replicator, + method => Method, + url => FullUrl, + details => error_cause(Error) + }), couch_log:error("Replicator, request ~s to ~p failed due to error ~s", [Method, FullUrl, error_cause(Error)]). @@ -493,3 +522,27 @@ backoff_before_request(Worker, HttpDb, Params) -> Sleep when Sleep == 0 -> ok end. + + +merge_headers(Headers1, Headers2) when is_list(Headers1), is_list(Headers2) -> + Empty = mochiweb_headers:empty(), + Merged = mochiweb_headers:enter_from_list(Headers1 ++ Headers2, Empty), + mochiweb_headers:to_list(Merged). + + +-ifdef(TEST). + +-include_lib("couch/include/couch_eunit.hrl"). + + +merge_headers_test() -> + ?assertEqual([], merge_headers([], [])), + ?assertEqual([{"a", "x"}], merge_headers([], [{"a", "x"}])), + ?assertEqual([{"a", "x"}], merge_headers([{"a", "x"}], [])), + ?assertEqual([{"a", "y"}], merge_headers([{"A", "x"}], [{"a", "y"}])), + ?assertEqual([{"a", "y"}, {"B", "x"}], merge_headers([{"B", "x"}], + [{"a", "y"}])), + ?assertEqual([{"a", "y"}], merge_headers([{"A", "z"}, {"a", "y"}], [])), + ?assertEqual([{"a", "y"}], merge_headers([], [{"A", "z"}, {"a", "y"}])). + +-endif. |