Create verify_repl_deleteonfail.erl

tests/verify_repl_deleteonfail.erl

@@ -0,0 +1,273 @@
+%% -------------------------------------------------------------------
+%%
+%% 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.
+%%
+%% -------------------------------------------------------------------
+%% @doc When replicating deletes, if a node is down in the cluster, clusters
+%% will get out of sync, unless the deletion of tombstones is deferred when
+%% non-primaries are unavailable
+
+-module(verify_repl_deleteonfail).
+-export([confirm/0]).
+-import(location, [plan_and_wait/2]).
+
+-include_lib("eunit/include/eunit.hrl").
+
+-define(DEFAULT_RING_SIZE, 16).
+-define(SNK_WORKERS, 4).
+-define(MAX_RESULTS, 512).
+-define(NUM_KEYS_PERNODE, 5000).
+-define(MAX_LOOOPS, 20).
+
+
+-define(CFG_REPL(SrcQueueDefns, NVal),
+ [{riak_kv,
+ [
+ % Speedy AAE configuration
+ {anti_entropy, {off, []}},
+ {tictacaae_active, active},
+ {tictacaae_parallelstore, leveled_ko},
+ % if backend not leveled will use parallel key-ordered
+ % store
+ {tictacaae_rebuildwait, 4},
+ {tictacaae_rebuilddelay, 3600},
+ {tictacaae_rebuildtick, 3600000},
+ {replrtq_enablesrc, true},
+ {replrtq_srcqueue, SrcQueueDefns},
+ {log_readrepair, true},
+ {read_repair_log, true},
+ {delete_mode, immediate},
+ {defer_reap_on_failure, true}
+ ]},
+ {riak_core,
+ [
+ {ring_creation_size, ?DEFAULT_RING_SIZE},
+ {vnode_management_timer, 2000},
+ {vnode_inactivity_timeout, 4000},
+ {handoff_concurrency, 16},
+ {default_bucket_props,
+ [
+ {n_val, NVal},
+ {allow_mult, true},
+ {dvv_enabled, true}
+ ]}
+ ]}]
+ ).
+
+-define(SNK_CONFIG(ClusterName, PeerList), 
+ [{riak_kv, 
+ [{replrtq_enablesink, true},
+ {replrtq_sinkqueue, ClusterName},
+ {replrtq_sinkpeers, PeerList},
+ {replrtq_sinkworkers, ?SNK_WORKERS}]}]).
+
+-define(FS_CONFIG(PeerIP, PeerPort, LocalClusterName, RemoteClusterName),
+ [{riak_kv,
+ [{ttaaefs_scope, all},
+ {ttaaefs_localnval, 3},
+ {ttaaefs_remotenval, 1},
+ {ttaaefs_peerip, PeerIP},
+ {ttaaefs_peerport, PeerPort},
+ {ttaaefs_peerprotocol, pb},
+ {ttaaefs_allcheck, 0},
+ {ttaaefs_autocheck, 0},
+ {ttaaefs_daycheck, 0},
+ {ttaaefs_hourcheck, 0},
+ {ttaaefs_nocheck, 24},
+ {ttaaefs_maxresults, ?MAX_RESULTS},
+ {ttaaefs_queuename, LocalClusterName},
+ {ttaaefs_queuename_peer, RemoteClusterName},
+ {ttaaefs_logrepairs, true}]}]).
+
+confirm() ->
+ [ClusterA, ClusterB] =
+ rt:deploy_clusters([
+ {5, ?CFG_REPL("cluster_b:any", 3)},
+ {1, ?CFG_REPL("cluster_a:any", 1)}]),
+
+ lager:info("Discover Peer IP/ports and restart with peer config"),
+ reset_peer_config(ClusterA, ClusterB),
+
+ lists:foreach(
+ fun(N) -> rt:wait_until_ready(N), rt:wait_until_pingable(N) end,
+ ClusterA ++ ClusterB
+ ),
+
+ rt:join_cluster(ClusterA),
+ rt:join_cluster(ClusterB),
+
+ ok = setup_fullsync_peer(ClusterA, hd(ClusterB)),
+
+ ok = verify_repl_delete(ClusterA),
+ pass.
+
+
+verify_repl_delete(Nodes) ->
+
+ lager:info("Commencing object load"),
+ KeyLoadFun =
+ fun(B, V) ->
+ fun(Node, KeyCount) ->
+ lager:info("Loading from key ~w on node ~w", [KeyCount, Node]),
+ KVs = 
+ test_data(
+ KeyCount + 1, KeyCount + ?NUM_KEYS_PERNODE, V),
+ ok = write_data(Node, B, KVs, []),
+ KeyCount + ?NUM_KEYS_PERNODE
+ end
+ end,
+
+ lists:foldl(KeyLoadFun(<<"B1">>, <<"v1">>), 1, Nodes),
+ lager:info("Loaded ~w objects", [?NUM_KEYS_PERNODE * length(Nodes)]),
+ wait_for_queues_to_drain(Nodes, cluster_b),
+ lists:foldl(KeyLoadFun(<<"B2">>, <<"v1">>), 1, Nodes),
+ lager:info("Loaded ~w objects", [?NUM_KEYS_PERNODE * length(Nodes)]),
+ wait_for_queues_to_drain(Nodes, cluster_b),
+ lists:foldl(KeyLoadFun(<<"B3">>, <<"v1">>), 1, Nodes),
+ lager:info("Loaded ~w objects", [?NUM_KEYS_PERNODE * length(Nodes)]),
+ wait_for_queues_to_drain(Nodes, cluster_b),
+ lists:foldl(KeyLoadFun(<<"B4">>, <<"v1">>), 1, Nodes),
+ lager:info("Loaded ~w objects", [?NUM_KEYS_PERNODE * length(Nodes)]),
+ wait_for_queues_to_drain(Nodes, cluster_b),
+
+
+ lager:info(
+ "Stopping a node - now tombstones will not be auto-reaped in A"),
+ FiddlingNode = hd(tl(Nodes)),
+ rt:stop_and_wait(FiddlingNode),
+
+ lager:info("Delete all keys for B1 bucket"),
+ delete_data(
+ hd(Nodes), <<"B1">>, lists:seq(1, ?NUM_KEYS_PERNODE * length(Nodes))),
+ lager:info("Delete attempts completed"),
+ wait_for_queues_to_drain(Nodes -- [FiddlingNode], cluster_b),
+
+ rt:start_and_wait(FiddlingNode),
+
+ wait_for_reaps_to_drain(Nodes -- [FiddlingNode]),
+
+ LoopsToSync = count_loops_to_sync(Nodes),
+
+ ?assert(is_integer(LoopsToSync)),
+ ?assert(LoopsToSync < 5).
+
+
+to_key(N) ->
+ list_to_binary(io_lib:format("K~4..0B", [N])).
+
+test_data(Start, End, V) ->
+ Keys = [to_key(N) || N <- lists:seq(Start, End)],
+ [{K, <<K/binary, V/binary>>} || K <- Keys].
+
+
+write_data(Node, Bucket, KVs, Opts) ->
+ PB = rt:pbc(Node),
+ [begin
+ O =
+ case riakc_pb_socket:get(PB, Bucket, K) of
+ {ok, Prev} ->
+ riakc_obj:update_value(Prev, V);
+ _ ->
+ riakc_obj:new(Bucket, K, V)
+ end,
+ ?assertMatch(ok, riakc_pb_socket:put(PB, O, Opts))
+ end || {K, V} <- KVs],
+ riakc_pb_socket:stop(PB),
+ ok.
+
+delete_data(Node, Bucket, Ns) ->
+ PB = rt:pbc(Node),
+ lists:foreach(
+ fun(N) -> ok = riakc_pb_socket:delete(PB, Bucket, to_key(N)) end, Ns),
+ riakc_pb_socket:stop(PB),
+ ok.
+
+reset_peer_config(ClusterA, ClusterB) ->
+ FoldToPeerConfigPB = 
+ fun(Node, Acc) ->
+ {pb, {IP, Port}} =
+ lists:keyfind(pb, 1, rt:connection_info(Node)),
+ Acc0 = case Acc of "" -> ""; _ -> Acc ++ "|" end,
+ Acc0 ++ IP ++ ":" ++ integer_to_list(Port) ++ ":pb"
+ end,
+ ClusterASnkPL = lists:foldl(FoldToPeerConfigPB, "", ClusterB),
+ ClusterBSnkPL = lists:foldl(FoldToPeerConfigPB, "", ClusterA),
+ ClusterASNkCfg = ?SNK_CONFIG(cluster_a, ClusterASnkPL),
+ ClusterBSNkCfg = ?SNK_CONFIG(cluster_b, ClusterBSnkPL),
+ lists:foreach(
+ fun(N) -> rt:set_advanced_conf(N, ClusterASNkCfg) end, ClusterA),
+ lists:foreach(
+ fun(N) -> rt:set_advanced_conf(N, ClusterBSNkCfg) end, ClusterB), 
+
+ lists:foreach(
+ fun(N) -> rt:wait_for_service(N, riak_kv) end,
+ ClusterA ++ ClusterB).
+
+
+wait_for_queues_to_drain([], QueueName) ->
+ lager:info("Queue ~w drained on nodes", [QueueName]);
+wait_for_queues_to_drain([N|Rest], QueueName) ->
+ rt:wait_until(
+ fun() ->
+ {QueueName, {0, 0, 0}} ==
+ rpc:call(N, riak_kv_replrtq_src, length_rtq, [QueueName])
+ end
+ ),
+ wait_for_queues_to_drain(Rest, QueueName).
+
+
+wait_for_reaps_to_drain([]) ->
+ lager:info("Reaps drained on all nodes");
+wait_for_reaps_to_drain([N|Rest]) ->
+ rt:wait_until(
+ fun() ->
+ {mqueue_lengths, MQLs} =
+ lists:keyfind(
+ mqueue_lengths,
+ 1,
+ rpc:call(N, riak_kv_reaper, reap_stats, [])),
+ lager:info("Reap queue lengths ~w on ~w", [MQLs, N]),
+ lists:sum(lists:map(fun({_P, L}) -> L end, MQLs)) == 0
+ end
+ ),
+ lager:info("Reaps drained fron Node ~p", [N]),
+ wait_for_reaps_to_drain(Rest).
+
+setup_fullsync_peer(ClusterA, NodeB) ->
+ {pb, {IP, Port}} = lists:keyfind(pb, 1, rt:connection_info(NodeB)),
+ ClusterACfg = ?FS_CONFIG(IP, Port, cluster_b, cluster_a),
+ lists:foreach(
+ fun(N) -> rt:set_advanced_conf(N, ClusterACfg) end, ClusterA),
+
+ lager:info("Waiting for convergence."),
+ rt:wait_until_ring_converged(ClusterA),
+ lists:foreach(
+ fun(N) -> rt:wait_for_service(N, riak_kv) end, ClusterA).
+
+count_loops_to_sync(Nodes) ->
+ count_loops_to_sync(Nodes, 1).
+
+count_loops_to_sync(_Nodes, ?MAX_LOOOPS) ->
+ false;
+count_loops_to_sync(Nodes, L) ->
+ R = rpc:call(hd(Nodes), riak_client, ttaaefs_fullsync, [all_check, 60]),
+ lager:info("Full sync A -> B ~p", [R]),
+ case R of
+ {root_compare, 0} ->
+ L;
+ _ ->
+ wait_for_queues_to_drain(Nodes, cluster_b),
+ count_loops_to_sync(Nodes, L + 1)
+ end.