Allow missing shard stats for restarted nodes for _snapshot/_status

Returns an empty shard stats for shard entries where stats were
unavailable in the case where a node has been restarted or left
the cluster.  The change adds a 'description' field to the
SnapshotIndexShardStatus class that is used to include a message
indicating why the stats are empty. This change was motivated by
a desire to reduce latency for getting the stats for currently
running snapshots.  The stats can still be loaded from the
repository via a _snapshot/<repository>/snapshot/_status call.

Closes ES-10982

Author: JeremyDahlgren

server/src/internalClusterTest/java/org/elasticsearch/snapshots/SnapshotStatusApisIT.java

@@ -212,7 +212,8 @@ public void testGetSnapshotsWithoutIndices() throws Exception {
      * 1. Start snapshot of two shards (both located on separate data nodes).
      * 2. Have one of the shards snapshot completely and the other block
      * 3. Restart the data node that completed its shard snapshot
-     * 4. Make sure that snapshot status APIs show correct file-counts and -sizes
+     * 4. Make sure that snapshot status APIs show correct file-counts and -sizes for non-restarted nodes
+     * 5. Make sure the description string is set for shard snapshots on restarted nodes.
      *
      * @throws Exception on failure
      */
@@ -261,8 +262,9 @@ public void testCorrectCountsForDoneShards() throws Exception {
             indexTwo
         );
         assertThat(snapshotShardStateAfterNodeRestart.getStage(), is(SnapshotIndexShardStage.DONE));
-        assertThat(snapshotShardStateAfterNodeRestart.getStats().getTotalFileCount(), equalTo(totalFiles));
-        assertThat(snapshotShardStateAfterNodeRestart.getStats().getTotalSize(), equalTo(totalFileSize));
+        assertNotNull("expected a description string for missing stats", snapshotShardStateAfterNodeRestart.getDescription());
+        assertThat(snapshotShardStateAfterNodeRestart.getStats().getTotalFileCount(), equalTo(0));
+        assertThat(snapshotShardStateAfterNodeRestart.getStats().getTotalSize(), equalTo(0L));
 
         unblockAllDataNodes(repoName);
         assertThat(responseSnapshotOne.get().getSnapshotInfo().state(), is(SnapshotState.SUCCESS));

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -259,6 +259,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_TIME_SERIES_SOURCE_STATUS = def(9_076_0_00);
     public static final TransportVersion ESQL_HASH_OPERATOR_STATUS_OUTPUT_TIME = def(9_077_0_00);
     public static final TransportVersion ML_INFERENCE_HUGGING_FACE_CHAT_COMPLETION_ADDED = def(9_078_0_00);
+    public static final TransportVersion SNAPSHOT_INDEX_SHARD_STATUS_DESCRIPTION_ADDED = def(9_079_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -21,6 +21,8 @@
 import java.io.IOException;
 import java.util.Objects;
 
+import static org.elasticsearch.TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_DESCRIPTION_ADDED;
+
 public class SnapshotIndexShardStatus extends BroadcastShardResponse implements ToXContentFragment {
 
     private final SnapshotIndexShardStage stage;
@@ -31,12 +33,17 @@ public class SnapshotIndexShardStatus extends BroadcastShardResponse implements
 
     private String failure;
 
+    private String description;
+
     public SnapshotIndexShardStatus(StreamInput in) throws IOException {
         super(in);
         stage = SnapshotIndexShardStage.fromValue(in.readByte());
         stats = new SnapshotStats(in);
         nodeId = in.readOptionalString();
         failure = in.readOptionalString();
+        if (in.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_DESCRIPTION_ADDED)) {
+            description = in.readOptionalString();
+        }
     }
 
     SnapshotIndexShardStatus(ShardId shardId, SnapshotIndexShardStage stage) {
@@ -74,11 +81,23 @@ public SnapshotIndexShardStatus(StreamInput in) throws IOException {
     }
 
     SnapshotIndexShardStatus(ShardId shardId, SnapshotIndexShardStage stage, SnapshotStats stats, String nodeId, String failure) {
+        this(shardId, stage, stats, nodeId, failure, null);
+    }
+
+    SnapshotIndexShardStatus(
+        ShardId shardId,
+        SnapshotIndexShardStage stage,
+        SnapshotStats stats,
+        String nodeId,
+        String failure,
+        String description
+    ) {
         super(shardId);
         this.stage = stage;
         this.stats = stats;
         this.nodeId = nodeId;
         this.failure = failure;
+        this.description = description;
     }
 
     /**
@@ -109,19 +128,30 @@ public String getFailure() {
         return failure;
     }
 
+    /**
+     * Returns the optional description of the data values contained in the {@code stats} field.
+     */
+    public String getDescription() {
+        return description;
+    }
+
     @Override
     public void writeTo(StreamOutput out) throws IOException {
         super.writeTo(out);
         out.writeByte(stage.value());
         stats.writeTo(out);
         out.writeOptionalString(nodeId);
         out.writeOptionalString(failure);
+        if (out.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_DESCRIPTION_ADDED)) {
+            out.writeOptionalString(description);
+        }
     }
 
     static final class Fields {
         static final String STAGE = "stage";
         static final String REASON = "reason";
         static final String NODE = "node";
+        static final String DESCRIPTION = "description";
     }
 
     @Override
@@ -135,6 +165,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (getFailure() != null) {
             builder.field(Fields.REASON, getFailure());
         }
+        if (getDescription() != null) {
+            builder.field(Fields.DESCRIPTION, getDescription());
+        }
         builder.endObject();
         return builder;
     }
@@ -151,7 +184,8 @@ public boolean equals(Object o) {
         return stage == that.stage
             && Objects.equals(stats, that.stats)
             && Objects.equals(nodeId, that.nodeId)
-            && Objects.equals(failure, that.failure);
+            && Objects.equals(failure, that.failure)
+            && Objects.equals(description, that.description);
     }
 
     @Override
@@ -160,6 +194,7 @@ public int hashCode() {
         result = 31 * result + (stats != null ? stats.hashCode() : 0);
         result = 31 * result + (nodeId != null ? nodeId.hashCode() : 0);
         result = 31 * result + (failure != null ? failure.hashCode() : 0);
+        result = 31 * result + (description != null ? description.hashCode() : 0);
         return result;
     }
 

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

@@ -196,13 +196,13 @@ void buildResponse(
                     }
                     SnapshotsInProgress.ShardSnapshotStatus status = shardEntry.getValue();
                     if (status.nodeId() != null) {
-                        // We should have information about this shard from the shard:
+                        // We should have information about this shard from the node:
                         TransportNodesSnapshotsStatus.NodeSnapshotStatus nodeStatus = nodeSnapshotStatusMap.get(status.nodeId());
                         if (nodeStatus != null) {
-                            Map<ShardId, SnapshotIndexShardStatus> shardStatues = nodeStatus.status().get(entry.snapshot());
-                            if (shardStatues != null) {
+                            Map<ShardId, SnapshotIndexShardStatus> shardStatuses = nodeStatus.status().get(entry.snapshot());
+                            if (shardStatuses != null) {
                                 final ShardId sid = entry.shardId(shardEntry.getKey());
-                                SnapshotIndexShardStatus shardStatus = shardStatues.get(sid);
+                                SnapshotIndexShardStatus shardStatus = shardStatuses.get(sid);
                                 if (shardStatus != null) {
                                     // We have full information about this shard
                                     if (shardStatus.getStage() == SnapshotIndexShardStage.DONE
@@ -228,26 +228,25 @@ void buildResponse(
                     }
                     // We failed to find the status of the shard from the responses we received from data nodes.
                     // This can happen if nodes drop out of the cluster completely or restart during the snapshot.
-                    // We rebuild the information they would have provided from their in memory state from the cluster
-                    // state and the repository contents in the below logic
                     final SnapshotIndexShardStage stage = switch (shardEntry.getValue().state()) {
                         case FAILED, ABORTED, MISSING -> SnapshotIndexShardStage.FAILURE;
                         case INIT, WAITING, PAUSED_FOR_NODE_REMOVAL, QUEUED -> SnapshotIndexShardStage.STARTED;
                         case SUCCESS -> SnapshotIndexShardStage.DONE;
                     };
                     final SnapshotIndexShardStatus shardStatus;
                     if (stage == SnapshotIndexShardStage.DONE) {
-                        // Shard snapshot completed successfully so we should be able to load the exact statistics for this
-                        // shard from the repository already.
-                        final ShardId shardId = entry.shardId(shardEntry.getKey());
+                        // When processing currently running snapshots, instead of reading the statistics from the repository, which can be
+                        // expensive, we choose instead to provide a message to the caller explaining why the stats are missing and the API
+                        // that can be used to load them.
                         shardStatus = new SnapshotIndexShardStatus(
-                            shardId,
-                            repositoriesService.repository(entry.repository())
-                                .getShardSnapshotStatus(
-                                    entry.snapshot().getSnapshotId(),
-                                    entry.indices().get(shardId.getIndexName()),
-                                    shardId
-                                )
+                            entry.shardId(shardEntry.getKey()),
+                            stage,
+                            new SnapshotStats(),
+                            null,
+                            null,
+                            """
+                               Snapshot shard stats missing from a currently running snapshot due to a node leaving the cluster after \
+                               completing the snapshot; use /_snapshot/<repository>/<snapshot>/_status to load from the repository."""
                         );
                     } else {
                         shardStatus = new SnapshotIndexShardStatus(entry.shardId(shardEntry.getKey()), stage);

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/status/SnapshotIndexShardStatusTests.java

@@ -39,10 +39,13 @@ protected SnapshotIndexShardStatus createForIndex(String indexName) {
         SnapshotStats stats = new SnapshotStatsTests().createTestInstance();
         String nodeId = randomAlphaOfLength(20);
         String failure = null;
+        String description = null;
         if (rarely()) {
             failure = randomAlphaOfLength(200);
+        } else if (rarely()) {
+            description = randomAlphaOfLength(200);
         }
-        return new SnapshotIndexShardStatus(shardId, stage, stats, nodeId, failure);
+        return new SnapshotIndexShardStatus(shardId, stage, stats, nodeId, failure, description);
     }
 
     @Override
@@ -76,6 +79,7 @@ protected boolean supportsUnknownFields() {
                 String rawStage = (String) parsedObjects[i++];
                 String nodeId = (String) parsedObjects[i++];
                 String failure = (String) parsedObjects[i++];
+                String description = (String) parsedObjects[i++];
                 SnapshotStats stats = (SnapshotStats) parsedObjects[i];
 
                 SnapshotIndexShardStage stage;
@@ -89,12 +93,13 @@ protected boolean supportsUnknownFields() {
                         rawStage
                     );
                 }
-                return new SnapshotIndexShardStatus(shard, stage, stats, nodeId, failure);
+                return new SnapshotIndexShardStatus(shard, stage, stats, nodeId, failure, description);
             }
         );
         innerParser.declareString(constructorArg(), new ParseField(SnapshotIndexShardStatus.Fields.STAGE));
         innerParser.declareString(optionalConstructorArg(), new ParseField(SnapshotIndexShardStatus.Fields.NODE));
         innerParser.declareString(optionalConstructorArg(), new ParseField(SnapshotIndexShardStatus.Fields.REASON));
+        innerParser.declareString(optionalConstructorArg(), new ParseField(SnapshotIndexShardStatus.Fields.DESCRIPTION));
         innerParser.declareObject(constructorArg(), (p, c) -> SnapshotStats.fromXContent(p), new ParseField(SnapshotStats.Fields.STATS));
         PARSER = (p, indexId, shardName) -> {
             // Combine the index name in the context with the shard name passed in for the named object parser

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/status/TransportSnapshotsStatusActionTests.java

@@ -14,12 +14,16 @@
 import org.elasticsearch.client.internal.node.NodeClient;
 import org.elasticsearch.cluster.SnapshotsInProgress;
 import org.elasticsearch.cluster.metadata.ProjectId;
+import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.index.IndexVersion;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.repositories.IndexId;
 import org.elasticsearch.repositories.RepositoriesService;
 import org.elasticsearch.repositories.ShardGeneration;
+import org.elasticsearch.repositories.ShardSnapshotResult;
 import org.elasticsearch.snapshots.Snapshot;
 import org.elasticsearch.snapshots.SnapshotId;
 import org.elasticsearch.tasks.CancellableTask;
@@ -200,4 +204,142 @@ public void onFailure(Exception e) {
         );
         assertTrue("Expected listener to be invoked", listenerInvoked.get());
     }
+
+    public void testShardSnapshotMissingDataFromNodeWhenNodeHasBeenRestarted() {
+        final var snapshot = new Snapshot(ProjectId.DEFAULT, "test-repo", new SnapshotId("snapshot", "uuid"));
+        final var indexName = "test-index-name";
+        final var indexUuid = "test-index-uuid";
+        final var shardGeneration = new ShardGeneration("gen");
+        final var shardId2 = new ShardId(indexName, indexUuid, 2);
+        final var nowMsecs = System.currentTimeMillis();
+        final var eightKb = ByteSizeValue.ofKb(8).getBytes();
+
+        final var currentSnapshotEntries = List.of(
+            SnapshotsInProgress.Entry.snapshot(
+                snapshot,
+                randomBoolean(),
+                randomBoolean(),
+                SnapshotsInProgress.State.STARTED,
+                Map.of(indexName, new IndexId(indexName, indexUuid)),
+                List.of(),
+                List.of(),
+                randomNonNegativeLong(),
+                randomNonNegativeLong(),
+                Map.of(
+                    new ShardId(indexName, indexUuid, 0),
+                    SnapshotsInProgress.ShardSnapshotStatus.success(
+                        "nodeId0",
+                        new ShardSnapshotResult(shardGeneration, ByteSizeValue.ofKb(5), 1)
+                    ),
+                    new ShardId(indexName, indexUuid, 1),
+                    new SnapshotsInProgress.ShardSnapshotStatus("nodeId1", shardGeneration),
+                    shardId2,
+                    SnapshotsInProgress.ShardSnapshotStatus.success(
+                        "nodeId2",
+                        new ShardSnapshotResult(shardGeneration, ByteSizeValue.ofKb(8), 1)
+                    )
+                ),
+                null,
+                Map.of(),
+                IndexVersion.current()
+            )
+        );
+        final var nodeSnapshotStatuses = new TransportNodesSnapshotsStatus.NodesSnapshotStatus(
+            clusterService.getClusterName(),
+            List.of(
+                new TransportNodesSnapshotsStatus.NodeSnapshotStatus(
+                    new DiscoveryNode(
+                        "nodeName0",
+                        "nodeId0",
+                        new TransportAddress(TransportAddress.META_ADDRESS, 9000),
+                        Map.of(),
+                        Set.of(),
+                        null
+                    ),
+                    // Here we are missing the snapshot data for the shard on this node.
+                    Map.of()
+                ),
+                new TransportNodesSnapshotsStatus.NodeSnapshotStatus(
+                    new DiscoveryNode(
+                        "nodeName2",
+                        "nodeId2",
+                        new TransportAddress(TransportAddress.META_ADDRESS, 9002),
+                        Map.of(),
+                        Set.of(),
+                        null
+                    ),
+                    Map.of(
+                        snapshot,
+                        Map.of(
+                            shardId2,
+                            new SnapshotIndexShardStatus(
+                                new ShardId(indexName, indexUuid, 2),
+                                SnapshotIndexShardStage.DONE,
+                                new SnapshotStats(nowMsecs, 0, 1, 1, 1, eightKb, eightKb, eightKb),
+                                "nodeId2",
+                                null
+                            )
+                        )
+                    )
+                )
+            ),
+            List.of()
+        );
+
+        final Consumer<SnapshotsStatusResponse> verifyResponse = rsp -> {
+            assertNotNull(rsp);
+            final var snapshotStatuses = rsp.getSnapshots();
+            assertNotNull(snapshotStatuses);
+            assertEquals(
+                "expected 1 snapshot status, got " + snapshotStatuses.size() + ": " + snapshotStatuses,
+                1,
+                snapshotStatuses.size()
+            );
+            final var snapshotStatus = snapshotStatuses.getFirst();
+            assertEquals(SnapshotsInProgress.State.STARTED, snapshotStatus.getState());
+            final var shardStats = snapshotStatus.getShardsStats();
+            assertNotNull("expected non-null shard stats for SnapshotStatus: " + snapshotStatus, shardStats);
+            assertEquals(new SnapshotShardsStats(0, 1 /* started */, 0, 2 /* done */, 0, 3 /* total */), shardStats);
+            final var totalStats = snapshotStatus.getStats();
+            assertNotNull("expected non-null total stats for SnapshotStatus: " + snapshotStatus, snapshotStatus);
+            assertEquals("expected total file count to be 1 in the stats: " + totalStats, 1, totalStats.getTotalFileCount());
+            assertEquals("expected total size to be " + eightKb + " in the stats: " + totalStats, eightKb, totalStats.getTotalSize());
+            final var snapshotStatusIndices = snapshotStatus.getIndices();
+            assertNotNull("expected a non-null map from getIndices() from SnapshotStatus: " + snapshotStatus, snapshotStatusIndices);
+            final var snapshotIndexStatus = snapshotStatusIndices.get(indexName);
+            assertNotNull(
+                "no entry for indexName [" + indexName + "] found in snapshotStatusIndices: " + snapshotStatusIndices,
+                snapshotIndexStatus
+            );
+            final var shardMap = snapshotIndexStatus.getShards();
+            assertNotNull("expected a non-null shard map for SnapshotIndexStatus: " + snapshotIndexStatus, shardMap);
+            final var shard0Entry = shardMap.get(0);
+            assertNotNull("no entry for shard 0 found in indexName [" + indexName + "] shardMap: " + shardMap, shard0Entry);
+            assertNotNull("expected a description string for shard 0 with missing stats from node0", shard0Entry.getDescription());
+        };
+
+        final var listener = new ActionListener<SnapshotsStatusResponse>() {
+            @Override
+            public void onResponse(SnapshotsStatusResponse rsp) {
+                verifyResponse.accept(rsp);
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                fail("expected onResponse() instead of onFailure(" + e + ")");
+            }
+        };
+
+        final var listenerInvoked = new AtomicBoolean(false);
+
+        action.buildResponse(
+            SnapshotsInProgress.EMPTY,
+            new SnapshotsStatusRequest(TEST_REQUEST_TIMEOUT),
+            currentSnapshotEntries,
+            nodeSnapshotStatuses,
+            new CancellableTask(randomLong(), "type", "action", "desc", null, Map.of()),
+            ActionListener.runAfter(listener, () -> listenerInvoked.set(true))
+        );
+        assertTrue("Expected listener to be invoked", listenerInvoked.get());
+    }
 }