Skip to content

Commit bb94e84

Browse files
imRishNvinaykpud
imRishN
authored and
vinaykpud
committed
Change priority for scheduling reroute during timeout (opensearch-project#16445)
* Change priority for scheduling reroute in timeout Signed-off-by: Rishab Nahata <[email protected]> * Add setting for ESA Signed-off-by: Rishab Nahata <[email protected]> * Fix tests Signed-off-by: Rishab Nahata <[email protected]> * Trigger Build Signed-off-by: Rishab Nahata <[email protected]> * Trigger Build Signed-off-by: Rishab Nahata <[email protected]> * Add test Signed-off-by: Rishab Nahata <[email protected]> * Trigger Build Signed-off-by: Rishab Nahata <[email protected]> * Add changelog Signed-off-by: Rishab Nahata <[email protected]> * Trigger Build Signed-off-by: Rishab Nahata <[email protected]> * Trigger Build Signed-off-by: Rishab Nahata <[email protected]> * Trigger Build Signed-off-by: Rishab Nahata <[email protected]> --------- Signed-off-by: Rishab Nahata <[email protected]>
1 parent d70ed60 commit bb94e84

File tree

6 files changed

+248
-24
lines changed

6 files changed

+248
-24
lines changed

‎CHANGELOG.md

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
55

66
## [Unreleased 2.x]
77
### Added
8+
- Change priority for scheduling reroute during timeout([#16445](https://github.com/opensearch-project/OpenSearch/pull/16445))
89
- Renaming the node role search to warm ([#17573](https://github.com/opensearch-project/OpenSearch/pull/17573))
910
- Search Replica Allocation and Recovery ([#17457](https://github.com/opensearch-project/OpenSearch/pull/17457))
1011

‎server/src/main/java/org/opensearch/cluster/routing/allocation/allocator/BalancedShardsAllocator.java

Lines changed: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -62,9 +62,11 @@
6262
import java.util.HashMap;
6363
import java.util.HashSet;
6464
import java.util.Iterator;
65+
import java.util.Locale;
6566
import java.util.Map;
6667
import java.util.Set;
6768

69+
import static org.opensearch.cluster.action.shard.ShardStateAction.FOLLOW_UP_REROUTE_PRIORITY_SETTING;
6870
import static org.opensearch.cluster.routing.allocation.ConstraintTypes.CLUSTER_PRIMARY_SHARD_BALANCE_CONSTRAINT_ID;
6971
import static org.opensearch.cluster.routing.allocation.ConstraintTypes.CLUSTER_PRIMARY_SHARD_REBALANCE_CONSTRAINT_ID;
7072
import static org.opensearch.cluster.routing.allocation.ConstraintTypes.INDEX_PRIMARY_SHARD_BALANCE_CONSTRAINT_ID;
@@ -199,6 +201,32 @@ public class BalancedShardsAllocator implements ShardsAllocator {
199201
Setting.Property.Dynamic
200202
);
201203

204+
/**
205+
* Adjusts the priority of the followup reroute task when current round times out. NORMAL is right for reasonable clusters,
206+
* but for a cluster in a messed up state which is starving NORMAL priority tasks, it might be necessary to raise this higher
207+
* to allocate shards.
208+
*/
209+
public static final Setting<Priority> FOLLOW_UP_REROUTE_PRIORITY_SETTING = new Setting<>(
210+
"cluster.routing.allocation.balanced_shards_allocator.schedule_reroute.priority",
211+
Priority.NORMAL.toString(),
212+
BalancedShardsAllocator::parseReroutePriority,
213+
Setting.Property.NodeScope,
214+
Setting.Property.Dynamic
215+
);
216+
217+
private static Priority parseReroutePriority(String priorityString) {
218+
final Priority priority = Priority.valueOf(priorityString.toUpperCase(Locale.ROOT));
219+
switch (priority) {
220+
case NORMAL:
221+
case HIGH:
222+
case URGENT:
223+
return priority;
224+
}
225+
throw new IllegalArgumentException(
226+
"priority [" + priority + "] not supported for [" + FOLLOW_UP_REROUTE_PRIORITY_SETTING.getKey() + "]"
227+
);
228+
}
229+
202230
private volatile boolean movePrimaryFirst;
203231
private volatile ShardMovementStrategy shardMovementStrategy;
204232

@@ -213,6 +241,7 @@ public class BalancedShardsAllocator implements ShardsAllocator {
213241

214242
private volatile boolean ignoreThrottleInRestore;
215243
private volatile TimeValue allocatorTimeout;
244+
private volatile Priority followUpRerouteTaskPriority;
216245
private long startTime;
217246
private RerouteService rerouteService;
218247

@@ -233,6 +262,7 @@ public BalancedShardsAllocator(Settings settings, ClusterSettings clusterSetting
233262
setPreferPrimaryShardRebalance(PREFER_PRIMARY_SHARD_REBALANCE.get(settings));
234263
setShardMovementStrategy(SHARD_MOVEMENT_STRATEGY_SETTING.get(settings));
235264
setAllocatorTimeout(ALLOCATOR_TIMEOUT_SETTING.get(settings));
265+
setFollowUpRerouteTaskPriority(FOLLOW_UP_REROUTE_PRIORITY_SETTING.get(settings));
236266
clusterSettings.addSettingsUpdateConsumer(PREFER_PRIMARY_SHARD_BALANCE, this::setPreferPrimaryShardBalance);
237267
clusterSettings.addSettingsUpdateConsumer(SHARD_MOVE_PRIMARY_FIRST_SETTING, this::setMovePrimaryFirst);
238268
clusterSettings.addSettingsUpdateConsumer(SHARD_MOVEMENT_STRATEGY_SETTING, this::setShardMovementStrategy);
@@ -244,6 +274,7 @@ public BalancedShardsAllocator(Settings settings, ClusterSettings clusterSetting
244274
clusterSettings.addSettingsUpdateConsumer(PRIMARY_CONSTRAINT_THRESHOLD_SETTING, this::setPrimaryConstraintThresholdSetting);
245275
clusterSettings.addSettingsUpdateConsumer(IGNORE_THROTTLE_FOR_REMOTE_RESTORE, this::setIgnoreThrottleInRestore);
246276
clusterSettings.addSettingsUpdateConsumer(ALLOCATOR_TIMEOUT_SETTING, this::setAllocatorTimeout);
277+
clusterSettings.addSettingsUpdateConsumer(FOLLOW_UP_REROUTE_PRIORITY_SETTING, this::setFollowUpRerouteTaskPriority);
247278
}
248279

249280
@Override
@@ -342,6 +373,10 @@ private void setAllocatorTimeout(TimeValue allocatorTimeout) {
342373
this.allocatorTimeout = allocatorTimeout;
343374
}
344375

376+
private void setFollowUpRerouteTaskPriority(Priority followUpRerouteTaskPriority) {
377+
this.followUpRerouteTaskPriority = followUpRerouteTaskPriority;
378+
}
379+
345380
protected boolean allocatorTimedOut() {
346381
if (allocatorTimeout.equals(TimeValue.MINUS_ONE)) {
347382
if (logger.isTraceEnabled()) {
@@ -438,10 +473,13 @@ private void failAllocationOfNewPrimaries(RoutingAllocation allocation) {
438473

439474
private void scheduleRerouteIfAllocatorTimedOut() {
440475
if (allocatorTimedOut()) {
441-
assert rerouteService != null : "RerouteService not set to schedule reroute after allocator time out";
476+
if (rerouteService == null) {
477+
logger.info("RerouteService not set to schedule reroute after allocator time out");
478+
return;
479+
}
442480
rerouteService.reroute(
443481
"reroute after balanced shards allocator timed out",
444-
Priority.HIGH,
482+
followUpRerouteTaskPriority,
445483
ActionListener.wrap(
446484
r -> logger.trace("reroute after balanced shards allocator timed out completed"),
447485
e -> logger.debug("reroute after balanced shards allocator timed out failed", e)

‎server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -276,6 +276,7 @@ public void apply(Settings value, Settings current, Settings previous) {
276276
BalancedShardsAllocator.THRESHOLD_SETTING,
277277
BalancedShardsAllocator.IGNORE_THROTTLE_FOR_REMOTE_RESTORE,
278278
BalancedShardsAllocator.ALLOCATOR_TIMEOUT_SETTING,
279+
BalancedShardsAllocator.FOLLOW_UP_REROUTE_PRIORITY_SETTING,
279280
BalancedShardsAllocator.PRIMARY_CONSTRAINT_THRESHOLD_SETTING,
280281
BreakerSettings.CIRCUIT_BREAKER_LIMIT_SETTING,
281282
BreakerSettings.CIRCUIT_BREAKER_OVERHEAD_SETTING,
@@ -350,6 +351,7 @@ public void apply(Settings value, Settings current, Settings previous) {
350351
ShardsBatchGatewayAllocator.GATEWAY_ALLOCATOR_BATCH_SIZE,
351352
ShardsBatchGatewayAllocator.PRIMARY_BATCH_ALLOCATOR_TIMEOUT_SETTING,
352353
ShardsBatchGatewayAllocator.REPLICA_BATCH_ALLOCATOR_TIMEOUT_SETTING,
354+
ShardsBatchGatewayAllocator.FOLLOW_UP_REROUTE_PRIORITY_SETTING,
353355
PersistedClusterStateService.SLOW_WRITE_LOGGING_THRESHOLD,
354356
NetworkModule.HTTP_DEFAULT_TYPE_SETTING,
355357
NetworkModule.TRANSPORT_DEFAULT_TYPE_SETTING,

‎server/src/main/java/org/opensearch/gateway/ShardsBatchGatewayAllocator.java

Lines changed: 38 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@
5353
import java.util.HashSet;
5454
import java.util.Iterator;
5555
import java.util.List;
56+
import java.util.Locale;
5657
import java.util.Map;
5758
import java.util.Objects;
5859
import java.util.Set;
@@ -82,6 +83,7 @@ public class ShardsBatchGatewayAllocator implements ExistingShardsAllocator {
8283

8384
private TimeValue primaryShardsBatchGatewayAllocatorTimeout;
8485
private TimeValue replicaShardsBatchGatewayAllocatorTimeout;
86+
private volatile Priority followUpRerouteTaskPriority;
8587
public static final TimeValue MIN_ALLOCATOR_TIMEOUT = TimeValue.timeValueSeconds(20);
8688
private final ClusterManagerMetrics clusterManagerMetrics;
8789

@@ -145,6 +147,32 @@ public void validate(TimeValue timeValue) {
145147
Setting.Property.Dynamic
146148
);
147149

150+
/**
151+
* Adjusts the priority of the followup reroute task when current round times out. NORMAL is right for reasonable clusters,
152+
* but for a cluster in a messed up state which is starving NORMAL priority tasks, it might be necessary to raise this higher
153+
* to allocate existing shards.
154+
*/
155+
public static final Setting<Priority> FOLLOW_UP_REROUTE_PRIORITY_SETTING = new Setting<>(
156+
"cluster.routing.allocation.shards_batch_gateway_allocator.schedule_reroute.priority",
157+
Priority.NORMAL.toString(),
158+
ShardsBatchGatewayAllocator::parseReroutePriority,
159+
Setting.Property.NodeScope,
160+
Setting.Property.Dynamic
161+
);
162+
163+
private static Priority parseReroutePriority(String priorityString) {
164+
final Priority priority = Priority.valueOf(priorityString.toUpperCase(Locale.ROOT));
165+
switch (priority) {
166+
case NORMAL:
167+
case HIGH:
168+
case URGENT:
169+
return priority;
170+
}
171+
throw new IllegalArgumentException(
172+
"priority [" + priority + "] not supported for [" + FOLLOW_UP_REROUTE_PRIORITY_SETTING.getKey() + "]"
173+
);
174+
}
175+
148176
private final RerouteService rerouteService;
149177
private final PrimaryShardBatchAllocator primaryShardBatchAllocator;
150178
private final ReplicaShardBatchAllocator replicaShardBatchAllocator;
@@ -179,6 +207,8 @@ public ShardsBatchGatewayAllocator(
179207
this.replicaShardsBatchGatewayAllocatorTimeout = REPLICA_BATCH_ALLOCATOR_TIMEOUT_SETTING.get(settings);
180208
clusterSettings.addSettingsUpdateConsumer(REPLICA_BATCH_ALLOCATOR_TIMEOUT_SETTING, this::setReplicaBatchAllocatorTimeout);
181209
this.clusterManagerMetrics = clusterManagerMetrics;
210+
setFollowUpRerouteTaskPriority(FOLLOW_UP_REROUTE_PRIORITY_SETTING.get(settings));
211+
clusterSettings.addSettingsUpdateConsumer(FOLLOW_UP_REROUTE_PRIORITY_SETTING, this::setFollowUpRerouteTaskPriority);
182212
}
183213

184214
@Override
@@ -308,8 +338,8 @@ public void onComplete() {
308338
logger.trace("scheduling reroute after existing shards allocator timed out for primary shards");
309339
assert rerouteService != null;
310340
rerouteService.reroute(
311-
"reroute after existing shards allocator timed out",
312-
Priority.HIGH,
341+
"reroute after existing shards allocator [P] timed out",
342+
followUpRerouteTaskPriority,
313343
ActionListener.wrap(
314344
r -> logger.trace("reroute after existing shards allocator timed out completed"),
315345
e -> logger.debug("reroute after existing shards allocator timed out failed", e)
@@ -343,8 +373,8 @@ public void onComplete() {
343373
logger.trace("scheduling reroute after existing shards allocator timed out for replica shards");
344374
assert rerouteService != null;
345375
rerouteService.reroute(
346-
"reroute after existing shards allocator timed out",
347-
Priority.HIGH,
376+
"reroute after existing shards allocator [R] timed out",
377+
followUpRerouteTaskPriority,
348378
ActionListener.wrap(
349379
r -> logger.trace("reroute after existing shards allocator timed out completed"),
350380
e -> logger.debug("reroute after existing shards allocator timed out failed", e)
@@ -920,4 +950,8 @@ protected void setPrimaryBatchAllocatorTimeout(TimeValue primaryShardsBatchGatew
920950
protected void setReplicaBatchAllocatorTimeout(TimeValue replicaShardsBatchGatewayAllocatorTimeout) {
921951
this.replicaShardsBatchGatewayAllocatorTimeout = replicaShardsBatchGatewayAllocatorTimeout;
922952
}
953+
954+
protected void setFollowUpRerouteTaskPriority(Priority followUpRerouteTaskPriority) {
955+
this.followUpRerouteTaskPriority = followUpRerouteTaskPriority;
956+
}
923957
}

0 commit comments

Comments
 (0)