honeymoose
/
OpenSearch
mirror of https://github.com/honeymoose/OpenSearch.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Changing dynamic index and cluster settings should work on master-only nodes 

Fixes #2675
This commit is contained in:
Igor Motov 2013-02-25 18:39:59 -05:00
parent bd75b731c6
commit de243493c9
37 changed files with 714 additions and 551 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
src/main/java/org/elasticsearch/action/admin/cluster/settings/TransportClusterUpdateSettingsAction.java
View File

@ -29,6 +29,8 @@ import org.elasticsearch.cluster.block.ClusterBlocks;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.allocation.AllocationService;
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
import org.elasticsearch.cluster.settings.ClusterDynamicSettings;
import org.elasticsearch.cluster.settings.DynamicSettings;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.ImmutableSettings;
@ -49,11 +51,14 @@ public class TransportClusterUpdateSettingsAction extends TransportMasterNodeOpe
private final AllocationService allocationService;
private final DynamicSettings dynamicSettings;
@Inject
public TransportClusterUpdateSettingsAction(Settings settings, TransportService transportService, ClusterService clusterService, ThreadPool threadPool,
AllocationService allocationService) {
AllocationService allocationService, @ClusterDynamicSettings DynamicSettings dynamicSettings) {
super(settings, transportService, clusterService, threadPool);
this.allocationService = allocationService;
this.dynamicSettings = dynamicSettings;
}
@Override
@ -89,7 +94,7 @@ public class TransportClusterUpdateSettingsAction extends TransportMasterNodeOpe
ImmutableSettings.Builder transientSettings = ImmutableSettings.settingsBuilder();
transientSettings.put(currentState.metaData().transientSettings());
for (Map.Entry<String, String> entry : request.transientSettings().getAsMap().entrySet()) {
if (MetaData.hasDynamicSetting(entry.getKey()) || entry.getKey().startsWith("logger.")) {
if (dynamicSettings.hasDynamicSetting(entry.getKey()) || entry.getKey().startsWith("logger.")) {
transientSettings.put(entry.getKey(), entry.getValue());
changed = true;
} else {
@ -100,7 +105,7 @@ public class TransportClusterUpdateSettingsAction extends TransportMasterNodeOpe
ImmutableSettings.Builder persistentSettings = ImmutableSettings.settingsBuilder();
persistentSettings.put(currentState.metaData().persistentSettings());
for (Map.Entry<String, String> entry : request.persistentSettings().getAsMap().entrySet()) {
if (MetaData.hasDynamicSetting(entry.getKey()) || entry.getKey().startsWith("logger.")) {
if (dynamicSettings.hasDynamicSetting(entry.getKey()) || entry.getKey().startsWith("logger.")) {
changed = true;
persistentSettings.put(entry.getKey(), entry.getValue());
} else {

7
src/main/java/org/elasticsearch/cluster/ClusterModule.java
View File

@ -28,10 +28,12 @@ import org.elasticsearch.cluster.routing.RoutingService;
import org.elasticsearch.cluster.routing.allocation.AllocationModule;
import org.elasticsearch.cluster.routing.operation.OperationRoutingModule;
import org.elasticsearch.cluster.service.InternalClusterService;
import org.elasticsearch.cluster.settings.ClusterDynamicSettingsModule;
import org.elasticsearch.common.inject.AbstractModule;
import org.elasticsearch.common.inject.Module;
import org.elasticsearch.common.inject.SpawnModules;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.settings.IndexDynamicSettingsModule;
/**
*
@ -46,7 +48,10 @@ public class ClusterModule extends AbstractModule implements SpawnModules {
@Override
public Iterable<? extends Module> spawnModules() {
return ImmutableList.of(new AllocationModule(settings), new OperationRoutingModule(settings));
return ImmutableList.of(new AllocationModule(settings),
new OperationRoutingModule(settings),
new ClusterDynamicSettingsModule(),
new IndexDynamicSettingsModule());
}
@Override

37
src/main/java/org/elasticsearch/cluster/metadata/IndexMetaData.java
View File

@ -20,7 +20,6 @@
package org.elasticsearch.cluster.metadata;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import org.elasticsearch.ElasticSearchIllegalArgumentException;
import org.elasticsearch.ElasticSearchIllegalStateException;
import org.elasticsearch.cluster.block.ClusterBlock;
@ -32,7 +31,6 @@ import org.elasticsearch.common.collect.MapBuilder;
import org.elasticsearch.common.compress.CompressedString;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.regex.Regex;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.settings.loader.SettingsLoader;
@ -45,9 +43,12 @@ import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.warmer.IndexWarmersMetaData;
import java.io.IOException;
import java.util.*;
import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.*;
import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.AND;
import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.OR;
import static org.elasticsearch.common.settings.ImmutableSettings.*;
/**
@ -108,39 +109,11 @@ public class IndexMetaData {
return factory;
}
private static ImmutableSet<String> dynamicSettings = ImmutableSet.<String>builder()
.add(IndexMetaData.SETTING_NUMBER_OF_REPLICAS)
.add(IndexMetaData.SETTING_AUTO_EXPAND_REPLICAS)
.add(IndexMetaData.SETTING_READ_ONLY)
.add(IndexMetaData.SETTING_BLOCKS_READ)
.add(IndexMetaData.SETTING_BLOCKS_WRITE)
.add(IndexMetaData.SETTING_BLOCKS_METADATA)
.build();
public static final ClusterBlock INDEX_READ_ONLY_BLOCK = new ClusterBlock(5, "index read-only (api)", false, false, RestStatus.FORBIDDEN, ClusterBlockLevel.WRITE, ClusterBlockLevel.METADATA);
public static final ClusterBlock INDEX_READ_BLOCK = new ClusterBlock(7, "index read (api)", false, false, RestStatus.FORBIDDEN, ClusterBlockLevel.READ);
public static final ClusterBlock INDEX_WRITE_BLOCK = new ClusterBlock(8, "index write (api)", false, false, RestStatus.FORBIDDEN, ClusterBlockLevel.WRITE);
public static final ClusterBlock INDEX_METADATA_BLOCK = new ClusterBlock(9, "index metadata (api)", false, false, RestStatus.FORBIDDEN, ClusterBlockLevel.METADATA);
public static ImmutableSet<String> dynamicSettings() {
return dynamicSettings;
}
public static boolean hasDynamicSetting(String key) {
for (String dynamicSetting : dynamicSettings) {
if (Regex.simpleMatch(dynamicSetting, key)) {
return true;
}
}
return false;
}
public static synchronized void addDynamicSettings(String... settings) {
HashSet<String> updatedSettings = new HashSet<String>(dynamicSettings);
updatedSettings.addAll(Arrays.asList(settings));
dynamicSettings = ImmutableSet.copyOf(updatedSettings);
}
public static enum State {
OPEN((byte) 0),
CLOSE((byte) 1);

25
src/main/java/org/elasticsearch/cluster/metadata/MetaData.java
View File

@ -99,29 +99,6 @@ public class MetaData implements Iterable<IndexMetaData> {
public static final ClusterBlock CLUSTER_READ_ONLY_BLOCK = new ClusterBlock(6, "cluster read-only (api)", false, false, RestStatus.FORBIDDEN, ClusterBlockLevel.WRITE, ClusterBlockLevel.METADATA);
private static ImmutableSet<String> dynamicSettings = ImmutableSet.<String>builder()
.add(SETTING_READ_ONLY)
.build();
public static ImmutableSet<String> dynamicSettings() {
return dynamicSettings;
}
public static boolean hasDynamicSetting(String key) {
for (String dynamicSetting : dynamicSettings) {
if (Regex.simpleMatch(dynamicSetting, key)) {
return true;
}
}
return false;
}
public static synchronized void addDynamicSettings(String... settings) {
HashSet<String> updatedSettings = new HashSet<String>(dynamicSettings);
updatedSettings.addAll(Arrays.asList(settings));
dynamicSettings = ImmutableSet.copyOf(updatedSettings);
}
public static final MetaData EMPTY_META_DATA = newMetaDataBuilder().build();
private final long version;
@ -696,7 +673,7 @@ public class MetaData implements Iterable<IndexMetaData> {
public int getTotalNumberOfShards() {
return totalNumberOfShards();
}
public int numberOfShards() {
return this.numberOfShards;
}

9
src/main/java/org/elasticsearch/cluster/metadata/MetaDataUpdateSettingsService.java
View File

@ -26,12 +26,14 @@ import org.elasticsearch.cluster.block.ClusterBlocks;
import org.elasticsearch.cluster.routing.RoutingTable;
import org.elasticsearch.cluster.routing.allocation.AllocationService;
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
import org.elasticsearch.cluster.settings.DynamicSettings;
import org.elasticsearch.common.Booleans;
import org.elasticsearch.common.Priority;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.index.settings.IndexDynamicSettings;
import java.util.Map;
import java.util.Set;
@ -47,12 +49,15 @@ public class MetaDataUpdateSettingsService extends AbstractComponent implements
private final AllocationService allocationService;
private final DynamicSettings dynamicSettings;
@Inject
public MetaDataUpdateSettingsService(Settings settings, ClusterService clusterService, AllocationService allocationService) {
public MetaDataUpdateSettingsService(Settings settings, ClusterService clusterService, AllocationService allocationService, @IndexDynamicSettings DynamicSettings dynamicSettings) {
super(settings);
this.clusterService = clusterService;
this.clusterService.add(this);
this.allocationService = allocationService;
this.dynamicSettings = dynamicSettings;
}
@Override
@ -139,7 +144,7 @@ public class MetaDataUpdateSettingsService extends AbstractComponent implements
final Set<String> removedSettings = Sets.newHashSet();
for (String key : updatedSettingsBuilder.internalMap().keySet()) {
if (!IndexMetaData.hasDynamicSetting(key)) {
if (!dynamicSettings.hasDynamicSetting(key)) {
removedSettings.add(key);
}
}

321
src/main/java/org/elasticsearch/cluster/routing/allocation/allocator/BalancedShardsAllocator.java
View File

@ -19,26 +19,12 @@
package org.elasticsearch.cluster.routing.allocation.allocator;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import org.apache.lucene.util.SorterTemplate;
import org.elasticsearch.ElasticSearchIllegalArgumentException;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.MutableShardRouting;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.RoutingNodes;
import org.elasticsearch.cluster.routing.ShardRoutingState;
import org.elasticsearch.cluster.routing.*;
import org.elasticsearch.cluster.routing.allocation.FailedRerouteAllocation;
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
import org.elasticsearch.cluster.routing.allocation.StartedRerouteAllocation;
@ -51,46 +37,36 @@ import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.node.settings.NodeSettingsService;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import java.util.*;
import static org.elasticsearch.cluster.routing.ShardRoutingState.INITIALIZING;
import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
/**
* The {@link BalancedShardsAllocator} re-balances the nodes allocations
* within an cluster based on a {@link WeightFunction}. The clusters balance is defined by four parameters which can be set
* in the cluster update API that allows changes in real-time:
*
* <ul><li><code>cluster.routing.allocation.balance.shard</code> - The <b>shard balance</b> defines the weight factor
* for shards allocated on a {@link RoutingNode}</li>
* <li><code>cluster.routing.allocation.balance.index</code> - The <b>index balance</b> defines a factor to the number
* of {@link ShardRouting}s per index allocated on a specific node</li>
* <li><code>cluster.routing.allocation.balance.primary</code> - the <b>primary balance</b> defines a weight factor for
* the number of primaries of a specific index allocated on a node</li>
* <li><code>cluster.routing.allocation.balance.threshold</code> - A <b>threshold</b> to set the minimal optimization
* value of operations that should be performed</li>
* </ul>
*
* These parameters are combined in a {@link WeightFunction} that allows calculation of node weights which
* are used to re-balance shards based on global as well as per-index factors.
*/
* The {@link BalancedShardsAllocator} re-balances the nodes allocations
* within an cluster based on a {@link WeightFunction}. The clusters balance is defined by four parameters which can be set
* in the cluster update API that allows changes in real-time:
* <p/>
* <ul><li><code>cluster.routing.allocation.balance.shard</code> - The <b>shard balance</b> defines the weight factor
* for shards allocated on a {@link RoutingNode}</li>
* <li><code>cluster.routing.allocation.balance.index</code> - The <b>index balance</b> defines a factor to the number
* of {@link org.elasticsearch.cluster.routing.ShardRouting}s per index allocated on a specific node</li>
* <li><code>cluster.routing.allocation.balance.primary</code> - the <b>primary balance</b> defines a weight factor for
* the number of primaries of a specific index allocated on a node</li>
* <li><code>cluster.routing.allocation.balance.threshold</code> - A <b>threshold</b> to set the minimal optimization
* value of operations that should be performed</li>
* </ul>
* <p/>
* These parameters are combined in a {@link WeightFunction} that allows calculation of node weights which
* are used to re-balance shards based on global as well as per-index factors.
*/
public class BalancedShardsAllocator extends AbstractComponent implements ShardsAllocator {
public static final String SETTING_THRESHOLD = "cluster.routing.allocation.balance.threshold";
public static final String SETTING_INDEX_BALANCE_FACTOR = "cluster.routing.allocation.balance.index";
public static final String SETTING_SHARD_BALANCE_FACTOR = "cluster.routing.allocation.balance.shard";
public static final String SETTING_PRIMARY_BALANCE_FACTOR = "cluster.routing.allocation.balance.primary";
static {
MetaData.addDynamicSettings(
SETTING_INDEX_BALANCE_FACTOR,
SETTING_PRIMARY_BALANCE_FACTOR,
SETTING_SHARD_BALANCE_FACTOR,
SETTING_THRESHOLD
);
}
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
@ -105,14 +81,14 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
BalancedShardsAllocator.this.weightFunction = new WeightFunction(indexBalance, shardBalance, primaryBalance);
}
}
private volatile WeightFunction weightFunction;
private volatile float threshold;
public BalancedShardsAllocator(Settings settings) {
this(settings, new NodeSettingsService(settings));
}
@Inject
public BalancedShardsAllocator(Settings settings, NodeSettingsService nodeSettingsService) {
super(settings);
@ -146,7 +122,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
/**
* This class is the primary weight function used to create balanced over nodes and shards in the cluster.
* This class is the primary weight function used to create balanced over nodes and shards in the cluster.
* Currently this function has 3 properties:
* <ul>
* <li><code>index balance</code> - balance property over shards per index</li>
@ -155,8 +131,8 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
* </ul>
* <p>
* Each of these properties are expressed as factor such that the properties factor defines the relative importance of the property for the
* weight function. For example if the weight function should calculate the weights only based on a global (shard) balance the index and primary balance
* can be set to <tt>0.0</tt> and will in turn have no effect on the distribution.
* weight function. For example if the weight function should calculate the weights only based on a global (shard) balance the index and primary balance
* can be set to <tt>0.0</tt> and will in turn have no effect on the distribution.
* </p>
* The weight per index is calculated based on the following formula:
* <ul>
@ -171,9 +147,8 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
* </li>
* </ul>
* <code>weight(node, index) = weight<sub>index</sub>(node, index) + weight<sub>node</sub>(node, index) + weight<sub>primary</sub>(node, index)</code>
*
*/
public static class WeightFunction {
public static class WeightFunction {
private final float indexBalance;
private final float shardBalance;
@ -211,7 +186,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
private final float threshold;
private final MetaData metaData;
private final Predicate<MutableShardRouting> assignedFilter = new Predicate<MutableShardRouting>() {
@Override
public boolean apply(MutableShardRouting input) {
@ -234,7 +209,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
* Returns an array view on the nodes in the balancer. Nodes should not be removed from this list.
*/
private ModelNode[] nodesArray() {
return nodes.values().toArray(new ModelNode[nodes.size()]);
return nodes.values().toArray(new ModelNode[nodes.size()]);
}
/**
@ -250,14 +225,14 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
public float avgShardsPerNode() {
return ((float) metaData.totalNumberOfShards()) / nodes.size();
}
/**
* Returns the global average of primaries per node
*/
public float avgPrimariesPerNode() {
return ((float) metaData.numberOfShards()) / nodes.size();
}
/**
* Returns the average of primaries per node for the given index
*/
@ -268,20 +243,20 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
/**
* Returns a new {@link NodeSorter} that sorts the nodes based on their
* current weight with respect to the index passed to the sorter. The
* returned sorter is not sorted. Use {@link NodeSorter#reset(String)}
* returned sorter is not sorted. Use {@link NodeSorter#reset(String)}
* to sort based on an index.
*/
private NodeSorter newNodeSorter() {
final NodeSorter sorter = new NodeSorter(nodesArray(), weight, this);
return sorter;
}
private boolean initialize(RoutingNodes routing) {
Collection<MutableShardRouting> shards = new ArrayList<MutableShardRouting>();
if (logger.isTraceEnabled()) {
logger.trace("Start distributing Shards");
}
for (IndexRoutingTable index : allocation.routingTable().indicesRouting().values()) {
indices.add(index.index());
for (IndexShardRoutingTable shard : index.getShards().values()) {
@ -302,7 +277,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
* the threshold is set to <tt>1.0</tt> to enforce gaining relocation
* only, or in other words relocations that move the weight delta closer
* to <tt>0.0</tt>
*
*
* @return <code>true</code> if the current configuration has been
* changed, otherwise <code>false</code>
*/
@ -314,7 +289,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
if (logger.isTraceEnabled()) {
logger.trace("Start balancing cluster");
}
boolean changed = initialize(allocation.routingNodes());
NodeSorter sorter = newNodeSorter();
if (nodes.size() > 1) { /* skip if we only have one node */
@ -328,7 +303,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
final ModelNode minNode = modelNodes[lowIdx];
final ModelNode maxNode = modelNodes[highIdx];
if (maxNode.numShards(index) > 0) {
float delta = weights[highIdx] - weights[lowIdx];
float delta = weights[highIdx] - weights[lowIdx];
if (delta <= threshold) {
if (logger.isTraceEnabled()) {
logger.trace("Stop balancing index [{}] min_node [{}] weight: [{}] max_node [{}] weight: [{}] delta: [{}]",
@ -355,7 +330,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
highIdx = weights.length - 1;
changed = true;
continue;
}
}
}
if (lowIdx < highIdx - 1) {
/* we can't move from any shard from the min node lets move on to the next node
@ -368,7 +343,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
lowIdx = 0;
highIdx--;
} else {
/* we are done here, we either can't relocate anymore or we are balanced */
/* we are done here, we either can't relocate anymore or we are balanced */
break;
}
}
@ -376,7 +351,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
}
return changed;
}
/**
* This builds a initial index ordering where the indices are returned
* in most unbalanced first. We need this in order to prevent over
@ -428,7 +403,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
return indices;
}
/**
* This function executes a move operation moving the given shard from
* the given node to the minimal eligible node with respect to the
@ -436,7 +411,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
* {@link ShardRoutingState#RELOCATING} and a shadow instance of this
* shard is created with an incremented version in the state
* {@link ShardRoutingState#INITIALIZING}.
*
*
* @return <code>true</code> iff the shard has successfully been moved.
*/
public boolean move(MutableShardRouting shard, RoutingNode node) {
@ -498,7 +473,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
assert shard.assignedToNode();
/* we skip relocating shards here since we expect an initializing shard with the same id coming in */
if (shard.state() == RELOCATING) {
continue;
continue;
}
ModelNode node = nodes.get(shard.currentNodeId());
assert node != null;
@ -510,8 +485,8 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
}
/**
* Allocates all given shards on the minimal eligable node for the shards index
* with respect to the weight function. All given shards must be unassigned.
* Allocates all given shards on the minimal eligable node for the shards index
* with respect to the weight function. All given shards must be unassigned.
*/
private boolean allocateUnassigned(List<MutableShardRouting> unassigned, List<MutableShardRouting> ignoredUnassigned) {
assert !nodes.isEmpty();
@ -519,7 +494,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
logger.trace("Start allocating unassigned shards");
}
if (unassigned.isEmpty()) {
return false;
return false;
}
boolean changed = false;
@ -530,62 +505,62 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
final RoutingNodes routingNodes = allocation.routingNodes();
final AllocationDeciders deciders = allocation.deciders();
final Set<MutableShardRouting> currentRound = new TreeSet<MutableShardRouting>(new Comparator<MutableShardRouting>() {
@Override
public int compare(MutableShardRouting o1,
MutableShardRouting o2) {
final int indexCmp;
if ((indexCmp = o1.index().compareTo(o2.index())) == 0) {
if (o1.getId() - o2.getId() == 0) {
return o1.primary() ? -1 : o2.primary() ? 1 : 0;
}
return o1.getId() - o2.getId();
}
return indexCmp;
}
});
@Override
public int compare(MutableShardRouting o1,
MutableShardRouting o2) {
final int indexCmp;
if ((indexCmp = o1.index().compareTo(o2.index())) == 0) {
if (o1.getId() - o2.getId() == 0) {
return o1.primary() ? -1 : o2.primary() ? 1 : 0;
}
return o1.getId() - o2.getId();
}
return indexCmp;
}
});
do {
Iterator<MutableShardRouting> iterator = unassigned.iterator();
while(iterator.hasNext()) {
/* we treat every index equally here once chunk a time such that we fill up
* nodes with all indices at the same time. Only on shard of a shard a time.
Iterator<MutableShardRouting> iterator = unassigned.iterator();
while (iterator.hasNext()) {
/* we treat every index equally here once chunk a time such that we fill up
* nodes with all indices at the same time. Only on shard of a shard a time.
* Although there might be a primary and a shard of a shard in the set but
* primaries will be started first.*/
if (currentRound.add(iterator.next())) {
iterator.remove();
}
if (currentRound.add(iterator.next())) {
iterator.remove();
}
}
boolean iterationChanged = false;
for (MutableShardRouting shard : currentRound) {
assert !shard.assignedToNode();
/* find an node with minimal weight we can allocate on*/
float minWeight = Float.POSITIVE_INFINITY;
ModelNode minNode = null;
Decision decision = null;
for (ModelNode node : nodes.values()) {
boolean iterationChanged = false;
for (MutableShardRouting shard : currentRound) {
assert !shard.assignedToNode();
/* find an node with minimal weight we can allocate on*/
float minWeight = Float.POSITIVE_INFINITY;
ModelNode minNode = null;
Decision decision = null;
for (ModelNode node : nodes.values()) {
/*
* The shard we add is removed below to simulate the
* addition for weight calculation we use Decision.ALWAYS to
* not violate the not null condition.
*/
if (!node.containsShard(shard)) {
node.addShard(shard, Decision.ALWAYS);
float currentWeight = weight.weight(this, node, shard.index());
if (!node.containsShard(shard)) {
node.addShard(shard, Decision.ALWAYS);
float currentWeight = weight.weight(this, node, shard.index());
/*
* Remove the shard from the node again this is only a
* simulation
*/
Decision removed = node.removeShard(shard);
assert removed != null;
Decision removed = node.removeShard(shard);
assert removed != null;
/*
* Unless the operation is not providing any gains we
* don't check deciders
*/
if (currentWeight <= minWeight) {
Decision currentDecision = deciders.canAllocate(shard, routingNodes.node(node.getNodeId()), allocation);
NOUPDATE:
if (currentDecision.type() == Type.YES || currentDecision.type() == Type.THROTTLE) {
if (currentWeight == minWeight) {
if (currentWeight <= minWeight) {
Decision currentDecision = deciders.canAllocate(shard, routingNodes.node(node.getNodeId()), allocation);
NOUPDATE:
if (currentDecision.type() == Type.YES || currentDecision.type() == Type.THROTTLE) {
if (currentWeight == minWeight) {
/* we have an equal weight tie breaking:
* 1. if one decision is YES prefer it
* 2. prefer the node that holds the primary for this index with the next id in the ring ie.
@ -597,59 +572,61 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
* than the id of the shard we need to assign. This works find when new indices are created since
* primaries are added first and we only add one shard set a time in this algorithm.
*/
if (currentDecision.type() == decision.type()) {
final int repId = shard.id();
final int nodeHigh = node.highestPrimary(shard.index());
final int minNodeHigh = minNode.highestPrimary(shard.index());
if ((((nodeHigh > repId && minNodeHigh > repId) || (nodeHigh < repId && minNodeHigh < repId)) && (nodeHigh < minNodeHigh))
|| (nodeHigh > minNodeHigh && nodeHigh > repId && minNodeHigh < repId)) {
minNode = node;
minWeight = currentWeight;
decision = currentDecision;
} else { break NOUPDATE; }
} else if (currentDecision.type() != Type.YES) {
break NOUPDATE;
}
}
minNode = node;
minWeight = currentWeight;
decision = currentDecision;
}
}
}
}
assert decision != null && minNode != null || decision == null && minNode == null;
if (minNode != null) {
iterationChanged = true;
minNode.addShard(shard, decision);
if (decision.type() == Type.YES) {
if (logger.isTraceEnabled()) {
logger.trace("Assigned shard [{}] to [{}]", shard, minNode.getNodeId());
}
routingNodes.node(minNode.getNodeId()).add(shard);
changed = true;
continue; // don't add to ignoreUnassigned
}
} else if (logger.isTraceEnabled()) {
logger.trace("No Node found to assign shard [{}]", shard);
}
ignoredUnassigned.add(shard);
}
if (!iterationChanged && !unassigned.isEmpty()) {
ignoredUnassigned.addAll(unassigned);
unassigned.clear();
return changed;
}
currentRound.clear();
} while(!unassigned.isEmpty());
if (currentDecision.type() == decision.type()) {
final int repId = shard.id();
final int nodeHigh = node.highestPrimary(shard.index());
final int minNodeHigh = minNode.highestPrimary(shard.index());
if ((((nodeHigh > repId && minNodeHigh > repId) || (nodeHigh < repId && minNodeHigh < repId)) && (nodeHigh < minNodeHigh))
|| (nodeHigh > minNodeHigh && nodeHigh > repId && minNodeHigh < repId)) {
minNode = node;
minWeight = currentWeight;
decision = currentDecision;
} else {
break NOUPDATE;
}
} else if (currentDecision.type() != Type.YES) {
break NOUPDATE;
}
}
minNode = node;
minWeight = currentWeight;
decision = currentDecision;
}
}
}
}
assert decision != null && minNode != null || decision == null && minNode == null;
if (minNode != null) {
iterationChanged = true;
minNode.addShard(shard, decision);
if (decision.type() == Type.YES) {
if (logger.isTraceEnabled()) {
logger.trace("Assigned shard [{}] to [{}]", shard, minNode.getNodeId());
}
routingNodes.node(minNode.getNodeId()).add(shard);
changed = true;
continue; // don't add to ignoreUnassigned
}
} else if (logger.isTraceEnabled()) {
logger.trace("No Node found to assign shard [{}]", shard);
}
ignoredUnassigned.add(shard);
}
if (!iterationChanged && !unassigned.isEmpty()) {
ignoredUnassigned.addAll(unassigned);
unassigned.clear();
return changed;
}
currentRound.clear();
} while (!unassigned.isEmpty());
// clear everything we have either added it or moved to ingoreUnassigned
return changed;
}
/**
* Tries to find a relocation from the max node to the minimal node for an arbitrary shard of the given index on the
* balance model. Iff this method returns a <code>true</code> the relocation has already been executed on the
* simulation model as well as on the cluster.
* Tries to find a relocation from the max node to the minimal node for an arbitrary shard of the given index on the
* balance model. Iff this method returns a <code>true</code> the relocation has already been executed on the
* simulation model as well as on the cluster.
*/
private boolean tryRelocateShard(ModelNode minNode, ModelNode maxNode, String idx, float minCost) {
final ModelIndex index = maxNode.getIndex(idx);
@ -669,7 +646,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
// skip initializing, unassigned and relocating shards we can't relocate them anyway
Decision allocationDecision = deciders.canAllocate(shard, node, allocation);
Decision rebalanceDecission = deciders.canRebalance(shard, allocation);
if (((allocationDecision.type() == Type.YES) || (allocationDecision.type() == Type.THROTTLE))
&& ((rebalanceDecission.type() == Type.YES) || (rebalanceDecission.type() == Type.THROTTLE))) {
Decision srcDecision;
@ -704,15 +681,15 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
lowRoutingNode.add(new MutableShardRouting(candidate.index(), candidate.id(), lowRoutingNode.nodeId(), candidate
.currentNodeId(), candidate.primary(), INITIALIZING, candidate.version() + 1));
candidate.relocate(lowRoutingNode.nodeId());
} else {
assert candidate.unassigned();
allocation.routingNodes().node(minNode.getNodeId()).add(candidate);
}
return true;
}
}
}
}
}
if (logger.isTraceEnabled()) {
logger.trace("Couldn't find shard to relocate from node [{}] to node [{}]", maxNode.getNodeId(),
@ -781,7 +758,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
}
return result;
}
public int highestPrimary(String index) {
ModelIndex idx = indices.get(index);
if (idx != null) {
@ -812,7 +789,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
}
return removed;
}
public String toString() {
StringBuilder sb = new StringBuilder();
sb.append("Node(").append(id).append(")");
@ -903,7 +880,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
static final class NodeSorter extends SorterTemplate {
final ModelNode[] modelNodes;
/* the nodes weights with respect to the current weight function / index */
/* the nodes weights with respect to the current weight function / index */
final float[] weights;
private final WeightFunction function;
private String index;
@ -911,7 +888,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
private float pivotWeight;
public NodeSorter(ModelNode[] modelNodes, WeightFunction function, Balancer balancer) {
this.function = function;
this.balancer = balancer;
this.modelNodes = modelNodes;
@ -929,7 +906,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
}
quickSort(0, modelNodes.length - 1);
}
public float weight(ModelNode node) {
return function.weight(balancer, node, index);
}
@ -958,9 +935,9 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
protected int comparePivot(int j) {
return Float.compare(pivotWeight, weights[j]);
}
public float delta() {
return weights[weights.length-1] - weights[0];
return weights[weights.length - 1] - weights[0];
}
}
}

39
src/main/java/org/elasticsearch/cluster/routing/allocation/decider/AwarenessAllocationDecider.java
View File

@ -22,7 +22,6 @@ package org.elasticsearch.cluster.routing.allocation.decider;
import com.google.common.collect.Maps;
import gnu.trove.map.hash.TObjectIntHashMap;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.MutableShardRouting;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
@ -42,63 +41,59 @@ import java.util.Map;
* attributes like node or physical rack locations. Awareness attributes accept
* arbitrary configuration keys like a rack data-center identifier. For example
* the setting:
*
* <p/>
* <pre>
* cluster.routing.allocation.awareness.attributes: rack_id
* </pre>
*
* <p/>
* will cause allocations to be distributed over different racks such that
* ideally at least one replicas of the all shard is available on the same rack.
* To enable allocation awareness in this example nodes should contain a value
* for the <tt>rack_id</tt> key like:
*
* <p/>
* <pre>
* node.rack_id:1
* </pre>
*
* <p/>
* Awareness can also be used to prevent over-allocation in the case of node or
* even "zone" failure. For example in cloud-computing infrastructures like
* Amazone AWS a cluster might span over multiple "zones". Awareness can be used
* to distribute replicas to individual zones by setting:
*
* <p/>
* <pre>
* cluster.routing.allocation.awareness.attributes: zone
* </pre>
*
* <p/>
* and forcing allocation to be aware of the following zone the data resides in:
*
* <p/>
* <pre>
* cluster.routing.allocation.awareness.force.zone.values: zone1,zone2
* </pre>
*
* <p/>
* In contrast to regular awareness this setting will prevent over-allocation on
* <tt>zone1</tt> even if <tt>zone2</tt> fails partially or becomes entirely
* unavailable. Nodes that belong to a certain zone / group should be started
* with the zone id configured on the node-level settings like:
*
* <p/>
* <pre>
* node.zone: zone1
* </pre>
*/
public class AwarenessAllocationDecider extends AllocationDecider {
static {
MetaData.addDynamicSettings(
"cluster.routing.allocation.awareness.attributes",
"cluster.routing.allocation.awareness.force.*"
);
}
public static final String CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTES = "cluster.routing.allocation.awareness.attributes";
public static final String CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP = "cluster.routing.allocation.awareness.force.";
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
String[] awarenessAttributes = settings.getAsArray("cluster.routing.allocation.awareness.attributes", null);
String[] awarenessAttributes = settings.getAsArray(CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTES, null);
if (awarenessAttributes != null) {
logger.info("updating [cluster.routing.allocation.awareness.attributes] from [{}] to [{}]", AwarenessAllocationDecider.this.awarenessAttributes, awarenessAttributes);
AwarenessAllocationDecider.this.awarenessAttributes = awarenessAttributes;
}
Map<String, String[]> forcedAwarenessAttributes = new HashMap<String, String[]>(AwarenessAllocationDecider.this.forcedAwarenessAttributes);
Map<String, Settings> forceGroups = settings.getGroups("cluster.routing.allocation.awareness.force.");
Map<String, Settings> forceGroups = settings.getGroups(CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP);
if (!forceGroups.isEmpty()) {
for (Map.Entry<String, Settings> entry : forceGroups.entrySet()) {
String[] aValues = entry.getValue().getAsArray("values");
@ -124,6 +119,7 @@ public class AwarenessAllocationDecider extends AllocationDecider {
/**
* Creates a new {@link AwarenessAllocationDecider} instance from given settings
*
* @param settings {@link Settings} to use
*/
public AwarenessAllocationDecider(Settings settings) {
@ -133,10 +129,10 @@ public class AwarenessAllocationDecider extends AllocationDecider {
@Inject
public AwarenessAllocationDecider(Settings settings, NodeSettingsService nodeSettingsService) {
super(settings);
this.awarenessAttributes = settings.getAsArray("cluster.routing.allocation.awareness.attributes");
this.awarenessAttributes = settings.getAsArray(CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTES);
forcedAwarenessAttributes = Maps.newHashMap();
Map<String, Settings> forceGroups = settings.getGroups("cluster.routing.allocation.awareness.force.");
Map<String, Settings> forceGroups = settings.getGroups(CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP);
for (Map.Entry<String, Settings> entry : forceGroups.entrySet()) {
String[] aValues = entry.getValue().getAsArray("values");
if (aValues.length > 0) {
@ -148,7 +144,8 @@ public class AwarenessAllocationDecider extends AllocationDecider {
}
/**
* Get the attributes defined by this instance
* Get the attributes defined by this instance
*
* @return attributes defined by this instance
*/
public String[] awarenessAttributes() {

14
src/main/java/org/elasticsearch/cluster/routing/allocation/decider/ConcurrentRebalanceAllocationDecider.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.cluster.routing.allocation.decider;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.MutableShardRouting;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
@ -37,25 +36,20 @@ import java.util.List;
* re-balance (relocation) operations and restricts node allocations if the
* configured threashold is reached. The default number of concurrent rebalance
* operations is set to <tt>2</tt>
* <p>
* <p/>
* Re-balance operations can be controlled in real-time via the cluster update API using
* <tt>cluster.routing.allocation.cluster_concurrent_rebalance</tt>. Iff this
* setting is set to <tt>-1</tt> the number of concurrent re-balance operations
* are unlimited.
*
*/
public class ConcurrentRebalanceAllocationDecider extends AllocationDecider {
static {
MetaData.addDynamicSettings(
"cluster.routing.allocation.cluster_concurrent_rebalance"
);
}
public static final String CLUSTER_ROUTING_ALLOCATION_CLUSTER_CONCURRENT_REBALANCE = "cluster.routing.allocation.cluster_concurrent_rebalance";
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
int clusterConcurrentRebalance = settings.getAsInt("cluster.routing.allocation.cluster_concurrent_rebalance", ConcurrentRebalanceAllocationDecider.this.clusterConcurrentRebalance);
int clusterConcurrentRebalance = settings.getAsInt(CLUSTER_ROUTING_ALLOCATION_CLUSTER_CONCURRENT_REBALANCE, ConcurrentRebalanceAllocationDecider.this.clusterConcurrentRebalance);
if (clusterConcurrentRebalance != ConcurrentRebalanceAllocationDecider.this.clusterConcurrentRebalance) {
logger.info("updating [cluster.routing.allocation.cluster_concurrent_rebalance] from [{}], to [{}]", ConcurrentRebalanceAllocationDecider.this.clusterConcurrentRebalance, clusterConcurrentRebalance);
ConcurrentRebalanceAllocationDecider.this.clusterConcurrentRebalance = clusterConcurrentRebalance;
@ -68,7 +62,7 @@ public class ConcurrentRebalanceAllocationDecider extends AllocationDecider {
@Inject
public ConcurrentRebalanceAllocationDecider(Settings settings, NodeSettingsService nodeSettingsService) {
super(settings);
this.clusterConcurrentRebalance = settings.getAsInt("cluster.routing.allocation.cluster_concurrent_rebalance", 2);
this.clusterConcurrentRebalance = settings.getAsInt(CLUSTER_ROUTING_ALLOCATION_CLUSTER_CONCURRENT_REBALANCE, 2);
logger.debug("using [cluster_concurrent_rebalance] with [{}]", clusterConcurrentRebalance);
nodeSettingsService.addListener(new ApplySettings());
}

29
src/main/java/org/elasticsearch/cluster/routing/allocation/decider/DisableAllocationDecider.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.cluster.routing.allocation.decider;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.cluster.routing.allocation.RoutingAllocation;
@ -36,15 +35,15 @@ import org.elasticsearch.node.settings.NodeSettingsService;
* <code>true</code> no new shard-allocation are allowed. Note: this setting is
* only applied if the allocated shard is a primary and it has not been
* allocated before the this setting was applied.</li>
*
* <p/>
* <li><tt>cluster.routing.allocation.disable_allocation</tt> - if set to
* <code>true</code> cluster wide allocations are disabled</li>
*
* <p/>
* <li><tt>cluster.routing.allocation.disable_replica_allocation</tt> - if set
* to <code>true</code> cluster wide replica allocations are disabled while
* primary shards can still be allocated</li>
* </ul>
*
* <p/>
* <p>
* Note: all of the above settings might be ignored if the allocation happens on
* a shard that explicitly ignores disabled allocations via
@ -54,30 +53,26 @@ import org.elasticsearch.node.settings.NodeSettingsService;
*/
public class DisableAllocationDecider extends AllocationDecider {
static {
MetaData.addDynamicSettings(
"cluster.routing.allocation.disable_new_allocation",
"cluster.routing.allocation.disable_allocation",
"cluster.routing.allocation.disable_replica_allocation"
);
}
public static final String CLUSTER_ROUTING_ALLOCATION_DISABLE_NEW_ALLOCATION = "cluster.routing.allocation.disable_new_allocation";
public static final String CLUSTER_ROUTING_ALLOCATION_DISABLE_ALLOCATION = "cluster.routing.allocation.disable_allocation";
public static final String CLUSTER_ROUTING_ALLOCATION_DISABLE_REPLICA_ALLOCATION = "cluster.routing.allocation.disable_replica_allocation";
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
boolean disableNewAllocation = settings.getAsBoolean("cluster.routing.allocation.disable_new_allocation", DisableAllocationDecider.this.disableNewAllocation);
boolean disableNewAllocation = settings.getAsBoolean(CLUSTER_ROUTING_ALLOCATION_DISABLE_NEW_ALLOCATION, DisableAllocationDecider.this.disableNewAllocation);
if (disableNewAllocation != DisableAllocationDecider.this.disableNewAllocation) {
logger.info("updating [cluster.routing.allocation.disable_new_allocation] from [{}] to [{}]", DisableAllocationDecider.this.disableNewAllocation, disableNewAllocation);
DisableAllocationDecider.this.disableNewAllocation = disableNewAllocation;
}
boolean disableAllocation = settings.getAsBoolean("cluster.routing.allocation.disable_allocation", DisableAllocationDecider.this.disableAllocation);
boolean disableAllocation = settings.getAsBoolean(CLUSTER_ROUTING_ALLOCATION_DISABLE_ALLOCATION, DisableAllocationDecider.this.disableAllocation);
if (disableAllocation != DisableAllocationDecider.this.disableAllocation) {
logger.info("updating [cluster.routing.allocation.disable_allocation] from [{}] to [{}]", DisableAllocationDecider.this.disableAllocation, disableAllocation);
DisableAllocationDecider.this.disableAllocation = disableAllocation;
}
boolean disableReplicaAllocation = settings.getAsBoolean("cluster.routing.allocation.disable_replica_allocation", DisableAllocationDecider.this.disableReplicaAllocation);
boolean disableReplicaAllocation = settings.getAsBoolean(CLUSTER_ROUTING_ALLOCATION_DISABLE_REPLICA_ALLOCATION, DisableAllocationDecider.this.disableReplicaAllocation);
if (disableReplicaAllocation != DisableAllocationDecider.this.disableReplicaAllocation) {
logger.info("updating [cluster.routing.allocation.disable_replica_allocation] from [{}] to [{}]", DisableAllocationDecider.this.disableReplicaAllocation, disableReplicaAllocation);
DisableAllocationDecider.this.disableReplicaAllocation = disableReplicaAllocation;
@ -92,9 +87,9 @@ public class DisableAllocationDecider extends AllocationDecider {
@Inject
public DisableAllocationDecider(Settings settings, NodeSettingsService nodeSettingsService) {
super(settings);
this.disableNewAllocation = settings.getAsBoolean("cluster.routing.allocation.disable_new_allocation", false);
this.disableAllocation = settings.getAsBoolean("cluster.routing.allocation.disable_allocation", false);
this.disableReplicaAllocation = settings.getAsBoolean("cluster.routing.allocation.disable_replica_allocation", false);
this.disableNewAllocation = settings.getAsBoolean(CLUSTER_ROUTING_ALLOCATION_DISABLE_NEW_ALLOCATION, false);
this.disableAllocation = settings.getAsBoolean(CLUSTER_ROUTING_ALLOCATION_DISABLE_ALLOCATION, false);
this.disableReplicaAllocation = settings.getAsBoolean(CLUSTER_ROUTING_ALLOCATION_DISABLE_REPLICA_ALLOCATION, false);
nodeSettingsService.addListener(new ApplySettings());
}

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

@ -21,7 +21,6 @@ package org.elasticsearch.cluster.routing.allocation.decider;
import com.google.common.collect.ImmutableMap;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.node.DiscoveryNodeFilters;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
@ -30,7 +29,8 @@ import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.node.settings.NodeSettingsService;
import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.*;
import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.AND;
import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.OR;
/**
* This {@link AllocationDecider} control shard allocation by include and
@ -48,32 +48,25 @@ import static org.elasticsearch.cluster.node.DiscoveryNodeFilters.OpType.*;
* settings it wont be allocated on a node if the cluster specific settings
* would disallow the allocation. Filters are applied in the following order:
* <ol>
* <li><tt>required</tt> - filters required allocations.
* <li><tt>required</tt> - filters required allocations.
* If any <tt>required</tt> filters are set the allocation is denied if the index is <b>not</b> in the set of <tt>required</tt> to allocate on the filtered node</li>
*
* <li><tt>include</tt> - filters "allowed" allocations.
* <p/>
* <li><tt>include</tt> - filters "allowed" allocations.
* If any <tt>include</tt> filters are set the allocation is denied if the index is <b>not</b> in the set of <tt>include</tt> filters for the filtered node</li>
*
* <li><tt>exclude</tt> - filters "prohibited" allocations.
* <p/>
* <li><tt>exclude</tt> - filters "prohibited" allocations.
* If any <tt>exclude</tt> filters are set the allocation is denied if the index is in the set of <tt>exclude</tt> filters for the filtered node</li>
* </ol>
*
*
*/
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.*"
);
}
public static final String INDEX_ROUTING_REQUIRE_GROUP = "index.routing.allocation.require.";
public static final String INDEX_ROUTING_INCLUDE_GROUP = "index.routing.allocation.include.";
public static final String INDEX_ROUTING_EXCLUDE_GROUP = "index.routing.allocation.exclude.";
public static final String CLUSTER_ROUTING_REQUIRE_GROUP = "cluster.routing.allocation.require.";
public static final String CLUSTER_ROUTING_INCLUDE_GROUP = "cluster.routing.allocation.include.";
public static final String CLUSTER_ROUTING_EXCLUDE_GROUP = "cluster.routing.allocation.exclude.";
private volatile DiscoveryNodeFilters clusterRequireFilters;
private volatile DiscoveryNodeFilters clusterIncludeFilters;
@ -82,19 +75,19 @@ public class FilterAllocationDecider extends AllocationDecider {
@Inject
public FilterAllocationDecider(Settings settings, NodeSettingsService nodeSettingsService) {
super(settings);
ImmutableMap<String, String> requireMap = settings.getByPrefix("cluster.routing.allocation.require.").getAsMap();
ImmutableMap<String, String> requireMap = settings.getByPrefix(CLUSTER_ROUTING_REQUIRE_GROUP).getAsMap();
if (requireMap.isEmpty()) {
clusterRequireFilters = null;
} else {
clusterRequireFilters = DiscoveryNodeFilters.buildFromKeyValue(AND, requireMap);
}
ImmutableMap<String, String> includeMap = settings.getByPrefix("cluster.routing.allocation.include.").getAsMap();
ImmutableMap<String, String> includeMap = settings.getByPrefix(CLUSTER_ROUTING_INCLUDE_GROUP).getAsMap();
if (includeMap.isEmpty()) {
clusterIncludeFilters = null;
} else {
clusterIncludeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, includeMap);
}
ImmutableMap<String, String> excludeMap = settings.getByPrefix("cluster.routing.allocation.exclude.").getAsMap();
ImmutableMap<String, String> excludeMap = settings.getByPrefix(CLUSTER_ROUTING_EXCLUDE_GROUP).getAsMap();
if (excludeMap.isEmpty()) {
clusterExcludeFilters = null;
} else {
@ -153,15 +146,15 @@ public class FilterAllocationDecider extends AllocationDecider {
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
ImmutableMap<String, String> requireMap = settings.getByPrefix("cluster.routing.allocation.require.").getAsMap();
ImmutableMap<String, String> requireMap = settings.getByPrefix(CLUSTER_ROUTING_REQUIRE_GROUP).getAsMap();
if (!requireMap.isEmpty()) {
clusterRequireFilters = DiscoveryNodeFilters.buildFromKeyValue(AND, requireMap);
}
ImmutableMap<String, String> includeMap = settings.getByPrefix("cluster.routing.allocation.include.").getAsMap();
ImmutableMap<String, String> includeMap = settings.getByPrefix(CLUSTER_ROUTING_INCLUDE_GROUP).getAsMap();
if (!includeMap.isEmpty()) {
clusterIncludeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, includeMap);
}
ImmutableMap<String, String> excludeMap = settings.getByPrefix("cluster.routing.allocation.exclude.").getAsMap();
ImmutableMap<String, String> excludeMap = settings.getByPrefix(CLUSTER_ROUTING_EXCLUDE_GROUP).getAsMap();
if (!excludeMap.isEmpty()) {
clusterExcludeFilters = DiscoveryNodeFilters.buildFromKeyValue(OR, excludeMap);
}

6
src/main/java/org/elasticsearch/cluster/routing/allocation/decider/ShardsLimitAllocationDecider.java
View File

@ -57,12 +57,6 @@ public class ShardsLimitAllocationDecider extends AllocationDecider {
*/
public static final String INDEX_TOTAL_SHARDS_PER_NODE = "index.routing.allocation.total_shards_per_node";
static {
IndexMetaData.addDynamicSettings(
INDEX_TOTAL_SHARDS_PER_NODE
);
}
@Inject
public ShardsLimitAllocationDecider(Settings settings) {
super(settings);

23
src/main/java/org/elasticsearch/cluster/routing/allocation/decider/ThrottlingAllocationDecider.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.cluster.routing.allocation.decider;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.MutableShardRouting;
import org.elasticsearch.cluster.routing.RoutingNode;
import org.elasticsearch.cluster.routing.ShardRouting;
@ -35,17 +34,17 @@ import java.util.List;
* {@link ThrottlingAllocationDecider} controls the recovery process per node in
* the cluster. It exposes two settings via the cluster update API that allow
* changes in real-time:
*
* <p/>
* <ul>
* <li><tt>cluster.routing.allocation.node_initial_primaries_recoveries</tt> -
* restricts the number of initial primary shard recovery operations on a single
* node. The default is <tt>4</tt></li>
*
* <p/>
* <li><tt>cluster.routing.allocation.node_concurrent_recoveries</tt> -
* restricts the number of concurrent recovery operations on a single node. The
* default is <tt>2</tt></li>
* </ul>
*
* <p/>
* If one of the above thresholds is exceeded per node this allocation decider
* will return {@link Decision#THROTTLE} as a hit to upstream logic to throttle
* the allocation process to prevent overloading nodes due to too many concurrent recovery
@ -53,12 +52,8 @@ import java.util.List;
*/
public class ThrottlingAllocationDecider extends AllocationDecider {
static {
MetaData.addDynamicSettings(
"cluster.routing.allocation.node_initial_primaries_recoveries",
"cluster.routing.allocation.node_concurrent_recoveries"
);
}
public static final String CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES = "cluster.routing.allocation.node_initial_primaries_recoveries";
public static final String CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_RECOVERIES = "cluster.routing.allocation.node_concurrent_recoveries";
private volatile int primariesInitialRecoveries;
private volatile int concurrentRecoveries;
@ -67,8 +62,8 @@ public class ThrottlingAllocationDecider extends AllocationDecider {
public ThrottlingAllocationDecider(Settings settings, NodeSettingsService nodeSettingsService) {
super(settings);
this.primariesInitialRecoveries = settings.getAsInt("cluster.routing.allocation.node_initial_primaries_recoveries", settings.getAsInt("cluster.routing.allocation.node_initial_primaries_recoveries", 4));
this.concurrentRecoveries = settings.getAsInt("cluster.routing.allocation.concurrent_recoveries", settings.getAsInt("cluster.routing.allocation.node_concurrent_recoveries", 2));
this.primariesInitialRecoveries = settings.getAsInt(CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES, 4);
this.concurrentRecoveries = settings.getAsInt("cluster.routing.allocation.concurrent_recoveries", settings.getAsInt(CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_RECOVERIES, 2));
logger.debug("using node_concurrent_recoveries [{}], node_initial_primaries_recoveries [{}]", concurrentRecoveries, primariesInitialRecoveries);
nodeSettingsService.addListener(new ApplySettings());
@ -124,13 +119,13 @@ public class ThrottlingAllocationDecider extends AllocationDecider {
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
int primariesInitialRecoveries = settings.getAsInt("cluster.routing.allocation.node_initial_primaries_recoveries", ThrottlingAllocationDecider.this.primariesInitialRecoveries);
int primariesInitialRecoveries = settings.getAsInt(CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES, ThrottlingAllocationDecider.this.primariesInitialRecoveries);
if (primariesInitialRecoveries != ThrottlingAllocationDecider.this.primariesInitialRecoveries) {
logger.info("updating [cluster.routing.allocation.node_initial_primaries_recoveries] from [{}] to [{}]", ThrottlingAllocationDecider.this.primariesInitialRecoveries, primariesInitialRecoveries);
ThrottlingAllocationDecider.this.primariesInitialRecoveries = primariesInitialRecoveries;
}
int concurrentRecoveries = settings.getAsInt("cluster.routing.allocation.node_concurrent_recoveries", ThrottlingAllocationDecider.this.concurrentRecoveries);
int concurrentRecoveries = settings.getAsInt(CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_RECOVERIES, ThrottlingAllocationDecider.this.concurrentRecoveries);
if (concurrentRecoveries != ThrottlingAllocationDecider.this.concurrentRecoveries) {
logger.info("updating [cluster.routing.allocation.node_concurrent_recoveries] from [{}] to [{}]", ThrottlingAllocationDecider.this.concurrentRecoveries, concurrentRecoveries);
ThrottlingAllocationDecider.this.concurrentRecoveries = concurrentRecoveries;

38
src/main/java/org/elasticsearch/cluster/settings/ClusterDynamicSettings.java Normal file
View File

@ -0,0 +1,38 @@
/*
* Licensed to ElasticSearch and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. ElasticSearch licenses this
* file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.cluster.settings;
import org.elasticsearch.common.inject.BindingAnnotation;
import java.lang.annotation.Documented;
import java.lang.annotation.Retention;
import java.lang.annotation.Target;
import static java.lang.annotation.ElementType.FIELD;
import static java.lang.annotation.ElementType.PARAMETER;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
@BindingAnnotation
@Target({FIELD, PARAMETER})
@Retention(RUNTIME)
@Documented
public @interface ClusterDynamicSettings {
}

82
src/main/java/org/elasticsearch/cluster/settings/ClusterDynamicSettingsModule.java Normal file
View File

@ -0,0 +1,82 @@
/*
* Licensed to ElasticSearch and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. ElasticSearch licenses this
* file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.cluster.settings;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.allocation.allocator.BalancedShardsAllocator;
import org.elasticsearch.cluster.routing.allocation.decider.*;
import org.elasticsearch.common.inject.AbstractModule;
import org.elasticsearch.discovery.zen.elect.ElectMasterService;
import org.elasticsearch.indices.cache.filter.IndicesFilterCache;
import org.elasticsearch.indices.recovery.RecoverySettings;
import org.elasticsearch.indices.store.IndicesStore;
import org.elasticsearch.indices.ttl.IndicesTTLService;
import org.elasticsearch.threadpool.ThreadPool;
/**
*/
public class ClusterDynamicSettingsModule extends AbstractModule {
private final DynamicSettings clusterDynamicSettings;
public ClusterDynamicSettingsModule() {
clusterDynamicSettings = new DynamicSettings();
clusterDynamicSettings.addDynamicSettings(
AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTES,
AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP + "*",
BalancedShardsAllocator.SETTING_INDEX_BALANCE_FACTOR,
BalancedShardsAllocator.SETTING_PRIMARY_BALANCE_FACTOR,
BalancedShardsAllocator.SETTING_SHARD_BALANCE_FACTOR,
BalancedShardsAllocator.SETTING_THRESHOLD,
ConcurrentRebalanceAllocationDecider.CLUSTER_ROUTING_ALLOCATION_CLUSTER_CONCURRENT_REBALANCE,
DisableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_DISABLE_NEW_ALLOCATION,
DisableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_DISABLE_ALLOCATION,
DisableAllocationDecider.CLUSTER_ROUTING_ALLOCATION_DISABLE_REPLICA_ALLOCATION,
ElectMasterService.DISCOVERY_ZEN_MINIMUM_MASTER_NODES,
FilterAllocationDecider.CLUSTER_ROUTING_INCLUDE_GROUP + "*",
FilterAllocationDecider.CLUSTER_ROUTING_EXCLUDE_GROUP + "*",
FilterAllocationDecider.CLUSTER_ROUTING_REQUIRE_GROUP + "*",
IndicesFilterCache.INDICES_CACHE_FILTER_SIZE,
IndicesFilterCache.INDICES_CACHE_FILTER_EXPIRE,
IndicesStore.INDICES_STORE_THROTTLE_TYPE,
IndicesStore.INDICES_STORE_THROTTLE_MAX_BYTES_PER_SEC,
IndicesTTLService.INDICES_TTL_INTERVAL,
MetaData.SETTING_READ_ONLY,
RecoverySettings.INDICES_RECOVERY_FILE_CHUNK_SIZE,
RecoverySettings.INDICES_RECOVERY_TRANSLOG_OPS,
RecoverySettings.INDICES_RECOVERY_TRANSLOG_SIZE,
RecoverySettings.INDICES_RECOVERY_COMPRESS,
RecoverySettings.INDICES_RECOVERY_CONCURRENT_STREAMS,
RecoverySettings.INDICES_RECOVERY_MAX_SIZE_PER_SEC,
ThreadPool.THREADPOOL_GROUP + "*",
ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_INITIAL_PRIMARIES_RECOVERIES,
ThrottlingAllocationDecider.CLUSTER_ROUTING_ALLOCATION_NODE_CONCURRENT_RECOVERIES
);
}
public void addDynamicSetting(String... settings) {
clusterDynamicSettings.addDynamicSettings(settings);
}
@Override
protected void configure() {
bind(DynamicSettings.class).annotatedWith(ClusterDynamicSettings.class).toInstance(clusterDynamicSettings);
}
}

49
src/main/java/org/elasticsearch/cluster/settings/DynamicSettings.java Normal file
View File

@ -0,0 +1,49 @@
/*
* Licensed to ElasticSearch and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. ElasticSearch licenses this
* file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.cluster.settings;
import com.google.common.collect.ImmutableSet;
import org.elasticsearch.common.regex.Regex;
import java.util.Arrays;
import java.util.HashSet;
/**
*/
public class DynamicSettings {
private ImmutableSet<String> dynamicSettings = ImmutableSet.of();
public boolean hasDynamicSetting(String key) {
for (String dynamicSetting : dynamicSettings) {
if (Regex.simpleMatch(dynamicSetting, key)) {
return true;
}
}
return false;
}
public synchronized void addDynamicSettings(String... settings) {
HashSet<String> updatedSettings = new HashSet<String>(dynamicSettings);
updatedSettings.addAll(Arrays.asList(settings));
dynamicSettings = ImmutableSet.copyOf(updatedSettings);
}
}

7
src/main/java/org/elasticsearch/discovery/zen/elect/ElectMasterService.java
View File

@ -20,7 +20,6 @@
package org.elasticsearch.discovery.zen.elect;
import com.google.common.collect.Lists;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.settings.Settings;
@ -35,9 +34,7 @@ import java.util.List;
*/
public class ElectMasterService extends AbstractComponent {
static {
MetaData.addDynamicSettings("discovery.zen.minimum_master_nodes");
}
public static final String DISCOVERY_ZEN_MINIMUM_MASTER_NODES = "discovery.zen.minimum_master_nodes";
private final NodeComparator nodeComparator = new NodeComparator();
@ -45,7 +42,7 @@ public class ElectMasterService extends AbstractComponent {
public ElectMasterService(Settings settings) {
super(settings);
this.minimumMasterNodes = settings.getAsInt("discovery.zen.minimum_master_nodes", -1);
this.minimumMasterNodes = settings.getAsInt(DISCOVERY_ZEN_MINIMUM_MASTER_NODES, -1);
logger.debug("using minimum_master_nodes [{}]", minimumMasterNodes);
}

6
src/main/java/org/elasticsearch/gateway/local/LocalGatewayAllocator.java
View File

@ -58,9 +58,7 @@ import java.util.concurrent.ConcurrentMap;
*/
public class LocalGatewayAllocator extends AbstractComponent implements GatewayAllocator {
static {
IndexMetaData.addDynamicSettings("index.recovery.initial_shards");
}
public static final String INDEX_RECOVERY_INITIAL_SHARDS = "index.recovery.initial_shards";
private final TransportNodesListGatewayStartedShards listGatewayStartedShards;
@ -156,7 +154,7 @@ public class LocalGatewayAllocator extends AbstractComponent implements GatewayA
int requiredAllocation = 1;
try {
IndexMetaData indexMetaData = routingNodes.metaData().index(shard.index());
String initialShards = indexMetaData.settings().get("index.recovery.initial_shards", settings.get("index.recovery.initial_shards", this.initialShards));
String initialShards = indexMetaData.settings().get(INDEX_RECOVERY_INITIAL_SHARDS, settings.get(INDEX_RECOVERY_INITIAL_SHARDS, this.initialShards));
if ("quorum".equals(initialShards)) {
if (indexMetaData.numberOfReplicas() > 1) {
requiredAllocation = ((1 + indexMetaData.numberOfReplicas()) / 2) + 1;

36
src/main/java/org/elasticsearch/index/engine/robin/RobinEngine.java
View File

@ -28,7 +28,6 @@ import org.apache.lucene.search.SearcherManager;
import org.apache.lucene.store.AlreadyClosedException;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.ElasticSearchIllegalStateException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.operation.hash.djb.DjbHashFunction;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.Preconditions;
@ -159,11 +158,11 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
Preconditions.checkNotNull(deletionPolicy, "Snapshot deletion policy must be provided to the engine");
Preconditions.checkNotNull(translog, "Translog must be provided to the engine");
this.gcDeletesInMillis = indexSettings.getAsTime("index.gc_deletes", TimeValue.timeValueSeconds(60)).millis();
this.gcDeletesInMillis = indexSettings.getAsTime(INDEX_GC_DELETES, TimeValue.timeValueSeconds(60)).millis();
this.indexingBufferSize = componentSettings.getAsBytesSize("index_buffer_size", new ByteSizeValue(64, ByteSizeUnit.MB)); // not really important, as it is set by the IndexingMemory manager
this.termIndexInterval = indexSettings.getAsInt("index.term_index_interval", IndexWriterConfig.DEFAULT_TERM_INDEX_INTERVAL);
this.termIndexDivisor = indexSettings.getAsInt("index.term_index_divisor", 1); // IndexReader#DEFAULT_TERMS_INDEX_DIVISOR
this.codecName = indexSettings.get("index.codec", "default");
this.termIndexInterval = indexSettings.getAsInt(INDEX_TERM_INDEX_INTERVAL, IndexWriterConfig.DEFAULT_TERM_INDEX_INTERVAL);
this.termIndexDivisor = indexSettings.getAsInt(INDEX_TERM_INDEX_DIVISOR, 1); // IndexReader#DEFAULT_TERMS_INDEX_DIVISOR
this.codecName = indexSettings.get(INDEX_CODEC, "default");
this.threadPool = threadPool;
this.indexSettingsService = indexSettingsService;
@ -178,7 +177,7 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
this.similarityService = similarityService;
this.codecService = codecService;
this.indexConcurrency = indexSettings.getAsInt("index.index_concurrency", IndexWriterConfig.DEFAULT_MAX_THREAD_STATES);
this.indexConcurrency = indexSettings.getAsInt(INDEX_INDEX_CONCURRENCY, IndexWriterConfig.DEFAULT_MAX_THREAD_STATES);
this.versionMap = ConcurrentCollections.newConcurrentMap();
this.dirtyLocks = new Object[indexConcurrency * 50]; // we multiply it to have enough...
for (int i = 0; i < dirtyLocks.length; i++) {
@ -1327,30 +1326,25 @@ public class RobinEngine extends AbstractIndexShardComponent implements Engine {
return indexWriter;
}
static {
IndexMetaData.addDynamicSettings(
"index.term_index_interval",
"index.term_index_divisor",
"index.index_concurrency",
"index.gc_deletes",
"index.codec"
);
}
public static final String INDEX_TERM_INDEX_INTERVAL = "index.term_index_interval";
public static final String INDEX_TERM_INDEX_DIVISOR = "index.term_index_divisor";
public static final String INDEX_INDEX_CONCURRENCY = "index.index_concurrency";
public static final String INDEX_GC_DELETES = "index.gc_deletes";
public static final String INDEX_CODEC = "index.codec";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
long gcDeletesInMillis = indexSettings.getAsTime("index.gc_deletes", TimeValue.timeValueMillis(RobinEngine.this.gcDeletesInMillis)).millis();
long gcDeletesInMillis = indexSettings.getAsTime(INDEX_GC_DELETES, TimeValue.timeValueMillis(RobinEngine.this.gcDeletesInMillis)).millis();
if (gcDeletesInMillis != RobinEngine.this.gcDeletesInMillis) {
logger.info("updating index.gc_deletes from [{}] to [{}]", TimeValue.timeValueMillis(RobinEngine.this.gcDeletesInMillis), TimeValue.timeValueMillis(gcDeletesInMillis));
RobinEngine.this.gcDeletesInMillis = gcDeletesInMillis;
}
int termIndexInterval = settings.getAsInt("index.term_index_interval", RobinEngine.this.termIndexInterval);
int termIndexDivisor = settings.getAsInt("index.term_index_divisor", RobinEngine.this.termIndexDivisor); // IndexReader#DEFAULT_TERMS_INDEX_DIVISOR
int indexConcurrency = settings.getAsInt("index.index_concurrency", RobinEngine.this.indexConcurrency);
String codecName = settings.get("index.codec", RobinEngine.this.codecName);
int termIndexInterval = settings.getAsInt(INDEX_TERM_INDEX_INTERVAL, RobinEngine.this.termIndexInterval);
int termIndexDivisor = settings.getAsInt(INDEX_TERM_INDEX_DIVISOR, RobinEngine.this.termIndexDivisor); // IndexReader#DEFAULT_TERMS_INDEX_DIVISOR
int indexConcurrency = settings.getAsInt(INDEX_INDEX_CONCURRENCY, RobinEngine.this.indexConcurrency);
String codecName = settings.get(INDEX_CODEC, RobinEngine.this.codecName);
boolean requiresFlushing = false;
if (termIndexInterval != RobinEngine.this.termIndexInterval || termIndexDivisor != RobinEngine.this.termIndexDivisor || indexConcurrency != RobinEngine.this.indexConcurrency || !codecName.equals(RobinEngine.this.codecName)) {
rwl.readLock().lock();

7
src/main/java/org/elasticsearch/index/gateway/IndexShardGatewayService.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.index.gateway;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
@ -92,14 +91,12 @@ public class IndexShardGatewayService extends AbstractIndexShardComponent implem
indexSettingsService.addListener(applySettings);
}
static {
IndexMetaData.addDynamicSettings("index.gateway.snapshot_interval");
}
public static final String INDEX_GATEWAY_SNAPSHOT_INTERVAL = "index.gateway.snapshot_interval";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
TimeValue snapshotInterval = settings.getAsTime("index.gateway.snapshot_interval", IndexShardGatewayService.this.snapshotInterval);
TimeValue snapshotInterval = settings.getAsTime(INDEX_GATEWAY_SNAPSHOT_INTERVAL, IndexShardGatewayService.this.snapshotInterval);
if (!snapshotInterval.equals(IndexShardGatewayService.this.snapshotInterval)) {
logger.info("updating snapshot_interval from [{}] to [{}]", IndexShardGatewayService.this.snapshotInterval, snapshotInterval);
IndexShardGatewayService.this.snapshotInterval = snapshotInterval;

29
src/main/java/org/elasticsearch/index/indexing/slowlog/ShardSlowLogIndexingService.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.index.indexing.slowlog;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.logging.ESLogger;
import org.elasticsearch.common.logging.Loggers;
@ -52,45 +51,41 @@ public class ShardSlowLogIndexingService extends AbstractIndexShardComponent {
private final ESLogger indexLogger;
private final ESLogger deleteLogger;
static {
IndexMetaData.addDynamicSettings(
"index.indexing.slowlog.threshold.index.warn",
"index.indexing.slowlog.threshold.index.info",
"index.indexing.slowlog.threshold.index.debug",
"index.indexing.slowlog.threshold.index.trace",
"index.indexing.slowlog.reformat",
"index.indexing.slowlog.level"
);
}
public static final String INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_WARN = "index.indexing.slowlog.threshold.index.warn";
public static final String INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_INFO = "index.indexing.slowlog.threshold.index.info";
public static final String INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_DEBUG = "index.indexing.slowlog.threshold.index.debug";
public static final String INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_TRACE = "index.indexing.slowlog.threshold.index.trace";
public static final String INDEX_INDEXING_SLOWLOG_REFORMAT = "index.indexing.slowlog.reformat";
public static final String INDEX_INDEXING_SLOWLOG_LEVEL = "index.indexing.slowlog.level";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public synchronized void onRefreshSettings(Settings settings) {
long indexWarnThreshold = settings.getAsTime("index.indexing.slowlog.threshold.index.warn", TimeValue.timeValueNanos(ShardSlowLogIndexingService.this.indexWarnThreshold)).nanos();
long indexWarnThreshold = settings.getAsTime(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_WARN, TimeValue.timeValueNanos(ShardSlowLogIndexingService.this.indexWarnThreshold)).nanos();
if (indexWarnThreshold != ShardSlowLogIndexingService.this.indexWarnThreshold) {
ShardSlowLogIndexingService.this.indexWarnThreshold = indexWarnThreshold;
}
long indexInfoThreshold = settings.getAsTime("index.indexing.slowlog.threshold.index.info", TimeValue.timeValueNanos(ShardSlowLogIndexingService.this.indexInfoThreshold)).nanos();
long indexInfoThreshold = settings.getAsTime(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_INFO, TimeValue.timeValueNanos(ShardSlowLogIndexingService.this.indexInfoThreshold)).nanos();
if (indexInfoThreshold != ShardSlowLogIndexingService.this.indexInfoThreshold) {
ShardSlowLogIndexingService.this.indexInfoThreshold = indexInfoThreshold;
}
long indexDebugThreshold = settings.getAsTime("index.indexing.slowlog.threshold.index.debug", TimeValue.timeValueNanos(ShardSlowLogIndexingService.this.indexDebugThreshold)).nanos();
long indexDebugThreshold = settings.getAsTime(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_DEBUG, TimeValue.timeValueNanos(ShardSlowLogIndexingService.this.indexDebugThreshold)).nanos();
if (indexDebugThreshold != ShardSlowLogIndexingService.this.indexDebugThreshold) {
ShardSlowLogIndexingService.this.indexDebugThreshold = indexDebugThreshold;
}
long indexTraceThreshold = settings.getAsTime("index.indexing.slowlog.threshold.index.trace", TimeValue.timeValueNanos(ShardSlowLogIndexingService.this.indexTraceThreshold)).nanos();
long indexTraceThreshold = settings.getAsTime(INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_TRACE, TimeValue.timeValueNanos(ShardSlowLogIndexingService.this.indexTraceThreshold)).nanos();
if (indexTraceThreshold != ShardSlowLogIndexingService.this.indexTraceThreshold) {
ShardSlowLogIndexingService.this.indexTraceThreshold = indexTraceThreshold;
}
String level = settings.get("index.indexing.slowlog.level", ShardSlowLogIndexingService.this.level);
String level = settings.get(INDEX_INDEXING_SLOWLOG_LEVEL, ShardSlowLogIndexingService.this.level);
if (!level.equals(ShardSlowLogIndexingService.this.level)) {
ShardSlowLogIndexingService.this.indexLogger.setLevel(level.toUpperCase());
ShardSlowLogIndexingService.this.deleteLogger.setLevel(level.toUpperCase());
ShardSlowLogIndexingService.this.level = level;
}
boolean reformat = settings.getAsBoolean("index.indexing.slowlog.reformat", ShardSlowLogIndexingService.this.reformat);
boolean reformat = settings.getAsBoolean(INDEX_INDEXING_SLOWLOG_REFORMAT, ShardSlowLogIndexingService.this.reformat);
if (reformat != ShardSlowLogIndexingService.this.reformat) {
ShardSlowLogIndexingService.this.reformat = reformat;
}

27
src/main/java/org/elasticsearch/index/merge/policy/LogByteSizeMergePolicyProvider.java
View File

@ -22,7 +22,6 @@ package org.elasticsearch.index.merge.policy;
import org.apache.lucene.index.LogByteSizeMergePolicy;
import org.apache.lucene.index.SegmentInfos;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.Preconditions;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
@ -61,7 +60,7 @@ public class LogByteSizeMergePolicyProvider extends AbstractIndexShardComponent
Preconditions.checkNotNull(store, "Store must be provided to merge policy");
this.indexSettingsService = indexSettingsService;
this.compoundFormat = indexSettings.getAsBoolean("index.compound_format", store.suggestUseCompoundFile());
this.compoundFormat = indexSettings.getAsBoolean(INDEX_COMPOUND_FORMAT, store.suggestUseCompoundFile());
this.minMergeSize = componentSettings.getAsBytesSize("min_merge_size", new ByteSizeValue((long) (LogByteSizeMergePolicy.DEFAULT_MIN_MERGE_MB * 1024 * 1024), ByteSizeUnit.BYTES));
this.maxMergeSize = componentSettings.getAsBytesSize("max_merge_size", new ByteSizeValue((long) LogByteSizeMergePolicy.DEFAULT_MAX_MERGE_MB, ByteSizeUnit.MB));
this.mergeFactor = componentSettings.getAsInt("merge_factor", LogByteSizeMergePolicy.DEFAULT_MERGE_FACTOR);
@ -98,20 +97,16 @@ public class LogByteSizeMergePolicyProvider extends AbstractIndexShardComponent
indexSettingsService.removeListener(applySettings);
}
static {
IndexMetaData.addDynamicSettings(
"index.merge.policy.min_merge_size",
"index.merge.policy.max_merge_size",
"index.merge.policy.max_merge_docs",
"index.merge.policy.merge_factor",
"index.compound_format"
);
}
public static final String INDEX_MERGE_POLICY_MIN_MERGE_SIZE = "index.merge.policy.min_merge_size";
public static final String INDEX_MERGE_POLICY_MAX_MERGE_SIZE = "index.merge.policy.max_merge_size";
public static final String INDEX_MERGE_POLICY_MAX_MERGE_DOCS = "index.merge.policy.max_merge_docs";
public static final String INDEX_MERGE_POLICY_MERGE_FACTOR = "index.merge.policy.merge_factor";
public static final String INDEX_COMPOUND_FORMAT = "index.compound_format";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
ByteSizeValue minMergeSize = settings.getAsBytesSize("index.merge.policy.min_merge_size", LogByteSizeMergePolicyProvider.this.minMergeSize);
ByteSizeValue minMergeSize = settings.getAsBytesSize(INDEX_MERGE_POLICY_MIN_MERGE_SIZE, LogByteSizeMergePolicyProvider.this.minMergeSize);
if (!minMergeSize.equals(LogByteSizeMergePolicyProvider.this.minMergeSize)) {
logger.info("updating min_merge_size from [{}] to [{}]", LogByteSizeMergePolicyProvider.this.minMergeSize, minMergeSize);
LogByteSizeMergePolicyProvider.this.minMergeSize = minMergeSize;
@ -120,7 +115,7 @@ public class LogByteSizeMergePolicyProvider extends AbstractIndexShardComponent
}
}
ByteSizeValue maxMergeSize = settings.getAsBytesSize("index.merge.policy.max_merge_size", LogByteSizeMergePolicyProvider.this.maxMergeSize);
ByteSizeValue maxMergeSize = settings.getAsBytesSize(INDEX_MERGE_POLICY_MAX_MERGE_SIZE, LogByteSizeMergePolicyProvider.this.maxMergeSize);
if (!maxMergeSize.equals(LogByteSizeMergePolicyProvider.this.maxMergeSize)) {
logger.info("updating max_merge_size from [{}] to [{}]", LogByteSizeMergePolicyProvider.this.maxMergeSize, maxMergeSize);
LogByteSizeMergePolicyProvider.this.maxMergeSize = maxMergeSize;
@ -129,7 +124,7 @@ public class LogByteSizeMergePolicyProvider extends AbstractIndexShardComponent
}
}
int maxMergeDocs = settings.getAsInt("index.merge.policy.max_merge_docs", LogByteSizeMergePolicyProvider.this.maxMergeDocs);
int maxMergeDocs = settings.getAsInt(INDEX_MERGE_POLICY_MAX_MERGE_DOCS, LogByteSizeMergePolicyProvider.this.maxMergeDocs);
if (maxMergeDocs != LogByteSizeMergePolicyProvider.this.maxMergeDocs) {
logger.info("updating max_merge_docs from [{}] to [{}]", LogByteSizeMergePolicyProvider.this.maxMergeDocs, maxMergeDocs);
LogByteSizeMergePolicyProvider.this.maxMergeDocs = maxMergeDocs;
@ -138,7 +133,7 @@ public class LogByteSizeMergePolicyProvider extends AbstractIndexShardComponent
}
}
int mergeFactor = settings.getAsInt("index.merge.policy.merge_factor", LogByteSizeMergePolicyProvider.this.mergeFactor);
int mergeFactor = settings.getAsInt(INDEX_MERGE_POLICY_MERGE_FACTOR, LogByteSizeMergePolicyProvider.this.mergeFactor);
if (mergeFactor != LogByteSizeMergePolicyProvider.this.mergeFactor) {
logger.info("updating merge_factor from [{}] to [{}]", LogByteSizeMergePolicyProvider.this.mergeFactor, mergeFactor);
LogByteSizeMergePolicyProvider.this.mergeFactor = mergeFactor;
@ -147,7 +142,7 @@ public class LogByteSizeMergePolicyProvider extends AbstractIndexShardComponent
}
}
boolean compoundFormat = settings.getAsBoolean("index.compound_format", LogByteSizeMergePolicyProvider.this.compoundFormat);
boolean compoundFormat = settings.getAsBoolean(INDEX_COMPOUND_FORMAT, LogByteSizeMergePolicyProvider.this.compoundFormat);
if (compoundFormat != LogByteSizeMergePolicyProvider.this.compoundFormat) {
logger.info("updating index.compound_format from [{}] to [{}]", LogByteSizeMergePolicyProvider.this.compoundFormat, compoundFormat);
LogByteSizeMergePolicyProvider.this.compoundFormat = compoundFormat;

24
src/main/java/org/elasticsearch/index/merge/policy/LogDocMergePolicyProvider.java
View File

@ -22,7 +22,6 @@ package org.elasticsearch.index.merge.policy;
import org.apache.lucene.index.LogDocMergePolicy;
import org.apache.lucene.index.SegmentInfos;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.Preconditions;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
@ -58,7 +57,7 @@ public class LogDocMergePolicyProvider extends AbstractIndexShardComponent imple
Preconditions.checkNotNull(store, "Store must be provided to merge policy");
this.indexSettingsService = indexSettingsService;
this.compoundFormat = indexSettings.getAsBoolean("index.compound_format", store.suggestUseCompoundFile());
this.compoundFormat = indexSettings.getAsBoolean(INDEX_COMPOUND_FORMAT, store.suggestUseCompoundFile());
this.minMergeDocs = componentSettings.getAsInt("min_merge_docs", LogDocMergePolicy.DEFAULT_MIN_MERGE_DOCS);
this.maxMergeDocs = componentSettings.getAsInt("max_merge_docs", LogDocMergePolicy.DEFAULT_MAX_MERGE_DOCS);
this.mergeFactor = componentSettings.getAsInt("merge_factor", LogDocMergePolicy.DEFAULT_MERGE_FACTOR);
@ -92,19 +91,16 @@ public class LogDocMergePolicyProvider extends AbstractIndexShardComponent imple
return mergePolicy;
}
static {
IndexMetaData.addDynamicSettings(
"index.merge.policy.min_merge_docs",
"index.merge.policy.max_merge_docs",
"index.merge.policy.merge_factor",
"index.compound_format"
);
}
public static final String INDEX_MERGE_POLICY_MIN_MERGE_DOCS = "index.merge.policy.min_merge_docs";
public static final String INDEX_MERGE_POLICY_MAX_MERGE_DOCS = "index.merge.policy.max_merge_docs";
public static final String INDEX_MERGE_POLICY_MERGE_FACTOR = "index.merge.policy.merge_factor";
public static final String INDEX_COMPOUND_FORMAT = "index.compound_format";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
int minMergeDocs = settings.getAsInt("index.merge.policy.min_merge_docs", LogDocMergePolicyProvider.this.minMergeDocs);
int minMergeDocs = settings.getAsInt(INDEX_MERGE_POLICY_MIN_MERGE_DOCS, LogDocMergePolicyProvider.this.minMergeDocs);
if (minMergeDocs != LogDocMergePolicyProvider.this.minMergeDocs) {
logger.info("updating min_merge_docs from [{}] to [{}]", LogDocMergePolicyProvider.this.minMergeDocs, minMergeDocs);
LogDocMergePolicyProvider.this.minMergeDocs = minMergeDocs;
@ -113,7 +109,7 @@ public class LogDocMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
int maxMergeDocs = settings.getAsInt("index.merge.policy.max_merge_docs", LogDocMergePolicyProvider.this.maxMergeDocs);
int maxMergeDocs = settings.getAsInt(INDEX_MERGE_POLICY_MAX_MERGE_DOCS, LogDocMergePolicyProvider.this.maxMergeDocs);
if (maxMergeDocs != LogDocMergePolicyProvider.this.maxMergeDocs) {
logger.info("updating max_merge_docs from [{}] to [{}]", LogDocMergePolicyProvider.this.maxMergeDocs, maxMergeDocs);
LogDocMergePolicyProvider.this.maxMergeDocs = maxMergeDocs;
@ -122,7 +118,7 @@ public class LogDocMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
int mergeFactor = settings.getAsInt("index.merge.policy.merge_factor", LogDocMergePolicyProvider.this.mergeFactor);
int mergeFactor = settings.getAsInt(INDEX_MERGE_POLICY_MERGE_FACTOR, LogDocMergePolicyProvider.this.mergeFactor);
if (mergeFactor != LogDocMergePolicyProvider.this.mergeFactor) {
logger.info("updating merge_factor from [{}] to [{}]", LogDocMergePolicyProvider.this.mergeFactor, mergeFactor);
LogDocMergePolicyProvider.this.mergeFactor = mergeFactor;
@ -131,7 +127,7 @@ public class LogDocMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
boolean compoundFormat = settings.getAsBoolean("index.compound_format", LogDocMergePolicyProvider.this.compoundFormat);
boolean compoundFormat = settings.getAsBoolean(INDEX_COMPOUND_FORMAT, LogDocMergePolicyProvider.this.compoundFormat);
if (compoundFormat != LogDocMergePolicyProvider.this.compoundFormat) {
logger.info("updating index.compound_format from [{}] to [{}]", LogDocMergePolicyProvider.this.compoundFormat, compoundFormat);
LogDocMergePolicyProvider.this.compoundFormat = compoundFormat;

39
src/main/java/org/elasticsearch/index/merge/policy/TieredMergePolicyProvider.java
View File

@ -23,7 +23,6 @@ import org.apache.lucene.index.MergePolicy;
import org.apache.lucene.index.SegmentInfos;
import org.apache.lucene.index.TieredMergePolicy;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
@ -59,7 +58,7 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
super(store.shardId(), store.indexSettings());
this.indexSettingsService = indexSettingsService;
this.compoundFormat = indexSettings.getAsBoolean("index.compound_format", store.suggestUseCompoundFile());
this.compoundFormat = indexSettings.getAsBoolean(INDEX_COMPOUND_FORMAT, store.suggestUseCompoundFile());
this.asyncMerge = indexSettings.getAsBoolean("index.merge.async", true);
this.forceMergeDeletesPctAllowed = componentSettings.getAsDouble("expunge_deletes_allowed", 10d); // percentage
this.floorSegment = componentSettings.getAsBytesSize("floor_segment", new ByteSizeValue(2, ByteSizeUnit.MB));
@ -116,23 +115,19 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
indexSettingsService.removeListener(applySettings);
}
static {
IndexMetaData.addDynamicSettings(
"index.merge.policy.expunge_deletes_allowed",
"index.merge.policy.floor_segment",
"index.merge.policy.max_merge_at_once",
"index.merge.policy.max_merge_at_once_explicit",
"index.merge.policy.max_merged_segment",
"index.merge.policy.segments_per_tier",
"index.merge.policy.reclaim_deletes_weight",
"index.compound_format"
);
}
public static final String INDEX_MERGE_POLICY_EXPUNGE_DELETES_ALLOWED = "index.merge.policy.expunge_deletes_allowed";
public static final String INDEX_MERGE_POLICY_FLOOR_SEGMENT = "index.merge.policy.floor_segment";
public static final String INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE = "index.merge.policy.max_merge_at_once";
public static final String INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_EXPLICIT = "index.merge.policy.max_merge_at_once_explicit";
public static final String INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT = "index.merge.policy.max_merged_segment";
public static final String INDEX_MERGE_POLICY_SEGMENTS_PER_TIER = "index.merge.policy.segments_per_tier";
public static final String INDEX_MERGE_POLICY_RECLAIM_DELETES_WEIGHT = "index.merge.policy.reclaim_deletes_weight";
public static final String INDEX_COMPOUND_FORMAT = "index.compound_format";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
double expungeDeletesPctAllowed = settings.getAsDouble("index.merge.policy.expunge_deletes_allowed", TieredMergePolicyProvider.this.forceMergeDeletesPctAllowed);
double expungeDeletesPctAllowed = settings.getAsDouble(INDEX_MERGE_POLICY_EXPUNGE_DELETES_ALLOWED, TieredMergePolicyProvider.this.forceMergeDeletesPctAllowed);
if (expungeDeletesPctAllowed != TieredMergePolicyProvider.this.forceMergeDeletesPctAllowed) {
logger.info("updating [expunge_deletes_allowed] from [{}] to [{}]", TieredMergePolicyProvider.this.forceMergeDeletesPctAllowed, expungeDeletesPctAllowed);
TieredMergePolicyProvider.this.forceMergeDeletesPctAllowed = expungeDeletesPctAllowed;
@ -141,7 +136,7 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
ByteSizeValue floorSegment = settings.getAsBytesSize("index.merge.policy.floor_segment", TieredMergePolicyProvider.this.floorSegment);
ByteSizeValue floorSegment = settings.getAsBytesSize(INDEX_MERGE_POLICY_FLOOR_SEGMENT, TieredMergePolicyProvider.this.floorSegment);
if (!floorSegment.equals(TieredMergePolicyProvider.this.floorSegment)) {
logger.info("updating [floor_segment] from [{}] to [{}]", TieredMergePolicyProvider.this.floorSegment, floorSegment);
TieredMergePolicyProvider.this.floorSegment = floorSegment;
@ -150,7 +145,7 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
int maxMergeAtOnce = settings.getAsInt("index.merge.policy.max_merge_at_once", TieredMergePolicyProvider.this.maxMergeAtOnce);
int maxMergeAtOnce = settings.getAsInt(INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE, TieredMergePolicyProvider.this.maxMergeAtOnce);
if (maxMergeAtOnce != TieredMergePolicyProvider.this.maxMergeAtOnce) {
logger.info("updating [max_merge_at_once] from [{}] to [{}]", TieredMergePolicyProvider.this.maxMergeAtOnce, maxMergeAtOnce);
TieredMergePolicyProvider.this.maxMergeAtOnce = maxMergeAtOnce;
@ -159,7 +154,7 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
int maxMergeAtOnceExplicit = settings.getAsInt("index.merge.policy.max_merge_at_once_explicit", TieredMergePolicyProvider.this.maxMergeAtOnceExplicit);
int maxMergeAtOnceExplicit = settings.getAsInt(INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_EXPLICIT, TieredMergePolicyProvider.this.maxMergeAtOnceExplicit);
if (maxMergeAtOnceExplicit != TieredMergePolicyProvider.this.maxMergeAtOnceExplicit) {
logger.info("updating [max_merge_at_once_explicit] from [{}] to [{}]", TieredMergePolicyProvider.this.maxMergeAtOnceExplicit, maxMergeAtOnceExplicit);
TieredMergePolicyProvider.this.maxMergeAtOnceExplicit = maxMergeAtOnceExplicit;
@ -168,7 +163,7 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
ByteSizeValue maxMergedSegment = settings.getAsBytesSize("index.merge.policy.max_merged_segment", TieredMergePolicyProvider.this.maxMergedSegment);
ByteSizeValue maxMergedSegment = settings.getAsBytesSize(INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT, TieredMergePolicyProvider.this.maxMergedSegment);
if (!maxMergedSegment.equals(TieredMergePolicyProvider.this.maxMergedSegment)) {
logger.info("updating [max_merged_segment] from [{}] to [{}]", TieredMergePolicyProvider.this.maxMergedSegment, maxMergedSegment);
TieredMergePolicyProvider.this.maxMergedSegment = maxMergedSegment;
@ -177,7 +172,7 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
double segmentsPerTier = settings.getAsDouble("index.merge.policy.segments_per_tier", TieredMergePolicyProvider.this.segmentsPerTier);
double segmentsPerTier = settings.getAsDouble(INDEX_MERGE_POLICY_SEGMENTS_PER_TIER, TieredMergePolicyProvider.this.segmentsPerTier);
if (segmentsPerTier != TieredMergePolicyProvider.this.segmentsPerTier) {
logger.info("updating [segments_per_tier] from [{}] to [{}]", TieredMergePolicyProvider.this.segmentsPerTier, segmentsPerTier);
TieredMergePolicyProvider.this.segmentsPerTier = segmentsPerTier;
@ -186,7 +181,7 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
double reclaimDeletesWeight = settings.getAsDouble("index.merge.policy.reclaim_deletes_weight", TieredMergePolicyProvider.this.reclaimDeletesWeight);
double reclaimDeletesWeight = settings.getAsDouble(INDEX_MERGE_POLICY_RECLAIM_DELETES_WEIGHT, TieredMergePolicyProvider.this.reclaimDeletesWeight);
if (reclaimDeletesWeight != TieredMergePolicyProvider.this.reclaimDeletesWeight) {
logger.info("updating [reclaim_deletes_weight] from [{}] to [{}]", TieredMergePolicyProvider.this.reclaimDeletesWeight, reclaimDeletesWeight);
TieredMergePolicyProvider.this.reclaimDeletesWeight = reclaimDeletesWeight;
@ -195,7 +190,7 @@ public class TieredMergePolicyProvider extends AbstractIndexShardComponent imple
}
}
boolean compoundFormat = settings.getAsBoolean("index.compound_format", TieredMergePolicyProvider.this.compoundFormat);
boolean compoundFormat = settings.getAsBoolean(INDEX_COMPOUND_FORMAT, TieredMergePolicyProvider.this.compoundFormat);
if (compoundFormat != TieredMergePolicyProvider.this.compoundFormat) {
logger.info("updating index.compound_format from [{}] to [{}]", TieredMergePolicyProvider.this.compoundFormat, compoundFormat);
TieredMergePolicyProvider.this.compoundFormat = compoundFormat;

45
src/main/java/org/elasticsearch/index/search/slowlog/ShardSlowLogSearchService.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.index.search.slowlog;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.logging.ESLogger;
@ -57,66 +56,62 @@ public class ShardSlowLogSearchService extends AbstractIndexShardComponent {
private final ESLogger queryLogger;
private final ESLogger fetchLogger;
static {
IndexMetaData.addDynamicSettings(
"index.search.slowlog.threshold.query.warn",
"index.search.slowlog.threshold.query.info",
"index.search.slowlog.threshold.query.debug",
"index.search.slowlog.threshold.query.trace",
"index.search.slowlog.threshold.fetch.warn",
"index.search.slowlog.threshold.fetch.info",
"index.search.slowlog.threshold.fetch.debug",
"index.search.slowlog.threshold.fetch.trace",
"index.search.slowlog.reformat",
"index.search.slowlog.level"
);
}
public static final String INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_WARN = "index.search.slowlog.threshold.query.warn";
public static final String INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_INFO = "index.search.slowlog.threshold.query.info";
public static final String INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_DEBUG = "index.search.slowlog.threshold.query.debug";
public static final String INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_TRACE = "index.search.slowlog.threshold.query.trace";
public static final String INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_WARN = "index.search.slowlog.threshold.fetch.warn";
public static final String INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_INFO = "index.search.slowlog.threshold.fetch.info";
public static final String INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_DEBUG = "index.search.slowlog.threshold.fetch.debug";
public static final String INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_TRACE = "index.search.slowlog.threshold.fetch.trace";
public static final String INDEX_SEARCH_SLOWLOG_REFORMAT = "index.search.slowlog.reformat";
public static final String INDEX_SEARCH_SLOWLOG_LEVEL = "index.search.slowlog.level";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public synchronized void onRefreshSettings(Settings settings) {
long queryWarnThreshold = settings.getAsTime("index.search.slowlog.threshold.query.warn", TimeValue.timeValueNanos(ShardSlowLogSearchService.this.queryWarnThreshold)).nanos();
long queryWarnThreshold = settings.getAsTime(INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_WARN, TimeValue.timeValueNanos(ShardSlowLogSearchService.this.queryWarnThreshold)).nanos();
if (queryWarnThreshold != ShardSlowLogSearchService.this.queryWarnThreshold) {
ShardSlowLogSearchService.this.queryWarnThreshold = queryWarnThreshold;
}
long queryInfoThreshold = settings.getAsTime("index.search.slowlog.threshold.query.info", TimeValue.timeValueNanos(ShardSlowLogSearchService.this.queryInfoThreshold)).nanos();
long queryInfoThreshold = settings.getAsTime(INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_INFO, TimeValue.timeValueNanos(ShardSlowLogSearchService.this.queryInfoThreshold)).nanos();
if (queryInfoThreshold != ShardSlowLogSearchService.this.queryInfoThreshold) {
ShardSlowLogSearchService.this.queryInfoThreshold = queryInfoThreshold;
}
long queryDebugThreshold = settings.getAsTime("index.search.slowlog.threshold.query.debug", TimeValue.timeValueNanos(ShardSlowLogSearchService.this.queryDebugThreshold)).nanos();
long queryDebugThreshold = settings.getAsTime(INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_DEBUG, TimeValue.timeValueNanos(ShardSlowLogSearchService.this.queryDebugThreshold)).nanos();
if (queryDebugThreshold != ShardSlowLogSearchService.this.queryDebugThreshold) {
ShardSlowLogSearchService.this.queryDebugThreshold = queryDebugThreshold;
}
long queryTraceThreshold = settings.getAsTime("index.search.slowlog.threshold.query.trace", TimeValue.timeValueNanos(ShardSlowLogSearchService.this.queryTraceThreshold)).nanos();
long queryTraceThreshold = settings.getAsTime(INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_TRACE, TimeValue.timeValueNanos(ShardSlowLogSearchService.this.queryTraceThreshold)).nanos();
if (queryTraceThreshold != ShardSlowLogSearchService.this.queryTraceThreshold) {
ShardSlowLogSearchService.this.queryTraceThreshold = queryTraceThreshold;
}
long fetchWarnThreshold = settings.getAsTime("index.search.slowlog.threshold.fetch.warn", TimeValue.timeValueNanos(ShardSlowLogSearchService.this.fetchWarnThreshold)).nanos();
long fetchWarnThreshold = settings.getAsTime(INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_WARN, TimeValue.timeValueNanos(ShardSlowLogSearchService.this.fetchWarnThreshold)).nanos();
if (fetchWarnThreshold != ShardSlowLogSearchService.this.fetchWarnThreshold) {
ShardSlowLogSearchService.this.fetchWarnThreshold = fetchWarnThreshold;
}
long fetchInfoThreshold = settings.getAsTime("index.search.slowlog.threshold.fetch.info", TimeValue.timeValueNanos(ShardSlowLogSearchService.this.fetchInfoThreshold)).nanos();
long fetchInfoThreshold = settings.getAsTime(INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_INFO, TimeValue.timeValueNanos(ShardSlowLogSearchService.this.fetchInfoThreshold)).nanos();
if (fetchInfoThreshold != ShardSlowLogSearchService.this.fetchInfoThreshold) {
ShardSlowLogSearchService.this.fetchInfoThreshold = fetchInfoThreshold;
}
long fetchDebugThreshold = settings.getAsTime("index.search.slowlog.threshold.fetch.debug", TimeValue.timeValueNanos(ShardSlowLogSearchService.this.fetchDebugThreshold)).nanos();
long fetchDebugThreshold = settings.getAsTime(INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_DEBUG, TimeValue.timeValueNanos(ShardSlowLogSearchService.this.fetchDebugThreshold)).nanos();
if (fetchDebugThreshold != ShardSlowLogSearchService.this.fetchDebugThreshold) {
ShardSlowLogSearchService.this.fetchDebugThreshold = fetchDebugThreshold;
}
long fetchTraceThreshold = settings.getAsTime("index.search.slowlog.threshold.fetch.trace", TimeValue.timeValueNanos(ShardSlowLogSearchService.this.fetchTraceThreshold)).nanos();
long fetchTraceThreshold = settings.getAsTime(INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_TRACE, TimeValue.timeValueNanos(ShardSlowLogSearchService.this.fetchTraceThreshold)).nanos();
if (fetchTraceThreshold != ShardSlowLogSearchService.this.fetchTraceThreshold) {
ShardSlowLogSearchService.this.fetchTraceThreshold = fetchTraceThreshold;
}
String level = settings.get("index.search.slowlog.level", ShardSlowLogSearchService.this.level);
String level = settings.get(INDEX_SEARCH_SLOWLOG_LEVEL, ShardSlowLogSearchService.this.level);
if (!level.equals(ShardSlowLogSearchService.this.level)) {
ShardSlowLogSearchService.this.queryLogger.setLevel(level.toUpperCase());
ShardSlowLogSearchService.this.fetchLogger.setLevel(level.toUpperCase());
ShardSlowLogSearchService.this.level = level;
}
boolean reformat = settings.getAsBoolean("index.search.slowlog.reformat", ShardSlowLogSearchService.this.reformat);
boolean reformat = settings.getAsBoolean(INDEX_SEARCH_SLOWLOG_REFORMAT, ShardSlowLogSearchService.this.reformat);
if (reformat != ShardSlowLogSearchService.this.reformat) {
ShardSlowLogSearchService.this.reformat = reformat;
}

39
src/main/java/org/elasticsearch/index/settings/IndexDynamicSettings.java Normal file
View File

@ -0,0 +1,39 @@
/*
* Licensed to ElasticSearch and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. ElasticSearch licenses this
* file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.index.settings;
import org.elasticsearch.common.inject.BindingAnnotation;
import java.lang.annotation.Documented;
import java.lang.annotation.Retention;
import java.lang.annotation.Target;
import static java.lang.annotation.ElementType.FIELD;
import static java.lang.annotation.ElementType.PARAMETER;
import static java.lang.annotation.RetentionPolicy.RUNTIME;
@BindingAnnotation
@Target({FIELD, PARAMETER})
@Retention(RUNTIME)
@Documented
public @interface IndexDynamicSettings {
}

121
src/main/java/org/elasticsearch/index/settings/IndexDynamicSettingsModule.java Normal file
View File

@ -0,0 +1,121 @@
/*
* Licensed to ElasticSearch and Shay Banon under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. ElasticSearch licenses this
* file to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package org.elasticsearch.index.settings;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.allocation.decider.FilterAllocationDecider;
import org.elasticsearch.cluster.routing.allocation.decider.ShardsLimitAllocationDecider;
import org.elasticsearch.cluster.settings.DynamicSettings;
import org.elasticsearch.common.inject.AbstractModule;
import org.elasticsearch.gateway.local.LocalGatewayAllocator;
import org.elasticsearch.index.engine.robin.RobinEngine;
import org.elasticsearch.index.gateway.IndexShardGatewayService;
import org.elasticsearch.index.indexing.slowlog.ShardSlowLogIndexingService;
import org.elasticsearch.index.merge.policy.LogByteSizeMergePolicyProvider;
import org.elasticsearch.index.merge.policy.LogDocMergePolicyProvider;
import org.elasticsearch.index.merge.policy.TieredMergePolicyProvider;
import org.elasticsearch.index.search.slowlog.ShardSlowLogSearchService;
import org.elasticsearch.index.shard.service.InternalIndexShard;
import org.elasticsearch.index.store.support.AbstractIndexStore;
import org.elasticsearch.index.translog.TranslogService;
import org.elasticsearch.index.translog.fs.FsTranslog;
import org.elasticsearch.indices.ttl.IndicesTTLService;
/**
*/
public class IndexDynamicSettingsModule extends AbstractModule {
private final DynamicSettings indexDynamicSettings;
public IndexDynamicSettingsModule() {
indexDynamicSettings = new DynamicSettings();
indexDynamicSettings.addDynamicSettings(
AbstractIndexStore.INDEX_STORE_THROTTLE_MAX_BYTES_PER_SEC,
AbstractIndexStore.INDEX_STORE_THROTTLE_TYPE,
FilterAllocationDecider.INDEX_ROUTING_REQUIRE_GROUP + "*",
FilterAllocationDecider.INDEX_ROUTING_INCLUDE_GROUP + "*",
FilterAllocationDecider.INDEX_ROUTING_EXCLUDE_GROUP + "*",
FsTranslog.INDEX_TRANSLOG_FS_TYPE,
FsTranslog.INDEX_TRANSLOG_FS_BUFFER_SIZE,
FsTranslog.INDEX_TRANSLOG_FS_TRANSIENT_BUFFER_SIZE,
IndexMetaData.SETTING_NUMBER_OF_REPLICAS,
IndexMetaData.SETTING_AUTO_EXPAND_REPLICAS,
IndexMetaData.SETTING_READ_ONLY,
IndexMetaData.SETTING_BLOCKS_READ,
IndexMetaData.SETTING_BLOCKS_WRITE,
IndexMetaData.SETTING_BLOCKS_METADATA,
IndexShardGatewayService.INDEX_GATEWAY_SNAPSHOT_INTERVAL,
IndicesTTLService.INDEX_TTL_DISABLE_PURGE,
InternalIndexShard.INDEX_REFRESH_INTERVAL,
LocalGatewayAllocator.INDEX_RECOVERY_INITIAL_SHARDS,
LogByteSizeMergePolicyProvider.INDEX_MERGE_POLICY_MIN_MERGE_SIZE,
LogByteSizeMergePolicyProvider.INDEX_MERGE_POLICY_MAX_MERGE_SIZE,
LogByteSizeMergePolicyProvider.INDEX_MERGE_POLICY_MAX_MERGE_DOCS,
LogByteSizeMergePolicyProvider.INDEX_MERGE_POLICY_MERGE_FACTOR,
LogByteSizeMergePolicyProvider.INDEX_COMPOUND_FORMAT,
LogDocMergePolicyProvider.INDEX_MERGE_POLICY_MIN_MERGE_DOCS,
LogDocMergePolicyProvider.INDEX_MERGE_POLICY_MAX_MERGE_DOCS,
LogDocMergePolicyProvider.INDEX_MERGE_POLICY_MERGE_FACTOR,
LogDocMergePolicyProvider.INDEX_COMPOUND_FORMAT,
RobinEngine.INDEX_TERM_INDEX_INTERVAL,
RobinEngine.INDEX_TERM_INDEX_DIVISOR,
RobinEngine.INDEX_INDEX_CONCURRENCY,
RobinEngine.INDEX_GC_DELETES,
RobinEngine.INDEX_CODEC,
ShardSlowLogIndexingService.INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_WARN,
ShardSlowLogIndexingService.INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_INFO,
ShardSlowLogIndexingService.INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_DEBUG,
ShardSlowLogIndexingService.INDEX_INDEXING_SLOWLOG_THRESHOLD_INDEX_TRACE,
ShardSlowLogIndexingService.INDEX_INDEXING_SLOWLOG_REFORMAT,
ShardSlowLogIndexingService.INDEX_INDEXING_SLOWLOG_LEVEL,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_WARN,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_INFO,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_DEBUG,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_THRESHOLD_QUERY_TRACE,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_WARN,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_INFO,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_DEBUG,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_THRESHOLD_FETCH_TRACE,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_REFORMAT,
ShardSlowLogSearchService.INDEX_SEARCH_SLOWLOG_LEVEL,
ShardsLimitAllocationDecider.INDEX_TOTAL_SHARDS_PER_NODE,
TieredMergePolicyProvider.INDEX_MERGE_POLICY_EXPUNGE_DELETES_ALLOWED,
TieredMergePolicyProvider.INDEX_MERGE_POLICY_FLOOR_SEGMENT,
TieredMergePolicyProvider.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE,
TieredMergePolicyProvider.INDEX_MERGE_POLICY_MAX_MERGE_AT_ONCE_EXPLICIT,
TieredMergePolicyProvider.INDEX_MERGE_POLICY_MAX_MERGED_SEGMENT,
TieredMergePolicyProvider.INDEX_MERGE_POLICY_SEGMENTS_PER_TIER,
TieredMergePolicyProvider.INDEX_MERGE_POLICY_RECLAIM_DELETES_WEIGHT,
TieredMergePolicyProvider.INDEX_COMPOUND_FORMAT,
TranslogService.INDEX_TRANSLOG_FLUSH_THRESHOLD_OPS,
TranslogService.INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE,
TranslogService.INDEX_TRANSLOG_FLUSH_THRESHOLD_PERIOD,
TranslogService.INDEX_TRANSLOG_DISABLE_FLUSH);
}
public void addDynamicSetting(String... settings) {
indexDynamicSettings.addDynamicSettings(settings);
}
@Override
protected void configure() {
bind(DynamicSettings.class).annotatedWith(IndexDynamicSettings.class).toInstance(indexDynamicSettings);
}
}

3
src/main/java/org/elasticsearch/index/settings/IndexSettingsService.java
View File

@ -62,6 +62,9 @@ public class IndexSettingsService extends AbstractIndexComponent {
return this.settings;
}
/**
* Only settings registered in {@link IndexDynamicSettingsModule} can be changed dynamically.
*/
public void addListener(Listener listener) {
this.listeners.add(listener);
}

9
src/main/java/org/elasticsearch/index/shard/service/InternalIndexShard.java
View File

@ -28,7 +28,6 @@ import org.apache.lucene.util.ThreadInterruptedException;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.ElasticSearchIllegalArgumentException;
import org.elasticsearch.ElasticSearchIllegalStateException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.routing.ShardRouting;
import org.elasticsearch.common.Booleans;
import org.elasticsearch.common.Nullable;
@ -157,7 +156,7 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
this.shardWarmerService = shardWarmerService;
state = IndexShardState.CREATED;
this.refreshInterval = indexSettings.getAsTime("engine.robin.refresh_interval", indexSettings.getAsTime("index.refresh_interval", engine.defaultRefreshInterval()));
this.refreshInterval = indexSettings.getAsTime("engine.robin.refresh_interval", indexSettings.getAsTime(INDEX_REFRESH_INTERVAL, engine.defaultRefreshInterval()));
this.mergeInterval = indexSettings.getAsTime("index.merge.async_interval", TimeValue.timeValueSeconds(1));
indexSettingsService.addListener(applyRefreshSettings);
@ -686,9 +685,7 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
return query;
}
static {
IndexMetaData.addDynamicSettings("index.refresh_interval");
}
public static final String INDEX_REFRESH_INTERVAL = "index.refresh_interval";
private class ApplyRefreshSettings implements IndexSettingsService.Listener {
@Override
@ -697,7 +694,7 @@ public class InternalIndexShard extends AbstractIndexShardComponent implements I
if (state == IndexShardState.CLOSED) {
return;
}
TimeValue refreshInterval = settings.getAsTime("engine.robin.refresh_interval", settings.getAsTime("index.refresh_interval", InternalIndexShard.this.refreshInterval));
TimeValue refreshInterval = settings.getAsTime("engine.robin.refresh_interval", settings.getAsTime(INDEX_REFRESH_INTERVAL, InternalIndexShard.this.refreshInterval));
if (!refreshInterval.equals(InternalIndexShard.this.refreshInterval)) {
logger.info("updating refresh_interval from [{}] to [{}]", InternalIndexShard.this.refreshInterval, refreshInterval);
if (refreshScheduledFuture != null) {

17
src/main/java/org/elasticsearch/index/store/support/AbstractIndexStore.java
View File

@ -21,7 +21,6 @@ package org.elasticsearch.index.store.support;
import org.apache.lucene.store.StoreRateLimiting;
import org.elasticsearch.ElasticSearchException;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.index.AbstractIndexComponent;
@ -40,17 +39,13 @@ import java.io.IOException;
*/
public abstract class AbstractIndexStore extends AbstractIndexComponent implements IndexStore {
static {
IndexMetaData.addDynamicSettings(
"index.store.throttle.type",
"index.store.throttle.max_bytes_per_sec"
);
}
public static final String INDEX_STORE_THROTTLE_TYPE = "index.store.throttle.type";
public static final String INDEX_STORE_THROTTLE_MAX_BYTES_PER_SEC = "index.store.throttle.max_bytes_per_sec";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
String rateLimitingType = indexSettings.get("index.store.throttle.type", AbstractIndexStore.this.rateLimitingType);
String rateLimitingType = indexSettings.get(INDEX_STORE_THROTTLE_TYPE, AbstractIndexStore.this.rateLimitingType);
if (!rateLimitingType.equals(AbstractIndexStore.this.rateLimitingType)) {
logger.info("updating index.store.throttle.type from [{}] to [{}]", AbstractIndexStore.this.rateLimitingType, rateLimitingType);
if (rateLimitingType.equalsIgnoreCase("node")) {
@ -64,7 +59,7 @@ public abstract class AbstractIndexStore extends AbstractIndexComponent implemen
}
}
ByteSizeValue rateLimitingThrottle = settings.getAsBytesSize("index.store.throttle.max_bytes_per_sec", AbstractIndexStore.this.rateLimitingThrottle);
ByteSizeValue rateLimitingThrottle = settings.getAsBytesSize(INDEX_STORE_THROTTLE_MAX_BYTES_PER_SEC, AbstractIndexStore.this.rateLimitingThrottle);
if (!rateLimitingThrottle.equals(AbstractIndexStore.this.rateLimitingThrottle)) {
logger.info("updating index.store.throttle.max_bytes_per_sec from [{}] to [{}], note, type is [{}]", AbstractIndexStore.this.rateLimitingThrottle, rateLimitingThrottle, AbstractIndexStore.this.rateLimitingType);
AbstractIndexStore.this.rateLimitingThrottle = rateLimitingThrottle;
@ -91,14 +86,14 @@ public abstract class AbstractIndexStore extends AbstractIndexComponent implemen
this.indexService = indexService;
this.indicesStore = indicesStore;
this.rateLimitingType = indexSettings.get("index.store.throttle.type", "node");
this.rateLimitingType = indexSettings.get(INDEX_STORE_THROTTLE_TYPE, "node");
if (rateLimitingType.equalsIgnoreCase("node")) {
nodeRateLimiting = true;
} else {
nodeRateLimiting = false;
rateLimiting.setType(rateLimitingType);
}
this.rateLimitingThrottle = indexSettings.getAsBytesSize("index.store.throttle.max_bytes_per_sec", new ByteSizeValue(0));
this.rateLimitingThrottle = indexSettings.getAsBytesSize(INDEX_STORE_THROTTLE_MAX_BYTES_PER_SEC, new ByteSizeValue(0));
rateLimiting.setMaxRate(rateLimitingThrottle);
logger.debug("using index.store.throttle.type [{}], with index.store.throttle.max_bytes_per_sec [{}]", rateLimitingType, rateLimitingThrottle);

21
src/main/java/org/elasticsearch/index/translog/TranslogService.java
View File

@ -19,7 +19,6 @@
package org.elasticsearch.index.translog;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.unit.ByteSizeUnit;
@ -94,34 +93,30 @@ public class TranslogService extends AbstractIndexShardComponent {
this.future.cancel(true);
}
static {
IndexMetaData.addDynamicSettings(
"index.translog.flush_threshold_ops",
"index.translog.flush_threshold_size",
"index.translog.flush_threshold_period",
"index.translog.disable_flush"
);
}
public static final String INDEX_TRANSLOG_FLUSH_THRESHOLD_OPS = "index.translog.flush_threshold_ops";
public static final String INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE = "index.translog.flush_threshold_size";
public static final String INDEX_TRANSLOG_FLUSH_THRESHOLD_PERIOD = "index.translog.flush_threshold_period";
public static final String INDEX_TRANSLOG_DISABLE_FLUSH = "index.translog.disable_flush";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
int flushThresholdOperations = settings.getAsInt("index.translog.flush_threshold_ops", TranslogService.this.flushThresholdOperations);
int flushThresholdOperations = settings.getAsInt(INDEX_TRANSLOG_FLUSH_THRESHOLD_OPS, TranslogService.this.flushThresholdOperations);
if (flushThresholdOperations != TranslogService.this.flushThresholdOperations) {
logger.info("updating flush_threshold_ops from [{}] to [{}]", TranslogService.this.flushThresholdOperations, flushThresholdOperations);
TranslogService.this.flushThresholdOperations = flushThresholdOperations;
}
ByteSizeValue flushThresholdSize = settings.getAsBytesSize("index.translog.flush_threshold_size", TranslogService.this.flushThresholdSize);
ByteSizeValue flushThresholdSize = settings.getAsBytesSize(INDEX_TRANSLOG_FLUSH_THRESHOLD_SIZE, TranslogService.this.flushThresholdSize);
if (!flushThresholdSize.equals(TranslogService.this.flushThresholdSize)) {
logger.info("updating flush_threshold_size from [{}] to [{}]", TranslogService.this.flushThresholdSize, flushThresholdSize);
TranslogService.this.flushThresholdSize = flushThresholdSize;
}
TimeValue flushThresholdPeriod = settings.getAsTime("index.translog.flush_threshold_period", TranslogService.this.flushThresholdPeriod);
TimeValue flushThresholdPeriod = settings.getAsTime(INDEX_TRANSLOG_FLUSH_THRESHOLD_PERIOD, TranslogService.this.flushThresholdPeriod);
if (!flushThresholdPeriod.equals(TranslogService.this.flushThresholdPeriod)) {
logger.info("updating flush_threshold_period from [{}] to [{}]", TranslogService.this.flushThresholdPeriod, flushThresholdPeriod);
TranslogService.this.flushThresholdPeriod = flushThresholdPeriod;
}
boolean disableFlush = settings.getAsBoolean("index.translog.disable_flush", TranslogService.this.disableFlush);
boolean disableFlush = settings.getAsBoolean(INDEX_TRANSLOG_DISABLE_FLUSH, TranslogService.this.disableFlush);
if (disableFlush != TranslogService.this.disableFlush) {
logger.info("updating disable_flush from [{}] to [{}]", TranslogService.this.disableFlush, disableFlush);
TranslogService.this.disableFlush = disableFlush;

17
src/main/java/org/elasticsearch/index/translog/fs/FsTranslog.java
View File

@ -20,7 +20,6 @@
package org.elasticsearch.index.translog.fs;
import jsr166y.ThreadLocalRandom;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.io.FileSystemUtils;
import org.elasticsearch.common.io.stream.BytesStreamOutput;
@ -47,30 +46,26 @@ import java.util.concurrent.locks.ReentrantReadWriteLock;
*/
public class FsTranslog extends AbstractIndexShardComponent implements Translog {
static {
IndexMetaData.addDynamicSettings(
"index.translog.fs.type",
"index.translog.fs.buffer_size",
"index.translog.fs.transient_buffer_size"
);
}
public static final String INDEX_TRANSLOG_FS_TYPE = "index.translog.fs.type";
public static final String INDEX_TRANSLOG_FS_BUFFER_SIZE = "index.translog.fs.buffer_size";
public static final String INDEX_TRANSLOG_FS_TRANSIENT_BUFFER_SIZE = "index.translog.fs.transient_buffer_size";
class ApplySettings implements IndexSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
int bufferSize = (int) settings.getAsBytesSize("index.translog.fs.buffer_size", new ByteSizeValue(FsTranslog.this.bufferSize)).bytes();
int bufferSize = (int) settings.getAsBytesSize(INDEX_TRANSLOG_FS_BUFFER_SIZE, new ByteSizeValue(FsTranslog.this.bufferSize)).bytes();
if (bufferSize != FsTranslog.this.bufferSize) {
logger.info("updating buffer_size from [{}] to [{}]", new ByteSizeValue(FsTranslog.this.bufferSize), new ByteSizeValue(bufferSize));
FsTranslog.this.bufferSize = bufferSize;
}
int transientBufferSize = (int) settings.getAsBytesSize("index.translog.fs.transient_buffer_size", new ByteSizeValue(FsTranslog.this.transientBufferSize)).bytes();
int transientBufferSize = (int) settings.getAsBytesSize(INDEX_TRANSLOG_FS_TRANSIENT_BUFFER_SIZE, new ByteSizeValue(FsTranslog.this.transientBufferSize)).bytes();
if (transientBufferSize != FsTranslog.this.transientBufferSize) {
logger.info("updating transient_buffer_size from [{}] to [{}]", new ByteSizeValue(FsTranslog.this.transientBufferSize), new ByteSizeValue(transientBufferSize));
FsTranslog.this.transientBufferSize = transientBufferSize;
}
FsTranslogFile.Type type = FsTranslogFile.Type.fromString(settings.get("index.translog.fs.type", FsTranslog.this.type.name()));
FsTranslogFile.Type type = FsTranslogFile.Type.fromString(settings.get(INDEX_TRANSLOG_FS_TYPE, FsTranslog.this.type.name()));
if (type != FsTranslog.this.type) {
logger.info("updating type from [{}] to [{}]", FsTranslog.this.type, type);
FsTranslog.this.type = type;

13
src/main/java/org/elasticsearch/indices/cache/filter/IndicesFilterCache.java vendored
View File

@ -26,7 +26,6 @@ import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import gnu.trove.set.hash.THashSet;
import org.apache.lucene.search.DocIdSet;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.common.CacheRecycler;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
@ -60,24 +59,20 @@ public class IndicesFilterCache extends AbstractComponent implements RemovalList
private volatile boolean closed;
static {
MetaData.addDynamicSettings(
"indices.cache.filter.size",
"indices.cache.filter.expire"
);
}
public static final String INDICES_CACHE_FILTER_SIZE = "indices.cache.filter.size";
public static final String INDICES_CACHE_FILTER_EXPIRE = "indices.cache.filter.expire";
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
boolean replace = false;
String size = settings.get("indices.cache.filter.size", IndicesFilterCache.this.size);
String size = settings.get(INDICES_CACHE_FILTER_SIZE, IndicesFilterCache.this.size);
if (!size.equals(IndicesFilterCache.this.size)) {
logger.info("updating [indices.cache.filter.size] from [{}] to [{}]", IndicesFilterCache.this.size, size);
IndicesFilterCache.this.size = size;
replace = true;
}
TimeValue expire = settings.getAsTime("indices.cache.filter.expire", IndicesFilterCache.this.expire);
TimeValue expire = settings.getAsTime(INDICES_CACHE_FILTER_EXPIRE, IndicesFilterCache.this.expire);
if (!Objects.equal(expire, IndicesFilterCache.this.expire)) {
logger.info("updating [indices.cache.filter.expire] from [{}] to [{}]", IndicesFilterCache.this.expire, expire);
IndicesFilterCache.this.expire = expire;

28
src/main/java/org/elasticsearch/indices/recovery/RecoverySettings.java
View File

@ -20,9 +20,7 @@
package org.elasticsearch.indices.recovery;
import com.google.common.base.Objects;
import org.apache.lucene.store.RateLimiter;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
@ -38,14 +36,12 @@ import java.util.concurrent.TimeUnit;
*/
public class RecoverySettings extends AbstractComponent {
static {
MetaData.addDynamicSettings("indices.recovery.file_chunk_size");
MetaData.addDynamicSettings("indices.recovery.translog_ops");
MetaData.addDynamicSettings("indices.recovery.translog_size");
MetaData.addDynamicSettings("indices.recovery.compress");
MetaData.addDynamicSettings("indices.recovery.concurrent_streams");
MetaData.addDynamicSettings("indices.recovery.max_size_per_sec");
}
public static final String INDICES_RECOVERY_FILE_CHUNK_SIZE = "indices.recovery.file_chunk_size";
public static final String INDICES_RECOVERY_TRANSLOG_OPS = "indices.recovery.translog_ops";
public static final String INDICES_RECOVERY_TRANSLOG_SIZE = "indices.recovery.translog_size";
public static final String INDICES_RECOVERY_COMPRESS = "indices.recovery.compress";
public static final String INDICES_RECOVERY_CONCURRENT_STREAMS = "indices.recovery.concurrent_streams";
public static final String INDICES_RECOVERY_MAX_SIZE_PER_SEC = "indices.recovery.max_size_per_sec";
private volatile ByteSizeValue fileChunkSize;
@ -125,7 +121,7 @@ public class RecoverySettings extends AbstractComponent {
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
ByteSizeValue maxSizePerSec = settings.getAsBytesSize("indices.recovery.max_size_per_sec", RecoverySettings.this.maxSizePerSec);
ByteSizeValue maxSizePerSec = settings.getAsBytesSize(INDICES_RECOVERY_MAX_SIZE_PER_SEC, RecoverySettings.this.maxSizePerSec);
if (!Objects.equal(maxSizePerSec, RecoverySettings.this.maxSizePerSec)) {
logger.info("updating [indices.recovery.max_size_per_sec] from [{}] to [{}]", RecoverySettings.this.maxSizePerSec, maxSizePerSec);
RecoverySettings.this.maxSizePerSec = maxSizePerSec;
@ -138,31 +134,31 @@ public class RecoverySettings extends AbstractComponent {
}
}
ByteSizeValue fileChunkSize = settings.getAsBytesSize("indices.recovery.file_chunk_size", RecoverySettings.this.fileChunkSize);
ByteSizeValue fileChunkSize = settings.getAsBytesSize(INDICES_RECOVERY_FILE_CHUNK_SIZE, RecoverySettings.this.fileChunkSize);
if (!fileChunkSize.equals(RecoverySettings.this.fileChunkSize)) {
logger.info("updating [indices.recovery.file_chunk_size] from [{}] to [{}]", RecoverySettings.this.fileChunkSize, fileChunkSize);
RecoverySettings.this.fileChunkSize = fileChunkSize;
}
int translogOps = settings.getAsInt("indices.recovery.translog_ops", RecoverySettings.this.translogOps);
int translogOps = settings.getAsInt(INDICES_RECOVERY_TRANSLOG_OPS, RecoverySettings.this.translogOps);
if (translogOps != RecoverySettings.this.translogOps) {
logger.info("updating [indices.recovery.translog_ops] from [{}] to [{}]", RecoverySettings.this.translogOps, translogOps);
RecoverySettings.this.translogOps = translogOps;
}
ByteSizeValue translogSize = settings.getAsBytesSize("indices.recovery.translog_size", RecoverySettings.this.translogSize);
ByteSizeValue translogSize = settings.getAsBytesSize(INDICES_RECOVERY_TRANSLOG_SIZE, RecoverySettings.this.translogSize);
if (!translogSize.equals(RecoverySettings.this.translogSize)) {
logger.info("updating [indices.recovery.translog_size] from [{}] to [{}]", RecoverySettings.this.translogSize, translogSize);
RecoverySettings.this.translogSize = translogSize;
}
boolean compress = settings.getAsBoolean("indices.recovery.compress", RecoverySettings.this.compress);
boolean compress = settings.getAsBoolean(INDICES_RECOVERY_COMPRESS, RecoverySettings.this.compress);
if (compress != RecoverySettings.this.compress) {
logger.info("updating [indices.recovery.compress] from [{}] to [{}]", RecoverySettings.this.compress, compress);
RecoverySettings.this.compress = compress;
}
int concurrentStreams = settings.getAsInt("indices.recovery.concurrent_streams", RecoverySettings.this.concurrentStreams);
int concurrentStreams = settings.getAsInt(INDICES_RECOVERY_CONCURRENT_STREAMS, RecoverySettings.this.concurrentStreams);
if (concurrentStreams != RecoverySettings.this.concurrentStreams) {
logger.info("updating [indices.recovery.concurrent_streams] from [{}] to [{}]", RecoverySettings.this.concurrentStreams, concurrentStreams);
RecoverySettings.this.concurrentStreams = concurrentStreams;

13
src/main/java/org/elasticsearch/indices/store/IndicesStore.java
View File

@ -23,7 +23,6 @@ import org.apache.lucene.store.StoreRateLimiting;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterService;
import org.elasticsearch.cluster.ClusterStateListener;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.cluster.routing.IndexRoutingTable;
import org.elasticsearch.cluster.routing.IndexShardRoutingTable;
import org.elasticsearch.cluster.routing.ShardRouting;
@ -46,17 +45,13 @@ import java.io.File;
*/
public class IndicesStore extends AbstractComponent implements ClusterStateListener {
static {
MetaData.addDynamicSettings(
"indices.store.throttle.type",
"indices.store.throttle.max_bytes_per_sec"
);
}
public static final String INDICES_STORE_THROTTLE_TYPE = "indices.store.throttle.type";
public static final String INDICES_STORE_THROTTLE_MAX_BYTES_PER_SEC = "indices.store.throttle.max_bytes_per_sec";
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
String rateLimitingType = settings.get("indices.store.throttle.type", IndicesStore.this.rateLimitingType);
String rateLimitingType = settings.get(INDICES_STORE_THROTTLE_TYPE, IndicesStore.this.rateLimitingType);
// try and parse the type
StoreRateLimiting.Type.fromString(rateLimitingType);
if (!rateLimitingType.equals(IndicesStore.this.rateLimitingType)) {
@ -65,7 +60,7 @@ public class IndicesStore extends AbstractComponent implements ClusterStateListe
IndicesStore.this.rateLimiting.setType(rateLimitingType);
}
ByteSizeValue rateLimitingThrottle = settings.getAsBytesSize("indices.store.throttle.max_bytes_per_sec", IndicesStore.this.rateLimitingThrottle);
ByteSizeValue rateLimitingThrottle = settings.getAsBytesSize(INDICES_STORE_THROTTLE_MAX_BYTES_PER_SEC, IndicesStore.this.rateLimitingThrottle);
if (!rateLimitingThrottle.equals(IndicesStore.this.rateLimitingThrottle)) {
logger.info("updating indices.store.throttle.max_bytes_per_sec from [{}] to [{}], note, type is [{}]", IndicesStore.this.rateLimitingThrottle, rateLimitingThrottle, IndicesStore.this.rateLimitingType);
IndicesStore.this.rateLimitingThrottle = rateLimitingThrottle;

15
src/main/java/org/elasticsearch/indices/ttl/IndicesTTLService.java
View File

@ -64,15 +64,8 @@ import java.util.List;
*/
public class IndicesTTLService extends AbstractLifecycleComponent<IndicesTTLService> {
static {
MetaData.addDynamicSettings(
"indices.ttl.interval"
);
IndexMetaData.addDynamicSettings(
"index.ttl.disable_purge"
);
}
public static final String INDICES_TTL_INTERVAL = "indices.ttl.interval";
public static final String INDEX_TTL_DISABLE_PURGE = "index.ttl.disable_purge";
private final ClusterService clusterService;
private final IndicesService indicesService;
@ -153,7 +146,7 @@ public class IndicesTTLService extends AbstractLifecycleComponent<IndicesTTLServ
if (indexMetaData == null) {
continue;
}
boolean disablePurge = indexMetaData.settings().getAsBoolean("index.ttl.disable_purge", false);
boolean disablePurge = indexMetaData.settings().getAsBoolean(INDEX_TTL_DISABLE_PURGE, false);
if (disablePurge) {
continue;
}
@ -282,7 +275,7 @@ public class IndicesTTLService extends AbstractLifecycleComponent<IndicesTTLServ
class ApplySettings implements NodeSettingsService.Listener {
@Override
public void onRefreshSettings(Settings settings) {
TimeValue interval = settings.getAsTime("indices.ttl.interval", IndicesTTLService.this.interval);
TimeValue interval = settings.getAsTime(INDICES_TTL_INTERVAL, IndicesTTLService.this.interval);
if (!interval.equals(IndicesTTLService.this.interval)) {
logger.info("updating indices.ttl.interval from [{}] to [{}]", IndicesTTLService.this.interval, interval);
IndicesTTLService.this.interval = interval;

3
src/main/java/org/elasticsearch/node/settings/NodeSettingsService.java
View File

@ -106,6 +106,9 @@ public class NodeSettingsService extends AbstractComponent implements ClusterSta
globalSettings = lastSettingsApplied;
}
/**
* Only settings registered in {@link org.elasticsearch.cluster.settings.ClusterDynamicSettingsModule} can be changed dynamically.
*/
public void addListener(Listener listener) {
this.listeners.add(listener);
}

9
src/main/java/org/elasticsearch/threadpool/ThreadPool.java
View File

@ -24,7 +24,6 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
import com.google.common.util.concurrent.MoreExecutors;
import org.elasticsearch.ElasticSearchIllegalArgumentException;
import org.elasticsearch.cluster.metadata.MetaData;
import org.elasticsearch.common.Nullable;
import org.elasticsearch.common.component.AbstractComponent;
import org.elasticsearch.common.inject.Inject;
@ -75,11 +74,7 @@ public class ThreadPool extends AbstractComponent {
public static final String SNAPSHOT = "snapshot";
}
static {
MetaData.addDynamicSettings(
"threadpool.*"
);
}
public static final String THREADPOOL_GROUP = "threadpool.";
private volatile ImmutableMap<String, ExecutorHolder> executors;
@ -99,7 +94,7 @@ public class ThreadPool extends AbstractComponent {
public ThreadPool(Settings settings, @Nullable NodeSettingsService nodeSettingsService) {
super(settings);
Map<String, Settings> groupSettings = settings.getGroups("threadpool");
Map<String, Settings> groupSettings = settings.getGroups(THREADPOOL_GROUP);
defaultExecutorTypeSettings = ImmutableMap.<String, Settings>builder()
.put(Names.GENERIC, settingsBuilder().put("type", "cached").put("keep_alive", "30s").build())