Store indexes in the ring and sync during ring events

An index is a cluster-wide resource.  If the create index API is
called on one node then it should be called on all nodes--even if they
aren't part of the cluster during the time of the call.  A metadata
entry in the ring is used to store the list of indexes.  During a ring
event the local list of lindexes is checked against that in the ring.
The delta is calculated and missing indexes are added.  Removal is not
yet implemented.

An internal index named `_yz` has been added.  Currently, it's only
use is to determine if Solr is up by pinging the `_yz` index.
However, it may be useful to use this index as a place to store data
internal to Yokozuna or Riak.

Finally, some changes were made to the process that starts Solr.  The
current working directory was changed to be the same as Riak's start
script, the lib dir is now passed at startup, and the solr log now
resides in same location as Riak logs, and make sure to wait until
Solr is fully started or things break when a ring event is handled.

Author: rzezeski

include/yokozuna.hrl

@@ -55,8 +55,11 @@
 -type value() :: term().
 -type field() :: {name(), value()}.
 -type fields() :: [field()].
+-type index_name() :: string().
 -type doc() :: {doc, fields()}.
 -type base64() :: base64:ascii_string().
+-type ordset() :: ordsets:ordset().
+-type ring() :: riak_core_ring:riak_core_ring().
 -type solr_vclocks() :: #solr_vclocks{}.
 -type iso8601() :: string().
 -type tree_name() :: atom().
@@ -75,6 +78,10 @@
 %% Mapping from logical partition to partition
 -type logical_idx() :: [{lp(), p()}].
 
+-type node_event() :: {node_event, node(), up | down}.
+-type ring_event() :: {ring_event, riak_core_ring:riak_core_ring()}.
+-type event() :: node_event() | ring_event().
+
 
 %%%===================================================================
 %%% Macros
@@ -85,7 +92,9 @@
 
 -define(INT_TO_BIN(I), list_to_binary(integer_to_list(I))).
 
+-define(YZ_INDEX, "_yz").
 -define(YZ_DEFAULT_SOLR_PORT, "8983").
+-define(YZ_DEFAULT_SOLR_STARTUP_WAIT, 15).
 -define(YZ_EVENTS_TAB, yz_events_tab).
 -define(YZ_ENTROPY_DATA_FIELD, '_yz_ed').
 -define(YZ_ROOT_DIR, app_helper:get_env(?YZ_APP_NAME, root_dir, "data/yz")).
@@ -97,3 +106,4 @@
 -define(YZ_APP_NAME, yokozuna).
 -define(YZ_SVC_NAME, yokozuna).
 -define(YZ_VNODE_MASTER, yokozuna_vnode_master).
+-define(YZ_META_INDEXES, yokozuna_indexes).

priv/conf/config.xml

@@ -73,7 +73,7 @@
   <lib dir="../../dist/" regex="apache-solr-langid-\d.*\.jar" />
   <lib dir="../../contrib/langid/lib/" regex=".*\.jar" />
 
-  <lib dir="${yz_java_lib_dir}" />
+  <lib dir="${yz.lib.dir}" />
 
   <!-- an exact 'path' can be used instead of a 'dir' to specify a
        specific file.  This will cause a serious error to be logged if

priv/logging.properties

@@ -3,7 +3,7 @@
 handlers = java.util.logging.FileHandler
 
 java.util.logging.FileHandler.formatter = java.util.logging.SimpleFormatter
-java.util.logging.FileHandler.pattern = ./logs/solr-%g.log
+java.util.logging.FileHandler.pattern = {{platform_log_dir}}/solr-%g.log
 java.util.logging.FileHandler.append = true
 java.util.logging.FileHandler.count = 5
 java.util.logging.FileHandler.limit = 10485760

priv/solr.xml

@@ -29,6 +29,7 @@
     If 'null' (or absent), cores will not be manageable via request handler
   -->
   <cores adminPath="/admin/cores">
-    <!-- <core name="default" instanceDir="default" /> -->
+    <core name="_yz" schema="schema.xml" config="config.xml" instanceDir="_yz">
+    </core>
   </cores>
 </solr>

src/yokozuna_app.erl

@@ -31,6 +31,7 @@
 
 start(_StartType, _StartArgs) ->
     riak_core:wait_for_service(riak_kv),
+    yz_index:add_to_ring(?YZ_INDEX),
     case yokozuna_sup:start_link() of
         {ok, Pid} ->
             register_app(),

src/yz_events.erl

@@ -18,13 +18,17 @@
 %%
 %% -------------------------------------------------------------------
 
+%% @doc Functionality related to events.  This is the single producer of
+%% writes to the ETS table `yz_events`.
+
 -module(yz_events).
--compile(export_all).
 -behavior(gen_server).
+-compile(export_all).
+-export([handle_cast/2,
+         init/1,
+         terminate/2]).
 -include("yokozuna.hrl").
 
-%% @doc Functionality related to events.  This is the single producer of
-%% writes to the ETS table `yz_events`.
 
 %%%===================================================================
 %%% API
@@ -51,23 +55,45 @@ init([]) ->
     ok = create_events_table(),
     {ok, none}.
 
-handle_cast({node_event, Node, Status}, S) ->
+handle_cast({node_event, _Node, _Status}=NE, S) ->
     Mapping = get_mapping(),
-    Mapping2 = handle_node_event(Node, Status, Mapping),
+    Mapping2 = new_mapping(NE, Mapping),
     ok = set_mapping(Mapping2),
+
     {noreply, S};
 
-handle_cast({ring_event, Ring}, S) ->
+handle_cast({ring_event, Ring}=RE, S) ->
     Mapping = get_mapping(),
-    Mapping2 = handle_ring_event(Ring, Mapping),
+    Mapping2 = new_mapping(RE, Mapping),
     ok = set_mapping(Mapping2),
+
+    Local = yz_index:indexes(),
+    Cluster = yz_index:get_indexes_from_ring(Ring),
+    {Removed, Added, Same} = index_delta(Local, Cluster),
+    ok = sync_indexes(Removed, Added, Same),
+
     {noreply, S}.
 
+terminate(_Reason, _S) ->
+    ok = destroy_events_table().
+
 
 %%%===================================================================
 %%% Private
 %%%===================================================================
 
+-spec add_index(index_name()) -> ok.
+add_index(Name) ->
+    case yz_index:exists(Name) of
+        true -> ok;
+        false -> ok = yz_index:create(Name)
+    end.
+
+-spec add_indexes(ordset()) -> ok.
+add_indexes(Names) ->
+    lists:foreach(fun add_index/1, Names),
+    ok.
+
 -spec add_node(node(), list()) -> list().
 add_node(Node, Mapping) ->
     HostPort = host_port(Node),
@@ -88,19 +114,10 @@ create_events_table() ->
     ?YZ_EVENTS_TAB = ets:new(?YZ_EVENTS_TAB, Opts),
     ok.
 
--spec handle_node_event(node(), up | down, list()) -> list().
-handle_node_event(Node, down, Mapping) ->
-    remove_node(Node, Mapping);
-handle_node_event(Node, up, Mapping) ->
-    add_node(Node, Mapping).
-
--spec handle_ring_event(riak_core_ring:ring(), list()) -> list().
-handle_ring_event(Ring, Mapping) ->
-    Nodes = riak_core_ring:all_members(Ring),
-    {Removed, Added} = node_ops(Mapping, Nodes),
-    Mapping2 = remove_nodes(Removed, Mapping),
-    Mapping3 = add_nodes(Added, Mapping2),
-    check_unkown(Mapping3).
+-spec destroy_events_table() -> ok.
+destroy_events_table() ->
+    true = ets:delete(?YZ_EVENTS_TAB),
+    ok.
 
 -spec host_port(node()) -> {string(), non_neg_integer() | unknown}.
 host_port(Node) ->
@@ -118,6 +135,14 @@ hostname(Node) ->
     [_, Host] = re:split(S, "@", [{return, list}]),
     Host.
 
+-spec index_delta(ordset(), ordset()) ->
+                         {Removed::ordset(), Added::ordset(), Same::ordset()}.
+index_delta(Local, Cluster) ->
+    Removed = ordsets:subtract(Local, Cluster),
+    Added = ordsets:subtract(Cluster, Local),
+    Same = ordsets:intersection(Cluster, Local),
+    {Removed, Added, Same}.
+
 -spec is_unknown(tuple()) -> boolean().
 is_unknown({_, {_, unknown}}) -> true;
 is_unknown({_, {_, Port}}) when is_list(Port) -> false.
@@ -126,6 +151,18 @@ is_unknown({_, {_, Port}}) when is_list(Port) -> false.
 just_nodes(Mapping) ->
     [Node || {Node, _} <- Mapping].
 
+-spec new_mapping(event(), list()) -> list().
+new_mapping({node_event, Node, down}, Mapping) ->
+    remove_node(Node, Mapping);
+new_mapping({node_event, Node, up}, Mapping) ->
+    add_node(Node, Mapping);
+new_mapping({ring_event, Ring}, Mapping) ->
+    Nodes = riak_core_ring:all_members(Ring),
+    {Removed, Added} = node_ops(Mapping, Nodes),
+    Mapping2 = remove_nodes(Removed, Mapping),
+    Mapping3 = add_nodes(Added, Mapping2),
+    check_unkown(Mapping3).
+
 -spec node_ops(list(), list()) -> {Removed::list(), Added::list()}.
 node_ops(Mapping, Nodes) ->
     MappingNodesSet = sets:from_list(just_nodes(Mapping)),
@@ -134,6 +171,9 @@ node_ops(Mapping, Nodes) ->
     Added = sets:subtract(NodesSet, MappingNodesSet),
     {sets:to_list(Removed), sets:to_list(Added)}.
 
+remove_indexes(_Names) ->
+    throw(implement_remove_indexes).
+
 -spec remove_node(node(), list()) -> list().
 remove_node(Node, Mapping) ->
     proplists:delete(Node, Mapping).
@@ -154,6 +194,10 @@ set_mapping(Mapping) ->
     true = ets:insert(?YZ_EVENTS_TAB, [{mapping, Mapping}]),
     ok.
 
+sync_indexes(_Removed, Added, Same) ->
+    %% ok = remove_indexes(Removed),
+    ok = add_indexes(Added ++ Same).
+
 watch_node_events() ->
     riak_core_node_watcher_events:add_sup_callback(fun send_node_update/1).
 

src/yz_index.erl

@@ -31,31 +31,59 @@
 %%% API
 %%%===================================================================
 
+-spec add_to_ring(string()) -> ok.
+add_to_ring(Name) ->
+    {ok, Ring} = riak_core_ring_manager:get_raw_ring(),
+    ok = add_to_ring(Ring, Name).
+
 %% TODO: Allow data dir to be changed
 -spec create(string()) -> ok.
 create(Name) ->
     IndexDir = index_dir(Name),
     ConfDir = filename:join([IndexDir, "conf"]),
     ConfFiles = filelib:wildcard(filename:join([?YZ_PRIV, "conf", "*"])),
     DataDir = filename:join([IndexDir, "data"]),
-    JavaLibDir = java_lib_dir(),
 
-    make_dirs([ConfDir, DataDir, JavaLibDir]),
+    make_dirs([ConfDir, DataDir]),
     copy_files(ConfFiles, ConfDir),
 
     CoreProps = [
                  {name, Name},
                  {index_dir, IndexDir},
                  {cfg_file, ?YZ_CORE_CFG_FILE},
-                 {java_lib_dir, JavaLibDir},
                  {schema_file, ?YZ_SCHEMA_FILE}
                 ],
-    ok = yz_solr:core(create, CoreProps).
+    {ok, _, _} = yz_solr:core(create, CoreProps),
+    ok.
+
+-spec exists(string()) -> boolean().
+exists(Name) ->
+    true == yz_solr:ping(Name).
+
+-spec get_indexes_from_ring(ring()) -> ordset().
+get_indexes_from_ring(Ring) ->
+    case riak_core_ring:get_meta(?YZ_META_INDEXES, Ring) of
+        {ok, Indexes} -> Indexes;
+        undefined -> []
+    end.
+
+-spec indexes() -> ordset().
+indexes() ->
+    {ok, _, Body} = yz_solr:core(status, [{wt,json}]),
+    Status = yz_solr:get_path(mochijson2:decode(Body), [<<"status">>]),
+    ordsets:from_list([binary_to_list(Name) || {Name, _} <- Status]).
 
 %%%===================================================================
 %%% Private
 %%%===================================================================
 
+add_to_ring(Ring, Name) ->
+    Indexes = get_indexes_from_ring(Ring),
+    Indexes2 = ordsets:add_element(Name, Indexes),
+    Ring2 = riak_core_ring:update_meta(?YZ_META_INDEXES, Indexes2, Ring),
+    {ok, _Ring3} = riak_core_ring_manager:ring_trans(set_ring_trans(Ring2), []),
+    ok.
+
 copy_files([], _) ->
     ok;
 copy_files([File|Rest], Dir) ->
@@ -75,9 +103,6 @@ index_dir(Name) ->
     YZDir = app_helper:get_env(?YZ_APP_NAME, yz_dir, ?YZ_DEFAULT_DIR),
     filename:absname(filename:join([YZDir, Name])).
 
-java_lib_dir() ->
-    ?YZ_PRIV ++ "/java_lib".
-
 make_dir(Dir) ->
     case filelib:is_dir(Dir) of
         true ->
@@ -96,6 +121,10 @@ make_dirs([Dir|Rest]) ->
 schema_file() ->
     ?YZ_PRIV ++ "/" ++ ?YZ_SCHEMA_FILE.
 
+set_ring_trans(Ring) ->
+    fun(_,_) -> {new_ring, Ring} end.
+
+
 %% Old stuff relate to exception handling ideas I was playing with
 
 %% create2(Name, _Schema) ->

src/yz_solr.erl

@@ -24,7 +24,6 @@
 
 -define(CORE_ALIASES, [{index_dir, instanceDir},
                        {cfg_file, config},
-                       {java_lib_dir, "property.yz_java_lib_dir"},
                        {schema_file, schema}]).
 -define(DEFAULT_URL, "http://localhost:8983/solr").
 -define(DEFAULT_VCLOCK_N, 1000).
@@ -61,10 +60,10 @@ core(Action, Props) ->
     URL = BaseURL ++ "?" ++ Encoded,
 
     case ibrowse:send_req(URL, [], get, [], Opts) of
-        {ok, "200", _Headers, _Body} ->
-            ok;
+        {ok, "200", Headers, Body} ->
+            {ok, Headers, Body};
         X ->
-            throw({error_calling_solr, create_core, X})
+            throw({error_calling_solr, core, Action, X})
     end.
 
 delete(Core, DocID) ->
@@ -174,7 +173,8 @@ filter_to_str({Partition, FPFilter}) ->
     "(" ++ PNQ ++ " AND (" ++ FPQ2 ++ "))".
 
 
-convert_action(create) -> "CREATE".
+convert_action(create) -> "CREATE";
+convert_action(status) -> "STATUS".
 
 %% TODO: Encoding functions copied from esolr, redo this.
 encode_commit() ->
@@ -214,6 +214,16 @@ get_pairs(R) ->
 to_pair({struct, [{_,DocId},{_,Base64VClock}]}) ->
     {DocId, Base64VClock}.
 
+get_path({struct, PL}, Path) ->
+    get_path(PL, Path);
+get_path(PL, [Name]) ->
+    case proplists:get_value(Name, PL) of
+        {struct, Obj} -> Obj;
+        Val -> Val
+    end;
+get_path(PL, [Name|Path]) ->
+    get_path(proplists:get_value(Name, PL), Path).
+
 get_response(R) ->
     json_get_key(<<"response">>, R).