Add index.routing.allocation.require.... and cluster.routing.allocation.require.... settings

Fixes elastic#2404

Author: imotov

src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java

@@ -47,6 +47,7 @@
 import java.io.IOException;
 import java.util.*;
 
+import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.*;
 import static org.elasticsearch.common.settings.ImmutableSettings.*;
 
 /**
@@ -197,6 +198,7 @@ public static State fromString(String state) {
 
  private transient final int totalNumberOfShards;
 
+ private final DiscoveryNodeFilters requireFilters;
  private final DiscoveryNodeFilters includeFilters;
  private final DiscoveryNodeFilters excludeFilters;
 
@@ -213,17 +215,23 @@ private IndexMetaData(String index, long version, State state, Settings settings
 
  this.aliases = aliases;
 
+ ImmutableMap<String, String> requireMap = settings.getByPrefix("index.routing.allocation.require.").getAsMap();
+ if (requireMap.isEmpty()) {
+ requireFilters = null;
+ } else {
+ requireFilters = DiscoveryNodeFilters.buildFromKeyValue(AND, requireMap);
+ }
  ImmutableMap<String, String> includeMap = settings.getByPrefix("index.routing.allocation.include.").getAsMap();
  if (includeMap.isEmpty()) {
  includeFilters = null;
  } else {
- includeFilters = DiscoveryNodeFilters.buildFromKeyValue(includeMap);
+ includeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, includeMap);
  }
  ImmutableMap<String, String> excludeMap = settings.getByPrefix("index.routing.allocation.exclude.").getAsMap();
  if (excludeMap.isEmpty()) {
  excludeFilters = null;
  } else {
- excludeFilters = DiscoveryNodeFilters.buildFromKeyValue(excludeMap);
+ excludeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, excludeMap);
  }
  }
 
@@ -332,6 +340,11 @@ public <T extends Custom> T custom(String type) {
  return (T) customs.get(type);
  }
 
+ @Nullable
+ public DiscoveryNodeFilters requireFilters() {
+ return requireFilters;
+ }
+
  @Nullable
  public DiscoveryNodeFilters includeFilters() {
  return includeFilters;

src/main/java/org/elasticsearch/cluster/node/DiscoveryNodeFilters.java

@@ -19,7 +19,6 @@
 
 package org.elasticsearch.cluster.node;
 
-import com.google.common.collect.ImmutableMap;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.regex.Regex;
 import org.elasticsearch.common.settings.Settings;
@@ -32,13 +31,16 @@
  */
 public class DiscoveryNodeFilters {
 
- public static final DiscoveryNodeFilters NO_FILTERS = new DiscoveryNodeFilters(ImmutableMap.<String, String[]>of());
+ public static enum OpType {
+ AND,
+ OR
+ };
 
- public static DiscoveryNodeFilters buildFromSettings(String prefix, Settings settings) {
- return buildFromKeyValue(settings.getByPrefix(prefix).getAsMap());
+ public static DiscoveryNodeFilters buildFromSettings(OpType opType, String prefix, Settings settings) {
+ return buildFromKeyValue(opType, settings.getByPrefix(prefix).getAsMap());
  }
 
- public static DiscoveryNodeFilters buildFromKeyValue(Map<String, String> filters) {
+ public static DiscoveryNodeFilters buildFromKeyValue(OpType opType, Map<String, String> filters) {
  Map<String, String[]> bFilters = new HashMap<String, String[]>();
  for (Map.Entry<String, String> entry : filters.entrySet()) {
  String[] values = Strings.splitStringByCommaToArray(entry.getValue());
@@ -47,76 +49,123 @@ public static DiscoveryNodeFilters buildFromKeyValue(Map<String, String> filters
  }
  }
  if (bFilters.isEmpty()) {
- return NO_FILTERS;
+ return null;
  }
- return new DiscoveryNodeFilters(bFilters);
+ return new DiscoveryNodeFilters(opType, bFilters);
  }
 
  private final Map<String, String[]> filters;
 
- DiscoveryNodeFilters(Map<String, String[]> filters) {
+ private final OpType opType;
+
+ DiscoveryNodeFilters(OpType opType, Map<String, String[]> filters) {
+ this.opType = opType;
  this.filters = filters;
  }
 
  public boolean match(DiscoveryNode node) {
- if (filters.isEmpty()) {
- return true;
- }
  for (Map.Entry<String, String[]> entry : filters.entrySet()) {
  String attr = entry.getKey();
  String[] values = entry.getValue();
  if ("_ip".equals(attr)) {
  if (!(node.address() instanceof InetSocketTransportAddress)) {
- return false;
+ if (opType == OpType.AND) {
+ return false;
+ } else {
+ continue;
+ }
  }
  InetSocketTransportAddress inetAddress = (InetSocketTransportAddress) node.address();
  for (String value : values) {
  if (Regex.simpleMatch(value, inetAddress.address().getAddress().getHostAddress())) {
- return true;
+ if (opType == OpType.OR) {
+ return true;
+ }
+ } else {
+ if (opType == OpType.AND) {
+ return false;
+ }
  }
  }
- return false;
  } else if ("_host".equals(attr)) {
  if (!(node.address() instanceof InetSocketTransportAddress)) {
- return false;
+ if (opType == OpType.AND) {
+ return false;
+ } else {
+ continue;
+ }
  }
  InetSocketTransportAddress inetAddress = (InetSocketTransportAddress) node.address();
  for (String value : values) {
  if (Regex.simpleMatch(value, inetAddress.address().getHostName())) {
- return true;
+ if (opType == OpType.OR) {
+ return true;
+ }
+ } else {
+ if (opType == OpType.AND) {
+ return false;
+ }
  }
  if (Regex.simpleMatch(value, inetAddress.address().getAddress().getHostAddress())) {
- return true;
+ if (opType == OpType.OR) {
+ return true;
+ }
+ } else {
+ if (opType == OpType.AND) {
+ return false;
+ }
  }
  }
- return false;
  } else if ("_id".equals(attr)) {
  for (String value : values) {
  if (node.id().equals(value)) {
- return true;
+ if (opType == OpType.OR) {
+ return true;
+ }
+ } else {
+ if (opType == OpType.AND) {
+ return false;
+ }
  }
  }
- return false;
  } else if ("_name".equals(attr) || "name".equals(attr)) {
  for (String value : values) {
  if (Regex.simpleMatch(value, node.name())) {
- return true;
+ if (opType == OpType.OR) {
+ return true;
+ }
+ } else {
+ if (opType == OpType.AND) {
+ return false;
+ }
  }
  }
- return false;
  } else {
  String nodeAttributeValue = node.attributes().get(attr);
  if (nodeAttributeValue == null) {
- return false;
+ if (opType == OpType.AND) {
+ return false;
+ } else {
+ continue;
+ }
  }
  for (String value : values) {
  if (Regex.simpleMatch(value, nodeAttributeValue)) {
- return true;
+ if (opType == OpType.OR) {
+ return true;
+ }
+ } else {
+ if (opType == OpType.AND) {
+ return false;
+ }
  }
  }
- return false;
  }
  }
- return true;
+ if (opType == OpType.OR) {
+ return false;
+ } else {
+ return true;
+ }
  }
 }

src/main/java/org/elasticsearch/cluster/routing/allocation/decider/FilterAllocationDecider.java

@@ -30,38 +30,49 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.node.settings.NodeSettingsService;
 
+import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.*;
+
 /**
  */
 public class FilterAllocationDecider extends AllocationDecider {
 
  static {
  MetaData.addDynamicSettings(
+ "cluster.routing.allocation.require.*",
  "cluster.routing.allocation.include.*",
  "cluster.routing.allocation.exclude.*"
  );
  IndexMetaData.addDynamicSettings(
+ "index.routing.allocation.require.*",
  "index.routing.allocation.include.*",
  "index.routing.allocation.exclude.*"
  );
  }
 
+ private volatile DiscoveryNodeFilters clusterRequireFilters;
  private volatile DiscoveryNodeFilters clusterIncludeFilters;
  private volatile DiscoveryNodeFilters clusterExcludeFilters;
 
  @Inject
  public FilterAllocationDecider(Settings settings, NodeSettingsService nodeSettingsService) {
  super(settings);
+ ImmutableMap<String, String> requireMap = settings.getByPrefix("cluster.routing.allocation.require.").getAsMap();
+ if (requireMap.isEmpty()) {
+ clusterRequireFilters = null;
+ } else {
+ clusterRequireFilters = DiscoveryNodeFilters.buildFromKeyValue(AND, requireMap);
+ }
  ImmutableMap<String, String> includeMap = settings.getByPrefix("cluster.routing.allocation.include.").getAsMap();
  if (includeMap.isEmpty()) {
  clusterIncludeFilters = null;
  } else {
- clusterIncludeFilters = DiscoveryNodeFilters.buildFromKeyValue(includeMap);
+ clusterIncludeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, includeMap);
  }
  ImmutableMap<String, String> excludeMap = settings.getByPrefix("cluster.routing.allocation.exclude.").getAsMap();
  if (excludeMap.isEmpty()) {
  clusterExcludeFilters = null;
  } else {
- clusterExcludeFilters = DiscoveryNodeFilters.buildFromKeyValue(excludeMap);
+ clusterExcludeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, excludeMap);
  }
  nodeSettingsService.addListener(new ApplySettings());
  }
@@ -77,6 +88,11 @@ public Decision canRemain(ShardRouting shardRouting, RoutingNode node, RoutingAl
  }
 
  private boolean shouldFilter(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
+ if (clusterRequireFilters != null) {
+ if (!clusterRequireFilters.match(node.node())) {
+ return true;
+ }
+ }
  if (clusterIncludeFilters != null) {
  if (!clusterIncludeFilters.match(node.node())) {
  return true;
@@ -89,6 +105,11 @@ private boolean shouldFilter(ShardRouting shardRouting, RoutingNode node, Routin
  }
 
  IndexMetaData indexMd = allocation.routingNodes().metaData().index(shardRouting.index());
+ if (indexMd.requireFilters() != null) {
+ if (!indexMd.requireFilters().match(node.node())) {
+ return true;
+ }
+ }
  if (indexMd.includeFilters() != null) {
  if (!indexMd.includeFilters().match(node.node())) {
  return true;
@@ -106,13 +127,17 @@ private boolean shouldFilter(ShardRouting shardRouting, RoutingNode node, Routin
  class ApplySettings implements NodeSettingsService.Listener {
  @Override
  public void onRefreshSettings(Settings settings) {
+ ImmutableMap<String, String> requireMap = settings.getByPrefix("cluster.routing.allocation.require.").getAsMap();
+ if (!requireMap.isEmpty()) {
+ clusterRequireFilters = DiscoveryNodeFilters.buildFromKeyValue(AND, requireMap);
+ }
  ImmutableMap<String, String> includeMap = settings.getByPrefix("cluster.routing.allocation.include.").getAsMap();
  if (!includeMap.isEmpty()) {
- clusterIncludeFilters = DiscoveryNodeFilters.buildFromKeyValue(includeMap);
+ clusterIncludeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, includeMap);
  }
  ImmutableMap<String, String> excludeMap = settings.getByPrefix("cluster.routing.allocation.exclude.").getAsMap();
  if (!excludeMap.isEmpty()) {
- clusterExcludeFilters = DiscoveryNodeFilters.buildFromKeyValue(excludeMap);
+ clusterExcludeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, excludeMap);
  }
  }
  }

src/test/java/org/elasticsearch/test/integration/cluster/allocation/FilteringAllocationTests.java

@@ -93,4 +93,66 @@ public void testDecommissionNodeNoReplicas() throws Exception {
  client("node1").admin().indices().prepareRefresh().execute().actionGet();
  assertThat(client("node1").prepareCount().setQuery(QueryBuilders.matchAllQuery()).execute().actionGet().count(), equalTo(100l));
  }
+
+ @Test
+ public void testDisablingAllocationFiltering() throws Exception {
+ logger.info("--> starting 2 nodes");
+ startNode("node1");
+ startNode("node2");
+
+ logger.info("--> creating an index with no replicas");
+ client("node1").admin().indices().prepareCreate("test")
+ .setSettings(settingsBuilder().put("index.number_of_replicas", 0))
+ .execute().actionGet();
+
+ ClusterHealthResponse clusterHealthResponse = client("node1").admin().cluster().prepareHealth().setWaitForGreenStatus().execute().actionGet();
+ assertThat(clusterHealthResponse.timedOut(), equalTo(false));
+
+ logger.info("--> index some data");
+ for (int i = 0; i < 100; i++) {
+ client("node1").prepareIndex("test", "type", Integer.toString(i)).setSource("field", "value" + i).execute().actionGet();
+ }
+ client("node1").admin().indices().prepareRefresh().execute().actionGet();
+ assertThat(client("node1").prepareCount().setQuery(QueryBuilders.matchAllQuery()).execute().actionGet().count(), equalTo(100l));
+
+ logger.info("--> remove index from the first node");
+ client("node1").admin().indices().prepareUpdateSettings("test")
+ .setSettings(settingsBuilder().put("index.routing.allocation.exclude._name", "node1"))
+ .execute().actionGet();
+
+ Thread.sleep(200);
+
+ clusterHealthResponse = client("node1").admin().cluster().prepareHealth()
+ .setWaitForGreenStatus()
+ .setWaitForRelocatingShards(0)
+ .execute().actionGet();
+ assertThat(clusterHealthResponse.timedOut(), equalTo(false));
+
+ logger.info("--> verify all shards are allocated on node2 now");
+ ClusterState clusterState = client("node1").admin().cluster().prepareState().execute().actionGet().state();
+ IndexRoutingTable indexRoutingTable = clusterState.routingTable().index("test");
+ for (IndexShardRoutingTable indexShardRoutingTable : indexRoutingTable) {
+ for (ShardRouting shardRouting : indexShardRoutingTable) {
+ assertThat(clusterState.nodes().get(shardRouting.currentNodeId()).name(), equalTo("node2"));
+ }
+ }
+
+ logger.info("--> disable allocation filtering ");
+ client("node1").admin().indices().prepareUpdateSettings("test")
+ .setSettings(settingsBuilder().put("index.routing.allocation.exclude._name", ""))
+ .execute().actionGet();
+
+ Thread.sleep(200);
+
+ clusterHealthResponse = client("node1").admin().cluster().prepareHealth()
+ .setWaitForGreenStatus()
+ .setWaitForRelocatingShards(0)
+ .execute().actionGet();
+ assertThat(clusterHealthResponse.timedOut(), equalTo(false));
+
+ logger.info("--> verify that there are shards allocated on both nodes now");
+ clusterState = client("node1").admin().cluster().prepareState().execute().actionGet().state();
+ assertThat(clusterState.routingTable().index("test").numberOfNodesShardsAreAllocatedOn(), equalTo(2));
+ }
 }
+