Consistent updates of IndexShardSnapshotStatus (#28130)

This commit changes IndexShardSnapshotStatus so that the Stage is updated
coherently with any required information. It also provides a asCopy()
method that returns the status of a IndexShardSnapshotStatus at a given
point in time, ensuring that all information are coherent.

Closes #26480

Author: tlrx

core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/SnapshotIndexShardStatus.java

@@ -22,7 +22,6 @@
 import org.elasticsearch.action.support.broadcast.BroadcastShardResponse;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.xcontent.ToXContent.Params;
 import org.elasticsearch.common.xcontent.ToXContentFragment;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.shard.ShardId;
@@ -49,13 +48,13 @@ private SnapshotIndexShardStatus() {
         this.stats = new SnapshotStats();
     }
 
-    SnapshotIndexShardStatus(ShardId shardId, IndexShardSnapshotStatus indexShardStatus) {
+    SnapshotIndexShardStatus(ShardId shardId, IndexShardSnapshotStatus.Copy indexShardStatus) {
         this(shardId, indexShardStatus, null);
     }
 
-    SnapshotIndexShardStatus(ShardId shardId, IndexShardSnapshotStatus indexShardStatus, String nodeId) {
+    SnapshotIndexShardStatus(ShardId shardId, IndexShardSnapshotStatus.Copy indexShardStatus, String nodeId) {
         super(shardId);
-        switch (indexShardStatus.stage()) {
+        switch (indexShardStatus.getStage()) {
             case INIT:
                 stage = SnapshotIndexShardStage.INIT;
                 break;
@@ -72,10 +71,12 @@ private SnapshotIndexShardStatus() {
                 stage = SnapshotIndexShardStage.FAILURE;
                 break;
             default:
-                throw new IllegalArgumentException("Unknown stage type " + indexShardStatus.stage());
+                throw new IllegalArgumentException("Unknown stage type " + indexShardStatus.getStage());
         }
-        stats = new SnapshotStats(indexShardStatus);
-        failure = indexShardStatus.failure();
+        this.stats = new SnapshotStats(indexShardStatus.getStartTime(), indexShardStatus.getTotalTime(),
+                                        indexShardStatus.getNumberOfFiles(), indexShardStatus.getProcessedFiles(),
+                                        indexShardStatus.getTotalSize(), indexShardStatus.getProcessedSize());
+        this.failure = indexShardStatus.getFailure();
         this.nodeId = nodeId;
     }
 

core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/SnapshotStats.java

@@ -25,33 +25,28 @@
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.ToXContentFragment;
 import org.elasticsearch.common.xcontent.XContentBuilder;
-import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
 
 import java.io.IOException;
 
 public class SnapshotStats implements Streamable, ToXContentFragment {
-    private long startTime;
 
+    private long startTime;
     private long time;
-
     private int numberOfFiles;
-
     private int processedFiles;
-
     private long totalSize;
-
     private long processedSize;
 
     SnapshotStats() {
     }
 
-    SnapshotStats(IndexShardSnapshotStatus indexShardStatus) {
-        startTime = indexShardStatus.startTime();
-        time = indexShardStatus.time();
-        numberOfFiles = indexShardStatus.numberOfFiles();
-        processedFiles = indexShardStatus.processedFiles();
-        totalSize = indexShardStatus.totalSize();
-        processedSize = indexShardStatus.processedSize();
+    SnapshotStats(long startTime, long time, int numberOfFiles, int processedFiles, long totalSize, long processedSize) {
+        this.startTime = startTime;
+        this.time = time;
+        this.numberOfFiles = numberOfFiles;
+        this.processedFiles = processedFiles;
+        this.totalSize = totalSize;
+        this.processedSize = processedSize;
     }
 
     /**

core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportNodesSnapshotsStatus.java

@@ -96,23 +96,25 @@ protected NodesSnapshotStatus newResponse(Request request, List<NodeSnapshotStat
     protected NodeSnapshotStatus nodeOperation(NodeRequest request) {
         Map<Snapshot, Map<ShardId, SnapshotIndexShardStatus>> snapshotMapBuilder = new HashMap<>();
         try {
-            String nodeId = clusterService.localNode().getId();
+            final String nodeId = clusterService.localNode().getId();
             for (Snapshot snapshot : request.snapshots) {
                 Map<ShardId, IndexShardSnapshotStatus> shardsStatus = snapshotShardsService.currentSnapshotShards(snapshot);
                 if (shardsStatus == null) {
                     continue;
                 }
                 Map<ShardId, SnapshotIndexShardStatus> shardMapBuilder = new HashMap<>();
                 for (Map.Entry<ShardId, IndexShardSnapshotStatus> shardEntry : shardsStatus.entrySet()) {
-                    SnapshotIndexShardStatus shardStatus;
-                    IndexShardSnapshotStatus.Stage stage = shardEntry.getValue().stage();
+                    final ShardId shardId = shardEntry.getKey();
+
+                    final IndexShardSnapshotStatus.Copy lastSnapshotStatus = shardEntry.getValue().asCopy();
+                    final IndexShardSnapshotStatus.Stage stage = lastSnapshotStatus.getStage();
+
+                    String shardNodeId = null;
                     if (stage != IndexShardSnapshotStatus.Stage.DONE && stage != IndexShardSnapshotStatus.Stage.FAILURE) {
                         // Store node id for the snapshots that are currently running.
-                        shardStatus = new SnapshotIndexShardStatus(shardEntry.getKey(), shardEntry.getValue(), nodeId);
-                    } else {
-                        shardStatus = new SnapshotIndexShardStatus(shardEntry.getKey(), shardEntry.getValue());
+                        shardNodeId = nodeId;
                     }
-                    shardMapBuilder.put(shardEntry.getKey(), shardStatus);
+                    shardMapBuilder.put(shardEntry.getKey(), new SnapshotIndexShardStatus(shardId, lastSnapshotStatus, shardNodeId));
                 }
                 snapshotMapBuilder.put(snapshot, unmodifiableMap(shardMapBuilder));
             }

core/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusAction.java

@@ -233,7 +233,8 @@ private SnapshotsStatusResponse buildResponse(SnapshotsStatusRequest request, Li
                     Map<ShardId, IndexShardSnapshotStatus> shardStatues =
                         snapshotsService.snapshotShards(request.repository(), snapshotInfo);
                     for (Map.Entry<ShardId, IndexShardSnapshotStatus> shardStatus : shardStatues.entrySet()) {
-                        shardStatusBuilder.add(new SnapshotIndexShardStatus(shardStatus.getKey(), shardStatus.getValue()));
+                        IndexShardSnapshotStatus.Copy lastSnapshotStatus = shardStatus.getValue().asCopy();
+                        shardStatusBuilder.add(new SnapshotIndexShardStatus(shardStatus.getKey(), lastSnapshotStatus));
                     }
                     final SnapshotsInProgress.State state;
                     switch (snapshotInfo.state()) {