honeymoose/OpenSearch
1
0
Fork 0
mirror of https://github.com/honeymoose/OpenSearch.git synced 2025-03-29 19:38:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve Snapshot State Machine Performance (#62000) (#62049)

Browse Source 
Just a few random things to optimize motivated by somewhat sub-standard performance
for large snapshot cluster states with many concurrent snapshots observed in production.
This commit is contained in:
Armin Braun 2020-09-07 13:25:40 +02:00 committed by GitHub
parent fa8e76abb1
commit 395538f508
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 17 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
server/src/main/java/org/elasticsearch/snapshots/SnapshotShardsService.java
View File

@ -205,8 +205,8 @@ public class SnapshotShardsService extends AbstractLifecycleComponent implements
// Add all new shards to start processing on // Add all new shards to start processing on
final ShardId shardId = shard.key; final ShardId shardId = shard.key;
final ShardSnapshotStatus shardSnapshotStatus = shard.value; final ShardSnapshotStatus shardSnapshotStatus = shard.value;
if (localNodeId.equals(shardSnapshotStatus.nodeId()) if (shardSnapshotStatus.state() == ShardState.INIT
&& shardSnapshotStatus.state() == ShardState.INIT && localNodeId.equals(shardSnapshotStatus.nodeId())
&& snapshotShards.containsKey(shardId) == false) { && snapshotShards.containsKey(shardId) == false) {
logger.trace("[{}] - Adding shard to the queue", shardId); logger.trace("[{}] - Adding shard to the queue", shardId);
if (startedShards == null) { if (startedShards == null) {

19
server/src/main/java/org/elasticsearch/snapshots/SnapshotsService.java
View File

@ -2171,6 +2171,8 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
boolean changed = false; boolean changed = false;
final String repoName = deleteEntry.repository(); final String repoName = deleteEntry.repository();
// Computing the new assignments can be quite costly, only do it once below if actually needed
ImmutableOpenMap<ShardId, ShardSnapshotStatus> shardAssignments = null;
for (SnapshotsInProgress.Entry entry : snapshotsInProgress.entries()) { for (SnapshotsInProgress.Entry entry : snapshotsInProgress.entries()) {
if (entry.repository().equals(repoName)) { if (entry.repository().equals(repoName)) {
if (entry.state().completed() == false) { if (entry.state().completed() == false) {
@ -2186,9 +2188,11 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
// No shards can be updated in this snapshot so we just add it as is again // No shards can be updated in this snapshot so we just add it as is again
snapshotEntries.add(entry); snapshotEntries.add(entry);
} else { } else {
final ImmutableOpenMap<ShardId, ShardSnapshotStatus> shardAssignments = shards(snapshotsInProgress, if (shardAssignments == null) {
updatedDeletions, currentState.metadata(), currentState.routingTable(), entry.indices(), shardAssignments = shards(snapshotsInProgress,
entry.version().onOrAfter(SHARD_GEN_IN_REPO_DATA_VERSION), repositoryData, repoName); updatedDeletions, currentState.metadata(), currentState.routingTable(), entry.indices(),
entry.version().onOrAfter(SHARD_GEN_IN_REPO_DATA_VERSION), repositoryData, repoName);
}
final ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> updatedAssignmentsBuilder = final ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> updatedAssignmentsBuilder =
ImmutableOpenMap.builder(entry.shards()); ImmutableOpenMap.builder(entry.shards());
for (ShardId shardId : canBeUpdated) { for (ShardId shardId : canBeUpdated) {
@ -2296,7 +2300,7 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
} else { } else {
final IndexRoutingTable indexRoutingTable = routingTable.index(indexName); final IndexRoutingTable indexRoutingTable = routingTable.index(indexName);
for (int i = 0; i < indexMetadata.getNumberOfShards(); i++) { for (int i = 0; i < indexMetadata.getNumberOfShards(); i++) {
ShardId shardId = new ShardId(indexMetadata.getIndex(), i); final ShardId shardId = indexRoutingTable.shard(i).shardId();
final String shardRepoGeneration; final String shardRepoGeneration;
if (useShardGenerations) { if (useShardGenerations) {
if (isNewIndex) { if (isNewIndex) {
@ -2474,11 +2478,10 @@ public class SnapshotsService extends AbstractLifecycleComponent implements Clus
} else { } else {
final String updatedRepository = updateSnapshotState.snapshot().getRepository(); final String updatedRepository = updateSnapshotState.snapshot().getRepository();
final Set<ShardId> reusedShardIds = reusedShardIdsByRepo.computeIfAbsent(updatedRepository, k -> new HashSet<>()); final Set<ShardId> reusedShardIds = reusedShardIdsByRepo.computeIfAbsent(updatedRepository, k -> new HashSet<>());
if (entry.repository().equals(updatedRepository) && if (entry.state().completed() == false && entry.repository().equals(updatedRepository)
entry.state().completed() == false && reusedShardIds.contains(finishedShardId) == false && reusedShardIds.contains(finishedShardId) == false) {
&& entry.shards().keys().contains(finishedShardId)) {
final ShardSnapshotStatus existingStatus = entry.shards().get(finishedShardId); final ShardSnapshotStatus existingStatus = entry.shards().get(finishedShardId);
if (existingStatus.state() != ShardState.QUEUED) { if (existingStatus == null || existingStatus.state() != ShardState.QUEUED) {
continue; continue;
} }
if (updated == false) { if (updated == false) {

6
server/src/test/java/org/elasticsearch/snapshots/SnapshotsInProgressSerializationTests.java
View File

@ -74,8 +74,10 @@ public class SnapshotsInProgressSerializationTests extends AbstractDiffableWireS
ShardId shardId = new ShardId(idx, j); ShardId shardId = new ShardId(idx, j);
String nodeId = randomAlphaOfLength(10); String nodeId = randomAlphaOfLength(10);
ShardState shardState = randomFrom(ShardState.values()); ShardState shardState = randomFrom(ShardState.values());
builder.put(shardId, new SnapshotsInProgress.ShardSnapshotStatus(nodeId, shardState, builder.put(shardId,
shardState.failed() ? randomAlphaOfLength(10) : null, "1")); shardState == ShardState.QUEUED ? SnapshotsInProgress.ShardSnapshotStatus.UNASSIGNED_QUEUED :
new SnapshotsInProgress.ShardSnapshotStatus(nodeId, shardState,
shardState.failed() ? randomAlphaOfLength(10) : null, "1"));
} }
} }
ImmutableOpenMap<ShardId, SnapshotsInProgress.ShardSnapshotStatus> shards = builder.build(); ImmutableOpenMap<ShardId, SnapshotsInProgress.ShardSnapshotStatus> shards = builder.build();