Add participate_in_2i_coverage riak option (#917)

Allows marking of nodes 'down' in calculations for 2i coverage
queries.  In create_plan, it queries each node for the value of the
participate_in_2i_coverage and treats those with the value set
explicitly to `false` as 'down' for the calculation of avaiable
vnodes.

See related riak_kv commit that adds a default value to the KV schema.
This is an operator feature. The operator must bring down the node,
change the config, and start the node, for it to have effect.

Author: russelldb

src/riak_core.erl

@@ -159,7 +159,13 @@ standard_join(Node, Ring, Rejoin, Auto) ->
                                                   node(),
                                                   gossip_vsn,
                                                   GossipVsn),
-            {_, Ring5} = riak_core_capability:update_ring(Ring4),
+            ParticipateInCoverage = app_helper:get_env(riak_core,participate_in_coverage),
+            Ring4a =
+                riak_core_ring:update_member_meta(node(),
+                                                  Ring4,
+                                                  node(),
+                                                  participate_in_coverage, ParticipateInCoverage),
+            {_, Ring5} = riak_core_capability:update_ring(Ring4a),
             Ring6 = maybe_auto_join(Auto, node(), Ring5),
             riak_core_ring_manager:set_my_ring(Ring6),
             riak_core_gossip:send_ring(Node, node())

src/riak_core_apl.erl

@@ -29,7 +29,7 @@
          get_apl_ann_with_pnum/1,
          get_primary_apl/3, get_primary_apl/4,
          get_primary_apl_chbin/4,
-         first_up/2, offline_owners/1, offline_owners/2
+         first_up/2, offline_owners/1, offline_owners/2, offline_owners/3
         ]).
 
 -export_type([preflist/0, preflist_ann/0, preflist_with_pnum_ann/0]).
@@ -175,9 +175,12 @@ offline_owners(Service) ->
     offline_owners(Service, CHBin).
 
 offline_owners(Service, CHBin) ->
+    offline_owners(Service, CHBin, []).
+
+offline_owners(Service, CHBin, OtherDownNodes) ->
     UpSet = ordsets:from_list(riak_core_node_watcher:nodes(Service)),
     DownVNodes = chashbin:to_list_filter(fun({_Index, Node}) ->
-                                                 not is_up(Node, UpSet)
+                                                 (not is_up(Node, UpSet) or lists:member(Node,OtherDownNodes))
                                          end, CHBin),
     DownVNodes.
 

src/riak_core_coverage_fsm.erl

@@ -180,7 +180,6 @@ init([Mod,
 init({test, Args, StateProps}) ->
     %% Call normal init
     {ok, initialize, StateData, 0} = init(Args),
-
     %% Then tweak the state record with entries provided by StateProps
     Fields = record_info(fields, state),
     FieldPos = lists:zip(Fields, lists:seq(2, length(Fields)+1)),

src/riak_core_coverage_plan.erl

@@ -47,12 +47,18 @@
 create_plan(VNodeSelector, NVal, PVC, ReqId, Service) ->
     {ok, CHBin} = riak_core_ring_manager:get_chash_bin(),
     PartitionCount = chashbin:num_partitions(CHBin),
+    {ok, Ring} = riak_core_ring_manager:get_my_ring(),
     %% Create a coverage plan with the requested primary
     %% preference list VNode coverage.
     %% Get a list of the VNodes owned by any unavailble nodes
+    Members = riak_core_ring:all_members(Ring),
+    NonCoverageNodes = [Node || Node <- Members,
+                                      riak_core_ring:get_member_meta(Ring, Node, participate_in_coverage) == false],
+
     DownVNodes = [Index ||
                      {Index, _Node}
-                         <- riak_core_apl:offline_owners(Service, CHBin)],
+                         <- riak_core_apl:offline_owners(Service, CHBin, NonCoverageNodes)],
+
     %% Calculate an offset based on the request id to offer
     %% the possibility of different sets of VNodes being
     %% used even when all nodes are available.

src/riak_core_ring_manager.erl

@@ -331,8 +331,9 @@ stop() ->
 init([Mode]) ->
     setup_ets(Mode),
     Ring = reload_ring(Mode),
-    State = set_ring(Ring, #state{mode = Mode}),
-    riak_core_ring_events:ring_update(Ring),
+    Ring2 = node_level_config(Ring),
+    State = set_ring(Ring2, #state{mode = Mode}),
+    riak_core_ring_events:ring_update(Ring2),
     {ok, State}.
 
 reload_ring(test) ->
@@ -507,6 +508,14 @@ run_fixups([{App, Fixup}|T], BucketName, BucketProps) ->
     end,
     run_fixups(T, BucketName, BP).
 
+%% Add node level configs to ring
+-spec node_level_config(riak_core_ring:riak_core_ring()) -> riak_core_ring:riak_core_ring().
+node_level_config(Ring) ->
+    %% Check node participation in coverage queries and update ring
+    Node = node(),
+    ParticipateInCoverage = app_helper:get_env(riak_core,participate_in_coverage),
+    riak_core_ring:update_member_meta(Node, Ring, Node, participate_in_coverage, ParticipateInCoverage).
+
 set_ring(Ring, State) ->
     set_ring_global(Ring),
     Now = os:timestamp(),