Update shardGenerations for all indices on snapshot finalization #128650
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -89,6 +89,7 @@ | |
| import org.elasticsearch.indices.SystemDataStreamDescriptor; | ||
| import org.elasticsearch.indices.SystemIndices; | ||
| import org.elasticsearch.repositories.FinalizeSnapshotContext; | ||
| import org.elasticsearch.repositories.FinalizeSnapshotContext.UpdatedShardGenerations; | ||
| import org.elasticsearch.repositories.IndexId; | ||
| import org.elasticsearch.repositories.RepositoriesService; | ||
| import org.elasticsearch.repositories.Repository; | ||
|
|
@@ -469,7 +470,7 @@ private void startCloning(Repository repository, SnapshotsInProgress.Entry clone | |
| endingSnapshots.add(targetSnapshot); | ||
| initializingClones.remove(targetSnapshot); | ||
| logger.info(() -> "Failed to start snapshot clone [" + cloneEntry + "]", e); | ||
| removeFailedSnapshotFromClusterState(targetSnapshot, e, null, UpdatedShardGenerations.EMPTY); | ||
| }; | ||
|
|
||
| // 1. step, load SnapshotInfo to make sure that source snapshot was successful for the indices we want to clone | ||
|
|
@@ -748,21 +749,23 @@ private static void validate(final String repositoryName, final String snapshotN | |
| } | ||
| } | ||
|
|
||
| private static UpdatedShardGenerations buildGenerations(SnapshotsInProgress.Entry snapshot, Metadata metadata) { | ||
| ShardGenerations.Builder builder = ShardGenerations.builder(); | ||
| ShardGenerations.Builder deletedBuilder = ShardGenerations.builder(); | ||
| if (snapshot.isClone()) { | ||
| snapshot.shardSnapshotStatusByRepoShardId().forEach((key, value) -> builder.put(key.index(), key.shardId(), value)); | ||
| } else { | ||
| snapshot.shardSnapshotStatusByRepoShardId().forEach((key, value) -> { | ||
| final Index index = snapshot.indexByName(key.indexName()); | ||
| if (metadata.findIndex(index).isEmpty()) { | ||
| assert snapshot.partial() : "Index [" + index + "] was deleted during a snapshot but snapshot was not partial."; | ||
| deletedBuilder.put(key.index(), key.shardId(), value); | ||
| return; | ||
| } | ||
| builder.put(key.index(), key.shardId(), value); | ||
| }); | ||
| } | ||
| return new UpdatedShardGenerations(builder.build(), deletedBuilder.build()); | ||
|
||
| } | ||
|
|
||
| private static Metadata metadataForSnapshot(SnapshotsInProgress.Entry snapshot, Metadata metadata) { | ||
|
|
@@ -1360,7 +1363,7 @@ private void endSnapshot(SnapshotsInProgress.Entry entry, Metadata metadata, @Nu | |
| snapshot, | ||
| new SnapshotException(snapshot, entry.failure()), | ||
| null, | ||
| UpdatedShardGenerations.EMPTY | ||
| ); | ||
| } | ||
| return; | ||
|
|
@@ -1454,8 +1457,9 @@ protected void doRun() { | |
| SnapshotsInProgress.Entry entry = SnapshotsInProgress.get(clusterService.state()).snapshot(snapshot); | ||
| final String failure = entry.failure(); | ||
| logger.trace("[{}] finalizing snapshot in repository, state: [{}], failure[{}]", snapshot, entry.state(), failure); | ||
| final var snapshotShardGenerations = buildGenerations(entry, metadata); | ||
| final ShardGenerations updatedShardGenerations = snapshotShardGenerations.liveIndices(); | ||
| final List<String> finalIndices = updatedShardGenerations.indices().stream().map(IndexId::getName).toList(); | ||
| final Set<String> indexNames = new HashSet<>(finalIndices); | ||
| ArrayList<SnapshotShardFailure> shardFailures = new ArrayList<>(); | ||
| for (Map.Entry<RepositoryShardId, ShardSnapshotStatus> shardStatus : entry.shardSnapshotStatusByRepoShardId().entrySet()) { | ||
|
|
@@ -1552,7 +1556,7 @@ protected void doRun() { | |
| entry.partial() ? onlySuccessfulFeatureStates(entry, finalIndices) : entry.featureStates(), | ||
| failure, | ||
| threadPool.absoluteTimeInMillis(), | ||
| entry.partial() ? updatedShardGenerations.totalShards() : entry.shardSnapshotStatusByRepoShardId().size(), | ||
| shardFailures, | ||
| entry.includeGlobalState(), | ||
| entry.userMetadata(), | ||
|
|
@@ -1562,7 +1566,7 @@ protected void doRun() { | |
| final ListenableFuture<List<ActionListener<SnapshotInfo>>> snapshotListeners = new ListenableFuture<>(); | ||
| repo.finalizeSnapshot( | ||
| new FinalizeSnapshotContext( | ||
| snapshotShardGenerations, | ||
| repositoryData.getGenId(), | ||
| metaForSnapshot, | ||
| snapshotInfo, | ||
|
|
@@ -1579,7 +1583,7 @@ protected void doRun() { | |
| snapshot, | ||
| repositoryData, | ||
| // we might have written the new root blob before failing here, so we must use the updated shardGenerations | ||
| snapshotShardGenerations | ||
| ) | ||
| ), | ||
| () -> snapshotListeners.addListener(new ActionListener<>() { | ||
|
|
@@ -1604,7 +1608,7 @@ public void onFailure(Exception e) { | |
| repositoryData, | ||
| // a failure here means the root blob was not updated, but the updated shard generation blobs are all in place so we can | ||
| // use the updated shardGenerations for all pending shard snapshots | ||
| snapshotShardGenerations | ||
| ) | ||
| )); | ||
| } | ||
|
|
@@ -1613,7 +1617,7 @@ public void onFailure(Exception e) { | |
| public void onRejection(Exception e) { | ||
| if (e instanceof EsRejectedExecutionException esre && esre.isExecutorShutdown()) { | ||
| logger.debug("failing finalization of {} due to shutdown", snapshot); | ||
| handleFinalizationFailure(e, snapshot, repositoryData, UpdatedShardGenerations.EMPTY); | ||
| } else { | ||
| onFailure(e); | ||
| } | ||
|
|
@@ -1623,7 +1627,7 @@ public void onRejection(Exception e) { | |
| public void onFailure(Exception e) { | ||
| logger.error(Strings.format("unexpected failure finalizing %s", snapshot), e); | ||
| assert false : new AssertionError("unexpected failure finalizing " + snapshot, e); | ||
| handleFinalizationFailure(e, snapshot, repositoryData, UpdatedShardGenerations.EMPTY); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -1679,7 +1683,7 @@ private void handleFinalizationFailure( | |
| Exception e, | ||
| Snapshot snapshot, | ||
| RepositoryData repositoryData, | ||
| UpdatedShardGenerations updatedShardGenerations | ||
| ) { | ||
| if (ExceptionsHelper.unwrap(e, NotMasterException.class, FailedToCommitClusterStateException.class) != null) { | ||
| // Failure due to not being master any more, don't try to remove snapshot from cluster state the next master | ||
|
|
@@ -1693,7 +1697,7 @@ private void handleFinalizationFailure( | |
| failAllListenersOnMasterFailOver(e); | ||
| } else { | ||
| logger.warn(() -> "[" + snapshot + "] failed to finalize snapshot", e); | ||
| removeFailedSnapshotFromClusterState(snapshot, e, repositoryData, updatedShardGenerations); | ||
| } | ||
| } | ||
|
|
||
|
|
@@ -1817,7 +1821,11 @@ private static Tuple<ClusterState, List<SnapshotDeletionsInProgress.Entry>> read | |
| * @param snapshot snapshot for which to remove the snapshot operation | ||
| * @return updated cluster state | ||
| */ | ||
| public static ClusterState stateWithoutSnapshot( | ||
| ClusterState state, | ||
| Snapshot snapshot, | ||
| UpdatedShardGenerations updatedShardGenerations | ||
| ) { | ||
| final SnapshotsInProgress inProgressSnapshots = SnapshotsInProgress.get(state); | ||
| ClusterState result = state; | ||
| int indexOfEntry = -1; | ||
|
|
@@ -1883,7 +1891,7 @@ public static ClusterState stateWithoutSnapshot(ClusterState state, Snapshot sna | |
| final RepositoryShardId repositoryShardId = finishedShardEntry.getKey(); | ||
| if (shardState.state() != ShardState.SUCCESS | ||
| || previousEntry.shardSnapshotStatusByRepoShardId().containsKey(repositoryShardId) == false | ||
| || updatedShardGenerations.hasShardGen(finishedShardEntry.getKey()) == false) { | ||
| continue; | ||
| } | ||
| updatedShardAssignments = maybeAddUpdatedAssignment( | ||
|
|
@@ -1902,7 +1910,7 @@ public static ClusterState stateWithoutSnapshot(ClusterState state, Snapshot sna | |
| final ShardSnapshotStatus shardState = finishedShardEntry.getValue(); | ||
| if (shardState.state() == ShardState.SUCCESS | ||
| && previousEntry.shardSnapshotStatusByRepoShardId().containsKey(finishedShardEntry.getKey()) | ||
| && updatedShardGenerations.hasShardGen(finishedShardEntry.getKey())) { | ||
| updatedShardAssignments = maybeAddUpdatedAssignment( | ||
| updatedShardAssignments, | ||
| shardState, | ||
|
|
@@ -1992,14 +2000,14 @@ private void removeFailedSnapshotFromClusterState( | |
| Snapshot snapshot, | ||
| Exception failure, | ||
| @Nullable RepositoryData repositoryData, | ||
| UpdatedShardGenerations updatedShardGenerations | ||
| ) { | ||
| assert failure != null : "Failure must be supplied"; | ||
| submitUnbatchedTask(REMOVE_SNAPSHOT_METADATA_TASK_SOURCE, new ClusterStateUpdateTask() { | ||
|
|
||
| @Override | ||
| public ClusterState execute(ClusterState currentState) { | ||
| final ClusterState updatedState = stateWithoutSnapshot(currentState, snapshot, updatedShardGenerations); | ||
| assert updatedState == currentState || endingSnapshots.contains(snapshot) | ||
| : "did not track [" + snapshot + "] in ending snapshots while removing it from the cluster state"; | ||
| // now check if there are any delete operations that refer to the just failed snapshot and remove the snapshot from them | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.