Skip to content

Commit a0d7b5a

Browse files
andreidanandreidan
committed
Online prewarming service interface docs and usage in SearchService
This adds the interface for search online prewarming with a default NOOP implementation. This also hooks the interface in the SearchService after we fork the query phase to the search thread pool.
1 parent d12eb8d commit a0d7b5a

File tree

7 files changed

+113
-7
lines changed

7 files changed

+113
-7
lines changed

‎server/src/main/java/org/elasticsearch/action/search/OnlinePrewarmingService.java‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,32 @@
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.action.search;
11+
12+
import org.elasticsearch.index.shard.IndexShard;
13+
14+
/**
15+
* Interface for prewarming the segments of a shard, tailored for consumption at
16+
* higher volumes than alternative warming strategies (i.e. offline / recovery warming)
17+
* that are more speculative.
18+
*/
19+
public interface OnlinePrewarmingService {
20+
OnlinePrewarmingService NOOP = (indexShard, skipPrewarmingCondition) -> {};
21+
22+
/**
23+
* Prewarms resources (typically segments) for the given index shard.
24+
*
25+
* @param indexShard the index shard for which resources should be prewarmed
26+
* @param skipPrewarming a flag indicating whether prewarming should be skipped.
27+
* Callers should decide if certain prewarming calls
28+
* should be skipped and indicate this decision via this
29+
* flag.
30+
*/
31+
void prewarm(IndexShard indexShard, boolean skipPrewarming);
32+
}

‎server/src/main/java/org/elasticsearch/action/search/OnlinePrewarmingServiceProvider.java‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
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.action.search;
11+
12+
import org.elasticsearch.cluster.service.ClusterService;
13+
import org.elasticsearch.common.settings.Settings;
14+
import org.elasticsearch.threadpool.ThreadPool;
15+
16+
public interface OnlinePrewarmingServiceProvider {
17+
OnlinePrewarmingServiceProvider DEFAULT = (settings, threadPool, clusterService) -> OnlinePrewarmingService.NOOP;
18+
19+
OnlinePrewarmingService create(Settings settings, ThreadPool threadPool, ClusterService clusterService);
20+
}

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -485,6 +485,7 @@ public void apply(Settings value, Settings current, Settings previous) {
485485
SearchService.CCS_VERSION_CHECK_SETTING,
486486
SearchService.CCS_COLLECT_TELEMETRY,
487487
SearchService.BATCHED_QUERY_PHASE,
488+
SearchService.PREWARMING_THRESHOLD_THREADPOOL_SIZE_FACTOR_POOL_SIZE,
488489
MultiBucketConsumerService.MAX_BUCKET_SETTING,
489490
SearchService.LOW_LEVEL_CANCELLATION_SETTING,
490491
SearchService.MAX_OPEN_SCROLL_CONTEXT,

‎server/src/main/java/org/elasticsearch/node/NodeServiceProvider.java‎

Lines changed: 8 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99

1010
package org.elasticsearch.node;
1111

12+
import org.elasticsearch.action.search.OnlinePrewarmingService;
13+
import org.elasticsearch.action.search.OnlinePrewarmingServiceProvider;
1214
import org.elasticsearch.client.internal.node.NodeClient;
1315
import org.elasticsearch.cluster.ClusterInfoService;
1416
import org.elasticsearch.cluster.InternalClusterInfoService;
@@ -123,6 +125,10 @@ SearchService newSearchService(
123125
ExecutorSelector executorSelector,
124126
Tracer tracer
125127
) {
128+
OnlinePrewarmingService onlinePrewarmingService = pluginsService.loadSingletonServiceProvider(
129+
OnlinePrewarmingServiceProvider.class,
130+
() -> OnlinePrewarmingServiceProvider.DEFAULT
131+
).create(clusterService.getSettings(), threadPool, clusterService);
126132
return new SearchService(
127133
clusterService,
128134
indicesService,
@@ -132,7 +138,8 @@ SearchService newSearchService(
132138
fetchPhase,
133139
circuitBreakerService,
134140
executorSelector,
135-
tracer
141+
tracer,
142+
onlinePrewarmingService
136143
);
137144
}
138145

‎server/src/main/java/org/elasticsearch/search/SearchService.java‎

Lines changed: 46 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.action.ResolvedIndices;
2525
import org.elasticsearch.action.search.CanMatchNodeRequest;
2626
import org.elasticsearch.action.search.CanMatchNodeResponse;
27+
import org.elasticsearch.action.search.OnlinePrewarmingService;
2728
import org.elasticsearch.action.search.SearchShardTask;
2829
import org.elasticsearch.action.search.SearchType;
2930
import org.elasticsearch.action.support.TransportActions;
@@ -147,6 +148,7 @@
147148
import java.util.Set;
148149
import java.util.concurrent.ExecutionException;
149150
import java.util.concurrent.Executor;
151+
import java.util.concurrent.ThreadPoolExecutor;
150152
import java.util.concurrent.TimeoutException;
151153
import java.util.concurrent.atomic.AtomicBoolean;
152154
import java.util.concurrent.atomic.AtomicInteger;
@@ -283,6 +285,16 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
283285
Property.NodeScope
284286
);
285287

288+
// This setting ensures that we skip online prewarming tasks if the queuing in the search thread pool
289+
// reaches the configured factor X number of max threads in the search thread pool, such that
290+
// the system has a chance to catch up and prewarming doesn't take over the network bandwidth
291+
public static final Setting<Integer> PREWARMING_THRESHOLD_THREADPOOL_SIZE_FACTOR_POOL_SIZE = Setting.intSetting(
292+
"search.online_prewarming_threshold_poolsize_factor",
293+
10,
294+
0, // 0 would mean we only execute online prewarming if there's no queuing in the search tp
295+
Setting.Property.NodeScope
296+
);
297+
286298
private static final boolean BATCHED_QUERY_PHASE_FEATURE_FLAG = new FeatureFlag("batched_query_phase").isEnabled();
287299

288300
/**
@@ -317,6 +329,8 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
317329

318330
private final FetchPhase fetchPhase;
319331
private final CircuitBreaker circuitBreaker;
332+
private final OnlinePrewarmingService onlinePrewarmingService;
333+
private final int prewarmingMaxPoolFactorThreshold;
320334
private volatile Executor searchExecutor;
321335
private volatile boolean enableQueryPhaseParallelCollection;
322336

@@ -362,7 +376,8 @@ public SearchService(
362376
FetchPhase fetchPhase,
363377
CircuitBreakerService circuitBreakerService,
364378
ExecutorSelector executorSelector,
365-
Tracer tracer
379+
Tracer tracer,
380+
OnlinePrewarmingService onlinePrewarmingService
366381
) {
367382
Settings settings = clusterService.getSettings();
368383
this.threadPool = threadPool;
@@ -375,7 +390,7 @@ public SearchService(
375390
this.multiBucketConsumerService = new MultiBucketConsumerService(clusterService, settings, circuitBreaker);
376391
this.executorSelector = executorSelector;
377392
this.tracer = tracer;
378-
393+
this.onlinePrewarmingService = onlinePrewarmingService;
379394
TimeValue keepAliveInterval = KEEPALIVE_INTERVAL_SETTING.get(settings);
380395
setKeepAlives(DEFAULT_KEEPALIVE_SETTING.get(settings), MAX_KEEPALIVE_SETTING.get(settings));
381396

@@ -427,6 +442,7 @@ public SearchService(
427442
memoryAccountingBufferSize = MEMORY_ACCOUNTING_BUFFER_SIZE.get(settings).getBytes();
428443
clusterService.getClusterSettings()
429444
.addSettingsUpdateConsumer(MEMORY_ACCOUNTING_BUFFER_SIZE, newValue -> this.memoryAccountingBufferSize = newValue.getBytes());
445+
prewarmingMaxPoolFactorThreshold = PREWARMING_THRESHOLD_THREADPOOL_SIZE_FACTOR_POOL_SIZE.get(settings);
430446
}
431447

432448
public CircuitBreaker getCircuitBreaker() {
@@ -702,6 +718,12 @@ private <T extends RefCounted> void ensureAfterSeqNoRefreshed(
702718
try {
703719
if (waitForCheckpoint <= UNASSIGNED_SEQ_NO) {
704720
runAsync(executor, executable, listener);
721+
// we successfully submitted the async task to the search pool so let's prewarm the shard
722+
onlinePrewarmingService.prewarm(
723+
shard,
724+
executor instanceof ThreadPoolExecutor tpe
725+
&& ((tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size())
726+
);
705727
return;
706728
}
707729
if (shard.indexSettings().getRefreshInterval().getMillis() <= 0) {
@@ -778,6 +800,12 @@ private void searchReady() {
778800
timeoutTask.cancel();
779801
}
780802
runAsync(executor, executable, listener);
803+
// we successfully submitted the async task to the search pool so let's prewarm the shard
804+
onlinePrewarmingService.prewarm(
805+
shard,
806+
executor instanceof ThreadPoolExecutor tpe
807+
&& ((tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size())
808+
);
781809
}
782810
}
783811
});
@@ -939,7 +967,8 @@ public void executeQueryPhase(
939967
freeReaderContext(readerContext.id());
940968
throw e;
941969
}
942-
runAsync(getExecutor(readerContext.indexShard()), () -> {
970+
Executor executor = getExecutor(readerContext.indexShard());
971+
runAsync(executor, () -> {
943972
final ShardSearchRequest shardSearchRequest = readerContext.getShardSearchRequest(null);
944973
try (SearchContext searchContext = createContext(readerContext, shardSearchRequest, task, ResultsType.QUERY, false);) {
945974
var opsListener = searchContext.indexShard().getSearchOperationListener();
@@ -965,6 +994,12 @@ public void executeQueryPhase(
965994
throw e;
966995
}
967996
}, wrapFailureListener(listener, readerContext, markAsUsed));
997+
// we successfully submitted the async task to the search pool so let's prewarm the shard
998+
onlinePrewarmingService.prewarm(
999+
readerContext.indexShard(),
1000+
executor instanceof ThreadPoolExecutor tpe
1001+
&& ((tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size())
1002+
);
9681003
}
9691004

9701005
/**
@@ -991,7 +1026,8 @@ public void executeQueryPhase(
9911026
final Releasable markAsUsed = readerContext.markAsUsed(getKeepAlive(shardSearchRequest));
9921027
rewriteAndFetchShardRequest(readerContext.indexShard(), shardSearchRequest, listener.delegateFailure((l, rewritten) -> {
9931028
// fork the execution in the search thread pool
994-
runAsync(getExecutor(readerContext.indexShard()), () -> {
1029+
Executor executor = getExecutor(readerContext.indexShard());
1030+
runAsync(executor, () -> {
9951031
readerContext.setAggregatedDfs(request.dfs());
9961032
try (SearchContext searchContext = createContext(readerContext, shardSearchRequest, task, ResultsType.QUERY, true);) {
9971033
final QuerySearchResult queryResult;
@@ -1029,6 +1065,12 @@ public void executeQueryPhase(
10291065
throw e;
10301066
}
10311067
}, wrapFailureListener(l, readerContext, markAsUsed));
1068+
// we successfully submitted the async task to the search pool so let's prewarm the shard
1069+
onlinePrewarmingService.prewarm(
1070+
readerContext.indexShard(),
1071+
executor instanceof ThreadPoolExecutor tpe
1072+
&& ((tpe.getMaximumPoolSize() * prewarmingMaxPoolFactorThreshold) < tpe.getQueue().size())
1073+
);
10321074
}));
10331075
}
10341076

‎server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,7 @@
4949
import org.elasticsearch.action.bulk.TransportShardBulkAction;
5050
import org.elasticsearch.action.index.IndexRequest;
5151
import org.elasticsearch.action.resync.TransportResyncReplicationAction;
52+
import org.elasticsearch.action.search.OnlinePrewarmingService;
5253
import org.elasticsearch.action.search.SearchExecutionStatsCollector;
5354
import org.elasticsearch.action.search.SearchPhaseController;
5455
import org.elasticsearch.action.search.SearchRequest;
@@ -2314,7 +2315,8 @@ public RecyclerBytesStreamOutput newNetworkBytesStream() {
23142315
new FetchPhase(Collections.emptyList()),
23152316
new NoneCircuitBreakerService(),
23162317
EmptySystemIndices.INSTANCE.getExecutorSelector(),
2317-
Tracer.NOOP
2318+
Tracer.NOOP,
2319+
OnlinePrewarmingService.NOOP
23182320
);
23192321

23202322
final SnapshotFilesProvider snapshotFilesProvider = new SnapshotFilesProvider(repositoriesService);

‎test/framework/src/main/java/org/elasticsearch/search/MockSearchService.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99

1010
package org.elasticsearch.search;
1111

12+
import org.elasticsearch.action.search.OnlinePrewarmingService;
1213
import org.elasticsearch.cluster.service.ClusterService;
1314
import org.elasticsearch.common.util.BigArrays;
1415
import org.elasticsearch.core.TimeValue;
@@ -95,7 +96,8 @@ public MockSearchService(
9596
fetchPhase,
9697
circuitBreakerService,
9798
executorSelector,
98-
tracer
99+
tracer,
100+
OnlinePrewarmingService.NOOP
99101
);
100102
}
101103

0 commit comments

Comments
 (0)