Track Shard Snapshot Generation in Cluster State

Backport on elastic/elasticsearch#46864

Author: mkleen

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

@@ -33,6 +33,7 @@
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.snapshots.Snapshot;
+import org.elasticsearch.snapshots.SnapshotsService;
 
 import javax.annotation.Nullable;
 import java.io.IOException;
@@ -262,26 +263,37 @@ public static class ShardSnapshotStatus {
         private final String nodeId;
         private final String reason;
 
-        public ShardSnapshotStatus(String nodeId) {
-            this(nodeId, ShardState.INIT);
+        @Nullable
+        private final String generation;
+
+        public ShardSnapshotStatus(String nodeId, String generation) {
+            this(nodeId, ShardState.INIT, generation);
         }
 
-        public ShardSnapshotStatus(String nodeId, ShardState state) {
-            this(nodeId, state, null);
+        public ShardSnapshotStatus(String nodeId, ShardState state, String generation) {
+            this(nodeId, state, null, generation);
         }
 
-        public ShardSnapshotStatus(String nodeId, ShardState state, String reason) {
+        public ShardSnapshotStatus(String nodeId, ShardState state, String reason, String generation) {
             this.nodeId = nodeId;
             this.state = state;
             this.reason = reason;
+            this.generation = generation;
             // If the state is failed we have to have a reason for this failure
             assert state.failed() == false || reason != null;
         }
 
         public ShardSnapshotStatus(StreamInput in) throws IOException {
             nodeId = in.readOptionalString();
             state = ShardState.fromValue(in.readByte());
+            if (in.getVersion().onOrAfter(SnapshotsService.SHARD_GEN_IN_REPO_DATA_VERSION)) {
+                generation = in.readOptionalString();
+                assert generation != null || state != ShardState.SUCCESS : "Received null generation for shard state [" + state + "]";
+            } else {
+                generation = null;
+            }
             reason = in.readOptionalString();
+
         }
 
         public ShardState state() {
@@ -296,9 +308,17 @@ public String reason() {
             return reason;
         }
 
+        @Nullable
+        public String generation() {
+            return generation;
+        }
+
         public void writeTo(StreamOutput out) throws IOException {
             out.writeOptionalString(nodeId);
             out.writeByte(state.value);
+            if (out.getVersion().onOrAfter(SnapshotsService.SHARD_GEN_IN_REPO_DATA_VERSION)) {
+                out.writeOptionalString(generation);
+            }
             out.writeOptionalString(reason);
         }
 
@@ -307,21 +327,24 @@ public boolean equals(Object o) {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             ShardSnapshotStatus status = (ShardSnapshotStatus) o;
-            return Objects.equals(nodeId, status.nodeId) && Objects.equals(reason, status.reason) && state == status.state;
-
+            return Objects.equals(nodeId, status.nodeId) &&
+                   Objects.equals(reason, status.reason) &&
+                   Objects.equals(generation, status.generation) &&
+                   state == status.state;
         }
 
         @Override
         public int hashCode() {
             int result = state != null ? state.hashCode() : 0;
             result = 31 * result + (nodeId != null ? nodeId.hashCode() : 0);
             result = 31 * result + (reason != null ? reason.hashCode() : 0);
+            result = 31 * result + (generation != null ? generation.hashCode() : 0);
             return result;
         }
 
         @Override
         public String toString() {
-            return "ShardSnapshotStatus[state=" + state + ", nodeId=" + nodeId + ", reason=" + reason + "]";
+            return "ShardSnapshotStatus[state=" + state + ", nodeId=" + nodeId + ", reason=" + reason + ", generation=" + generation + "]";
         }
     }
 

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/index/snapshots/IndexShardSnapshotStatus.java

@@ -58,6 +58,7 @@ public enum Stage {
     }
 
     private final AtomicReference<Stage> stage;
+    private final AtomicReference<String> generation;
     private long startTime;
     private long totalTime;
     private int incrementalFileCount;
@@ -69,10 +70,19 @@ public enum Stage {
     private long indexVersion;
     private String failure;
 
-    private IndexShardSnapshotStatus(final Stage stage, final long startTime, final long totalTime,
-                                     final int incrementalFileCount, final int totalFileCount, final int processedFileCount,
-                                     final long incrementalSize, final long totalSize, final long processedSize, final String failure) {
+    private IndexShardSnapshotStatus(final Stage stage,
+                                     final long startTime,
+                                     final long totalTime,
+                                     final int incrementalFileCount,
+                                     final int totalFileCount,
+                                     final int processedFileCount,
+                                     final long incrementalSize,
+                                     final long totalSize,
+                                     final long processedSize,
+                                     final String failure,
+                                     final String generation) {
         this.stage = new AtomicReference<>(Objects.requireNonNull(stage));
+        this.generation = new AtomicReference<>(generation);
         this.startTime = startTime;
         this.totalTime = totalTime;
         this.incrementalFileCount = incrementalFileCount;
@@ -109,21 +119,21 @@ public synchronized Copy moveToFinalize(final long indexVersion) {
         return asCopy();
     }
 
-    public synchronized Copy moveToDone(final long endTime) {
+    public synchronized void moveToDone(final long endTime, final String newGeneration) {
+        assert newGeneration != null;
         if (stage.compareAndSet(Stage.FINALIZE, Stage.DONE)) {
             this.totalTime = Math.max(0L, endTime - startTime);
+            this.generation.set(newGeneration);
         } else {
             throw new IllegalStateException("Unable to move the shard snapshot status to [DONE]: " +
                 "expecting [FINALIZE] but got [" + stage.get() + "]");
         }
-        return asCopy();
     }
 
-    public synchronized Copy abortIfNotCompleted(final String failure) {
+    public synchronized void abortIfNotCompleted(final String failure) {
         if (stage.compareAndSet(Stage.INIT, Stage.ABORTED) || stage.compareAndSet(Stage.STARTED, Stage.ABORTED)) {
             this.failure = failure;
         }
-        return asCopy();
     }
 
     public synchronized void moveToFailed(final long endTime, final String failure) {
@@ -133,6 +143,10 @@ public synchronized void moveToFailed(final long endTime, final String failure)
         }
     }
 
+    public String generation() {
+        return generation.get();
+    }
+
     public boolean isAborted() {
         return stage.get() == Stage.ABORTED;
     }
@@ -158,24 +172,24 @@ public synchronized IndexShardSnapshotStatus.Copy asCopy() {
             indexVersion, failure);
     }
 
-    public static IndexShardSnapshotStatus newInitializing() {
-        return new IndexShardSnapshotStatus(Stage.INIT, 0L, 0L, 0, 0, 0, 0, 0, 0, null);
+    public static IndexShardSnapshotStatus newInitializing(String generation) {
+        return new IndexShardSnapshotStatus(Stage.INIT, 0L, 0L, 0, 0, 0, 0, 0, 0, null, generation);
     }
 
     public static IndexShardSnapshotStatus newFailed(final String failure) {
         assert failure != null : "expecting non null failure for a failed IndexShardSnapshotStatus";
         if (failure == null) {
             throw new IllegalArgumentException("A failure description is required for a failed IndexShardSnapshotStatus");
         }
-        return new IndexShardSnapshotStatus(Stage.FAILURE, 0L, 0L, 0, 0, 0, 0, 0, 0, failure);
+        return new IndexShardSnapshotStatus(Stage.FAILURE, 0L, 0L, 0, 0, 0, 0, 0, 0, failure, null);
     }
 
     public static IndexShardSnapshotStatus newDone(final long startTime, final long totalTime,
                                                    final int incrementalFileCount, final int fileCount,
-                                                   final long incrementalSize, final long size) {
+                                                   final long incrementalSize, final long size, String generation) {
         // The snapshot is done which means the number of processed files is the same as total
         return new IndexShardSnapshotStatus(Stage.DONE, startTime, totalTime, incrementalFileCount, fileCount, incrementalFileCount,
-            incrementalSize, size, incrementalSize, null);
+                                            incrementalSize, size, incrementalSize, null, generation);
     }
 
     /**

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

@@ -224,7 +224,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<Void> listener);
+                       IndexShardSnapshotStatus snapshotStatus, ActionListener<String> listener);
 
     /**
      * Restores snapshot of the shard.