Skip to content

Commit 26ada7e

Browse files
Harsh-87Harsh-87
authored
Replace CPU load average logic with AverageTracker class and modify default thresholds (#18666)
Signed-off-by: Harsh Kothari <[email protected]>
1 parent efc3e5d commit 26ada7e

File tree

7 files changed

+222
-84
lines changed

7 files changed

+222
-84
lines changed

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
5656
- Fix the visit of sub queries for HasParentQuery and HasChildQuery ([#18621](https://github.com/opensearch-project/OpenSearch/pull/18621))
5757
- Fix the backward compatibility regression with COMPLEMENT for Regexp queries introduced in OpenSearch 3.0 ([#18640](https://github.com/opensearch-project/OpenSearch/pull/18640))
5858
- Fix Replication lag computation ([#18602](https://github.com/opensearch-project/OpenSearch/pull/18602))
59+
- Fixed Staggered merge - load average replace with AverageTrackers, some Default thresholds modified ([#18666](https://github.com/opensearch-project/OpenSearch/pull/18666))
5960

6061
### Security
6162

‎server/src/internalClusterTest/java/org/opensearch/index/autoforcemerge/AutoForceMergeManagerIT.java‎

Lines changed: 6 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -45,11 +45,6 @@ public class AutoForceMergeManagerIT extends RemoteStoreBaseIntegTestCase {
4545
private static final String MERGE_DELAY = "1s";
4646
private static final Integer SEGMENT_COUNT = 1;
4747

48-
@Override
49-
protected boolean addMockIndexStorePlugin() {
50-
return false;
51-
}
52-
5348
@Override
5449
protected Settings nodeSettings(int nodeOrdinal) {
5550
ByteSizeValue cacheSize = new ByteSizeValue(16, ByteSizeUnit.GB);
@@ -158,8 +153,8 @@ public void testAutoForceMergeTriggeringBasicWithOneShard() throws Exception {
158153
SegmentsStats segmentsStatsBefore = shard.segmentStats(false, false);
159154
waitUntil(() -> shard.segmentStats(false, false).getCount() == SEGMENT_COUNT, 1, TimeUnit.MINUTES);
160155
SegmentsStats segmentsStatsAfter = shard.segmentStats(false, false);
161-
// assertTrue((int) segmentsStatsBefore.getCount() > segmentsStatsAfter.getCount());
162-
// assertEquals((int) SEGMENT_COUNT, segmentsStatsAfter.getCount());
156+
assertTrue((int) segmentsStatsBefore.getCount() > segmentsStatsAfter.getCount());
157+
assertEquals((int) SEGMENT_COUNT, segmentsStatsAfter.getCount());
163158
assertAcked(client().admin().indices().prepareDelete(INDEX_NAME_1).get());
164159
}
165160

@@ -221,11 +216,11 @@ public void testAutoForceMergeTriggeringBasicWithFiveShardsOfTwoIndex() throws E
221216
SegmentsStats segmentsStatsForShard3Before = shard3.segmentStats(false, false);
222217
SegmentsStats segmentsStatsForShard4Before = shard4.segmentStats(false, false);
223218
SegmentsStats segmentsStatsForShard5Before = shard5.segmentStats(false, false);
224-
AtomicLong totalSegments = new AtomicLong(
219+
AtomicLong totalSegmentsBefore = new AtomicLong(
225220
segmentsStatsForShard1Before.getCount() + segmentsStatsForShard2Before.getCount() + segmentsStatsForShard3Before.getCount()
226221
+ segmentsStatsForShard4Before.getCount() + segmentsStatsForShard5Before.getCount()
227222
);
228-
assertTrue(totalSegments.get() > 5);
223+
assertTrue(totalSegmentsBefore.get() > 5);
229224
waitUntil(() -> shard1.segmentStats(false, false).getCount() == SEGMENT_COUNT, 1, TimeUnit.MINUTES);
230225
waitUntil(() -> shard2.segmentStats(false, false).getCount() == SEGMENT_COUNT, 1, TimeUnit.MINUTES);
231226
waitUntil(() -> shard3.segmentStats(false, false).getCount() == SEGMENT_COUNT, 1, TimeUnit.MINUTES);
@@ -236,11 +231,11 @@ public void testAutoForceMergeTriggeringBasicWithFiveShardsOfTwoIndex() throws E
236231
SegmentsStats segmentsStatsForShard3After = shard3.segmentStats(false, false);
237232
SegmentsStats segmentsStatsForShard4After = shard4.segmentStats(false, false);
238233
SegmentsStats segmentsStatsForShard5After = shard5.segmentStats(false, false);
239-
totalSegments.set(
234+
AtomicLong totalSegmentsAfter = new AtomicLong(
240235
segmentsStatsForShard1After.getCount() + segmentsStatsForShard2After.getCount() + segmentsStatsForShard3After.getCount()
241236
+ segmentsStatsForShard4After.getCount() + segmentsStatsForShard5After.getCount()
242237
);
243-
// assertEquals(5, totalSegments.get());
238+
assertTrue(totalSegmentsBefore.get() > totalSegmentsAfter.get());
244239
assertAcked(client().admin().indices().prepareDelete(INDEX_NAME_1).get());
245240
assertAcked(client().admin().indices().prepareDelete(INDEX_NAME_2).get());
246241
}

‎server/src/main/java/org/opensearch/index/autoforcemerge/AutoForceMergeManager.java‎

Lines changed: 77 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -66,10 +66,11 @@ public class AutoForceMergeManager extends AbstractLifecycleComponent {
6666
private ConfigurationValidator configurationValidator;
6767
private NodeValidator nodeValidator;
6868
private ShardValidator shardValidator;
69+
private Integer allocatedProcessors;
70+
private ResourceTrackerProvider.ResourceTrackers resourceTrackers;
6971
private final ForceMergeManagerSettings forceMergeManagerSettings;
7072
private final CommonStatsFlags flags = new CommonStatsFlags(CommonStatsFlags.Flag.Segments, CommonStatsFlags.Flag.Translog);
7173
private final Set<Integer> mergingShards;
72-
private Integer allocatedProcessors;
7374

7475
private static final Logger logger = LogManager.getLogger(AutoForceMergeManager.class);
7576

@@ -96,6 +97,7 @@ protected void doStart() {
9697
this.nodeValidator = new NodeValidator();
9798
this.shardValidator = new ShardValidator();
9899
this.allocatedProcessors = OpenSearchExecutors.allocatedProcessors(clusterService.getSettings());
100+
this.resourceTrackers = ResourceTrackerProvider.create(threadPool);
99101
}
100102

101103
@Override
@@ -117,43 +119,65 @@ private void modifySchedulerInterval(TimeValue schedulerInterval) {
117119
}
118120

119121
private void triggerForceMerge() {
122+
if (isValidForForceMerge() == false) {
123+
return;
124+
}
125+
executeForceMergeOnShards();
126+
}
127+
128+
private boolean isValidForForceMerge() {
120129
if (configurationValidator.hasWarmNodes() == false) {
130+
resourceTrackers.stop();
121131
logger.debug("No warm nodes found. Skipping Auto Force merge.");
122-
return;
132+
return false;
123133
}
124134
if (nodeValidator.validate().isAllowed() == false) {
125135
logger.debug("Node capacity constraints are not allowing to trigger auto ForceMerge");
126-
return;
136+
return false;
127137
}
128-
int iteration = nodeValidator.getMaxConcurrentForceMerges();
138+
return true;
139+
}
140+
141+
private void executeForceMergeOnShards() {
142+
int remainingIterations = nodeValidator.getMaxConcurrentForceMerges();
129143
for (IndexShard shard : getShardsBasedOnSorting(indicesService)) {
130-
if (iteration == 0) {
144+
if (remainingIterations == 0 || !nodeValidator.validate().isAllowed()) {
145+
if (remainingIterations > 0) {
146+
logger.debug("Node conditions no longer suitable for force merge.");
147+
}
131148
break;
132149
}
133-
if (nodeValidator.validate().isAllowed() == false) {
134-
logger.debug("Node conditions no longer suitable for force merge.");
150+
remainingIterations--;
151+
executeForceMergeForShard(shard);
152+
if (!waitBetweenShards()) {
135153
break;
136154
}
137-
iteration--;
138-
CompletableFuture.runAsync(() -> {
139-
try {
140-
mergingShards.add(shard.shardId().getId());
141-
shard.forceMerge(new ForceMergeRequest().maxNumSegments(forceMergeManagerSettings.getSegmentCount()));
142-
logger.debug("Merging is completed successfully for the shard {}", shard.shardId());
143-
} catch (Exception e) {
144-
logger.error("Error during force merge for shard {}\nException: {}", shard.shardId(), e);
145-
} finally {
146-
mergingShards.remove(shard.shardId().getId());
147-
}
148-
}, threadPool.executor(ThreadPool.Names.FORCE_MERGE));
149-
logger.info("Successfully triggered force merge for shard {}", shard.shardId());
155+
}
156+
}
157+
158+
private void executeForceMergeForShard(IndexShard shard) {
159+
CompletableFuture.runAsync(() -> {
150160
try {
151-
Thread.sleep(forceMergeManagerSettings.getForcemergeDelay().getMillis());
152-
} catch (InterruptedException e) {
153-
Thread.currentThread().interrupt();
154-
logger.error("Timer was interrupted while waiting between shards", e);
155-
break;
161+
mergingShards.add(shard.shardId().getId());
162+
shard.forceMerge(new ForceMergeRequest().maxNumSegments(forceMergeManagerSettings.getSegmentCount()));
163+
logger.debug("Merging is completed successfully for the shard {}", shard.shardId());
164+
} catch (Exception e) {
165+
logger.error("Error during force merge for shard {}\nException: {}", shard.shardId(), e);
166+
} finally {
167+
mergingShards.remove(shard.shardId().getId());
156168
}
169+
}, threadPool.executor(ThreadPool.Names.FORCE_MERGE));
170+
logger.info("Successfully triggered force merge for shard {}", shard.shardId());
171+
}
172+
173+
private boolean waitBetweenShards() {
174+
try {
175+
Thread.sleep(forceMergeManagerSettings.getForcemergeDelay().getMillis());
176+
return true;
177+
} catch (InterruptedException e) {
178+
Thread.currentThread().interrupt();
179+
logger.error("Timer was interrupted while waiting between shards", e);
180+
return false;
157181
}
158182
}
159183

@@ -264,15 +288,14 @@ protected class NodeValidator implements ValidationStrategy {
264288

265289
@Override
266290
public ValidationResult validate() {
291+
resourceTrackers.start();
267292
if (isCpuUsageOverThreshold()) {
268293
return new ValidationResult(false);
269294
}
270295
if (isDiskUsageOverThreshold()) {
271296
return new ValidationResult(false);
272297
}
273-
double jvmUsedPercent = jvmService.stats().getMem().getHeapUsedPercent();
274-
if (jvmUsedPercent >= forceMergeManagerSettings.getJvmThreshold()) {
275-
logger.debug("JVM memory: {}% breached the threshold: {}", jvmUsedPercent, forceMergeManagerSettings.getJvmThreshold());
298+
if (isJvmUsageOverThreshold()) {
276299
return new ValidationResult(false);
277300
}
278301
if (areForceMergeThreadsAvailable() == false) {
@@ -291,24 +314,34 @@ private boolean areForceMergeThreadsAvailable() {
291314
return false;
292315
}
293316

317+
private boolean isJvmUsageOverThreshold() {
318+
double jvmAverage = resourceTrackers.jvmFiveMinute.getAverage();
319+
if (jvmAverage >= forceMergeManagerSettings.getJvmThreshold()) {
320+
logger.debug("JVM Average: 5m({}%) breached the threshold: {}", jvmAverage, forceMergeManagerSettings.getJvmThreshold());
321+
return true;
322+
}
323+
jvmAverage = resourceTrackers.jvmOneMinute.getAverage();
324+
if (jvmAverage >= forceMergeManagerSettings.getJvmThreshold()) {
325+
logger.debug("JVM Average: 1m({}%) breached the threshold: {}", jvmAverage, forceMergeManagerSettings.getJvmThreshold());
326+
return true;
327+
}
328+
double jvmUsedPercent = jvmService.stats().getMem().getHeapUsedPercent();
329+
if (jvmUsedPercent >= forceMergeManagerSettings.getJvmThreshold()) {
330+
logger.debug("JVM memory: {}% breached the threshold: {}", jvmUsedPercent, forceMergeManagerSettings.getJvmThreshold());
331+
return true;
332+
}
333+
return false;
334+
}
335+
294336
private boolean isCpuUsageOverThreshold() {
295-
double[] loadAverage = osService.stats().getCpu().getLoadAverage();
296-
double loadAverage5m = (loadAverage[1] / (double) allocatedProcessors) * 100;
297-
if (loadAverage5m >= forceMergeManagerSettings.getCpuThreshold()) {
298-
logger.debug(
299-
"Load Average: 5m({}%) breached the threshold: {}",
300-
loadAverage5m,
301-
forceMergeManagerSettings.getCpuThreshold()
302-
);
337+
double cpuAverage = resourceTrackers.cpuFiveMinute.getAverage();
338+
if (cpuAverage >= forceMergeManagerSettings.getCpuThreshold()) {
339+
logger.debug("CPU Average: 5m({}%) breached the threshold: {}", cpuAverage, forceMergeManagerSettings.getCpuThreshold());
303340
return true;
304341
}
305-
double loadAverage1m = (loadAverage[0] / (double) allocatedProcessors) * 100;
306-
if (loadAverage1m >= forceMergeManagerSettings.getCpuThreshold()) {
307-
logger.debug(
308-
"Load Average: 1m({}%) breached the threshold: {}",
309-
loadAverage1m,
310-
forceMergeManagerSettings.getCpuThreshold()
311-
);
342+
cpuAverage = resourceTrackers.cpuOneMinute.getAverage();
343+
if (cpuAverage >= forceMergeManagerSettings.getCpuThreshold()) {
344+
logger.debug("CPU Average: 1m({}%) breached the threshold: {}", cpuAverage, forceMergeManagerSettings.getCpuThreshold());
312345
return true;
313346
}
314347
double cpuPercent = osService.stats().getCpu().getPercent();
@@ -445,6 +478,7 @@ protected boolean mustReschedule() {
445478
@Override
446479
protected void runInternal() {
447480
if (configurationValidator.validate().isAllowed() == false) {
481+
resourceTrackers.stop();
448482
return;
449483
}
450484
triggerForceMerge();

‎server/src/main/java/org/opensearch/index/autoforcemerge/ForceMergeManagerSettings.java‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -56,11 +56,11 @@ public class ForceMergeManagerSettings {
5656
);
5757

5858
/**
59-
* Setting for wait time between force merge operations (default: 10s).
59+
* Setting for wait time between force merge operations (default: 15s).
6060
*/
6161
public static final Setting<TimeValue> MERGE_DELAY_BETWEEN_SHARDS_FOR_AUTO_FORCE_MERGE = Setting.timeSetting(
6262
"node.auto_force_merge.merge_delay",
63-
TimeValue.timeValueSeconds(10),
63+
TimeValue.timeValueSeconds(15),
6464
TimeValue.timeValueSeconds(1),
6565
TimeValue.timeValueSeconds(60),
6666
Setting.Property.Dynamic,
@@ -92,23 +92,23 @@ public class ForceMergeManagerSettings {
9292
);
9393

9494
/**
95-
* Setting for cpu threshold. (default: 80)
95+
* Setting for cpu threshold. (default: 75)
9696
*/
9797
public static final Setting<Double> CPU_THRESHOLD_PERCENTAGE_FOR_AUTO_FORCE_MERGE = Setting.doubleSetting(
9898
"node.auto_force_merge.cpu.threshold",
99-
80.0,
99+
75.0,
100100
10,
101101
100,
102102
Setting.Property.Dynamic,
103103
Setting.Property.NodeScope
104104
);
105105

106106
/**
107-
* Setting for memory threshold. (default: 90)
107+
* Setting for disk threshold. (default: 85)
108108
*/
109109
public static final Setting<Double> DISK_THRESHOLD_PERCENTAGE_FOR_AUTO_FORCE_MERGE = Setting.doubleSetting(
110110
"node.auto_force_merge.disk.threshold",
111-
90.0,
111+
85.0,
112112
10,
113113
100,
114114
Setting.Property.Dynamic,

‎server/src/main/java/org/opensearch/index/autoforcemerge/ResourceTrackerProvider.java‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,85 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
*
4+
* The OpenSearch Contributors require contributions made to
5+
* this file be licensed under the Apache-2.0 license or a
6+
* compatible open source license.
7+
*/
8+
9+
package org.opensearch.index.autoforcemerge;
10+
11+
import org.opensearch.common.unit.TimeValue;
12+
import org.opensearch.node.resource.tracker.AverageCpuUsageTracker;
13+
import org.opensearch.node.resource.tracker.AverageMemoryUsageTracker;
14+
import org.opensearch.threadpool.ThreadPool;
15+
16+
/**
17+
* Provider for creating resource usage trackers used in auto force merge operations.
18+
*
19+
* @opensearch.internal
20+
*/
21+
public class ResourceTrackerProvider {
22+
23+
public static final TimeValue SHORT_POLL_INTERVAL = TimeValue.timeValueSeconds(6);
24+
public static final TimeValue LONG_POLL_INTERVAL = TimeValue.timeValueSeconds(30);
25+
public static final TimeValue SHORT_AVERAGE_WINDOW = TimeValue.timeValueMinutes(1);
26+
public static final TimeValue LONG_AVERAGE_WINDOW = TimeValue.timeValueMinutes(5);
27+
28+
public static ResourceTrackers resourceTrackers;
29+
30+
public static ResourceTrackers create(ThreadPool threadPool) {
31+
return resourceTrackers = new ResourceTrackers(
32+
new AverageCpuUsageTracker(threadPool, SHORT_POLL_INTERVAL, SHORT_AVERAGE_WINDOW),
33+
new AverageCpuUsageTracker(threadPool, LONG_POLL_INTERVAL, LONG_AVERAGE_WINDOW),
34+
new AverageMemoryUsageTracker(threadPool, SHORT_POLL_INTERVAL, SHORT_AVERAGE_WINDOW),
35+
new AverageMemoryUsageTracker(threadPool, LONG_POLL_INTERVAL, LONG_AVERAGE_WINDOW)
36+
);
37+
}
38+
39+
/**
40+
* Container for resource usage trackers used in auto force merge operations.
41+
* Provides access to CPU and JVM memory usage trackers with different time windows.
42+
*
43+
* @opensearch.internal
44+
*/
45+
public static class ResourceTrackers {
46+
public final AverageCpuUsageTracker cpuOneMinute;
47+
public final AverageCpuUsageTracker cpuFiveMinute;
48+
public final AverageMemoryUsageTracker jvmOneMinute;
49+
public final AverageMemoryUsageTracker jvmFiveMinute;
50+
51+
/**
52+
* Creates a new ResourceTrackers instance.
53+
*
54+
* @param cpuOneMinute CPU tracker with 1-minute window
55+
* @param cpuFiveMinute CPU tracker with 5-minute window
56+
* @param jvmOneMinute JVM memory tracker with 1-minute window
57+
* @param jvmFiveMinute JVM memory tracker with 5-minute window
58+
*/
59+
ResourceTrackers(
60+
AverageCpuUsageTracker cpuOneMinute,
61+
AverageCpuUsageTracker cpuFiveMinute,
62+
AverageMemoryUsageTracker jvmOneMinute,
63+
AverageMemoryUsageTracker jvmFiveMinute
64+
) {
65+
this.cpuOneMinute = cpuOneMinute;
66+
this.cpuFiveMinute = cpuFiveMinute;
67+
this.jvmOneMinute = jvmOneMinute;
68+
this.jvmFiveMinute = jvmFiveMinute;
69+
}
70+
71+
public void start() {
72+
cpuOneMinute.start();
73+
cpuFiveMinute.start();
74+
jvmOneMinute.start();
75+
jvmFiveMinute.start();
76+
}
77+
78+
public void stop() {
79+
cpuOneMinute.stop();
80+
cpuFiveMinute.stop();
81+
jvmOneMinute.stop();
82+
jvmFiveMinute.stop();
83+
}
84+
}
85+
}

0 commit comments

Comments
 (0)