Skip to content

Commit 11667b7

Browse files
pxsalehipxsalehi
committed
Expose merge events and their memory usage estimate
1 parent ca095be commit 11667b7

File tree

5 files changed

+72
-19
lines changed

5 files changed

+72
-19
lines changed

‎server/src/main/java/org/elasticsearch/index/engine/InternalEngine.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2870,7 +2870,7 @@ private final class EngineThreadPoolMergeScheduler extends ThreadPoolMergeSchedu
28702870
IndexSettings indexSettings,
28712871
ThreadPoolMergeExecutorService threadPoolMergeExecutorService
28722872
) {
2873-
super(shardId, indexSettings, threadPoolMergeExecutorService);
2873+
super(shardId, indexSettings, threadPoolMergeExecutorService, InternalEngine.this::estimateMergeBytes);
28742874
}
28752875

28762876
@Override

‎server/src/main/java/org/elasticsearch/index/engine/MergeMemoryEstimateProvider.java‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the "Elastic License
4+
* 2.0", the "GNU Affero General Public License v3.0 only", and the "Server Side
5+
* Public License v 1"; you may not use this file except in compliance with, at
6+
* your election, the "Elastic License 2.0", the "GNU Affero General Public
7+
* License v3.0 only", or the "Server Side Public License, v 1".
8+
*/
9+
10+
package org.elasticsearch.index.engine;
11+
12+
import org.apache.lucene.index.MergePolicy;
13+
14+
@FunctionalInterface
15+
public interface MergeMemoryEstimateProvider {
16+
17+
/**
18+
* Returns an estimate of the memory needed to perform a merge
19+
*/
20+
long estimateMergeMemoryBytes(MergePolicy.OneMerge merge);
21+
}

‎server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeExecutorService.java‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -130,13 +130,18 @@ boolean submitMergeTask(MergeTask mergeTask) {
130130
);
131131
}
132132
// then enqueue the merge task proper
133-
queuedMergeTasks.add(mergeTask);
133+
enqueueMergeTask(mergeTask);
134134
return true;
135135
}
136136
}
137137

138138
void reEnqueueBackloggedMergeTask(MergeTask mergeTask) {
139+
enqueueMergeTask(mergeTask);
140+
}
141+
142+
private void enqueueMergeTask(MergeTask mergeTask) {
139143
queuedMergeTasks.add(mergeTask);
144+
mergeEventConsumer.onMergeQueued(mergeTask.getOnGoingMerge(), mergeTask.getEstimateMergeMemoryBytes());
140145
}
141146

142147
public boolean allDone() {
@@ -204,6 +209,7 @@ private void runMergeTask(MergeTask mergeTask) {
204209
if (mergeTask.supportsIOThrottling()) {
205210
ioThrottledMergeTasksCount.decrementAndGet();
206211
}
212+
mergeEventConsumer.onMergeCompleted(mergeTask.getOnGoingMerge());
207213
}
208214
}
209215

@@ -216,6 +222,7 @@ private void abortMergeTask(MergeTask mergeTask) {
216222
if (mergeTask.supportsIOThrottling()) {
217223
ioThrottledMergeTasksCount.decrementAndGet();
218224
}
225+
mergeEventConsumer.onMergeAborted(mergeTask.getOnGoingMerge());
219226
}
220227
}
221228

@@ -287,9 +294,12 @@ public void registerMergeEventConsumer(MergeEventConsumer consumer) {
287294
}
288295

289296
public interface MergeEventConsumer {
297+
290298
void onMergeQueued(OnGoingMerge merge, long estimateMergeMemoryBytes);
291299

292300
void onMergeCompleted(OnGoingMerge merge);
301+
302+
void onMergeAborted(OnGoingMerge merge);
293303
}
294304

295305
// exposed for tests

‎server/src/main/java/org/elasticsearch/index/engine/ThreadPoolMergeScheduler.java‎

Lines changed: 18 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -65,12 +65,13 @@ public class ThreadPoolMergeScheduler extends MergeScheduler implements Elastics
6565
private final AtomicLong doneMergeTaskCount = new AtomicLong();
6666
private final CountDownLatch closedWithNoRunningMerges = new CountDownLatch(1);
6767
private volatile boolean closed = false;
68-
private final MergeMemoryEstimator mergeMemoryEstimator;
68+
private final MergeMemoryEstimateProvider mergeMemoryEstimateProvider;
6969

7070
public ThreadPoolMergeScheduler(
7171
ShardId shardId,
7272
IndexSettings indexSettings,
73-
ThreadPoolMergeExecutorService threadPoolMergeExecutorService
73+
ThreadPoolMergeExecutorService threadPoolMergeExecutorService,
74+
MergeMemoryEstimateProvider mergeMemoryEstimateProvider
7475
) {
7576
this.shardId = shardId;
7677
this.config = indexSettings.getMergeSchedulerConfig();
@@ -82,6 +83,7 @@ public ThreadPoolMergeScheduler(
8283
: Double.POSITIVE_INFINITY
8384
);
8485
this.threadPoolMergeExecutorService = threadPoolMergeExecutorService;
86+
this.mergeMemoryEstimateProvider = mergeMemoryEstimateProvider;
8587
}
8688

8789
@Override
@@ -177,11 +179,13 @@ MergeTask newMergeTask(MergeSource mergeSource, MergePolicy.OneMerge merge, Merg
177179
// forced merges, as well as merges triggered when closing a shard, always run un-IO-throttled
178180
boolean isAutoThrottle = mergeTrigger != MergeTrigger.CLOSING && merge.getStoreMergeInfo().mergeMaxNumSegments() == -1;
179181
// IO throttling cannot be toggled for existing merge tasks, only new merge tasks pick up the updated IO throttling setting
182+
long estimateMergeMemoryBytes = mergeMemoryEstimateProvider.estimateMergeMemoryBytes(merge);
180183
return new MergeTask(
181184
mergeSource,
182185
merge,
183186
isAutoThrottle && config.isAutoThrottle(),
184-
"Lucene Merge Task #" + submittedMergeTaskCount.incrementAndGet() + " for shard " + shardId
187+
"Lucene Merge Task #" + submittedMergeTaskCount.incrementAndGet() + " for shard " + shardId,
188+
estimateMergeMemoryBytes
185189
);
186190
}
187191

@@ -313,14 +317,22 @@ class MergeTask implements Runnable {
313317
private final OnGoingMerge onGoingMerge;
314318
private final MergeRateLimiter rateLimiter;
315319
private final boolean supportsIOThrottling;
316-
317-
MergeTask(MergeSource mergeSource, MergePolicy.OneMerge merge, boolean supportsIOThrottling, String name) {
320+
private final long estimateMergeMemoryBytes;
321+
322+
MergeTask(
323+
MergeSource mergeSource,
324+
MergePolicy.OneMerge merge,
325+
boolean supportsIOThrottling,
326+
String name,
327+
long estimateMergeMemoryBytes
328+
) {
318329
this.name = name;
319330
this.mergeStartTimeNS = new AtomicLong();
320331
this.mergeSource = mergeSource;
321332
this.onGoingMerge = new OnGoingMerge(merge);
322333
this.rateLimiter = new MergeRateLimiter(merge.getMergeProgress());
323334
this.supportsIOThrottling = supportsIOThrottling;
335+
this.estimateMergeMemoryBytes = estimateMergeMemoryBytes;
324336
}
325337

326338
Schedule schedule() {
@@ -451,7 +463,7 @@ long estimatedMergeSize() {
451463
}
452464

453465
public long getEstimateMergeMemoryBytes() {
454-
return mergeMemoryEstimator.estimateMergeMemoryBytes(onGoingMerge.getMerge());
466+
return estimateMergeMemoryBytes;
455467
}
456468

457469
public OnGoingMerge getOnGoingMerge() {

‎server/src/test/java/org/elasticsearch/index/engine/ThreadPoolMergeSchedulerTests.java‎

Lines changed: 21 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -61,7 +61,8 @@ public void testMergesExecuteInSizeOrder() throws IOException {
6161
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
6262
new ShardId("index", "_na_", 1),
6363
IndexSettingsModule.newIndexSettings("index", Settings.EMPTY),
64-
threadPoolMergeExecutorService
64+
threadPoolMergeExecutorService,
65+
merge -> 0
6566
)
6667
) {
6768
List<OneMerge> executedMergesList = new ArrayList<>();
@@ -103,7 +104,8 @@ public void testSimpleMergeTaskBacklogging() {
103104
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
104105
new ShardId("index", "_na_", 1),
105106
IndexSettingsModule.newIndexSettings("index", mergeSchedulerSettings),
106-
threadPoolMergeExecutorService
107+
threadPoolMergeExecutorService,
108+
merge -> 0
107109
);
108110
// more merge tasks than merge threads
109111
int mergeCount = mergeExecutorThreadCount + randomIntBetween(1, 5);
@@ -136,7 +138,8 @@ public void testSimpleMergeTaskReEnqueueingBySize() {
136138
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
137139
new ShardId("index", "_na_", 1),
138140
IndexSettingsModule.newIndexSettings("index", mergeSchedulerSettings),
139-
threadPoolMergeExecutorService
141+
threadPoolMergeExecutorService,
142+
merge -> 0
140143
);
141144
// sort backlogged merges by size
142145
PriorityQueue<MergeTask> backloggedMergeTasks = new PriorityQueue<>(16, Comparator.comparingLong(MergeTask::estimatedMergeSize));
@@ -347,7 +350,8 @@ public void testMergeSourceWithFollowUpMergesRunSequentially() throws Exception
347350
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
348351
new ShardId("index", "_na_", 1),
349352
IndexSettingsModule.newIndexSettings("index", settings),
350-
threadPoolMergeExecutorService
353+
threadPoolMergeExecutorService,
354+
merge -> 0
351355
)
352356
) {
353357
MergeSource mergeSource = mock(MergeSource.class);
@@ -420,7 +424,8 @@ public void testMergesRunConcurrently() throws Exception {
420424
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
421425
new ShardId("index", "_na_", 1),
422426
IndexSettingsModule.newIndexSettings("index", settings),
423-
threadPoolMergeExecutorService
427+
threadPoolMergeExecutorService,
428+
merge -> 0
424429
)
425430
) {
426431
// at least 1 extra merge than there are concurrently allowed
@@ -504,7 +509,8 @@ public void testSchedulerCloseWaitsForRunningMerge() throws Exception {
504509
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
505510
new ShardId("index", "_na_", 1),
506511
IndexSettingsModule.newIndexSettings("index", settings),
507-
threadPoolMergeExecutorService
512+
threadPoolMergeExecutorService,
513+
merge -> 0
508514
)
509515
) {
510516
CountDownLatch mergeDoneLatch = new CountDownLatch(1);
@@ -576,7 +582,8 @@ public void testAutoIOThrottleForMergeTasksWhenSchedulerDisablesIt() throws Exce
576582
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
577583
new ShardId("index", "_na_", 1),
578584
indexSettings,
579-
threadPoolMergeExecutorService
585+
threadPoolMergeExecutorService,
586+
merge -> 0
580587
)
581588
) {
582589
threadPoolMergeScheduler.merge(mergeSource, randomFrom(MergeTrigger.values()));
@@ -605,7 +612,8 @@ public void testAutoIOThrottleForMergeTasks() throws Exception {
605612
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
606613
new ShardId("index", "_na_", 1),
607614
indexSettings,
608-
threadPoolMergeExecutorService
615+
threadPoolMergeExecutorService,
616+
merge -> 0
609617
)
610618
) {
611619
threadPoolMergeScheduler.merge(mergeSource, randomFrom(MergeTrigger.values()));
@@ -621,7 +629,8 @@ public void testAutoIOThrottleForMergeTasks() throws Exception {
621629
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
622630
new ShardId("index", "_na_", 1),
623631
indexSettings,
624-
threadPoolMergeExecutorService
632+
threadPoolMergeExecutorService,
633+
merge -> 0
625634
)
626635
) {
627636
// merge submitted upon closing
@@ -637,7 +646,8 @@ public void testAutoIOThrottleForMergeTasks() throws Exception {
637646
ThreadPoolMergeScheduler threadPoolMergeScheduler = new ThreadPoolMergeScheduler(
638647
new ShardId("index", "_na_", 1),
639648
indexSettings,
640-
threadPoolMergeExecutorService
649+
threadPoolMergeExecutorService,
650+
merge -> 0
641651
)
642652
) {
643653
// merge submitted upon closing
@@ -668,7 +678,7 @@ static class TestThreadPoolMergeScheduler extends ThreadPoolMergeScheduler {
668678
IndexSettings indexSettings,
669679
ThreadPoolMergeExecutorService threadPoolMergeExecutorService
670680
) {
671-
super(shardId, indexSettings, threadPoolMergeExecutorService);
681+
super(shardId, indexSettings, threadPoolMergeExecutorService, merge -> 0);
672682
}
673683

674684
@Override

0 commit comments

Comments
 (0)