You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2018/03/16 20:00:07 UTC
[couchdb] 13/20: Use EPI to create local purge doc for indexers
This is an automated email from the ASF dual-hosted git repository.
davisp pushed a commit to branch COUCHDB-3326-clustered-purge-davisp-refactor
in repository https://gitbox.apache.org/repos/asf/couchdb.git
commit f8263882962745a2043d26c9b92ff6d4b8e3a93e
Author: jiangphcn <ji...@cn.ibm.com>
AuthorDate: Fri Mar 9 15:29:49 2018 +0800
Use EPI to create local purge doc for indexers
- Before performing compaction, local purge document needs to be
created so that compaction can detect whether old purge requests
can be compacted. Otherwise, if some old purge requests are
compacted and some of indexers didn't process purge requests,
it will cause index reset.
- use EPI approach to create local purge documents (if not) before
compacting purge requests. To do so, all local purge documents
against valid design documents can be created and used by
compactor to decide whether to compact old purge requests.
- In ASF stream, the extensible plugin interface is implemented
for secondary index (mrview index).
COUCHDB-3226
---
src/couch/src/couch_bt_engine_compactor.erl | 18 ++++
src/couch/src/couch_db_plugin.erl | 8 +-
src/couch_index/src/couch_index_epi.erl | 5 +-
...dex_epi.erl => couch_index_plugin_couch_db.erl} | 39 ++-------
src/couch_mrview/src/couch_mrview_index.erl | 95 +++++++++++++++++-----
.../test/couch_mrview_purge_docs_fabric_tests.erl | 76 ++++++++++++++++-
6 files changed, 186 insertions(+), 55 deletions(-)
diff --git a/src/couch/src/couch_bt_engine_compactor.erl b/src/couch/src/couch_bt_engine_compactor.erl
index 7b11b44..f6e79b4 100644
--- a/src/couch/src/couch_bt_engine_compactor.erl
+++ b/src/couch/src/couch_bt_engine_compactor.erl
@@ -44,6 +44,12 @@ start(#st{} = St, DbName, Options, Parent) ->
} = St,
couch_log:debug("Compaction process spawned for db \"~s\"", [DbName]),
+ {ok, DDocs} = design_docs(DbName),
+ lists:map(fun(DDoc) ->
+ JsonDDoc = couch_doc:from_json_obj(DDoc),
+ couch_db_plugin:maybe_init_index_purge_state(DbName, JsonDDoc)
+ end, DDocs),
+
{ok, NewSt, DName, DFd, MFd, Retry} =
open_compaction_files(Header, FilePath, Options),
erlang:monitor(process, MFd),
@@ -569,3 +575,15 @@ update_compact_task(NumChanges) ->
end,
couch_task_status:update([{changes_done, Changes2}, {progress, Progress}]).
+
+design_docs(DbName) ->
+ try
+ case fabric:design_docs(mem3:dbname(DbName)) of
+ {error, {maintenance_mode, _, _Node}} ->
+ {ok, []};
+ Else ->
+ Else
+ end
+ catch error:database_does_not_exist ->
+ {ok, []}
+ end.
diff --git a/src/couch/src/couch_db_plugin.erl b/src/couch/src/couch_db_plugin.erl
index 740b812..0107403 100644
--- a/src/couch/src/couch_db_plugin.erl
+++ b/src/couch/src/couch_db_plugin.erl
@@ -18,7 +18,8 @@
after_doc_read/2,
validate_docid/1,
check_is_admin/1,
- on_delete/2
+ on_delete/2,
+ maybe_init_index_purge_state/2
]).
-define(SERVICE_ID, couch_db).
@@ -60,6 +61,11 @@ on_delete(DbName, Options) ->
Handle = couch_epi:get_handle(?SERVICE_ID),
couch_epi:apply(Handle, ?SERVICE_ID, on_delete, [DbName, Options], []).
+maybe_init_index_purge_state(DbName, DDoc) ->
+ Handle = couch_epi:get_handle(?SERVICE_ID),
+ couch_epi:apply(Handle, ?SERVICE_ID, maybe_init_index_purge_state,
+ [DbName, DDoc], []).
+
%% ------------------------------------------------------------------
%% Internal Function Definitions
%% ------------------------------------------------------------------
diff --git a/src/couch_index/src/couch_index_epi.erl b/src/couch_index/src/couch_index_epi.erl
index 946a590..1c4eb95 100644
--- a/src/couch_index/src/couch_index_epi.erl
+++ b/src/couch_index/src/couch_index_epi.erl
@@ -28,8 +28,9 @@ app() ->
couch_index.
providers() ->
- [].
-
+ [
+ {couch_db, couch_index_plugin_couch_db}
+ ].
services() ->
[
diff --git a/src/couch_index/src/couch_index_epi.erl b/src/couch_index/src/couch_index_plugin_couch_db.erl
similarity index 52%
copy from src/couch_index/src/couch_index_epi.erl
copy to src/couch_index/src/couch_index_plugin_couch_db.erl
index 946a590..3e3b711 100644
--- a/src/couch_index/src/couch_index_epi.erl
+++ b/src/couch_index/src/couch_index_plugin_couch_db.erl
@@ -2,7 +2,7 @@
% 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
+% 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
@@ -10,40 +10,15 @@
% License for the specific language governing permissions and limitations under
% the License.
--module(couch_index_epi).
-
--behaviour(couch_epi_plugin).
+-module(couch_index_plugin_couch_db).
-export([
- app/0,
- providers/0,
- services/0,
- data_subscriptions/0,
- data_providers/0,
- processes/0,
- notify/3
+ maybe_init_index_purge_state/2
]).
-app() ->
- couch_index.
-
-providers() ->
- [].
-
-
-services() ->
- [
- {couch_index, couch_index_plugin}
- ].
-
-data_subscriptions() ->
- [].
-
-data_providers() ->
- [].
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_mrview/include/couch_mrview.hrl").
-processes() ->
- [].
-notify(_Key, _Old, _New) ->
- ok.
+maybe_init_index_purge_state(DbName, DDoc) ->
+ couch_mrview_index:maybe_create_local_purge_doc(DbName, DDoc).
diff --git a/src/couch_mrview/src/couch_mrview_index.erl b/src/couch_mrview/src/couch_mrview_index.erl
index 6ab0812..6c63eaf 100644
--- a/src/couch_mrview/src/couch_mrview_index.erl
+++ b/src/couch_mrview/src/couch_mrview_index.erl
@@ -19,6 +19,7 @@
-export([compact/3, swap_compacted/2, remove_compacted/1]).
-export([index_file_exists/1]).
-export([update_local_purge_doc/2, verify_index_exists/1]).
+-export([maybe_create_local_purge_doc/2]).
-include_lib("couch/include/couch_db.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
@@ -213,24 +214,6 @@ index_file_exists(State) ->
filelib:is_file(IndexFName).
-update_local_purge_doc(Db, State) ->
- Sig = couch_index_util:hexsig(get(signature, State)),
- Doc = couch_doc:from_json_obj({[
- {<<"_id">>, couch_mrview_util:get_local_purge_doc_id(Sig)},
- {<<"purge_seq">>, get(purge_seq, State)},
- {<<"timestamp_utc">>, couch_util:utc_string()},
- {<<"verify_module">>, <<"couch_mrview_index">>},
- {<<"verify_function">>, <<"verify_index_exists">>},
- {<<"verify_options">>, {[
- {<<"dbname">>, get(db_name, State)},
- {<<"ddoc_id">>, get(idx_name, State)},
- {<<"signature">>, Sig}
- ]}},
- {<<"type">>, <<"mrview">>}
- ]}),
- couch_db:update_doc(Db, Doc, []).
-
-
verify_index_exists(Options) ->
ShardDbName = couch_mrview_util:get_value_from_options(
<<"dbname">>,
@@ -274,7 +257,7 @@ verify_index_exists(Options) ->
end.
-maybe_create_local_purge_doc(Db, State) ->
+maybe_create_local_purge_doc(Db, #mrst{}=State) ->
Sig = couch_index_util:hexsig(get(signature, State)),
LocalPurgeDocId = couch_mrview_util:get_local_purge_doc_id(Sig),
case couch_db:open_doc(Db, LocalPurgeDocId, []) of
@@ -282,4 +265,78 @@ maybe_create_local_purge_doc(Db, State) ->
update_local_purge_doc(Db, State);
{ok, _LocalPurgeDoc} ->
ok
+ end;
+maybe_create_local_purge_doc(DbName, #doc{}=DDoc) ->
+ {ok, Db} = case couch_db:open_int(DbName, []) of
+ {ok, _} = Resp -> Resp;
+ Else -> exit(Else)
+ end,
+ try
+ {ok, DefaultPurgeSeq} = couch_db:get_purge_seq(Db),
+ case get_index_type(DDoc, <<"views">>) of true ->
+ try couch_mrview_util:ddoc_to_mrst(DbName, DDoc) of
+ {ok, MRSt} -> update_local_purge_doc(Db, MRSt, DefaultPurgeSeq)
+ catch _:_ ->
+ ok
+ end;
+ false ->
+ ok
+ end
+ catch E:T ->
+ Stack = erlang:get_stacktrace(),
+ couch_log:error(
+ "Error occurs when creating local purge document ~p ~p",
+ [{E, T}, Stack]
+ )
+ after
+ catch couch_db:close(Db)
+ end.
+
+
+update_local_purge_doc(Db, State) ->
+ Sig = couch_index_util:hexsig(get(signature, State)),
+ Doc = couch_doc:from_json_obj({[
+ {<<"_id">>, couch_mrview_util:get_local_purge_doc_id(Sig)},
+ {<<"purge_seq">>, get(purge_seq, State)},
+ {<<"timestamp_utc">>, couch_util:utc_string()},
+ {<<"verify_module">>, <<"couch_mrview_index">>},
+ {<<"verify_function">>, <<"verify_index_exists">>},
+ {<<"verify_options">>, {[
+ {<<"dbname">>, get(db_name, State)},
+ {<<"ddoc_id">>, get(idx_name, State)},
+ {<<"signature">>, Sig}
+ ]}},
+ {<<"type">>, <<"mrview">>}
+ ]}),
+ couch_db:update_doc(Db, Doc, []).
+
+
+update_local_purge_doc(Db, State, PSeq) ->
+ Sig = couch_index_util:hexsig(State#mrst.sig),
+ DocId = couch_mrview_util:get_local_purge_doc_id(Sig),
+ case couch_db:open_doc(Db, DocId, []) of
+ {not_found, _Reason} ->
+ Doc = couch_doc:from_json_obj({[
+ {<<"_id">>, DocId},
+ {<<"purge_seq">>, PSeq},
+ {<<"timestamp_utc">>, couch_util:utc_string()},
+ {<<"verify_module">>, <<"couch_mrview_index">>},
+ {<<"verify_function">>, <<"verify_index_exists">>},
+ {<<"verify_options">>, {[
+ {<<"dbname">>, State#mrst.db_name},
+ {<<"ddoc_id">>, State#mrst.idx_name},
+ {<<"signature">>, Sig}
+ ]}},
+ {<<"type">>, <<"mrview">>}
+ ]}),
+ couch_db:update_doc(Db, Doc, []);
+ {ok, _LocalPurgeDoc} ->
+ ok
+ end.
+
+
+get_index_type(#doc{body={Props}}, IndexType) ->
+ case couch_util:get_value(IndexType, Props) of
+ undefined -> false;
+ _ -> true
end.
diff --git a/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl b/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl
index 0fa2a7c..b36cad5 100644
--- a/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl
+++ b/src/couch_mrview/test/couch_mrview_purge_docs_fabric_tests.erl
@@ -14,6 +14,7 @@
-include_lib("couch/include/couch_eunit.hrl").
-include_lib("couch/include/couch_db.hrl").
+-include_lib("mem3/include/mem3.hrl").
-include_lib("couch_mrview/include/couch_mrview.hrl").
-define(TIMEOUT, 1000).
@@ -22,10 +23,15 @@
setup() ->
DbName = ?tempdb(),
ok = fabric:create_db(DbName, [?ADMIN_CTX]),
+ meck:new(couch_mrview_index, [passthrough]),
+ meck:expect(couch_mrview_index, maybe_create_local_purge_doc, fun(A, B) ->
+ meck:passthrough([A, B])
+ end),
DbName.
teardown(DbName) ->
+ meck:unload(couch_mrview_index),
ok = fabric:delete_db(DbName, [?ADMIN_CTX]).
@@ -40,7 +46,8 @@ view_purge_fabric_test_() ->
foreach,
fun setup/0, fun teardown/1,
[
- fun test_purge_verify_index/1
+ fun test_purge_verify_index/1,
+ fun test_purge_hook_before_compaction/1
]
}
}
@@ -82,6 +89,40 @@ test_purge_verify_index(DbName) ->
ok
end).
+test_purge_hook_before_compaction(DbName) ->
+ ?_test(begin
+ Docs1 = couch_mrview_test_util:make_docs(normal, 5),
+ {ok, _} = fabric:update_docs(DbName, Docs1, [?ADMIN_CTX]),
+ {ok, _} = fabric:update_doc(
+ DbName,
+ couch_mrview_test_util:ddoc(map),
+ [?ADMIN_CTX]
+ ),
+
+ purge_docs(DbName, [<<"1">>]),
+
+ Result1 = fabric:query_view(DbName, <<"bar">>, <<"baz">>, #mrargs{}),
+ Expect1 = {ok, [
+ {meta, [{total, 4}, {offset, 0}]},
+ {row, [{id, <<"2">>}, {key, 2}, {value, 2}]},
+ {row, [{id, <<"3">>}, {key, 3}, {value, 3}]},
+ {row, [{id, <<"4">>}, {key, 4}, {value, 4}]},
+ {row, [{id, <<"5">>}, {key, 5}, {value, 5}]}
+ ]},
+ ?assertEqual(Expect1, Result1),
+
+ [FirstShDbName | _RestShDbNames] = local_shards(DbName),
+ {ok, Db} = couch_db:open_int(FirstShDbName, []),
+ {ok, _CompactPid} = couch_db:start_compact(Db),
+ wait_compaction(FirstShDbName, "database", ?LINE),
+ ok = couch_db:close(Db),
+
+ ?assertEqual(ok, meck:wait(1, couch_mrview_index,
+ maybe_create_local_purge_doc, '_', 5000)
+ ),
+ ok
+ end).
+
get_rev(#full_doc_info{} = FDI) ->
#doc_info{
revs = [#rev_info{} = PrevRev | _]
@@ -95,3 +136,36 @@ purge_docs(DbName, DocIds) ->
Rev = get_rev(FDI),
{ok, [{ok, _}]} = fabric:purge_docs(DbName, [{DocId, [Rev]}], [])
end, DocIds).
+
+wait_compaction(DbName, Kind, Line) ->
+ WaitFun = fun() ->
+ case is_compaction_running(DbName) of
+ true -> wait;
+ false -> ok
+ end
+ end,
+ case test_util:wait(WaitFun, 10000) of
+ timeout ->
+ erlang:error({assertion_failed,
+ [{module, ?MODULE},
+ {line, Line},
+ {reason, "Timeout waiting for "
+ ++ Kind
+ ++ " database compaction"}]});
+ _ ->
+ ok
+ end.
+
+is_compaction_running(DbName) ->
+ {ok, Db} = couch_db:open_int(DbName, []),
+ {ok, DbInfo} = couch_db:get_db_info(Db),
+ couch_db:close(Db),
+ couch_util:get_value(compact_running, DbInfo).
+
+local_shards(DbName) ->
+ try
+ [ShardName || #shard{name = ShardName} <- mem3:local_shards(DbName)]
+ catch
+ error:database_does_not_exist ->
+ []
+ end.
--
To stop receiving notification emails like this one, please contact
davisp@apache.org.