Shard Allocation: Add node_initial_primaries_recoveries setting, closes #819.
This commit is contained in:
parent
f90b4e6fee
commit
14d98a7319
|
@ -31,13 +31,15 @@ import org.elasticsearch.common.settings.Settings;
|
|||
*/
|
||||
public class ThrottlingNodeAllocation extends NodeAllocation {
|
||||
|
||||
private final int primariesInitialRecoveries;
|
||||
private final int concurrentRecoveries;
|
||||
|
||||
@Inject public ThrottlingNodeAllocation(Settings settings) {
|
||||
super(settings);
|
||||
|
||||
this.primariesInitialRecoveries = componentSettings.getAsInt("node_initial_primaries_recoveries", 4);
|
||||
this.concurrentRecoveries = componentSettings.getAsInt("concurrent_recoveries", componentSettings.getAsInt("node_concurrent_recoveries", 2));
|
||||
logger.debug("using [concurrent_recoveries] with [{}]", concurrentRecoveries);
|
||||
logger.debug("using node_concurrent_recoveries [{}], node_initial_primaries_recoveries [{}]", concurrentRecoveries, primariesInitialRecoveries);
|
||||
}
|
||||
|
||||
@Override public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {
|
||||
|
@ -57,7 +59,7 @@ public class ThrottlingNodeAllocation extends NodeAllocation {
|
|||
primariesInRecovery++;
|
||||
}
|
||||
}
|
||||
if (primariesInRecovery >= concurrentRecoveries) {
|
||||
if (primariesInRecovery >= primariesInitialRecoveries) {
|
||||
return Decision.THROTTLE;
|
||||
} else {
|
||||
return Decision.YES;
|
||||
|
|
|
@ -166,7 +166,10 @@ public class FailedShardsRoutingTests {
|
|||
}
|
||||
|
||||
@Test public void test10ShardsWith1ReplicaFailure() {
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 10).build());
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
|
||||
.put("cluster.routing.allocation.node_concurrent_recoveries", 10)
|
||||
.put("cluster.routing.allocation.node_initial_primaries_recoveries", 10)
|
||||
.build());
|
||||
|
||||
logger.info("Building initial routing table");
|
||||
|
||||
|
|
|
@ -47,7 +47,10 @@ public class PrimaryNotRelocatedWhileBeingRecoveredTests {
|
|||
|
||||
|
||||
@Test public void testPrimaryNotRelocatedWhileBeingRecoveredFrom() {
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 10).build());
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
|
||||
.put("cluster.routing.allocation.node_concurrent_recoveries", 10)
|
||||
.put("cluster.routing.allocation.node_initial_primaries_recoveries", 10)
|
||||
.build());
|
||||
|
||||
logger.info("Building initial routing table");
|
||||
|
||||
|
|
|
@ -47,7 +47,8 @@ public class TenShardsOneReplicaRoutingTests {
|
|||
|
||||
@Test public void testSingleIndexFirstStartPrimaryThenBackups() {
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
|
||||
.put("cluster.routing.allocation.concurrent_recoveries", 10)
|
||||
.put("cluster.routing.allocation.node_concurrent_recoveries", 10)
|
||||
.put("cluster.routing.allocation.node_initial_primaries_recoveries", 10)
|
||||
.put("cluster.routing.allocation.allow_rebalance", "always")
|
||||
.put("cluster.routing.allocation.cluster_concurrent_rebalance", -1)
|
||||
.build());
|
||||
|
|
|
@ -45,7 +45,10 @@ public class ThrottlingAllocationTests {
|
|||
private final ESLogger logger = Loggers.getLogger(ThrottlingAllocationTests.class);
|
||||
|
||||
@Test public void testPrimaryRecoveryThrottling() {
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 3).build());
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
|
||||
.put("cluster.routing.allocation.node_concurrent_recoveries", 3)
|
||||
.put("cluster.routing.allocation.node_initial_primaries_recoveries", 3)
|
||||
.build());
|
||||
|
||||
logger.info("Building initial routing table");
|
||||
|
||||
|
@ -102,7 +105,10 @@ public class ThrottlingAllocationTests {
|
|||
}
|
||||
|
||||
@Test public void testReplicaAndPrimaryRecoveryThrottling() {
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder().put("cluster.routing.allocation.concurrent_recoveries", 3).build());
|
||||
ShardsAllocation strategy = new ShardsAllocation(settingsBuilder()
|
||||
.put("cluster.routing.allocation.concurrent_recoveries", 3)
|
||||
.put("cluster.routing.allocation.node_initial_primaries_recoveries", 3)
|
||||
.build());
|
||||
|
||||
logger.info("Building initial routing table");
|
||||
|
||||
|
|
Loading…
Reference in New Issue