* Sinusoidal Performance issue fix:

Added a callback to yz_index_hashtree:update exactly the same as the
callback added on the KV side in 2.0.7, because once the snapshot is
completed, we needed to (and can) allow the Solr queues to continue
writing to Solr.

* Refactor to partition/index per queue:
By making Solr queues responsible for only a single {partition, index}
pair we greatly simplify the code of the queue and the helper. Rather
than a pool of queues, and a pool of helpers, we now have one
queue/helper pair per partition/index. Beyond code clarity, this
provides many benefits:

  - We can now drain by partition, which means we can do parallel
  exchanges again.
  - Because draining a partition only affects the queues for that partition,
  we greatly reduce the risk of many vnodes being blocked by a single slow
  drain request.
  - All the code dealing with `dict`s of indexq records is now gone - each
  solrq has only a single index/partition, and the indexq record is now
  gone (collapsed into the main state record).
  - Additional refactoring around the interaction between helper and
  worker, simplifying handling of draining further.
  - If we no longer own a partition, we simply stop that worker/helper pair
  (which now has its own supervisor as well).

* Things we tried in fixing the performance issue that didn't work:
  - Throttling batches to Solr based on current in-flight queue lengths.
  This just allowed the queues to back up more.
  - Attempted to introduce backpressure by using the `riak_core_throttle`
  module in a pre-commit hook in `riak_kv_put_fsm`. This failed to resolve
  the issue, and in fact made it worse as we would hold up a large number
  of requests and then dump them all into KV and the Solr queues.

* Additional Fixes
  - Added queue length and aux_queue length to solrq_worker status
  - Fixed typespecs for fuse resets in yz_rt, and light refactor of find_representatives to make useable by yz_stat_test.
  - Remove PULSE from yz_solrq_eqc and supporting modules and Makefile
  - Meck sidejob:resource_exists so stats updates (also mecked) can work.
  - Fix format issues (places where we call ?EQC_DEBUG and forgot to pass an array) which would crash the running test, but not production code.
  - Add `?YZ_SHOULD_INDEX` macro so it can be used in a guard clause.
  - Make `yz_kv:should_index/1` use `?YZ_SHOULD_INDEX` as its implementaiton so we don't end up changing one and not the other.
  - Refactor `yz_solrq_helper:update_solr/3` to use `?YZ_SHOULD_INDEX` and reduce complexity/nested cases.
  - Clean up some deeply nested funs/maps in `yz_solrq_helper`. While we now will iterate over Entries twice, they are small (must be < max batch size) and this makes the code much easier to reason about.
  - Added stats for bad entries and extraction errors, and removed duplicate measurements of index failures.

Author: Doug Rohrer

Makefile

@@ -26,13 +26,6 @@ clean:
 distclean: clean
 	$(REBAR) delete-deps
 
-# You should 'clean' before your first run of this target
-# so that deps get built with PULSE where needed.
-pulse:
-	./rebar compile -D PULSE
-	./rebar eunit -D PULSE skip_deps=true suite=$(PULSE_TESTS)
-
-
 ##
 ## Dialyzer
 ##

debug/yz_perf.erl

@@ -350,6 +350,7 @@
 
 -define(TOMBSTONE, <<>>).
 -define(YZ_INDEX_TOMBSTONE, <<"_dont_index_">>).
+-define(YZ_SHOULD_INDEX(Index), Index =/= ?YZ_INDEX_TOMBSTONE).
 -define(YZ_INDEX, search_index).
 
 %%%===================================================================
@@ -506,7 +507,7 @@
 -type solrq_batch_max()            :: pos_integer().
 -type solrq_batch_flush_interval() :: non_neg_integer()|infinity.
 -type solrq_hwm()                  :: non_neg_integer().
--type purge_strategy()             :: ?PURGE_NONE|?PURGE_ONE|?PURGE_IDX|?PURGE_ALL.
+-type purge_strategy()             :: ?PURGE_NONE|?PURGE_ONE|?PURGE_IDX.
 
 %%%===================================================================
 %%% draining

include/yokozuna.hrl

@@ -197,8 +197,8 @@
 %%      but are guaranteed to be flushed within the
 %%      "search.queue.batch.flush_interval".
 {mapping, "search.queue.batch.minimum", "yokozuna.solrq_batch_min", [
-  {default, 1},
-  {commented, 1},
+  {default, 10},
+  {commented, 10},
   {datatype, integer},
   {validators, ["positive_integer"]}
 ]}.
@@ -211,8 +211,8 @@
 %%      "search.queue.batch.maximum object" will be delivered into Solr in any
 %%      given request.
 {mapping, "search.queue.batch.maximum", "yokozuna.solrq_batch_max", [
-  {default, 100},
-  {commented, 100},
+  {default, 500},
+  {commented, 500},
   {datatype, integer},
   {validators, ["positive_integer"]}
 ]}.
@@ -239,8 +239,8 @@
 %%      parameter exercises flow control between Riak and the Riak
 %%      Search batching subsystem if writes into Solr start to fall behind.
 {mapping, "search.queue.high_watermark", "yokozuna.solrq_hwm", [
-  {default, 10},
-  {commented, 10},
+  {default, 1000},
+  {commented, 1000},
   {datatype, integer},
   {validators, ["non_negative_integer"]}
 ]}.
@@ -264,30 +264,6 @@
   {datatype, {enum, [purge_one, purge_index, purge_all, off]}}
 ]}.
 
-%% @doc The number of solr queue workers to instantiate in the Riak
-%%      Search application. Solr queue workers are responsible for enqueuing
-%%      objects for insertion or update into Solr. Increasing the number of
-%%      solrq distributes the queuing of objects, and can lead to greater
-%%      throughput under high load, potentially at the expense of smaller batch
-%%      sizes.
-{mapping, "search.queue.worker_count", "yokozuna.solrq_worker_count", [
-  {default, 10},
-  {commented, 10},
-  {datatype, integer},
-  {validators, ["positive_integer"]}
-]}.
-
-%% @doc The number of solr queue helpers to instantiate in the Riak
-%%      Search application. Solr queue helpers are responsible for delivering
-%%      batches of data into Solr. Increasing the number of solrq helpers will
-%%      increase concurrent writes into Solr.
-{mapping, "search.queue.helper_count", "yokozuna.solrq_helper_count", [
-  {default, 10},
-  {commented, 10},
-  {datatype, integer},
-  {validators, ["positive_integer"]}
-]}.
-
 %% @doc The amount of time to wait before a drain operation times out.
 %%      If a drain times out during an AAE exchange, the exchange is cancelled
 %%      and retried at a later time.

priv/yokozuna.schema

@@ -27,5 +27,5 @@ entropy_data_cant_complete(Core, Filter) ->
 index_batch_call_orig(Core, Ops) ->
     ?M:index_batch_orig(Core, Ops).
 
-index_batch_throw_exception(_Core, _Ops) ->
-    throw({"Failed to index docs", other, error}).
+index_batch_returns_other_error(_Core, _Ops) ->
+    {error, other, "Failed to index docs"}.

riak_test/intercepts/yz_solr_intercepts.erl

@@ -63,12 +63,14 @@ confirm() ->
             verify_non_existent_index(Cluster, <<"froot">>),
             {0, _} = yz_rt:load_data(Cluster, ?BUCKET, YZBenchDir, ?NUM_KEYS),
             yz_rt:commit(Cluster, ?INDEX),
-            yz_rt:verify_num_match(Cluster, ?INDEX, ?NUM_KEYS),
+            verify_correct_solrqs(Cluster),
             %% Verify data exists before running join
+            yz_rt:verify_num_match(Cluster, ?INDEX, ?NUM_KEYS),
             Cluster2 = join_rest(Cluster, Nodes),
             rt:wait_for_cluster_service(Cluster2, riak_kv),
             rt:wait_for_cluster_service(Cluster2, yokozuna),
             verify_non_owned_data_deleted(Cluster, ?INDEX),
+            verify_correct_solrqs(Cluster2),
             wait_for_indexes(Cluster2),
             ok = test_tagging_http(Cluster2),
             ok = test_tagging_pb(Cluster2),
@@ -79,6 +81,7 @@ confirm() ->
             verify_deletes(Cluster2, ?INDEX, ?NUM_KEYS, KeysDeleted),
             ok = test_escaped_key(Cluster2),
             verify_unique_id(Cluster2, PBConns),
+            verify_deleted_index_stops_solrqs(Cluster2, PBConns),
             yz_rt:close_pb_conns(PBConns),
             pass;
         {error, bb_driver_build_failed} ->
@@ -404,3 +407,21 @@ verify_deletes(Cluster, Index, NumKeys, KeysDeleted) ->
     NumDeleted = length(KeysDeleted),
     lager:info("Verify ~p keys were deleted", [NumDeleted]),
     yz_rt:verify_num_match(Cluster, Index, NumKeys - NumDeleted).
+
+
+verify_deleted_index_stops_solrqs(Cluster, PBConns) ->
+    PBConn = hd(PBConns),
+    yz_rt:really_remove_index(Cluster, ?BUCKET, ?INDEX, PBConn),
+    verify_correct_solrqs(Cluster).
+
+verify_correct_solrqs(Cluster) ->
+    ?assertEqual(ok, rt:wait_until(Cluster, fun check_queues_match/1)).
+
+check_queues_match(Node) ->
+    CurrentIndexes = rpc:call(Node, yz_index, get_indexes_from_meta, []),
+    OwnedPartitions = rt:partitions_for_node(Node),
+    ActiveQueues = rpc:call(Node, yz_solrq_sup, active_queues, []),
+    ExpectedQueueus = [{Index, Partition} || Index <- CurrentIndexes, Partition <- OwnedPartitions],
+    lager:debug("Validating correct Solr Queues are running. Node: ~p, Expected: ~p, Active:  ~p", [Node, ExpectedQueueus, ActiveQueues]),
+    lists:sort(ExpectedQueueus) == lists:sort(ActiveQueues).
+

riak_test/yokozuna_essential.erl

@@ -174,14 +174,14 @@ confirm_admin_index(Cluster) ->
     [{Host, Port}] = host_entries(rt:connection_info([Node])),
     {ok, Pid} = riakc_pb_socket:start_link(Host, (Port-1)),
     F = fun(_) ->
-                %% Remove index from bucket props and delete it
-                yz_rt:remove_index(Node, Index),
-                DelResp = riakc_pb_socket:delete_search_index(Pid, Index),
-                case DelResp of
-                    ok -> true;
-                    {error,<<"notfound">>} -> true
-                end
-    end,
+        %% Remove index from bucket props and delete it
+        yz_rt:remove_index(Node, Index),
+        DelResp = riakc_pb_socket:delete_search_index(Pid, Index),
+        case DelResp of
+            ok -> true;
+            {error,<<"notfound">>} -> true
+        end
+        end,
     yz_rt:wait_until(Cluster, F),
     riakc_pb_socket:stop(Pid),
     ok.