Skip to content

Commit cb1dc52

Browse files
jaymodejaymode
authored
Dedicated threadpool for system index writes (#62792)
This commit adds a dedicated threadpool for system index write operations. The dedicated resources for system index writes serves as a means to ensure that user activity does not block important system operations from occurring such as the management of users and roles. Backport of #61655
1 parent 54d97ec commit cb1dc52

29 files changed

+431
-154
lines changed

‎docs/reference/modules/threadpool.asciidoc

+5
Original file line numberDiff line numberDiff line change
@@ -88,6 +88,11 @@ There are several thread pools, but the important ones include:
8888
Thread pool type is `fixed` and a default maximum size of
8989
`min(5, (`<<node.processors, `# of allocated processors`>>`) / 2)`.
9090

91+
`system_write`::
92+
For write operations on system indices.
93+
Thread pool type is `fixed` and a default maximum size of
94+
`min(5, (`<<node.processors, `# of allocated processors`>>`) / 2)`.
95+
9196
Changing a specific thread pool can be done by setting its type-specific
9297
parameters; for example, changing the number of threads in the `write` thread
9398
pool:

‎server/src/main/java/org/elasticsearch/action/bulk/BulkRequest.java

+5
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444

4545
import java.io.IOException;
4646
import java.util.ArrayList;
47+
import java.util.Collections;
4748
import java.util.HashSet;
4849
import java.util.List;
4950
import java.util.Objects;
@@ -491,4 +492,8 @@ private static Boolean valueOrDefault(Boolean value, Boolean globalDefault) {
491492
public long ramBytesUsed() {
492493
return SHALLOW_SIZE + requests.stream().mapToLong(Accountable::ramBytesUsed).sum();
493494
}
495+
496+
public Set<String> getIndices() {
497+
return Collections.unmodifiableSet(indices);
498+
}
494499
}

‎server/src/main/java/org/elasticsearch/action/bulk/TransportBulkAction.java

+43-20
Original file line numberDiff line numberDiff line change
@@ -62,15 +62,17 @@
6262
import org.elasticsearch.common.util.concurrent.AtomicArray;
6363
import org.elasticsearch.index.Index;
6464
import org.elasticsearch.index.IndexNotFoundException;
65-
import org.elasticsearch.index.VersionType;
6665
import org.elasticsearch.index.IndexingPressure;
66+
import org.elasticsearch.index.VersionType;
6767
import org.elasticsearch.index.seqno.SequenceNumbers;
6868
import org.elasticsearch.index.shard.ShardId;
6969
import org.elasticsearch.indices.IndexClosedException;
70+
import org.elasticsearch.indices.SystemIndices;
7071
import org.elasticsearch.ingest.IngestService;
7172
import org.elasticsearch.node.NodeClosedException;
7273
import org.elasticsearch.tasks.Task;
7374
import org.elasticsearch.threadpool.ThreadPool;
75+
import org.elasticsearch.threadpool.ThreadPool.Names;
7476
import org.elasticsearch.transport.TransportService;
7577

7678
import java.util.ArrayList;
@@ -82,6 +84,7 @@
8284
import java.util.Map;
8385
import java.util.Objects;
8486
import java.util.Set;
87+
import java.util.SortedMap;
8588
import java.util.concurrent.TimeUnit;
8689
import java.util.concurrent.atomic.AtomicInteger;
8790
import java.util.concurrent.atomic.AtomicIntegerArray;
@@ -112,22 +115,24 @@ public class TransportBulkAction extends HandledTransportAction<BulkRequest, Bul
112115
private final IndexNameExpressionResolver indexNameExpressionResolver;
113116
private static final String DROPPED_ITEM_WITH_AUTO_GENERATED_ID = "auto-generated";
114117
private final IndexingPressure indexingPressure;
118+
private final SystemIndices systemIndices;
115119

116120
@Inject
117121
public TransportBulkAction(ThreadPool threadPool, TransportService transportService,
118122
ClusterService clusterService, IngestService ingestService,
119-
TransportShardBulkAction shardBulkAction, NodeClient client,
120-
ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
121-
AutoCreateIndex autoCreateIndex, IndexingPressure indexingPressure) {
123+
TransportShardBulkAction shardBulkAction, NodeClient client, ActionFilters actionFilters,
124+
IndexNameExpressionResolver indexNameExpressionResolver,
125+
AutoCreateIndex autoCreateIndex, IndexingPressure indexingPressure, SystemIndices systemIndices) {
122126
this(threadPool, transportService, clusterService, ingestService, shardBulkAction, client, actionFilters,
123-
indexNameExpressionResolver, autoCreateIndex, indexingPressure, System::nanoTime);
127+
indexNameExpressionResolver, autoCreateIndex, indexingPressure, systemIndices, System::nanoTime);
124128
}
125129

126130
public TransportBulkAction(ThreadPool threadPool, TransportService transportService,
127131
ClusterService clusterService, IngestService ingestService,
128132
TransportShardBulkAction shardBulkAction, NodeClient client,
129133
ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver,
130-
AutoCreateIndex autoCreateIndex, IndexingPressure indexingPressure, LongSupplier relativeTimeProvider) {
134+
AutoCreateIndex autoCreateIndex, IndexingPressure indexingPressure, SystemIndices systemIndices,
135+
LongSupplier relativeTimeProvider) {
131136
super(BulkAction.NAME, transportService, actionFilters, BulkRequest::new, ThreadPool.Names.SAME);
132137
Objects.requireNonNull(relativeTimeProvider);
133138
this.threadPool = threadPool;
@@ -140,6 +145,7 @@ public TransportBulkAction(ThreadPool threadPool, TransportService transportServ
140145
this.client = client;
141146
this.indexNameExpressionResolver = indexNameExpressionResolver;
142147
this.indexingPressure = indexingPressure;
148+
this.systemIndices = systemIndices;
143149
clusterService.addStateApplier(this.ingestForwarder);
144150
}
145151

@@ -163,17 +169,19 @@ public static IndexRequest getIndexWriteRequest(DocWriteRequest<?> docWriteReque
163169

164170
@Override
165171
protected void doExecute(Task task, BulkRequest bulkRequest, ActionListener<BulkResponse> listener) {
166-
long indexingBytes = bulkRequest.ramBytesUsed();
167-
final Releasable releasable = indexingPressure.markCoordinatingOperationStarted(indexingBytes);
172+
final long indexingBytes = bulkRequest.ramBytesUsed();
173+
final boolean isOnlySystem = isOnlySystem(bulkRequest, clusterService.state().metadata().getIndicesLookup(), systemIndices);
174+
final Releasable releasable = indexingPressure.markCoordinatingOperationStarted(indexingBytes, isOnlySystem);
168175
final ActionListener<BulkResponse> releasingListener = ActionListener.runBefore(listener, releasable::close);
176+
final String executorName = isOnlySystem ? Names.SYSTEM_WRITE : Names.WRITE;
169177
try {
170-
doInternalExecute(task, bulkRequest, releasingListener);
178+
doInternalExecute(task, bulkRequest, executorName, releasingListener);
171179
} catch (Exception e) {
172180
releasingListener.onFailure(e);
173181
}
174182
}
175183

176-
protected void doInternalExecute(Task task, BulkRequest bulkRequest, ActionListener<BulkResponse> listener) {
184+
protected void doInternalExecute(Task task, BulkRequest bulkRequest, String executorName, ActionListener<BulkResponse> listener) {
177185
final long startTime = relativeTime();
178186
final AtomicArray<BulkItemResponse> responses = new AtomicArray<>(bulkRequest.requests.size());
179187

@@ -211,7 +219,7 @@ protected void doInternalExecute(Task task, BulkRequest bulkRequest, ActionListe
211219
assert arePipelinesResolved : bulkRequest;
212220
}
213221
if (clusterService.localNode().isIngestNode()) {
214-
processBulkIndexIngestRequest(task, bulkRequest, listener);
222+
processBulkIndexIngestRequest(task, bulkRequest, executorName, listener);
215223
} else {
216224
ingestForwarder.forwardIngestRequest(BulkAction.INSTANCE, bulkRequest, listener);
217225
}
@@ -261,7 +269,7 @@ protected void doInternalExecute(Task task, BulkRequest bulkRequest, ActionListe
261269
@Override
262270
public void onResponse(CreateIndexResponse result) {
263271
if (counter.decrementAndGet() == 0) {
264-
threadPool.executor(ThreadPool.Names.WRITE).execute(
272+
threadPool.executor(executorName).execute(
265273
() -> executeBulk(task, bulkRequest, startTime, listener, responses, indicesThatCannotBeCreated));
266274
}
267275
}
@@ -278,10 +286,11 @@ public void onFailure(Exception e) {
278286
}
279287
}
280288
if (counter.decrementAndGet() == 0) {
281-
executeBulk(task, bulkRequest, startTime, ActionListener.wrap(listener::onResponse, inner -> {
289+
threadPool.executor(executorName).execute(() -> executeBulk(task, bulkRequest, startTime,
290+
ActionListener.wrap(listener::onResponse, inner -> {
282291
inner.addSuppressed(e);
283292
listener.onFailure(inner);
284-
}), responses, indicesThatCannotBeCreated);
293+
}), responses, indicesThatCannotBeCreated));
285294
}
286295
}
287296
});
@@ -342,6 +351,18 @@ static void prohibitCustomRoutingOnDataStream(DocWriteRequest<?> writeRequest, M
342351
}
343352
}
344353

354+
boolean isOnlySystem(BulkRequest request, SortedMap<String, IndexAbstraction> indicesLookup, SystemIndices systemIndices) {
355+
final boolean onlySystem = request.getIndices().stream().allMatch(indexName -> {
356+
final IndexAbstraction abstraction = indicesLookup.get(indexName);
357+
if (abstraction != null) {
358+
return abstraction.isSystem();
359+
} else {
360+
return systemIndices.isSystemIndex(indexName);
361+
}
362+
});
363+
return onlySystem;
364+
}
365+
345366
boolean needToCheck() {
346367
return autoCreateIndex.needToCheck();
347368
}
@@ -662,7 +683,8 @@ private long relativeTime() {
662683
return relativeTimeProvider.getAsLong();
663684
}
664685

665-
private void processBulkIndexIngestRequest(Task task, BulkRequest original, ActionListener<BulkResponse> listener) {
686+
private void processBulkIndexIngestRequest(Task task, BulkRequest original, String executorName,
687+
ActionListener<BulkResponse> listener) {
666688
final long ingestStartTimeInNanos = System.nanoTime();
667689
final BulkRequestModifier bulkRequestModifier = new BulkRequestModifier(original);
668690
ingestService.executeBulkRequest(
@@ -687,18 +709,18 @@ private void processBulkIndexIngestRequest(Task task, BulkRequest original, Acti
687709
// If a processor went async and returned a response on a different thread then
688710
// before we continue the bulk request we should fork back on a write thread:
689711
if (originalThread == Thread.currentThread()) {
690-
assert Thread.currentThread().getName().contains(ThreadPool.Names.WRITE);
691-
doInternalExecute(task, bulkRequest, actionListener);
712+
assert Thread.currentThread().getName().contains(executorName);
713+
doInternalExecute(task, bulkRequest, executorName, actionListener);
692714
} else {
693-
threadPool.executor(ThreadPool.Names.WRITE).execute(new AbstractRunnable() {
715+
threadPool.executor(executorName).execute(new AbstractRunnable() {
694716
@Override
695717
public void onFailure(Exception e) {
696718
listener.onFailure(e);
697719
}
698720

699721
@Override
700722
protected void doRun() throws Exception {
701-
doInternalExecute(task, bulkRequest, actionListener);
723+
doInternalExecute(task, bulkRequest, executorName, actionListener);
702724
}
703725

704726
@Override
@@ -714,7 +736,8 @@ public boolean isForceExecution() {
714736
}
715737
}
716738
},
717-
bulkRequestModifier::markItemAsDropped
739+
bulkRequestModifier::markItemAsDropped,
740+
executorName
718741
);
719742
}
720743

‎server/src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

+16-5
Original file line numberDiff line numberDiff line change
@@ -67,15 +67,18 @@
6767
import org.elasticsearch.index.shard.ShardId;
6868
import org.elasticsearch.index.translog.Translog;
6969
import org.elasticsearch.indices.IndicesService;
70+
import org.elasticsearch.indices.SystemIndices;
7071
import org.elasticsearch.node.NodeClosedException;
7172
import org.elasticsearch.threadpool.ThreadPool;
73+
import org.elasticsearch.threadpool.ThreadPool.Names;
7274
import org.elasticsearch.transport.TransportRequestOptions;
7375
import org.elasticsearch.transport.TransportService;
7476

7577
import java.io.IOException;
7678
import java.util.Map;
7779
import java.util.concurrent.Executor;
7880
import java.util.function.Consumer;
81+
import java.util.function.Function;
7982
import java.util.function.LongSupplier;
8083

8184
/** Performs shard-level bulk (index, delete or update) operations */
@@ -84,6 +87,13 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
8487
public static final String ACTION_NAME = BulkAction.NAME + "[s]";
8588

8689
private static final Logger logger = LogManager.getLogger(TransportShardBulkAction.class);
90+
private static final Function<IndexShard, String> EXECUTOR_NAME_FUNCTION = shard -> {
91+
if (shard.indexSettings().getIndexMetadata().isSystem()) {
92+
return Names.SYSTEM_WRITE;
93+
} else {
94+
return Names.WRITE;
95+
}
96+
};
8797

8898
private final UpdateHelper updateHelper;
8999
private final MappingUpdatedAction mappingUpdatedAction;
@@ -92,9 +102,9 @@ public class TransportShardBulkAction extends TransportWriteAction<BulkShardRequ
92102
public TransportShardBulkAction(Settings settings, TransportService transportService, ClusterService clusterService,
93103
IndicesService indicesService, ThreadPool threadPool, ShardStateAction shardStateAction,
94104
MappingUpdatedAction mappingUpdatedAction, UpdateHelper updateHelper, ActionFilters actionFilters,
95-
IndexingPressure indexingPressure) {
105+
IndexingPressure indexingPressure, SystemIndices systemIndices) {
96106
super(settings, ACTION_NAME, transportService, clusterService, indicesService, threadPool, shardStateAction, actionFilters,
97-
BulkShardRequest::new, BulkShardRequest::new, ThreadPool.Names.WRITE, false, indexingPressure);
107+
BulkShardRequest::new, BulkShardRequest::new, EXECUTOR_NAME_FUNCTION, false, indexingPressure, systemIndices);
98108
this.updateHelper = updateHelper;
99109
this.mappingUpdatedAction = mappingUpdatedAction;
100110
}
@@ -134,7 +144,7 @@ public void onClusterServiceClose() {
134144
public void onTimeout(TimeValue timeout) {
135145
mappingUpdateListener.onFailure(new MapperException("timed out while waiting for a dynamic mapping update"));
136146
}
137-
}), listener, threadPool
147+
}), listener, threadPool, executor(primary)
138148
);
139149
}
140150

@@ -151,10 +161,11 @@ public static void performOnPrimary(
151161
MappingUpdatePerformer mappingUpdater,
152162
Consumer<ActionListener<Void>> waitForMappingUpdate,
153163
ActionListener<PrimaryResult<BulkShardRequest, BulkShardResponse>> listener,
154-
ThreadPool threadPool) {
164+
ThreadPool threadPool,
165+
String executorName) {
155166
new ActionRunnable<PrimaryResult<BulkShardRequest, BulkShardResponse>>(listener) {
156167

157-
private final Executor executor = threadPool.executor(ThreadPool.Names.WRITE);
168+
private final Executor executor = threadPool.executor(executorName);
158169

159170
private final BulkPrimaryExecutionContext context = new BulkPrimaryExecutionContext(request, primary);
160171

‎server/src/main/java/org/elasticsearch/action/resync/TransportResyncReplicationAction.java

+14-4
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@
2020

2121
import org.apache.logging.log4j.message.ParameterizedMessage;
2222
import org.elasticsearch.action.ActionListener;
23-
import org.elasticsearch.index.IndexingPressure;
2423
import org.elasticsearch.action.support.ActionFilters;
2524
import org.elasticsearch.action.support.replication.ReplicationOperation;
2625
import org.elasticsearch.action.support.replication.ReplicationResponse;
@@ -33,35 +32,46 @@
3332
import org.elasticsearch.common.inject.Inject;
3433
import org.elasticsearch.common.io.stream.StreamInput;
3534
import org.elasticsearch.common.settings.Settings;
35+
import org.elasticsearch.index.IndexingPressure;
3636
import org.elasticsearch.index.engine.Engine;
3737
import org.elasticsearch.index.seqno.SequenceNumbers;
3838
import org.elasticsearch.index.shard.IndexShard;
3939
import org.elasticsearch.index.shard.PrimaryReplicaSyncer;
4040
import org.elasticsearch.index.translog.Translog;
4141
import org.elasticsearch.indices.IndicesService;
42+
import org.elasticsearch.indices.SystemIndices;
4243
import org.elasticsearch.tasks.Task;
4344
import org.elasticsearch.threadpool.ThreadPool;
45+
import org.elasticsearch.threadpool.ThreadPool.Names;
4446
import org.elasticsearch.transport.TransportException;
4547
import org.elasticsearch.transport.TransportResponseHandler;
4648
import org.elasticsearch.transport.TransportService;
4749

4850
import java.io.IOException;
51+
import java.util.function.Function;
4952
import java.util.stream.Stream;
5053

5154
public class TransportResyncReplicationAction extends TransportWriteAction<ResyncReplicationRequest,
5255
ResyncReplicationRequest, ResyncReplicationResponse> implements PrimaryReplicaSyncer.SyncAction {
5356

5457
private static String ACTION_NAME = "internal:index/seq_no/resync";
58+
private static final Function<IndexShard, String> EXECUTOR_NAME_FUNCTION = shard -> {
59+
if (shard.indexSettings().getIndexMetadata().isSystem()) {
60+
return Names.SYSTEM_WRITE;
61+
} else {
62+
return Names.WRITE;
63+
}
64+
};
5565

5666
@Inject
5767
public TransportResyncReplicationAction(Settings settings, TransportService transportService,
5868
ClusterService clusterService, IndicesService indicesService, ThreadPool threadPool,
5969
ShardStateAction shardStateAction, ActionFilters actionFilters,
60-
IndexingPressure indexingPressure) {
70+
IndexingPressure indexingPressure, SystemIndices systemIndices) {
6171
super(settings, ACTION_NAME, transportService, clusterService, indicesService, threadPool, shardStateAction, actionFilters,
62-
ResyncReplicationRequest::new, ResyncReplicationRequest::new, ThreadPool.Names.WRITE,
72+
ResyncReplicationRequest::new, ResyncReplicationRequest::new, EXECUTOR_NAME_FUNCTION,
6373
true, /* we should never reject resync because of thread pool capacity on primary */
64-
indexingPressure);
74+
indexingPressure, systemIndices);
6575
}
6676

6777
@Override

0 commit comments

Comments
 (0)