Track Shard-Snapshot Index Generation at Repository Root

server/src/main/java/org/elasticsearch/action/admin/cluster/repositories/cleanup/TransportCleanupRepositoryAction.java

@@ -201,7 +201,8 @@ public void clusterStateProcessed(String source, ClusterState oldState, ClusterS
                     logger.debug("Initialized repository cleanup in cluster state for [{}][{}]", repositoryName, repositoryStateId);
                     threadPool.executor(ThreadPool.Names.SNAPSHOT).execute(ActionRunnable.wrap(listener,
                         l -> blobStoreRepository.cleanup(
-                            repositoryStateId, ActionListener.wrap(result -> after(null, result), e -> after(e, null)))));
+                            repositoryStateId, newState.nodes().getMinNodeVersion(),
+                            ActionListener.wrap(result -> after(null, result), e -> after(e, null)))));
                 }
 
                 private void after(@Nullable Exception failure, @Nullable RepositoryCleanupResult result) {

server/src/main/java/org/elasticsearch/repositories/FilterRepository.java

@@ -74,16 +74,17 @@ public RepositoryData getRepositoryData() {
     }
 
     @Override
-    public void finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTime, String failure, int totalShards,
-                                 List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState,
-                                 MetaData metaData, Map<String, Object> userMetadata, ActionListener<SnapshotInfo> listener) {
-        in.finalizeSnapshot(snapshotId, indices, startTime, failure, totalShards, shardFailures, repositoryStateId,
-            includeGlobalState, metaData, userMetadata, listener);
+    public void finalizeSnapshot(SnapshotId snapshotId, ShardGenerations shardGenerations, long startTime, String failure,
+                                         int totalShards, List<SnapshotShardFailure> shardFailures, long repositoryStateId,
+                                         boolean includeGlobalState, MetaData metaData, Map<String, Object> userMetadata,
+                                         Version version, ActionListener<SnapshotInfo> listener) {
+        in.finalizeSnapshot(snapshotId, shardGenerations, startTime, failure, totalShards, shardFailures, repositoryStateId,
+            includeGlobalState, metaData, userMetadata, version, listener);
     }
 
     @Override
-    public void deleteSnapshot(SnapshotId snapshotId, long repositoryStateId, ActionListener<Void> listener) {
-        in.deleteSnapshot(snapshotId, repositoryStateId, listener);
+    public void deleteSnapshot(SnapshotId snapshotId, long repositoryStateId, Version version, ActionListener<Void> listener) {
+        in.deleteSnapshot(snapshotId, repositoryStateId, version, listener);
     }
 
     @Override
@@ -118,8 +119,9 @@ public boolean isReadOnly() {
 
     @Override
     public void snapshotShard(Store store, MapperService mapperService, SnapshotId snapshotId, IndexId indexId,
-                              IndexCommit snapshotIndexCommit, IndexShardSnapshotStatus snapshotStatus, ActionListener<String> listener) {
-        in.snapshotShard(store, mapperService, snapshotId, indexId, snapshotIndexCommit, snapshotStatus, listener);
+                              IndexCommit snapshotIndexCommit, IndexShardSnapshotStatus snapshotStatus, Version version,
+                              ActionListener<String> listener) {
+        in.snapshotShard(store, mapperService, snapshotId, indexId, snapshotIndexCommit, snapshotStatus, version, listener);
     }
     @Override
     public void restoreShard(Store store, SnapshotId snapshotId,

server/src/main/java/org/elasticsearch/repositories/Repository.java

@@ -113,28 +113,33 @@ default Repository create(RepositoryMetaData metaData, Function<String, Reposito
      * <p>
      * This method is called on master after all shards are snapshotted.
      *
-     * @param snapshotId    snapshot id
-     * @param indices       list of indices in the snapshot
-     * @param startTime     start time of the snapshot
-     * @param failure       global failure reason or null
-     * @param totalShards   total number of shards
-     * @param shardFailures list of shard failures
-     * @param repositoryStateId the unique id identifying the state of the repository when the snapshot began
+     * @param snapshotId         snapshot id
+     * @param shardGenerations   updated shard generations
+     * @param startTime          start time of the snapshot
+     * @param failure            global failure reason or null
+     * @param totalShards        total number of shards
+     * @param shardFailures      list of shard failures
+     * @param repositoryStateId  the unique id identifying the state of the repository when the snapshot began
      * @param includeGlobalState include cluster global state
+     * @param clusterMetaData    cluster metadata
+     * @param userMetadata       user metadata
+     * @param version            minimum ES version the repository should be readable by
      * @param listener listener to be called on completion of the snapshot
      */
-    void finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTime, String failure, int totalShards,
-                          List<SnapshotShardFailure> shardFailures, long repositoryStateId, boolean includeGlobalState,
-                          MetaData clusterMetaData, Map<String, Object> userMetadata, ActionListener<SnapshotInfo> listener);
+    void finalizeSnapshot(SnapshotId snapshotId, ShardGenerations shardGenerations, long startTime, String failure,
+                          int totalShards, List<SnapshotShardFailure> shardFailures, long repositoryStateId,
+                          boolean includeGlobalState, MetaData clusterMetaData, Map<String, Object> userMetadata,
+                          Version version, ActionListener<SnapshotInfo> listener);
 
     /**
      * Deletes snapshot
      *
-     * @param snapshotId snapshot id
+     * @param snapshotId        snapshot id
      * @param repositoryStateId the unique id identifying the state of the repository when the snapshot deletion began
-     * @param listener completion listener
+     * @param version           minimum ES version the repository should be readable by
+     * @param listener          completion listener
      */
-    void deleteSnapshot(SnapshotId snapshotId, long repositoryStateId, ActionListener<Void> listener);
+    void deleteSnapshot(SnapshotId snapshotId, long repositoryStateId, Version version, ActionListener<Void> listener);
 
     /**
      * Returns snapshot throttle time in nanoseconds
@@ -196,7 +201,7 @@ void finalizeSnapshot(SnapshotId snapshotId, List<IndexId> indices, long startTi
      * @param listener            listener invoked on completion
      */
     void snapshotShard(Store store, MapperService mapperService, SnapshotId snapshotId, IndexId indexId, IndexCommit snapshotIndexCommit,
-                       IndexShardSnapshotStatus snapshotStatus, ActionListener<String> listener);
+                       IndexShardSnapshotStatus snapshotStatus,  Version version, ActionListener<String> listener);
 
     /**
      * Restores snapshot of the shard.

server/src/main/java/org/elasticsearch/repositories/RepositoryData.java

@@ -21,12 +21,15 @@
 
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.ResourceNotFoundException;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.UUIDs;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentParserUtils;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.snapshots.SnapshotState;
+import org.elasticsearch.snapshots.SnapshotsService;
 
 import java.io.IOException;
 import java.util.ArrayList;
@@ -55,7 +58,7 @@ public final class RepositoryData {
      * An instance initialized for an empty repository.
      */
     public static final RepositoryData EMPTY = new RepositoryData(EMPTY_REPO_GEN,
-        Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap());
+        Collections.emptyMap(), Collections.emptyMap(), Collections.emptyMap(), ShardGenerations.EMPTY);
 
     /**
      * The generational id of the index file from which the repository data was read.
@@ -78,19 +81,30 @@ public final class RepositoryData {
      */
     private final Map<IndexId, Set<SnapshotId>> indexSnapshots;
 
+    /**
+     * Shard generations.
+     */
+    private final ShardGenerations shardGenerations;
 
     public RepositoryData(long genId, Map<String, SnapshotId> snapshotIds, Map<String, SnapshotState> snapshotStates,
-                          Map<IndexId, Set<SnapshotId>> indexSnapshots) {
+                          Map<IndexId, Set<SnapshotId>> indexSnapshots, ShardGenerations shardGenerations) {
         this.genId = genId;
         this.snapshotIds = Collections.unmodifiableMap(snapshotIds);
         this.snapshotStates = Collections.unmodifiableMap(snapshotStates);
         this.indices = Collections.unmodifiableMap(indexSnapshots.keySet().stream()
             .collect(Collectors.toMap(IndexId::getName, Function.identity())));
         this.indexSnapshots = Collections.unmodifiableMap(indexSnapshots);
+        this.shardGenerations = Objects.requireNonNull(shardGenerations);
+        assert indices.values().containsAll(shardGenerations.indices()) : "ShardGenerations contained indices "
+            + shardGenerations.indices() + " but snapshots only reference indices " + indices.values();
     }
 
     protected RepositoryData copy() {
-        return new RepositoryData(genId, snapshotIds, snapshotStates, indexSnapshots);
+        return new RepositoryData(genId, snapshotIds, snapshotStates, indexSnapshots, shardGenerations);
+    }
+
+    public ShardGenerations shardGenerations() {
+        return shardGenerations;
     }
 
     /**
@@ -140,10 +154,15 @@ public List<IndexId> indicesToUpdateAfterRemovingSnapshot(SnapshotId snapshotId)
     /**
      * Add a snapshot and its indices to the repository; returns a new instance.  If the snapshot
      * already exists in the repository data, this method throws an IllegalArgumentException.
+     *
+     * @param snapshotId       Id of the new snapshot
+     * @param snapshotState    State of the new snapshot
+     * @param shardGenerations Updated shard generations in the new snapshot. For each index contained in the snapshot an array of new
+     *                         generations indexed by the shard id they correspond to must be supplied.
      */
     public RepositoryData addSnapshot(final SnapshotId snapshotId,
                                       final SnapshotState snapshotState,
-                                      final List<IndexId> snapshottedIndices) {
+                                      final ShardGenerations shardGenerations) {
         if (snapshotIds.containsKey(snapshotId.getUUID())) {
             // if the snapshot id already exists in the repository data, it means an old master
             // that is blocked from the cluster is trying to finalize a snapshot concurrently with
@@ -155,10 +174,11 @@ public RepositoryData addSnapshot(final SnapshotId snapshotId,
         Map<String, SnapshotState> newSnapshotStates = new HashMap<>(snapshotStates);
         newSnapshotStates.put(snapshotId.getUUID(), snapshotState);
         Map<IndexId, Set<SnapshotId>> allIndexSnapshots = new HashMap<>(indexSnapshots);
-        for (final IndexId indexId : snapshottedIndices) {
+        for (final IndexId indexId : shardGenerations.indices()) {
             allIndexSnapshots.computeIfAbsent(indexId, k -> new LinkedHashSet<>()).add(snapshotId);
         }
-        return new RepositoryData(genId, snapshots, newSnapshotStates, allIndexSnapshots);
+        return new RepositoryData(genId, snapshots, newSnapshotStates, allIndexSnapshots,
+            ShardGenerations.builder().putAll(this.shardGenerations).putAll(shardGenerations).build());
     }
 
     /**
@@ -171,13 +191,18 @@ public RepositoryData withGenId(long newGeneration) {
         if (newGeneration == genId) {
             return this;
         }
-        return new RepositoryData(newGeneration, this.snapshotIds, this.snapshotStates, this.indexSnapshots);
+        return new RepositoryData(newGeneration, this.snapshotIds, this.snapshotStates, this.indexSnapshots, this.shardGenerations);
     }
 
     /**
      * Remove a snapshot and remove any indices that no longer exist in the repository due to the deletion of the snapshot.
+     *
+     * @param snapshotId              Snapshot Id
+     * @param updatedShardGenerations Shard generations that changed as a result of removing the snapshot.
+     *                                The {@code String[]} passed for each {@link IndexId} contains the new shard generation id for each
+     *                                changed shard indexed by its shardId
      */
-    public RepositoryData removeSnapshot(final SnapshotId snapshotId) {
+    public RepositoryData removeSnapshot(final SnapshotId snapshotId, final ShardGenerations updatedShardGenerations) {
         Map<String, SnapshotId> newSnapshotIds = snapshotIds.values().stream()
             .filter(id -> !snapshotId.equals(id))
             .collect(Collectors.toMap(SnapshotId::getUUID, Function.identity()));
@@ -205,7 +230,10 @@ public RepositoryData removeSnapshot(final SnapshotId snapshotId) {
             indexSnapshots.put(indexId, set);
         }
 
-        return new RepositoryData(genId, newSnapshotIds, newSnapshotStates, indexSnapshots);
+        return new RepositoryData(genId, newSnapshotIds, newSnapshotStates, indexSnapshots,
+            ShardGenerations.builder().putAll(shardGenerations).putAll(updatedShardGenerations)
+                .retainIndices(indexSnapshots.keySet()).build()
+        );
     }
 
     /**
@@ -231,12 +259,13 @@ public boolean equals(Object obj) {
         return snapshotIds.equals(that.snapshotIds)
                    && snapshotStates.equals(that.snapshotStates)
                    && indices.equals(that.indices)
-                   && indexSnapshots.equals(that.indexSnapshots);
+                   && indexSnapshots.equals(that.indexSnapshots)
+                   && shardGenerations.equals(that.shardGenerations);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(snapshotIds, snapshotStates, indices, indexSnapshots);
+        return Objects.hash(snapshotIds, snapshotStates, indices, indexSnapshots, shardGenerations);
     }
 
     /**
@@ -284,6 +313,7 @@ public List<IndexId> resolveNewIndices(final List<String> indicesToResolve) {
         return snapshotIndices;
     }
 
+    private static final String SHARD_GENERATIONS = "shard_generations";
     private static final String SNAPSHOTS = "snapshots";
     private static final String INDICES = "indices";
     private static final String INDEX_ID = "id";
@@ -294,7 +324,11 @@ public List<IndexId> resolveNewIndices(final List<String> indicesToResolve) {
     /**
      * Writes the snapshots metadata and the related indices metadata to x-content.
      */
-    public XContentBuilder snapshotsToXContent(final XContentBuilder builder) throws IOException {
+    public XContentBuilder snapshotsToXContent(final XContentBuilder builder, final Version version) throws IOException {
+        final boolean shouldWriteShardGens = version.onOrAfter(SnapshotsService.SHARD_GEN_IN_REPO_DATA_VERSION);
+        assert shouldWriteShardGens || shardGenerations.indices().isEmpty() :
+            "Should not build shard generations in BwC mode but saw generations [" + shardGenerations + "]";
+
         builder.startObject();
         // write the snapshots list
         builder.startArray(SNAPSHOTS);
@@ -320,6 +354,13 @@ public XContentBuilder snapshotsToXContent(final XContentBuilder builder) throws
                 builder.value(snapshotId.getUUID());
             }
             builder.endArray();
+            if (shouldWriteShardGens) {
+                builder.startArray(SHARD_GENERATIONS);
+                for (String gen : shardGenerations.getGens(indexId)) {
+                    builder.value(gen);
+                }
+                builder.endArray();
+            }
             builder.endObject();
         }
         builder.endObject();
@@ -334,6 +375,7 @@ public static RepositoryData snapshotsFromXContent(final XContentParser parser,
         final Map<String, SnapshotId> snapshots = new HashMap<>();
         final Map<String, SnapshotState> snapshotStates = new HashMap<>();
         final Map<IndexId, Set<SnapshotId>> indexSnapshots = new HashMap<>();
+        final ShardGenerations.Builder shardGenerations = ShardGenerations.builder();
 
         if (parser.nextToken() == XContentParser.Token.START_OBJECT) {
             while (parser.nextToken() == XContentParser.Token.FIELD_NAME) {
@@ -379,6 +421,7 @@ public static RepositoryData snapshotsFromXContent(final XContentParser parser,
                     while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
                         final String indexName = parser.currentName();
                         final Set<SnapshotId> snapshotIds = new LinkedHashSet<>();
+                        final List<String> gens = new ArrayList<>();
 
                         IndexId indexId = null;
                         if (parser.nextToken() != XContentParser.Token.START_OBJECT) {
@@ -421,10 +464,20 @@ public static RepositoryData snapshotsFromXContent(final XContentParser parser,
                                             + " references an unknown snapshot uuid [" + uuid + "]");
                                     }
                                 }
+                            } else if (SHARD_GENERATIONS.equals(indexMetaFieldName)) {
+                                XContentParserUtils.ensureExpectedToken(
+                                    XContentParser.Token.START_ARRAY, parser.currentToken(), parser::getTokenLocation);
+                                while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
+                                    gens.add(parser.textOrNull());
+                                }
+
                             }
                         }
                         assert indexId != null;
                         indexSnapshots.put(indexId, snapshotIds);
+                        for (int i = 0; i < gens.size(); i++) {
+                            shardGenerations.put(indexId, i, gens.get(i));
+                        }
                     }
                 } else {
                     throw new ElasticsearchParseException("unknown field name  [" + field + "]");
@@ -433,7 +486,7 @@ public static RepositoryData snapshotsFromXContent(final XContentParser parser,
         } else {
             throw new ElasticsearchParseException("start object expected");
         }
-        return new RepositoryData(genId, snapshots, snapshotStates, indexSnapshots);
+        return new RepositoryData(genId, snapshots, snapshotStates, indexSnapshots, shardGenerations.build());
     }
 
 }