Merge branch 'develop' into issue-401

Conflicts:
	src/yz_solr_proc.erl

.gitignore

@@ -1,5 +1,6 @@
 *~
 deps
+doc
 ebin
 priv/java_lib
 priv/conf/lang
@@ -14,6 +15,7 @@ priv/solr/solr-webapp
 build/
 *.class
 *.tgz
+.rebar/*
 riak_test/ebin
 tests/20*
 tests/current
@@ -24,3 +26,5 @@ bb-*-fruit*
 .yokozuna_test_dialyzer_plt
 dialyzer_warnings
 dialyzer_unhandled_warnings
+/.eqc-info
+/current_counterexample.eqc

Makefile

@@ -1,10 +1,13 @@
+EXOMETER_PACKAGES = "(basic)"
+export EXOMETER_PACKAGES
+
 REBAR ?= $(shell pwd)/rebar
 
 .PHONY: deps rel stagedevrel test
 
 all: deps compile-riak-test
 
-compile:
+compile: deps
 	$(REBAR) compile
 
 compile-riak-test: compile

README.md

@@ -5,17 +5,9 @@ _Yokozuna_ - Horizontal rope.  The top rank in sumo, usually
 translated _Grand Champion_.  The name comes from the rope a yokozuna
 wears.
 
-The goal of the Yokozuna application is to integrate Apache Solr with
-Riak in order to find the "top rank" documents for a given query.
+Yokozuna is the new implementation of Riak Search built atop Apache Solr.
+Download [Riak 2.0][downloads] to try Yokozuna. See the [official
+documentation][search-docs] for more information.
 
-**NOTICE:** Yokozuna is the new search implementation for Riak 2.0.0,
-  replacing the old riak_search repository. Breaking changes may occur
-  at any point up until the final Riak 2.0.0 release. Download
-  [Riak 2.0.0beta1][riak2beta] to try Yokozuna today.
-
-
-For more information on using Yokozuna see the
-[official Riak Search documentation][search-docs].
-
-[riak2beta]: http://docs.basho.com/riak/2.0.0beta1/downloads/
-[search-docs]: http://docs.basho.com/riak/2.0.0beta1/dev/using/search/
+[downloads]: http://docs.basho.com/riak/latest/downloads/
+[search-docs]: http://docs.basho.com/riak/latest/dev/using/search/

include/yokozuna.hrl

@@ -129,6 +129,7 @@
 
 -define(DATA_DIR, application:get_env(riak_core, platform_data_dir)).
 
+-define(DEFAULT_IDX_CREATE_TIMEOUT, 30000).
 
 -define(MAYBE(Check, Expression, Default),
         case Check of
@@ -243,8 +244,14 @@
 %%% Riak KV
 %%%===================================================================
 
+-ifdef(namespaced_types).
+-type yz_dict() :: dict:dict().
+-else.
+-type yz_dict() :: dict().
+-endif.
+
 -type obj() :: riak_object:riak_object().
--type obj_metadata() :: dict().
+-type obj_metadata() :: yz_dict().
 
 %%%===================================================================
 %%% Docs

java_src/com/basho/yokozuna/handler/EntropyData.java

@@ -140,9 +140,9 @@ public void handleRequestBody(final SolrQueryRequest req, final SolrQueryRespons
 
                     vsn = vals[0];
                     docPartition = vals[1];
-                    riakBType = vals[2];
-                    riakBName = vals[3];
-                    riakKey = vals[4];
+                    riakBType = decodeBase64DocPart(vals[2]);
+                    riakBName = decodeBase64DocPart(vals[3]);
+                    riakKey = decodeBase64DocPart(vals[4]);
                     hash = vals[5];
 
                     if (partition.equals(docPartition)) {
@@ -208,4 +208,13 @@ public String getVersion() {
     public String getSource() {
         return "TODO: implement getSource";
     }
+
+    /**
+       @param base64EncodedVal base64 encoded string
+       @return a string of decoded base64 bytes
+     */
+    private String decodeBase64DocPart(String base64EncodedVal) {
+        byte[] bytes = Base64.decodeBase64(base64EncodedVal);
+        return new String(bytes);
+    }
 }

misc/bench/src/yz_driver.erl

@@ -9,6 +9,7 @@
 -include_lib("basho_bench/include/basho_bench.hrl").
 -record(state, {default_field, fruits, pb_conns, index, bucket, iurls, surls}).
 -define(DONT_VERIFY, dont_verify).
+-define(SPACER, "testfor spaces ").
 
 -define(M100,   100000000).
 -define(M10,    10000000).
@@ -159,6 +160,17 @@ run(load_fruit, KeyValGen, _, S=#state{iurls=URLs}) ->
         {error, Reason} -> {error, Reason, S2}
     end;
 
+run(load_fruit_plus_spaces, KeyValGen, _, S=#state{iurls=URLs}) ->
+    Base = get_base(URLs),
+    {Key, Val} = KeyValGen(),
+    Key2 = mochiweb_util:quote_plus(lists:concat([?SPACER, Key])),
+    URL = ?FMT("~s/~s", [Base, Key2]),
+    S2 = S#state{iurls=wrap(URLs)},
+    case http_put(URL, "text/plain", Val) of
+        ok -> {ok, S2};
+        {error, Reason} -> {error, Reason, S2}
+    end;
+
 run(load_fruit_pb, KeyValGen, _, S=#state{bucket=Bucket, pb_conns=Conns}) ->
     Conn = get_conn(Conns),
     {Key, Val} = KeyValGen(),

rebar.config

@@ -1,6 +1,7 @@
 {cover_enabled, true}.
 {erl_opts, [warnings_as_errors,
             debug_info,
+            {platform_define, "^[0-9]+", namespaced_types},
             {parse_transform, lager_transform}]}.
 {eunit_opts, [verbose]}.
 
@@ -10,9 +11,11 @@
 {deps,
  [
   {kvc, ".*",
-   {git, "git://github.com/etrepum/kvc.git", {tag, "v1.3.0"}}},
+   {git, "git://github.com/etrepum/kvc.git", {tag, "v1.5.0"}}},
   {riak_kv, ".*",
-   {git, "git://github.com/basho/riak_kv.git", {branch, "develop"}}}
+   {git, "git://github.com/basho/riak_kv.git", {branch, "develop"}}},
+  {ibrowse, "4.0.2",
+   {git, "git://github.com/cmullaparthi/ibrowse.git", {tag, "v4.0.2"}}}
  ]}.
 
 {pre_hooks, [{compile, "./tools/grab-solr.sh"}]}.

riak_test/aae_test.erl

@@ -3,10 +3,13 @@
 -include("yokozuna.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -define(NUM_KEYS, 10000).
+-define(NUM_KEYS_SPACES, 1000).
+-define(TOTAL_KEYS, ?NUM_KEYS + ?NUM_KEYS_SPACES).
 -define(BUCKET_TYPE, <<"data">>).
 -define(INDEX, <<"fruit_aae">>).
 -define(BUCKET, {?BUCKET_TYPE, ?INDEX}).
 -define(REPAIR_MFA, {yz_exchange_fsm, repair, 2}).
+-define(SPACER, "testfor spaces ").
 -define(CFG,
         [{riak_core,
           [
@@ -35,8 +38,11 @@ confirm() ->
             PBConn = yz_rt:select_random(PBConns),
             setup_index(Cluster, PBConn, YZBenchDir),
             {0, _} = yz_rt:load_data(Cluster, ?BUCKET, YZBenchDir, ?NUM_KEYS),
+            {0, _} = yz_rt:load_data(Cluster, ?BUCKET, YZBenchDir,
+                                     ?NUM_KEYS_SPACES,
+                                     [{load_fruit_plus_spaces, 1}]),
             lager:info("Verify data was indexed"),
-            verify_num_match(Cluster, ?NUM_KEYS),
+            verify_num_match(Cluster, ?TOTAL_KEYS),
             %% Wait for a full round of exchange and then get total repair
             %% count.  Need to do this because setting AAE so agressive means
             %% that the Solr soft-commit could race with AAE and thus repair
@@ -48,12 +54,19 @@ confirm() ->
             yz_rt:wait_for_full_exchange_round(Cluster, TS1),
             RepairCountBefore = get_cluster_repair_count(Cluster),
             yz_rt:count_calls(Cluster, ?REPAIR_MFA),
-            Keys = [{?BUCKET,K} || K <- yz_rt:random_keys(?NUM_KEYS)],
-            {DelKeys, _ChangeKeys} = lists:split(length(Keys) div 2, Keys),
-            lager:info("Deleting ~p keys", [length(DelKeys)]),
-            [delete_key_in_solr(Cluster, ?INDEX, K) || K <- DelKeys],
+            NumKeys = [{?BUCKET,K} || K <- yz_rt:random_keys(?NUM_KEYS)],
+            NumKeysSpaces = [{?BUCKET,add_space_to_key(K)} ||
+                                K <- yz_rt:random_keys(?NUM_KEYS_SPACES)],
+            {DelNumKeys, _ChangeKeys} = lists:split(length(NumKeys) div 2,
+                                                    NumKeys),
+            {DelNumKeysSpaces, _ChangeKeysSpaces} = lists:split(
+                                                length(NumKeysSpaces) div 2,
+                                                NumKeysSpaces),
+            AllDelKeys = DelNumKeys ++ DelNumKeysSpaces,
+            lager:info("Deleting ~p keys", [length(AllDelKeys)]),
+            [delete_key_in_solr(Cluster, ?INDEX, K) || K <- AllDelKeys],
             lager:info("Verify Solr indexes missing"),
-            verify_num_match(Cluster, ?NUM_KEYS - length(DelKeys)),
+            verify_num_match(Cluster, ?TOTAL_KEYS - length(AllDelKeys)),
             lager:info("Clear trees so AAE will notice missing indexes"),
             [ok = rpc:call(Node, yz_entropy_mgr, clear_trees, []) || Node <- Cluster],
             lager:info("Wait for all trees to re-build"),
@@ -63,9 +76,9 @@ confirm() ->
             TS2 = erlang:now(),
             yz_rt:wait_for_full_exchange_round(Cluster, TS2),
             lager:info("Verify AAE repairs missing Solr documents"),
-            verify_num_match(Cluster, ?NUM_KEYS),
+            verify_num_match(Cluster, ?TOTAL_KEYS),
             %% Multiply by 3 because of N value
-            ExpectedNumRepairs = length(DelKeys) * 3,
+            ExpectedNumRepairs = length(AllDelKeys) * 3,
             lager:info("Verify repair count = ~p", [ExpectedNumRepairs]),
             verify_repair_count(Cluster, RepairCountBefore + ExpectedNumRepairs),
             yz_rt:stop_tracing(),
@@ -124,7 +137,7 @@ create_orphan_postings(Cluster, Keys) ->
 -spec delete_key_in_solr([node()], index_name(), bkey()) -> [ok].
 delete_key_in_solr(Cluster, Index, BKey) ->
     [begin
-         lager:info("Deleting solr doc ~s/~s on node ~p", [Index, BKey, Node]),
+         lager:info("Deleting solr doc ~s/~p on node ~p", [Index, BKey, Node]),
          ok = rpc:call(Node, yz_solr, delete, [Index, [{bkey, BKey}]])
      end || Node <- Cluster].
 
@@ -151,7 +164,6 @@ setup_index(Cluster, PBConn, YZBenchDir) ->
     ok = yz_rt:wait_for_schema(Cluster, ?INDEX, RawSchema),
     ok = yz_rt:create_bucket_type(Node, ?BUCKET_TYPE),
     ok = yz_rt:create_index(Node, ?INDEX, ?INDEX),
-    ok = yz_rt:wait_for_index(Cluster, ?INDEX),
     ok = yz_rt:set_index(Node, ?BUCKET, ?INDEX).
 
 %% @doc Verify that expired trees do not prevent exchange from
@@ -279,3 +291,7 @@ verify_repair_count(Cluster, ExpectedNumRepairs) ->
     RepairCount = get_cluster_repair_count(Cluster),
     ?assertEqual(ExpectedNumRepairs, RepairCount),
     ok.
+
+-spec add_space_to_key(binary()) -> binary().
+add_space_to_key(Key) ->
+    list_to_binary(lists:concat([?SPACER, ?BIN_TO_INT(Key)])).

riak_test/intercepts/yz_solr_intercepts.erl

@@ -0,0 +1,6 @@
+-module(yz_solr_intercepts).
+-compile(export_all).
+
+slow_cores() ->
+    timer:sleep(6000),
+    {ok, []}.

riak_test/yokozuna_essential.erl

@@ -29,6 +29,14 @@
            %% more quickly.
            {vnode_management_timer, 1000}
           ]},
+         {riak_kv,
+          [
+           %% Max number of times that a secondary system can block
+           %% handoff of primary key-value data.
+           %% Set to prevent a race-condition when propagating cluster
+           %% meta to newly joined nodes.
+           {handoff_rejected_max, infinity}
+          ]},
          {yokozuna,
           [
           {enabled, true},
@@ -60,7 +68,7 @@ confirm() ->
             timer:sleep(30000),
             Cluster2 = join_rest(Cluster, Nodes),
             rt:wait_for_cluster_service(Cluster2, yokozuna),
-            check_status(wait_for(Ref)),
+            check_bb_status(wait_for(Ref)),
             verify_non_owned_data_deleted(Cluster, ?INDEX),
             ok = test_tagging(Cluster),
             KeysDeleted = delete_some_data(Cluster2, reap_sleep()),
@@ -229,7 +237,7 @@ verify_unique_id(Cluster, PBConns) ->
 make_query_fun(PBConns, Index, Query, Expected) ->
     fun() ->
             QueryRes = query_all(PBConns, Index, Query),
-            lists:all(fun(X) -> 
+            lists:all(fun(X) ->
                 lager:info("~p~n", [X]),
                 X == Expected end, QueryRes)
     end.
@@ -303,8 +311,38 @@ async_query(Cluster, YZBenchDir) ->
     write_terms(File, Cfg),
     run_bb(async, File).
 
-check_status({Status,_}) ->
-    ?assertEqual(?SUCCESS, Status).
+check_bb_status({Status,_}) ->
+    %% If we see an error, it'll be helpful to have the basho bench output here
+    %% in the riak_test log, since otherwise it may get wiped off of the build
+    %% machines before we get a chance to investigate the failure.
+    case Status of
+        ?SUCCESS ->
+            ok;
+        _ ->
+            lager:info("basho_bench returned error status ~p", [Status]),
+            lager:info("basho_bench log dump follows:"),
+            dump_bb_logs(),
+            ?assertEqual(?SUCCESS, Status)
+    end.
+
+dump_bb_logs() ->
+    Logs = filelib:wildcard("/tmp/yz-bb-results/current/*.log"),
+    lists:foreach(fun dump_log/1, Logs).
+
+dump_log(Log) ->
+    lager:info("--- Dumping log file ~p ---", [Log]),
+    {ok, File} = file:open(Log, [read]),
+    dump_file(File),
+    lager:info("--- End log file dump of ~p ---", [Log]).
+
+dump_file(File) ->
+    case file:read_line(File) of
+        eof ->
+            ok;
+        {ok, Data} ->
+            lager:info("~s", [Data]),
+            dump_file(File)
+    end.
 
 delete_key(Cluster, Key) ->
     Node = yz_rt:select_random(Cluster),
@@ -356,9 +394,6 @@ setup_indexing(Cluster, PBConns, YZBenchDir) ->
     ok = yz_rt:create_index(Node, <<"escaped">>),
     ok = yz_rt:create_index(Node, <<"unique">>),
 
-    [yz_rt:wait_for_index(Cluster, I)
-     || I <- [?INDEX, <<"tagging">>, <<"escaped">>, <<"unique">>]],
-
     yz_rt:set_index(Node, ?BUCKET, ?INDEX, ?INDEX_N_VAL),
     yz_rt:set_index(Node, {?BUCKET_TYPE, <<"tagging">>}, <<"tagging">>),
     yz_rt:set_index(Node, {?BUCKET_TYPE, <<"escaped">>}, <<"escaped">>).
@@ -383,7 +418,7 @@ verify_deletes(Cluster, KeysDeleted, YZBenchDir) ->
            {shutdown_on_error, true}],
     File = "bb-verify-deletes",
     write_terms(File, Cfg),
-    check_status(run_bb(sync, File)).
+    check_bb_status(run_bb(sync, File)).
 
 wait_for(Ref) ->
     rt:wait_for_cmd(Ref).

riak_test/yz_dt_test.erl

@@ -16,8 +16,7 @@
 -import(yz_rt, [
                 connection_info/1,
                 create_index/2,
-                search_expect/5,
-                wait_for_index/2
+                search_expect/5
                ]).
 
 -define(assertSearch(Solr, Index, Field, Query, Count),
@@ -34,7 +33,6 @@ confirm() ->
     [ begin
           %% Create an index for each type (default schema)
           create_index(Node, BType),
-          wait_for_index(Nodes, BType),
           %% Create bucket types for datatypes with given indexes
           rt:create_and_activate_bucket_type(Node, BType, [{datatype, Type},
                                                            {allow_mult, true},

riak_test/yz_errors.erl

@@ -119,7 +119,6 @@ create_index(Cluster, Index) ->
     URL = index_url(HP, Index),
     Headers = [{"content-type", "application/json"}],
     {ok, Status, _, _} = http(put, URL, Headers, ?NO_BODY),
-    yz_rt:wait_for_index(Cluster, Index),
     yz_rt:set_bucket_type_index(Node, Index),
     yz_rt:wait_for_bucket_type(Cluster, Index),
     ?assertEqual("204", Status).