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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| pr: 128650 | ||
| summary: Always update `shardGenerations` for previous in-progress snapshots | ||
| area: Snapshot/Restore | ||
| type: enhancement | ||
| issues: [] |
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 updatedShardGenerations = buildGenerations(entry, metadata); | ||
| final ShardGenerations updatedShardGensForLiveIndices = updatedShardGenerations.liveIndices(); | ||
| final List<String> finalIndices = updatedShardGensForLiveIndices.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() ? updatedShardGensForLiveIndices.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( | ||
| updatedShardGenerations, | ||
| 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 | ||
| updatedShardGenerations | ||
| ) | ||
| ), | ||
| () -> 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 | ||
| updatedShardGenerations | ||
| ) | ||
| )); | ||
| } | ||
|
|
@@ -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.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm this seems suspicious. We should be updating the shard generations for existing shards in
RepositoryDataand discarding the previous values, even if the index isn't included in the snapshot. AIUI the tripping assertion you mentioned in an earlier comment related to generations for shards that were totally absent fromRepositoryData. We should drop those shards entirely fromRepositoryData, but update any ones that do exist.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Ah yes you are absolutely right. We should conditionally update the shard generations in
RepositoryDatafor the deleted indices. The issue with my previous attempt is that the update is "unconditional" which triggered the assertion. It's now updated as suggested. Thanks a lot!