Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Add a Riak Test

This Riak Test module verifies essential behavior of Yokozuna.
  • Loading branch information...
commit 917cb9ad7f5d0d9a9ce64274849d34108fbc88a3 1 parent 3d678fc
Ryan Zezeski authored
3  .gitignore
View
@@ -8,3 +8,6 @@ priv/solr-jars
priv/lucene-solr
*.class
*.tgz
+riak_test/ebin
+tests/20*
+tests/current
BIN  rebar
View
Binary file not shown
10 rebar.config
View
@@ -12,8 +12,16 @@
{riak_core, ".*",
{git, "git://github.com/rzezeski/riak_core", {branch, "rz-yokozuna"}}},
{riak_kv, ".*",
- {git, "git://github.com/rzezeski/riak_kv", {branch, "rz-yokozuna"}}}
+ {git, "git://github.com/rzezeski/riak_kv", {branch, "rz-yokozuna"}}},
+ {riak_test, ".*",
+ {git, "git://github.com/basho/riak_test", {branch, "master"}}}
]}.
{pre_hooks, [{compile, "./priv/grab-solr.sh"},
{compile, "./priv/build-jar.sh"}]}.
+
+{plugins, [rebar_riak_test_plugin]}.
+{riak_test, [
+ {test_paths, ["riak_test/", "deps/riak_test/tests"]},
+ {test_output, "riak_test/ebin"}
+]}.
171 riak_test/yokozuna_essential.erl
View
@@ -0,0 +1,171 @@
+-module(yokozuna_essential).
+-compile(export_all).
+-include_lib("eunit/include/eunit.hrl").
+
+-define(FRUIT_SCHEMA_NAME, <<"fruit">>).
+-define(INDEX_S, "fruit").
+-define(INDEX_B, <<"fruit">>).
+-define(NUM_KEYS, 10000).
+-define(SUCCESS, 0).
+
+confirm() ->
+ YZBenchDir = rt:get_os_env("YZ_BENCH_DIR"),
+ random:seed(now()),
+ Nodes = rt:deploy_nodes(4),
+ Cluster = join_three(Nodes),
+ wait_for_joins(Cluster),
+ setup_indexing(Cluster, YZBenchDir),
+ load_data(Cluster, YZBenchDir),
+ Ref = async_query(Cluster, YZBenchDir),
+ Cluster2 = join_rest(Cluster, Nodes),
+ check_status(wait_for(Ref)),
+ KeysDeleted = delete_some_data(Cluster2, reap_sleep()),
+ verify_deletes(Cluster2, KeysDeleted, YZBenchDir),
+ pass.
+
+async_query(Cluster, YZBenchDir) ->
+ lager:info("Run async query against cluster ~p", [Cluster]),
+ Hosts = host_entries(rt:connection_info(Cluster)),
+ Concurrent = length(Hosts),
+ Apple = {search,"apple","id",?NUM_KEYS},
+ Cfg = [{mode, {rate,8}},
+ {duration, 2},
+ {concurrent, Concurrent},
+ {code_paths, [YZBenchDir]},
+ {driver, yz_driver},
+ {operations, [{Apple,1}]},
+ {http_conns, Hosts},
+ {pb_conns, []},
+ {search_path, "/search/" ++ ?INDEX_S},
+ {shutdown_on_error, true}],
+
+ File = "bb-query-fruit-" ++ ?INDEX_S,
+ write_terms(File, Cfg),
+ run_bb(async, File).
+
+check_status({Status,_}) ->
+ ?assertEqual(?SUCCESS, Status).
+
+create_index(Node, Index, SchemaName) ->
+ lager:info("Creating index ~s [~p]", [Index, Node]),
+ rpc:call(Node, yz_index, create, [Index, SchemaName]).
+
+delete_key(Cluster, Key) ->
+ Node = select_random(Cluster),
+ lager:info("Deleting key ~s", [Key]),
+ {ok, C} = riak:client_connect(Node),
+ C:delete(?INDEX_B, list_to_binary(Key)).
+
+delete_some_data(Cluster, ReapSleep) ->
+ Num = random:uniform(100),
+ lager:info("Deleting ~p keys", [Num]),
+ Keys = [integer_to_list(random:uniform(?NUM_KEYS))
+ || _ <- lists:seq(1, Num)],
+ [delete_key(Cluster, K) || K <- Keys],
+ lager:info("Sleeping ~ps to allow for reap", [ReapSleep]),
+ timer:sleep(timer:seconds(ReapSleep)),
+ Keys.
+
+host_entries(ClusterConnInfo) ->
+ [proplists:get_value(http, I) || {_,I} <- ClusterConnInfo].
+
+install_hook(Node, Index) ->
+ lager:info("Install index hook on bucket ~s [~p]", [Index, Node]),
+ rpc:call(Node, yz_kv, install_hook, [Index]).
+
+join_three(Nodes) ->
+ [NodeA|Others] = All = lists:sublist(Nodes, 3),
+ [rt:join(Node, NodeA) || Node <- Others],
+ All.
+
+join_rest([NodeA|_]=Cluster, Nodes) ->
+ ToJoin = Nodes -- Cluster,
+ [begin rt:join(Node, NodeA) end || Node <- ToJoin],
+ Nodes.
+
+load_data(Cluster, YZBenchDir) ->
+ lager:info("Load data onto cluster ~p", [Cluster]),
+ Hosts = host_entries(rt:connection_info(Cluster)),
+ KeyGen = {function, yz_driver, fruit_key_val_gen, [?NUM_KEYS]},
+ Cfg = [{mode,max},
+ {duration,5},
+ {concurrent, 3},
+ {code_paths, [YZBenchDir]},
+ {driver, yz_driver},
+ {index_path, "/riak/fruit"},
+ {http_conns, Hosts},
+ {pb_conns, []},
+ {key_generator, KeyGen},
+ {operations, [{load_fruit, 1}]}],
+ File = "bb-load-" ++ ?INDEX_S,
+ write_terms(File, Cfg),
+ run_bb(sync, File).
+
+read_schema(YZBenchDir) ->
+ Path = filename:join([YZBenchDir, "schemas", "fruit_schema.xml"]),
+ {ok, RawSchema} = file:read_file(Path),
+ RawSchema.
+
+reap_sleep() ->
+ %% NOTE: This is hardcoded to 5s now but if this test ever allows
+ %% configuation of deletion policy then this should be
+ %% calculated.
+ 10.
+
+run_bb(Method, File) ->
+ Fun = case Method of
+ sync -> cmd;
+ async -> spawn_cmd
+ end,
+ rt:Fun("$BASHO_BENCH/basho_bench " ++ File).
+
+select_random(List) ->
+ Length = length(List),
+ Idx = random:uniform(Length),
+ lists:nth(Idx, List).
+
+setup_indexing(Cluster, YZBenchDir) ->
+ Node = select_random(Cluster),
+ RawSchema = read_schema(YZBenchDir),
+ ok = store_schema(Node, ?FRUIT_SCHEMA_NAME, RawSchema),
+ ok = create_index(Node, ?INDEX_S, ?FRUIT_SCHEMA_NAME),
+ ok = install_hook(Node, ?INDEX_B),
+ %% Give Solr time to build index
+ timer:sleep(5000).
+
+store_schema(Node, Name, RawSchema) ->
+ ok = rpc:call(Node, yz_schema, store, [Name, RawSchema]).
+
+verify_deletes(Cluster, KeysDeleted, YZBenchDir) ->
+ NumDeleted = length(KeysDeleted),
+ lager:info("Verify ~p keys were deleted", [NumDeleted]),
+ Hosts = host_entries(rt:connection_info(Cluster)),
+ Concurrent = length(Hosts),
+ Apple = {search,"apple","id",?NUM_KEYS - NumDeleted},
+
+ Cfg = [{mode, max},
+ {duration, 1},
+ {concurrent, Concurrent},
+ {driver, yz_driver},
+ {code_paths, [YZBenchDir]},
+ {operations, [{Apple,1}]},
+ {http_conns, Hosts},
+ {pb_conns, []},
+ {search_path, "/search/" ++ ?INDEX_S},
+ {shutdown_on_error, true}],
+
+ File = "bb-verify-deletes-" ++ ?INDEX_S,
+ write_terms(File, Cfg),
+ check_status(run_bb(sync, File)).
+
+wait_for(Ref) ->
+ rt:wait_for_cmd(Ref).
+
+wait_for_joins(Cluster) ->
+ rt:wait_until_nodes_ready(Cluster),
+ rt:wait_until_no_pending_changes(Cluster).
+
+write_terms(File, Terms) ->
+ {ok, IO} = file:open(File, [write]),
+ [io:fwrite(IO, "~p.~n", [T]) || T <- Terms],
+ file:close(IO).
Please sign in to comment.
Something went wrong with that request. Please try again.