Track Shard-Snapshot Index Generation at Repository Root

Based on elastic/elasticsearch#46250

server/src/main/java/org/elasticsearch/action/ActionListener.java

@@ -247,4 +247,27 @@ public void onFailure(Exception e) {
             }
         };
     }
+
+    /**
+     * Creates a listener that delegates all responses it receives to another listener.
+     *
+     * @param delegate ActionListener to wrap and delegate any exception to
+     * @param bc BiConsumer invoked with delegate listener and exception
+     * @param <T> Type of the listener
+     * @return Delegating listener
+     */
+    static <T> ActionListener<T> delegateResponse(ActionListener<T> delegate, BiConsumer<ActionListener<T>, Exception> bc) {
+        return new ActionListener<T>() {
+
+            @Override
+            public void onResponse(T r) {
+                delegate.onResponse(r);
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                bc.accept(delegate, e);
+            }
+        };
+    }
 }

server/src/main/java/org/elasticsearch/action/ActionRunnable.java

@@ -19,7 +19,9 @@
 
 package org.elasticsearch.action;
 
+import io.crate.common.CheckedSupplier;
 import org.elasticsearch.common.CheckedConsumer;
+import org.elasticsearch.common.CheckedRunnable;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 
 /**
@@ -30,6 +32,32 @@ public abstract class ActionRunnable<Response> extends AbstractRunnable {
 
     protected final ActionListener<Response> listener;
 
+    /**
+     * Creates a {@link Runnable} that invokes the given listener with {@code null} after the given runnable has executed.
+     * @param listener Listener to invoke
+     * @param runnable Runnable to execute
+     * @return Wrapped {@code Runnable}
+     */
+    public static <T> ActionRunnable<T> run(ActionListener<T> listener, CheckedRunnable<Exception> runnable) {
+        return new ActionRunnable<>(listener) {
+            @Override
+            protected void doRun() throws Exception {
+                runnable.run();
+                listener.onResponse(null);
+            }
+        };
+    }
+
+    /**
+     * Creates a {@link Runnable} that invokes the given listener with the return of the given supplier.
+     * @param listener Listener to invoke
+     * @param supplier Supplier that provides value to pass to listener
+     * @return Wrapped {@code Runnable}
+     */
+    public static <T> ActionRunnable<T> supply(ActionListener<T> listener, CheckedSupplier<T, Exception> supplier) {
+        return ActionRunnable.wrap(listener, l -> l.onResponse(supplier.get()));
+    }
+
     /**
      * Creates a {@link Runnable} that wraps the given listener and a consumer of it that is executed when the {@link Runnable} is run.
      * Invokes {@link ActionListener#onFailure(Exception)} on it if an exception is thrown on executing the consumer.

server/src/main/java/org/elasticsearch/cluster/SnapshotsInProgress.java

@@ -93,11 +93,12 @@ public static class Entry {
         private final ImmutableOpenMap<String, List<ShardId>> waitingIndices;
         private final long startTime;
         private final long repositoryStateId;
+        private final boolean useShardGenerations;
         @Nullable private final String failure;
 
         public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, State state, List<IndexId> indices,
                      long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards,
-                     String failure) {
+                     String failure, boolean useShardGenerations) {
             this.state = state;
             this.snapshot = snapshot;
             this.includeGlobalState = includeGlobalState;
@@ -114,21 +115,22 @@ public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, Sta
             }
             this.repositoryStateId = repositoryStateId;
             this.failure = failure;
+            this.useShardGenerations = useShardGenerations;
         }
 
         public Entry(Snapshot snapshot, boolean includeGlobalState, boolean partial, State state, List<IndexId> indices,
-                     long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
-            this(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards, null);
+                     long startTime, long repositoryStateId, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards, boolean useShardGenerations) {
+            this(snapshot, includeGlobalState, partial, state, indices, startTime, repositoryStateId, shards, null, useShardGenerations);
         }
 
         public Entry(Entry entry, State state, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
             this(entry.snapshot, entry.includeGlobalState, entry.partial, state, entry.indices, entry.startTime,
-                entry.repositoryStateId, shards, entry.failure);
+                entry.repositoryStateId, shards, entry.failure, entry.useShardGenerations);
         }
 
         public Entry(Entry entry, State state, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards, String failure) {
             this(entry.snapshot, entry.includeGlobalState, entry.partial, state, entry.indices, entry.startTime,
-                entry.repositoryStateId, shards, failure);
+                entry.repositoryStateId, shards, failure, entry.useShardGenerations);
         }
 
         public Entry(Entry entry, ImmutableOpenMap<ShardId, ShardSnapshotStatus> shards) {
@@ -188,6 +190,16 @@ public String failure() {
             return failure;
         }
 
+        /**
+         * Whether to write to the repository in a format only understood by versions newer than
+         * {@link SnapshotsService#SHARD_GEN_IN_REPO_DATA_VERSION}.
+         *
+         * @return true if writing to repository in new format
+         */
+        public boolean useShardGenerations() {
+            return useShardGenerations;
+        }
+
         @Override
         public boolean equals(Object o) {
             if (this == o) return true;
@@ -203,6 +215,7 @@ public boolean equals(Object o) {
             if (!snapshot.equals(entry.snapshot)) return false;
             if (state != entry.state) return false;
             if (repositoryStateId != entry.repositoryStateId) return false;
+            if (useShardGenerations != entry.useShardGenerations) return false;
 
             return true;
         }
@@ -217,6 +230,8 @@ public int hashCode() {
             result = 31 * result + indices.hashCode();
             result = 31 * result + Long.hashCode(startTime);
             result = 31 * result + Long.hashCode(repositoryStateId);
+            result = 31 * result + (useShardGenerations ? 1 : 0);
+
             return result;
         }
 
@@ -445,24 +460,30 @@ public static NamedDiff<Custom> readDiffFrom(StreamInput in) throws IOException
     public SnapshotsInProgress(StreamInput in) throws IOException {
         Entry[] entries = new Entry[in.readVInt()];
         for (int i = 0; i < entries.length; i++) {
-            Snapshot snapshot = new Snapshot(in);
-            boolean includeGlobalState = in.readBoolean();
-            boolean partial = in.readBoolean();
-            State state = State.fromValue(in.readByte());
+            final Snapshot snapshot = new Snapshot(in);
+            final boolean includeGlobalState = in.readBoolean();
+            final boolean partial = in.readBoolean();
+            final State state = State.fromValue(in.readByte());
             int indices = in.readVInt();
             List<IndexId> indexBuilder = new ArrayList<>();
             for (int j = 0; j < indices; j++) {
                 indexBuilder.add(new IndexId(in.readString(), in.readString()));
             }
-            long startTime = in.readLong();
+            final long startTime = in.readLong();
             ImmutableOpenMap.Builder<ShardId, ShardSnapshotStatus> builder = ImmutableOpenMap.builder();
-            int shards = in.readVInt();
+            final int shards = in.readVInt();
             for (int j = 0; j < shards; j++) {
                 ShardId shardId = new ShardId(in);
                 builder.put(shardId, new ShardSnapshotStatus(in));
             }
-            long repositoryStateId = in.readLong();
+            final long repositoryStateId = in.readLong();
             final String failure = in.readOptionalString();
+            final boolean useShardGenerations;
+            if (in.getVersion().onOrAfter(SnapshotsService.SHARD_GEN_IN_REPO_DATA_VERSION)) {
+                useShardGenerations = in.readBoolean();
+            } else {
+                useShardGenerations = false;
+            }
             entries[i] = new Entry(snapshot,
                 includeGlobalState,
                 partial,
@@ -471,7 +492,8 @@ public SnapshotsInProgress(StreamInput in) throws IOException {
                 startTime,
                 repositoryStateId,
                 builder.build(),
-                failure);
+                failure,
+                useShardGenerations);
         }
         this.entries = Arrays.asList(entries);
     }
@@ -496,6 +518,9 @@ public void writeTo(StreamOutput out) throws IOException {
             }
             out.writeLong(entry.repositoryStateId);
             out.writeOptionalString(entry.failure);
+            if (out.getVersion().onOrAfter(SnapshotsService.SHARD_GEN_IN_REPO_DATA_VERSION)) {
+                out.writeBoolean(entry.useShardGenerations);
+            }
         }
     }
 

server/src/main/java/org/elasticsearch/cluster/metadata/MetaDataDeleteIndexService.java

@@ -38,6 +38,7 @@
 import org.elasticsearch.common.util.set.Sets;
 import org.elasticsearch.index.Index;
 import org.elasticsearch.snapshots.RestoreService;
+import org.elasticsearch.snapshots.SnapshotInProgressException;
 import org.elasticsearch.snapshots.SnapshotsService;
 
 import java.util.Arrays;
@@ -93,9 +94,15 @@ public ClusterState execute(final ClusterState currentState) {
      */
     public ClusterState deleteIndices(ClusterState currentState, Set<Index> indices) {
         final MetaData meta = currentState.metaData();
-        final Set<IndexMetaData> metaDatas = indices.stream().map(i -> meta.getIndexSafe(i)).collect(toSet());
+        final Set<Index> indicesToDelete = indices.stream().map(i -> meta.getIndexSafe(i).getIndex()).collect(toSet());
+
         // Check if index deletion conflicts with any running snapshots
-        SnapshotsService.checkIndexDeletion(currentState, metaDatas);
+        Set<Index> snapshottingIndices = SnapshotsService.snapshottingIndices(currentState, indicesToDelete);
+        if (snapshottingIndices.isEmpty() == false) {
+            throw new SnapshotInProgressException("Cannot delete indices that are being snapshotted: " + snapshottingIndices +
+                                                  ". Try again after snapshot finishes or cancel the currently running snapshot.");
+        }
+
         RoutingTable.Builder routingTableBuilder = RoutingTable.builder(currentState.routingTable());
         MetaData.Builder metaDataBuilder = MetaData.builder(meta);
         ClusterBlocks.Builder clusterBlocksBuilder = ClusterBlocks.builder().blocks(currentState.blocks());

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

@@ -133,36 +133,39 @@ 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 includeGlobalState    include cluster global state
-     * @param clusterMetaData       cluster metadata
-     * @param listener              listener to be called on completion of the snapshot
+     * @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 writeShardGens     if shard generations should be written to the repository
+     * @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,
-                                  ActionListener<SnapshotInfo> listener);
+                          ShardGenerations shardGenerations,
+                          long startTime,
+                          String failure,
+                          int totalShards,
+                          List<SnapshotShardFailure> shardFailures,
+                          long repositoryStateId,
+                          boolean includeGlobalState,
+                          MetaData clusterMetaData,
+                          boolean writeShardGens,
+                          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 writeShardGens    if shard generations should be written to the repository
+     * @param listener          completion listener
      */
-    void deleteSnapshot(SnapshotId snapshotId, long repositoryStateId, ActionListener<Void> listener);
+    void deleteSnapshot(SnapshotId snapshotId, long repositoryStateId, boolean writeShardGens, ActionListener<Void> listener);
 
     /**
      * Verifies repository on the master node and returns the verification token.
@@ -213,7 +216,7 @@ void finalizeSnapshot(SnapshotId snapshotId,
      * @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, boolean writeShardGens, ActionListener<String> listener);
 
     /**
      * Restores snapshot of the shard.
@@ -229,4 +232,14 @@ void snapshotShard(Store store, MapperService mapperService, SnapshotId snapshot
     void restoreShard(Store store, SnapshotId snapshotId, Version version, IndexId indexId, ShardId snapshotShardId,
                              RecoveryState recoveryState);
 
+    /**
+     * Retrieve shard snapshot status for the stored snapshot
+     *
+     * @param snapshotId snapshot id
+     * @param indexId    the snapshotted index id for the shard to get status for
+     * @param shardId    shard id
+     * @return snapshot status
+     */
+    IndexShardSnapshotStatus getShardSnapshotStatus(SnapshotId snapshotId, IndexId indexId, ShardId shardId);
+
 }