[ALLOCATION] Remove primary balance factor
The `cluster.routing.allocation.balance.primary` setting has caused a lot of confusion in the past while it has very little benefit form a shard allocatioon point of view. Users tend to modify this value to evently distribute primaries across the nodes which is dangerous since a prmiary flag on it's own can trigger relocations. The primary flag for a shard is should not have any impact on cluster performance unless the high level feature suffereing from primary hotspots is buggy. Yet, this setting was intended to be a tie-breaker which is not necessary anymore since the algorithm is deterministic. This commit removes this setting entriely.
This commit is contained in:
Simon Willnauer
parent
8948b489d6
commit
236e2491b4
|
|
@ -82,11 +82,6 @@ due to forced awareness or allocation filtering.
|
|||
tendency to equalize the number of shards per index across all nodes in
|
||||
the cluster.
|
||||
|
||||
`cluster.routing.allocation.balance.primary`::
|
||||
Defines a weight factor for the number of primaries of a specific index
|
||||
allocated on a node (float). `0.00f`. Raising this raises the tendency
|
||||
to equalize the number of primary shards across all nodes in the cluster. deprecated[1.3.8]
|
||||
|
||||
`cluster.routing.allocation.balance.threshold`::
|
||||
Minimal optimization value of operations that should be performed (non
|
||||
negative float). Defaults to `1.0f`. Raising this will cause the cluster
|
||||
|
|
|
|||
|
|
@ -55,8 +55,6 @@ import static org.elasticsearch.cluster.routing.ShardRoutingState.RELOCATING;
|
|||
* 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>
|
||||
|
|
@ -69,37 +67,25 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
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";
|
||||
|
||||
private static final float DEFAULT_INDEX_BALANCE_FACTOR = 0.55f;
|
||||
private static final float DEFAULT_SHARD_BALANCE_FACTOR = 0.45f;
|
||||
/**
|
||||
* The primary balance factor was introduces as a tie-breaker to make the initial allocation
|
||||
* more deterministic. Yet other mechanism have been added ensure that the algorithm is more deterministic such that this
|
||||
* setting is not needed anymore. Additionally, this setting was abused to balance shards based on their primary flag which can lead
|
||||
* to unexpected behavior when allocating or balancing the shards.
|
||||
*
|
||||
* @deprecated the threshold primary balance factor is deprecated and should not be used.
|
||||
*/
|
||||
@Deprecated
|
||||
private static final float DEFAULT_PRIMARY_BALANCE_FACTOR = 0.0f;
|
||||
|
||||
class ApplySettings implements NodeSettingsService.Listener {
|
||||
@Override
|
||||
public void onRefreshSettings(Settings settings) {
|
||||
final float indexBalance = settings.getAsFloat(SETTING_INDEX_BALANCE_FACTOR, weightFunction.indexBalance);
|
||||
final float shardBalance = settings.getAsFloat(SETTING_SHARD_BALANCE_FACTOR, weightFunction.shardBalance);
|
||||
final float primaryBalance = settings.getAsFloat(SETTING_PRIMARY_BALANCE_FACTOR, weightFunction.primaryBalance);
|
||||
float threshold = settings.getAsFloat(SETTING_THRESHOLD, BalancedShardsAllocator.this.threshold);
|
||||
if (threshold <= 0.0f) {
|
||||
throw new ElasticsearchIllegalArgumentException("threshold must be greater than 0.0f but was: " + threshold);
|
||||
}
|
||||
BalancedShardsAllocator.this.threshold = threshold;
|
||||
BalancedShardsAllocator.this.weightFunction = new WeightFunction(indexBalance, shardBalance, primaryBalance);
|
||||
BalancedShardsAllocator.this.weightFunction = new WeightFunction(indexBalance, shardBalance);
|
||||
}
|
||||
}
|
||||
|
||||
private volatile WeightFunction weightFunction = new WeightFunction(DEFAULT_INDEX_BALANCE_FACTOR, DEFAULT_SHARD_BALANCE_FACTOR, DEFAULT_PRIMARY_BALANCE_FACTOR);
|
||||
private volatile WeightFunction weightFunction = new WeightFunction(DEFAULT_INDEX_BALANCE_FACTOR, DEFAULT_SHARD_BALANCE_FACTOR);
|
||||
|
||||
private volatile float threshold = 1.0f;
|
||||
|
||||
|
|
@ -153,13 +139,6 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
return weightFunction.indexBalance;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the primary related weight factor.
|
||||
*/
|
||||
public float getPrimaryBalance() {
|
||||
return weightFunction.primaryBalance;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the shard related weight factor.
|
||||
*/
|
||||
|
|
@ -174,11 +153,10 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
* <ul>
|
||||
* <li><code>index balance</code> - balance property over shards per index</li>
|
||||
* <li><code>shard balance</code> - balance property over shards per cluster</li>
|
||||
* <li><code>primary balance</code> - balance property over primaries per cluster</li>
|
||||
* </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
|
||||
* weight function. For example if the weight function should calculate the weights only based on a global (shard) balance the index 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:
|
||||
|
|
@ -189,35 +167,31 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
* <li>
|
||||
* <code>weight<sub>node</sub>(node, index) = shardBalance * (node.numShards() - avgShardsPerNode)</code>
|
||||
* </li>
|
||||
* <li>
|
||||
* <code>weight<sub>primary</sub>(node, index) = primaryBalance * (node.numPrimaries() - avgPrimariesPerNode)</code>
|
||||
* </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>
|
||||
* <code>weight(node, index) = weight<sub>index</sub>(node, index) + weight<sub>node</sub>(node, index)</code>
|
||||
*/
|
||||
public static class WeightFunction {
|
||||
|
||||
private final float indexBalance;
|
||||
private final float shardBalance;
|
||||
private final float primaryBalance;
|
||||
private final float[] theta;
|
||||
|
||||
public WeightFunction(float indexBalance, float shardBalance, float primaryBalance) {
|
||||
float sum = indexBalance + shardBalance + primaryBalance;
|
||||
|
||||
public WeightFunction(float indexBalance, float shardBalance) {
|
||||
float sum = indexBalance + shardBalance;
|
||||
if (sum <= 0.0f) {
|
||||
throw new ElasticsearchIllegalArgumentException("Balance factors must sum to a value > 0 but was: " + sum);
|
||||
}
|
||||
theta = new float[]{shardBalance / sum, indexBalance / sum, primaryBalance / sum};
|
||||
theta = new float[]{shardBalance / sum, indexBalance / sum};
|
||||
this.indexBalance = indexBalance;
|
||||
this.shardBalance = shardBalance;
|
||||
this.primaryBalance = primaryBalance;
|
||||
}
|
||||
|
||||
public float weight(Operation operation, Balancer balancer, ModelNode node, String index) {
|
||||
final float weightShard = node.numShards() - balancer.avgShardsPerNode();
|
||||
final float weightIndex = node.numShards(index) - balancer.avgShardsPerNode(index);
|
||||
final float weightPrimary = node.numPrimaries() - balancer.avgPrimariesPerNode();
|
||||
return theta[0] * weightShard + theta[1] * weightIndex + theta[2] * weightPrimary;
|
||||
final float weightShard = (node.numShards() - balancer.avgShardsPerNode());
|
||||
final float weightIndex = (node.numShards(index) - balancer.avgShardsPerNode(index));
|
||||
assert theta != null;
|
||||
return theta[0] * weightShard + theta[1] * weightIndex;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -245,7 +219,6 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
* A {@link Balancer}
|
||||
*/
|
||||
public static class Balancer {
|
||||
|
||||
private final ESLogger logger;
|
||||
private final Map<String, ModelNode> nodes = new HashMap<>();
|
||||
private final HashSet<String> indices = new HashSet<>();
|
||||
|
|
@ -304,12 +277,6 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
return ((float) metaData.numberOfShards()) / nodes.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the average of primaries per node for the given index
|
||||
*/
|
||||
public float avgPrimariesPerNode(String index) {
|
||||
return ((float) metaData.index(index).numberOfShards()) / nodes.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns a new {@link NodeSorter} that sorts the nodes based on their
|
||||
|
|
@ -380,7 +347,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
final ModelNode maxNode = modelNodes[highIdx];
|
||||
advance_range:
|
||||
if (maxNode.numShards(index) > 0) {
|
||||
float delta = absDelta(weights[lowIdx], weights[highIdx]);
|
||||
final float delta = absDelta(weights[lowIdx], weights[highIdx]);
|
||||
if (lessThan(delta, threshold)) {
|
||||
if (lowIdx > 0 && highIdx-1 > 0 // is there a chance for a higher delta?
|
||||
&& (absDelta(weights[0], weights[highIdx-1]) > threshold) // check if we need to break at all
|
||||
|
|
@ -845,7 +812,6 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
private final Map<String, ModelIndex> indices = new HashMap<>();
|
||||
/* cached stats - invalidated on add/remove and lazily calculated */
|
||||
private int numShards = -1;
|
||||
private int numPrimaries = -1;
|
||||
|
||||
public ModelNode(String id) {
|
||||
this.id = id;
|
||||
|
|
@ -875,22 +841,6 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
return index == null ? 0 : index.numShards();
|
||||
}
|
||||
|
||||
public int numPrimaries(String idx) {
|
||||
ModelIndex index = indices.get(idx);
|
||||
return index == null ? 0 : index.numPrimaries();
|
||||
}
|
||||
|
||||
public int numPrimaries() {
|
||||
if (numPrimaries == -1) {
|
||||
int sum = 0;
|
||||
for (ModelIndex index : indices.values()) {
|
||||
sum += index.numPrimaries();
|
||||
}
|
||||
numPrimaries = sum;
|
||||
}
|
||||
return numPrimaries;
|
||||
}
|
||||
|
||||
public Collection<MutableShardRouting> shards() {
|
||||
Collection<MutableShardRouting> result = new ArrayList<>();
|
||||
for (ModelIndex index : indices.values()) {
|
||||
|
|
@ -908,7 +858,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
}
|
||||
|
||||
public void addShard(MutableShardRouting shard, Decision decision) {
|
||||
numPrimaries = numShards = -1;
|
||||
numShards = -1;
|
||||
ModelIndex index = indices.get(shard.index());
|
||||
if (index == null) {
|
||||
index = new ModelIndex(shard.index());
|
||||
|
|
@ -918,7 +868,7 @@ public class BalancedShardsAllocator extends AbstractComponent implements Shards
|
|||
}
|
||||
|
||||
public Decision removeShard(MutableShardRouting shard) {
|
||||
numPrimaries = numShards = -1;
|
||||
numShards = -1;
|
||||
ModelIndex index = indices.get(shard.index());
|
||||
Decision removed = null;
|
||||
if (index != null) {
|
||||
|
|
|
|||
|
|
@ -47,7 +47,6 @@ public class ClusterDynamicSettingsModule extends AbstractModule {
|
|||
clusterDynamicSettings.addDynamicSetting(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_ATTRIBUTES);
|
||||
clusterDynamicSettings.addDynamicSetting(AwarenessAllocationDecider.CLUSTER_ROUTING_ALLOCATION_AWARENESS_FORCE_GROUP + "*");
|
||||
clusterDynamicSettings.addDynamicSetting(BalancedShardsAllocator.SETTING_INDEX_BALANCE_FACTOR, Validator.FLOAT);
|
||||
clusterDynamicSettings.addDynamicSetting(BalancedShardsAllocator.SETTING_PRIMARY_BALANCE_FACTOR, Validator.FLOAT);
|
||||
clusterDynamicSettings.addDynamicSetting(BalancedShardsAllocator.SETTING_SHARD_BALANCE_FACTOR, Validator.FLOAT);
|
||||
clusterDynamicSettings.addDynamicSetting(BalancedShardsAllocator.SETTING_THRESHOLD, Validator.NON_NEGATIVE_FLOAT);
|
||||
clusterDynamicSettings.addDynamicSetting(ClusterRebalanceAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE,
|
||||
|
|
|
|||
|
|
@ -59,14 +59,12 @@ public class BalanceConfigurationTests extends ElasticsearchAllocationTestCase {
|
|||
/* Tests balance over indices only */
|
||||
final float indexBalance = 1.0f;
|
||||
final float replicaBalance = 0.0f;
|
||||
final float primaryBalance = 0.0f;
|
||||
final float balanceTreshold = 1.0f;
|
||||
|
||||
ImmutableSettings.Builder settings = settingsBuilder();
|
||||
settings.put(ClusterRebalanceAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE, ClusterRebalanceAllocationDecider.ClusterRebalanceType.ALWAYS.toString());
|
||||
settings.put(BalancedShardsAllocator.SETTING_INDEX_BALANCE_FACTOR, indexBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_SHARD_BALANCE_FACTOR, replicaBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_PRIMARY_BALANCE_FACTOR, primaryBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_THRESHOLD, balanceTreshold);
|
||||
|
||||
AllocationService strategy = createAllocationService(settings.build());
|
||||
|
|
@ -87,14 +85,12 @@ public class BalanceConfigurationTests extends ElasticsearchAllocationTestCase {
|
|||
/* Tests balance over replicas only */
|
||||
final float indexBalance = 0.0f;
|
||||
final float replicaBalance = 1.0f;
|
||||
final float primaryBalance = 0.0f;
|
||||
final float balanceTreshold = 1.0f;
|
||||
|
||||
ImmutableSettings.Builder settings = settingsBuilder();
|
||||
settings.put(ClusterRebalanceAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE, ClusterRebalanceAllocationDecider.ClusterRebalanceType.ALWAYS.toString());
|
||||
settings.put(BalancedShardsAllocator.SETTING_INDEX_BALANCE_FACTOR, indexBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_SHARD_BALANCE_FACTOR, replicaBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_PRIMARY_BALANCE_FACTOR, primaryBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_THRESHOLD, balanceTreshold);
|
||||
|
||||
AllocationService strategy = createAllocationService(settings.build());
|
||||
|
|
@ -110,33 +106,6 @@ public class BalanceConfigurationTests extends ElasticsearchAllocationTestCase {
|
|||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPrimaryBalance() {
|
||||
/* Tests balance over primaries only */
|
||||
final float indexBalance = 0.0f;
|
||||
final float replicaBalance = 0.0f;
|
||||
final float primaryBalance = 1.0f;
|
||||
final float balanceTreshold = 1.0f;
|
||||
|
||||
ImmutableSettings.Builder settings = settingsBuilder();
|
||||
settings.put(ClusterRebalanceAllocationDecider.CLUSTER_ROUTING_ALLOCATION_ALLOW_REBALANCE, ClusterRebalanceAllocationDecider.ClusterRebalanceType.ALWAYS.toString());
|
||||
settings.put(BalancedShardsAllocator.SETTING_INDEX_BALANCE_FACTOR, indexBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_SHARD_BALANCE_FACTOR, replicaBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_PRIMARY_BALANCE_FACTOR, primaryBalance);
|
||||
settings.put(BalancedShardsAllocator.SETTING_THRESHOLD, balanceTreshold);
|
||||
|
||||
AllocationService strategy = createAllocationService(settings.build());
|
||||
|
||||
ClusterState clusterstate = initCluster(strategy);
|
||||
assertPrimaryBalance(logger, clusterstate.getRoutingNodes(), numberOfNodes, numberOfIndices, numberOfReplicas, numberOfShards, balanceTreshold);
|
||||
|
||||
clusterstate = addNode(clusterstate, strategy);
|
||||
assertPrimaryBalance(logger, clusterstate.getRoutingNodes(), numberOfNodes + 1, numberOfIndices, numberOfReplicas, numberOfShards, balanceTreshold);
|
||||
|
||||
clusterstate = removeNodes(clusterstate, strategy);
|
||||
assertPrimaryBalance(logger, clusterstate.getRoutingNodes(), numberOfNodes + 1 - (numberOfNodes + 1) / 2, numberOfIndices, numberOfReplicas, numberOfShards, balanceTreshold);
|
||||
}
|
||||
|
||||
private ClusterState initCluster(AllocationService strategy) {
|
||||
MetaData.Builder metaDataBuilder = MetaData.builder();
|
||||
RoutingTable.Builder routingTableBuilder = RoutingTable.builder();
|
||||
|
|
@ -311,7 +280,6 @@ public class BalanceConfigurationTests extends ElasticsearchAllocationTestCase {
|
|||
ImmutableSettings.Builder settings = settingsBuilder();
|
||||
settings.put(BalancedShardsAllocator.SETTING_INDEX_BALANCE_FACTOR, 0.2);
|
||||
settings.put(BalancedShardsAllocator.SETTING_SHARD_BALANCE_FACTOR, 0.3);
|
||||
settings.put(BalancedShardsAllocator.SETTING_PRIMARY_BALANCE_FACTOR, 0.5);
|
||||
settings.put(BalancedShardsAllocator.SETTING_THRESHOLD, 2.0);
|
||||
final NodeSettingsService.Listener[] listeners = new NodeSettingsService.Listener[1];
|
||||
NodeSettingsService service = new NodeSettingsService(settingsBuilder().build()) {
|
||||
|
|
@ -326,7 +294,6 @@ public class BalanceConfigurationTests extends ElasticsearchAllocationTestCase {
|
|||
BalancedShardsAllocator allocator = new BalancedShardsAllocator(settings.build(), service);
|
||||
assertThat(allocator.getIndexBalance(), Matchers.equalTo(0.2f));
|
||||
assertThat(allocator.getShardBalance(), Matchers.equalTo(0.3f));
|
||||
assertThat(allocator.getPrimaryBalance(), Matchers.equalTo(0.5f));
|
||||
assertThat(allocator.getThreshold(), Matchers.equalTo(2.0f));
|
||||
|
||||
settings = settingsBuilder();
|
||||
|
|
@ -334,18 +301,15 @@ public class BalanceConfigurationTests extends ElasticsearchAllocationTestCase {
|
|||
listeners[0].onRefreshSettings(settings.build());
|
||||
assertThat(allocator.getIndexBalance(), Matchers.equalTo(0.2f));
|
||||
assertThat(allocator.getShardBalance(), Matchers.equalTo(0.3f));
|
||||
assertThat(allocator.getPrimaryBalance(), Matchers.equalTo(0.5f));
|
||||
assertThat(allocator.getThreshold(), Matchers.equalTo(2.0f));
|
||||
|
||||
settings = settingsBuilder();
|
||||
settings.put(BalancedShardsAllocator.SETTING_INDEX_BALANCE_FACTOR, 0.5);
|
||||
settings.put(BalancedShardsAllocator.SETTING_SHARD_BALANCE_FACTOR, 0.1);
|
||||
settings.put(BalancedShardsAllocator.SETTING_PRIMARY_BALANCE_FACTOR, 0.4);
|
||||
settings.put(BalancedShardsAllocator.SETTING_THRESHOLD, 3.0);
|
||||
listeners[0].onRefreshSettings(settings.build());
|
||||
assertThat(allocator.getIndexBalance(), Matchers.equalTo(0.5f));
|
||||
assertThat(allocator.getShardBalance(), Matchers.equalTo(0.1f));
|
||||
assertThat(allocator.getPrimaryBalance(), Matchers.equalTo(0.4f));
|
||||
assertThat(allocator.getThreshold(), Matchers.equalTo(3.0f));
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue