yokozuna_rt additions for types and better testing w/ allow-mult=true…

src/yokozuna_rt.erl

@@ -23,6 +23,7 @@
 -include("yokozuna_rt.hrl").
 
 -export([check_exists/2,
+         clear_trees/1,
          commit/2,
          expire_trees/1,
          gen_keys/1,
@@ -225,6 +226,15 @@ expire_trees(Cluster) ->
     timer:sleep(100),
     ok.
 
+%% @doc Expire YZ trees
+-spec clear_trees([node()]) -> ok.
+clear_trees(Cluster) ->
+    lager:info("Expire all trees"),
+    _ = [ok = rpc:call(Node, yz_entropy_mgr, clear_trees, [])
+         || Node <- Cluster],
+    ok.
+
+
 %% @doc Remove index directories, removing the index.
 -spec remove_index_dirs([node()], index_name()) -> ok.
 remove_index_dirs(Nodes, IndexName) ->
@@ -364,20 +374,24 @@ create_and_set_index(Cluster, Pid, Bucket, Index) ->
     ok = riakc_pb_socket:create_search_index(Pid, Index),
     %% For possible legacy upgrade reasons, wrap create index in a wait
     wait_for_index(Cluster, Index),
-    set_index(Pid, Bucket, Index).
+    set_index(Pid, hd(Cluster), Bucket, Index).
 -spec create_and_set_index([node()], pid(), bucket(), index_name(),
                            schema_name()) -> ok.
 create_and_set_index(Cluster, Pid, Bucket, Index, Schema) ->
     %% Create a search index and associate with a bucket
     lager:info("Create a search index ~s with a custom schema named ~s and " ++
-               "associate it with bucket ~s", [Index, Schema, Bucket]),
+               "associate it with bucket ~p", [Index, Schema, Bucket]),
     ok = riakc_pb_socket:create_search_index(Pid, Index, Schema, []),
     %% For possible legacy upgrade reasons, wrap create index in a wait
     wait_for_index(Cluster, Index),
-    set_index(Pid, Bucket, Index).
-
--spec set_index(pid(), bucket(), index_name()) -> ok.
-set_index(Pid, Bucket, Index) ->
+    set_index(Pid, hd(Cluster), Bucket, Index).
+
+-spec set_index(pid(), node(), bucket(), index_name()) -> ok.
+set_index(_Pid, Node, {BucketType, _Bucket}, Index) ->
+    lager:info("Create and activate map-based bucket type ~s and tie it to search_index ~s",
+               [BucketType, Index]),
+    rt:create_and_activate_bucket_type(Node, BucketType, [{search_index, Index}]);
+set_index(Pid, _Node, Bucket, Index) ->
     ok = riakc_pb_socket:set_search_index(Pid, Bucket, Index).
 
 internal_solr_url(Host, Port, Index) ->

tests/yz_extractors.erl

@@ -28,10 +28,12 @@
 -include_lib("riakc/include/riakc.hrl").
 
 -define(FMT(S, Args), lists:flatten(io_lib:format(S, Args))).
+-define(TYPE1, <<"extractors_in_paradise">>).
+-define(TYPE2, <<"extractors_in_paradiso">>).
 -define(INDEX1, <<"test_idx1">>).
--define(BUCKET1, <<"test_bkt1">>).
+-define(BUCKET1, {?TYPE1, <<"test_bkt1">>}).
 -define(INDEX2, <<"test_idx2">>).
--define(BUCKET2, <<"test_bkt2">>).
+-define(BUCKET2, {?TYPE2, <<"test_bkt2">>}).
 -define(SCHEMANAME, <<"test">>).
 -define(TEST_SCHEMA,
 <<"<schema name=\"test\" version=\"1.5\">
@@ -278,9 +280,9 @@ get_map(Node) ->
 verify_extractor(Node, PacketData, Mod) ->
     rpc:call(Node, yz_extractor, run, [PacketData, Mod]).
 
-bucket_url({Host,Port}, BName, Key) ->
-    ?FMT("http://~s:~B/buckets/~s/keys/~s",
-         [Host, Port, BName, Key]).
+bucket_url({Host,Port}, {BType, BName}, Key) ->
+    ?FMT("http://~s:~B/types/~s/buckets/~s/keys/~s",
+         [Host, Port, BType, BName, Key]).
 
 test_extractor_works(Cluster, Packet) ->
     [rt_intercept:add(ANode, {yz_noop_extractor,
@@ -304,7 +306,7 @@ test_extractor_with_aae_expire(Cluster, Index, Bucket, Packet) ->
     {Host, Port} = rt:select_random(yokozuna_rt:host_entries(
                                                rt:connection_info(
                                                  Cluster))),
-    URL = bucket_url({Host, Port}, mochiweb_util:quote_plus(Bucket),
+    URL = bucket_url({Host, Port}, Bucket,
                      mochiweb_util:quote_plus(Key)),
 
     CT = ?EXTRACTOR_CT,
@@ -326,8 +328,13 @@ test_extractor_with_aae_expire(Cluster, Index, Bucket, Packet) ->
     yokozuna_rt:override_schema(APid, Cluster, Index, ?SCHEMANAME,
                                 ?TEST_SCHEMA_UPGRADE),
 
+    {ok, "200", RHeaders, _} = ibrowse:send_req(URL, [{"Content-Type", CT}], get,
+                                                [], []),
+    VC = proplists:get_value("X-Riak-Vclock", RHeaders),
+
     {ok, "204", _, _} = ibrowse:send_req(
-                          URL, [{"Content-Type", CT}], put, Packet),
+                          URL, [{"Content-Type", CT}, {"X-Riak-Vclock", VC}],
+                          put, Packet),
     yokozuna_rt:commit(Cluster, Index),
 
     yokozuna_rt:search_expect({Host, Port}, Index, <<"method">>,

tests/yz_handoff.erl

@@ -99,8 +99,7 @@ confirm() ->
                            join_node = Node1,
                            admin_node = Node2}],
 
-    %% Run Shell Script to count/test # of replicas and leave/join
-    %% nodes from the cluster
+    %% Run set of leave/join trials and count/test #'s from the cluster
     [[begin
           check_data(Nodes, KeyCount, BucketURL, SearchURL, State),
           check_counts(Pid, KeyCount, BucketURL)