Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merge in cherry-pick from fd-remove-yz-dep #1721

Merged
merged 1 commit into from
Oct 4, 2019
Merged

Merge in cherry-pick from fd-remove-yz-dep #1721

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 72 additions & 0 deletions src/riak_kv_update_hook.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
%%
%% -------------------------------------------------------------------
%%
%% Copyright (c) 2016 Basho Technologies, Inc. All Rights Reserved.
%%
%% This file is provided to you 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(riak_kv_update_hook).

-include_lib("riak_core/include/riak_core_vnode.hrl").

-export_type([object_pair/0, update_reason/0, repair/0, partition/0, handoff_dest/0]).

-type object_pair() :: {riak_object:riak_object(), riak_object:riak_object() | no_old_object}.
-type repair() :: full_repair | tree_repair | failed_repair.
-type update_reason() ::
delete
| handoff
| put
| anti_entropy
| {delete, repair()}
| {anti_entropy, repair()}
| {anti_entropy_delete, repair()}
| anti_entropy_delete.


%% @doc Update a Riak object, given a reason and partition under which
%% the object is being indexed. The object pair contains the new
%% and old objects, in the case where a read-before-write resulted
%% in an old object.
-callback update(
object_pair(),
update_reason(),
partition()
) ->
ok.

%% @doc Update a Riak object encoded as an erlang binary. This function
%% is typically called from the write-once path, where there is no
%% old object to pass.
-callback update_binary(
riak_core_bucket:bucket(),
riak_object:key(),
binary(),
update_reason(),
partition()
) ->
ok.

%% @doc Determine whether a bucket requires an existing object,
%% based on its properties. If this function returns true,
%% this may result in a read-before-write in the vnode.
-callback requires_existing_object(riak_kv_bucket:props()) ->
boolean().

%% @doc Determine whether handoff should start.
-callback should_handoff(handoff_dest()) ->
boolean().
125 changes: 89 additions & 36 deletions src/riak_kv_vnode.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,25 +96,6 @@
-export([put_merge/6]). %% For fsm_eqc_vnode
-endif.

%% N.B. The ?INDEX macro should be called any time the object bytes on
%% disk are modified.
-ifdef(TEST).
%% Use values so that test compile doesn't give 'unused vars' warning.
-define(INDEX(A,B,C), _=element(1,{{_A1, _A2} = A,B,C}), ok).
-define(INDEX_BIN(A,B,C,D,E), _=element(1,{A,B,C,D,E}), ok).
-define(IS_SEARCH_ENABLED_FOR_BUCKET(BProps), _=element(1, {BProps}), false).
-else.
-define(INDEX(Objects, Reason, Partition), yz_kv:index(Objects, Reason, Partition)).
-define(INDEX_BIN(Bucket, Key, Obj, Reason, Partition), yz_kv:index_binary(Bucket, Key, Obj, Reason, Partition)).
-define(IS_SEARCH_ENABLED_FOR_BUCKET(BProps), yz_kv:is_search_enabled_for_bucket(BProps)).
-endif.

-ifdef(TEST).
-define(YZ_SHOULD_HANDOFF(X), true).
-else.
-define(YZ_SHOULD_HANDOFF(X), yz_kv:should_handoff(X)).
-endif.

-record(mrjob, {cachekey :: term(),
bkey :: term(),
reqid :: term(),
Expand All @@ -138,6 +119,8 @@
leasing = false :: boolean()
}).

-type update_hook() :: module() | undefined.
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hopefully no one would implement their hook as undefined.erl :)


-record(state, {idx :: partition(),
mod :: module(),
async_put :: boolean(),
Expand Down Expand Up @@ -168,7 +151,8 @@
tictac_startqueue = os:timestamp() :: erlang:timestamp(),
tictac_rebuilding = false :: erlang:timestamp()|false,
worker_pool_strategy = single :: none|single|dscp,
vnode_pool_pid :: undefined|pid()
vnode_pool_pid :: undefined|pid(),
update_hook :: update_hook()
}).

-type index_op() :: add | remove.
Expand Down Expand Up @@ -784,7 +768,8 @@ init([Index]) ->
mrjobs=dict:new(),
md_cache=MDCache,
md_cache_size=MDCacheSize,
worker_pool_strategy=WorkerPoolStrategy},
worker_pool_strategy=WorkerPoolStrategy,
update_hook=update_hook()},
try_set_vnode_lock_limit(Index),
case AsyncFolding of
true ->
Expand Down Expand Up @@ -1380,7 +1365,7 @@ handle_request(kv_w1c_put_request, Req, Sender, State=#state{async_put=true}) ->
{error, Reason, UpModState} ->
{reply, ?KV_W1C_PUT_REPLY{reply={error, Reason}, type=ReplicaType}, State#state{modstate=UpModState}}
end;
handle_request(kv_w1c_put_request, Req, _Sender, State=#state{async_put=false}) ->
handle_request(kv_w1c_put_request, Req, _Sender, State=#state{async_put=false, update_hook=UpdateHook}) ->
{Bucket, Key} = riak_kv_requests:get_bucket_key(Req),
EncodedVal = riak_kv_requests:get_encoded_obj(Req),
ReplicaType = riak_kv_requests:get_replica_type(Req),
Expand All @@ -1393,7 +1378,7 @@ handle_request(kv_w1c_put_request, Req, _Sender, State=#state{async_put=false})
aae_update(Bucket, Key, use_binary, assumed_no_old_object, EncodedVal, State),
% Write once path - and so should be a new object. If not this
% is an application fault
?INDEX_BIN(Bucket, Key, EncodedVal, put, Idx),
maybe_update_binary(UpdateHook, Bucket, Key, EncodedVal, put, Idx),
update_vnode_stats(vnode_put, Idx, StartTS),
{reply, ?KV_W1C_PUT_REPLY{reply=ok, type=ReplicaType}, State#state{modstate=UpModState}};
{error, Reason, UpModState} ->
Expand Down Expand Up @@ -1966,9 +1951,9 @@ do_request_hash(_, _) ->



handoff_starting({_HOType, TargetNode}=_X, State=#state{handoffs_rejected=RejectCount}) ->
handoff_starting({_HOType, TargetNode}=HandoffDest, State=#state{handoffs_rejected=RejectCount, update_hook=UpdateHook}) ->
MaxRejects = app_helper:get_env(riak_kv, handoff_rejected_max, 6),
case MaxRejects =< RejectCount orelse ?YZ_SHOULD_HANDOFF(_X) of
case MaxRejects =< RejectCount orelse maybe_should_handoff(UpdateHook, HandoffDest) of
true ->
{true, State#state{in_handoff=true, handoff_target=TargetNode}};
false ->
Expand Down Expand Up @@ -2083,11 +2068,11 @@ terminate(_Reason, #state{idx=Idx,
ok.

handle_info({{w1c_async_put, From, Type, Bucket, Key, EncodedVal, StartTS} = _Context, Reply},
State=#state{idx=Idx}) ->
State=#state{idx=Idx, update_hook=UpdateHook}) ->
aae_update(Bucket, Key, use_binary, assumed_no_old_object, EncodedVal, State),
% Write once path - and so should be a new object. If not this
% is an application fault
?INDEX_BIN(Bucket, Key, EncodedVal, put, Idx),
maybe_update_binary(UpdateHook, Bucket, Key, EncodedVal, put, Idx),
riak_core_vnode:reply(From, ?KV_W1C_PUT_REPLY{reply=Reply, type=Type}),
update_vnode_stats(vnode_put, Idx, StartTS),
{ok, State};
Expand Down Expand Up @@ -2311,9 +2296,14 @@ do_put(Sender, {Bucket,_Key}=BKey, RObj, ReqID, StartTime, Options, State) ->
update_index_write_stats(UpdPutArgs#putargs.is_index, UpdPutArgs#putargs.index_specs),
{Reply, UpdState}.

-spec do_backend_delete(
{riak_core_bucket:bucket(), riak_object:key()},
riak_object:riak_object(), #state{}
) -> #state{}.
do_backend_delete(BKey, RObj, State = #state{idx = Idx,
mod = Mod,
modstate = ModState}) ->
modstate = ModState,
update_hook = UpdateHook}) ->
%% object is a tombstone or all siblings are tombstones
%% Calculate the index specs to remove...
%% JDM: This should just be a tombstone by this point, but better
Expand All @@ -2324,7 +2314,7 @@ do_backend_delete(BKey, RObj, State = #state{idx = Idx,
{Bucket, Key} = BKey,
case Mod:delete(Bucket, Key, IndexSpecs, ModState) of
{ok, UpdModState} ->
?INDEX({RObj, no_old_object}, delete, Idx),
maybe_update(UpdateHook, {RObj, no_old_object}, delete, Idx),
aae_delete(Bucket, Key, RObj, State),
maybe_cache_evict(BKey, State),
update_index_delete_stats(IndexSpecs),
Expand All @@ -2350,7 +2340,8 @@ delete_hash(RObj) ->
putargs(), state()}.
prepare_put(State=#state{vnodeid=VId,
mod=Mod,
modstate=ModState},
modstate=ModState,
update_hook=UpdateHook},
PutArgs=#putargs{bkey={Bucket, _Key},
lww=LWW,
coord=Coord,
Expand All @@ -2362,7 +2353,7 @@ prepare_put(State=#state{vnodeid=VId,
%% no need to incur additional get. Otherwise, we need to read the
%% old object to know how the indexes have changed.
IndexBackend = is_indexed_backend(Mod, Bucket, ModState),
IsSearchable = ?IS_SEARCH_ENABLED_FOR_BUCKET(BProps),
IsSearchable = maybe_requires_existing_object(UpdateHook, BProps),
SkipReadBeforeWrite = LWW andalso (not IndexBackend) andalso (not IsSearchable),
case SkipReadBeforeWrite of
true ->
Expand Down Expand Up @@ -2595,13 +2586,14 @@ actual_put(BKey, {Obj, OldObj}, IndexSpecs, RB, ReqID, State) ->
actual_put(BKey={Bucket, Key}, {Obj, OldObj}, IndexSpecs, RB, ReqID, MaxCheckFlag,
State=#state{idx=Idx,
mod=Mod,
modstate=ModState}) ->
modstate=ModState,
update_hook=UpdateHook}) ->
case encode_and_put(Obj, Mod, Bucket, Key, IndexSpecs, ModState,
MaxCheckFlag) of
{{ok, UpdModState}, EncodedVal} ->
aae_update(Bucket, Key, Obj, OldObj, EncodedVal, State),
maybe_cache_object(BKey, Obj, State),
?INDEX({Obj, maybe_old_object(OldObj)}, put, Idx),
maybe_update(UpdateHook, {Obj, maybe_old_object(OldObj)}, put, Idx),
Reply = case RB of
true ->
{dw, Idx, Obj, ReqID};
Expand Down Expand Up @@ -3144,7 +3136,8 @@ do_get_vclock({Bucket, Key}, Mod, ModState) ->
do_diffobj_put({Bucket, Key}=BKey, DiffObj,
StateData=#state{mod=Mod,
modstate=ModState,
idx=Idx}) ->
idx=Idx,
update_hook=UpdateHook}) ->
StartTS = os:timestamp(),
{ok, Capabilities} = Mod:capabilities(Bucket, ModState),
IndexBackend = lists:member(indexes, Capabilities),
Expand All @@ -3166,7 +3159,7 @@ do_diffobj_put({Bucket, Key}=BKey, DiffObj,
StateData),
update_index_write_stats(IndexBackend, IndexSpecs),
update_vnode_stats(vnode_put, Idx, StartTS),
?INDEX({DiffObj2, no_old_object}, handoff, Idx),
maybe_update(UpdateHook, {DiffObj, no_old_object}, handoff, Idx),
{ok, State2#state{modstate=UpdModState}};
{{error, Reason, UpdModState}, _Val} ->
{error, Reason, State2#state{modstate=UpdModState}}
Expand Down Expand Up @@ -3194,7 +3187,9 @@ do_diffobj_put({Bucket, Key}=BKey, DiffObj,
StateData),
update_index_write_stats(IndexBackend, IndexSpecs),
update_vnode_stats(vnode_put, Idx, StartTS),
?INDEX({AMObj, maybe_old_object(OldObj)}, handoff, Idx),
maybe_update(UpdateHook,
{AMObj, maybe_old_object(OldObj)},
handoff, Idx),
{ok, State2#state{modstate=UpdModState}};
{{error, Reason, UpdModState}, _Val} ->
{error, Reason, State2#state{modstate=UpdModState}}
Expand Down Expand Up @@ -3998,6 +3993,64 @@ highest_actor(ActorBase, Obj) ->
%% get the greatest event for the highest/latest actor
{Actor, Epoch, riak_object:actor_counter(Actor, Obj)}.

%%
%% Technical note: The index_module configuration parameter should contain
%% a module name which must implement the following functions:
%%
%% - index(object_pair(), write_reason(), p()) -> ok.
%% - index_binary(bucket(), key(), binary(), write_reason(), p()) -> ok.
%% - is_searchable(riak_kv_bucket:props()) -> boolean().
%%
%% The indexing module will be called on puts, deletes, handoff, and
%% anti-entropy activity. In the case of puts, if an object is being over-written,
%% the old object will be passed as the second parameter in the object pair.
%% The indexing module may use this old object to optimize the update (e.g.,
%% to handle the special case of sibling writes, which may not map directly to
%% Riak puts).
%%
%% NB. Currently, yokozuna is the only repository that currently
%% implements this behavior. C.f., Yokozuna cuttlefish schema, to see
%% where this configuration is implicitly set.
%%

-spec update_hook()-> update_hook().
update_hook() ->
app_helper:get_env(riak_kv, update_hook).

-spec maybe_update(update_hook(),
riak_kv_update_hook:object_pair(),
riak_kv_update_hook:update_reason(),
riak_kv_update_hook:partition()) -> ok.
maybe_update(undefined, _RObjPair, _Reason, _Idx) ->
ok;
maybe_update(UpdateHook, RObjPair, Reason, Idx) ->
UpdateHook:update(RObjPair, Reason, Idx).

-spec maybe_update_binary(update_hook(),
riak_core_bucket:bucket(),
riak_object:key(),
binary(),
riak_kv_update_hook:update_reason(),
riak_kv_update_hook:partition()) -> ok.
maybe_update_binary(undefined, _Bucket, _Key, _Binary, _Reason, _Idx) ->
ok;
maybe_update_binary(UpdateHook, Bucket, Key, Binary, Reason, Idx) ->
UpdateHook:update_binary(Bucket, Key, Binary, Reason, Idx).

-spec maybe_requires_existing_object(update_hook(),
riak_kv_bucket:props()) -> boolean().
maybe_requires_existing_object(undefined, _BProps) ->
false;
maybe_requires_existing_object(UpdateHook, BProps) ->
UpdateHook:requires_existing_object(BProps).

-spec maybe_should_handoff(update_hook(),
riak_kv_update_hook:handoff_dest()) -> boolean().
maybe_should_handoff(undefined, _HandoffDest) ->
true;
maybe_should_handoff(UpdateHook, HandoffDest) ->
UpdateHook:should_handoff(HandoffDest).

-ifdef(TEST).

-define(MGR, riak_kv_vnode_status_mgr).
Expand Down