Allow missing shard stats for restarted nodes for _snapshot/_status (elastic#128399)

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

Co-authored-by: Dianna Hohensee <[email protected]>

Author: JeremyDahlgren

docs/changelog/128399.yaml

@@ -0,0 +1,5 @@
+pr: 128399
+summary: Allow missing shard stats for restarted nodes for `_snapshot/_status`
+area: Snapshot/Restore
+type: enhancement
+issues: []

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

@@ -13,7 +13,6 @@
 import org.elasticsearch.action.admin.cluster.snapshots.get.GetSnapshotsResponse;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotIndexShardStage;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotIndexShardStatus;
-import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStats;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotStatus;
 import org.elasticsearch.action.admin.cluster.snapshots.status.SnapshotsStatusResponse;
 import org.elasticsearch.action.support.GroupedActionListener;
@@ -212,7 +211,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
      */
@@ -248,21 +248,24 @@ public void testCorrectCountsForDoneShards() throws Exception {
             assertThat(snapshotShardState.getStage(), is(SnapshotIndexShardStage.DONE));
             assertThat(snapshotShardState.getStats().getTotalFileCount(), greaterThan(0));
             assertThat(snapshotShardState.getStats().getTotalSize(), greaterThan(0L));
+            assertNull("expected a null description for snapshot shard status: " + snapshotShardState, snapshotShardState.getDescription());
         }, 30L, TimeUnit.SECONDS);
 
-        final SnapshotStats snapshotShardStats = stateFirstShard(getSnapshotStatus(repoName, snapshotOne), indexTwo).getStats();
-        final int totalFiles = snapshotShardStats.getTotalFileCount();
-        final long totalFileSize = snapshotShardStats.getTotalSize();
-
         internalCluster().restartNode(dataNodeTwo);
 
-        final SnapshotIndexShardStatus snapshotShardStateAfterNodeRestart = stateFirstShard(
-            getSnapshotStatus(repoName, snapshotOne),
-            indexTwo
-        );
-        assertThat(snapshotShardStateAfterNodeRestart.getStage(), is(SnapshotIndexShardStage.DONE));
-        assertThat(snapshotShardStateAfterNodeRestart.getStats().getTotalFileCount(), equalTo(totalFiles));
-        assertThat(snapshotShardStateAfterNodeRestart.getStats().getTotalSize(), equalTo(totalFileSize));
+        final var snapshotStatusAfterRestart = getSnapshotStatus(repoName, snapshotOne);
+
+        final var snapshotShardStateIndexTwo = stateFirstShard(snapshotStatusAfterRestart, indexTwo);
+        assertThat(snapshotShardStateIndexTwo.getStage(), is(SnapshotIndexShardStage.DONE));
+        assertNotNull("expected a non-null description string for missing stats", snapshotShardStateIndexTwo.getDescription());
+        final var missingStats = snapshotShardStateIndexTwo.getStats();
+        assertThat(missingStats.getTotalFileCount(), equalTo(-1));
+        assertThat(missingStats.getTotalSize(), equalTo(-1L));
+
+        final var snapshotShardStateIndexOne = stateFirstShard(snapshotStatusAfterRestart, indexOne);
+        assertNull("expected a null description string for available stats", snapshotShardStateIndexOne.getDescription());
+        assertThat(snapshotShardStateIndexOne.getStats().getTotalFileCount(), greaterThan(0));
+        assertThat(snapshotShardStateIndexOne.getStats().getTotalSize(), greaterThan(0L));
 
         unblockAllDataNodes(repoName);
         assertThat(responseSnapshotOne.get().getSnapshotInfo().state(), is(SnapshotState.SUCCESS));

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

@@ -289,6 +289,8 @@ static TransportVersion def(int id) {
     public static final TransportVersion ML_INFERENCE_MISTRAL_CHAT_COMPLETION_ADDED = def(9_090_0_00);
     public static final TransportVersion IDP_CUSTOM_SAML_ATTRIBUTES_ALLOW_LIST = def(9_091_0_00);
     public static final TransportVersion SEARCH_SOURCE_EXCLUDE_VECTORS_PARAM = def(9_092_0_00);
+    public static final TransportVersion SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS = def(9_093_0_00);
+
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

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

@@ -13,6 +13,7 @@
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.index.shard.ShardId;
 import org.elasticsearch.index.snapshots.IndexShardSnapshotStatus;
 import org.elasticsearch.xcontent.ToXContentFragment;
@@ -21,6 +22,8 @@
 import java.io.IOException;
 import java.util.Objects;
 
+import static org.elasticsearch.TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS;
+
 public class SnapshotIndexShardStatus extends BroadcastShardResponse implements ToXContentFragment {
 
     private final SnapshotIndexShardStage stage;
@@ -31,12 +34,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_MISSING_STATS)) {
+            description = in.readOptionalString();
+        }
     }
 
     SnapshotIndexShardStatus(ShardId shardId, SnapshotIndexShardStage stage) {
@@ -74,11 +82,38 @@ 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,
+        @Nullable String description
+    ) {
         super(shardId);
         this.stage = stage;
         this.stats = stats;
         this.nodeId = nodeId;
         this.failure = failure;
+        this.description = description;
+    }
+
+    /**
+     * Creates an instance for scenarios where the snapshot is {@link SnapshotIndexShardStage#DONE} but the stats are unavailable, with a
+     * non-null description of why the stats are missing.
+     */
+    public static SnapshotIndexShardStatus forDoneButMissingStats(ShardId shardId, String description) {
+        return new SnapshotIndexShardStatus(
+            shardId,
+            SnapshotIndexShardStage.DONE,
+            SnapshotStats.forMissingStats(),
+            null,
+            null,
+            Objects.requireNonNull(description)
+        );
     }
 
     /**
@@ -109,19 +144,31 @@ public String getFailure() {
         return failure;
     }
 
+    /**
+     * Returns the optional description of the data values contained in the {@code stats} field.
+     */
+    @Nullable
+    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_MISSING_STATS)) {
+            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 +182,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 +201,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 +211,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/SnapshotStats.java

@@ -21,6 +21,8 @@
 
 import java.io.IOException;
 
+import static org.elasticsearch.TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS;
+
 public class SnapshotStats implements Writeable, ToXContentObject {
 
     private long startTime;
@@ -35,6 +37,19 @@ public class SnapshotStats implements Writeable, ToXContentObject {
     SnapshotStats() {}
 
     SnapshotStats(StreamInput in) throws IOException {
+        // We use a boolean to indicate if the stats are present (true) or missing (false), to skip writing all the values if missing.
+        if (in.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS) && in.readBoolean() == false) {
+            startTime = 0L;
+            time = 0L;
+            incrementalFileCount = -1;
+            processedFileCount = -1;
+            incrementalSize = -1L;
+            processedSize = -1L;
+            totalFileCount = -1;
+            totalSize = -1L;
+            return;
+        }
+
         startTime = in.readVLong();
         time = in.readVLong();
 
@@ -69,6 +84,20 @@ public class SnapshotStats implements Writeable, ToXContentObject {
         this.processedSize = processedSize;
     }
 
+    /**
+     * Returns a stats instance with invalid field values for use in situations where the snapshot stats are unavailable.
+     */
+    public static SnapshotStats forMissingStats() {
+        return new SnapshotStats(0L, 0L, -1, -1, -1, -1L, -1L, -1L);
+    }
+
+    /**
+     * Returns true if this instance is for a shard snapshot with unavailable stats.
+     */
+    public boolean isMissingStats() {
+        return incrementalFileCount == -1;
+    }
+
     /**
      * Returns time when snapshot started
      */
@@ -127,6 +156,23 @@ public long getProcessedSize() {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        if (out.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS)) {
+            // We use a boolean to indicate if the stats are present (true) or missing (false), to skip writing all the values if missing.
+            if (isMissingStats()) {
+                out.writeBoolean(false);
+                return;
+            }
+            out.writeBoolean(true);
+        } else if (isMissingStats()) {
+            throw new IllegalStateException(
+                "cannot serialize empty stats for transport version ["
+                    + out.getTransportVersion()
+                    + "] less than ["
+                    + SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS
+                    + "]"
+            );
+        }
+
         out.writeVLong(startTime);
         out.writeVLong(time);
 
@@ -196,6 +242,10 @@ public XContentBuilder toXContent(XContentBuilder builder, ToXContent.Params par
      * @param updateTimestamps Whether or not start time and duration should be updated
      */
     void add(SnapshotStats stats, boolean updateTimestamps) {
+        if (stats.isMissingStats()) {
+            return;
+        }
+
         incrementalFileCount += stats.incrementalFileCount;
         totalFileCount += stats.totalFileCount;
         processedFileCount += stats.processedFileCount;

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

@@ -11,6 +11,8 @@
 
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.ActionType;
 import org.elasticsearch.action.support.ActionFilters;
@@ -119,7 +121,7 @@ protected void masterOperation(
             Arrays.asList(request.snapshots())
         );
         if (currentSnapshots.isEmpty()) {
-            buildResponse(snapshotsInProgress, request, currentSnapshots, null, cancellableTask, listener);
+            buildResponse(snapshotsInProgress, request, currentSnapshots, null, state.getMinTransportVersion(), cancellableTask, listener);
             return;
         }
 
@@ -152,6 +154,7 @@ protected void masterOperation(
                             request,
                             currentSnapshots,
                             nodeSnapshotStatuses,
+                            state.getMinTransportVersion(),
                             cancellableTask,
                             l
                         )
@@ -160,7 +163,7 @@ protected void masterOperation(
             );
         } else {
             // We don't have any in-progress shards, just return current stats
-            buildResponse(snapshotsInProgress, request, currentSnapshots, null, cancellableTask, listener);
+            buildResponse(snapshotsInProgress, request, currentSnapshots, null, state.getMinTransportVersion(), cancellableTask, listener);
         }
 
     }
@@ -171,6 +174,7 @@ void buildResponse(
         SnapshotsStatusRequest request,
         List<SnapshotsInProgress.Entry> currentSnapshotEntries,
         TransportNodesSnapshotsStatus.NodesSnapshotStatus nodeSnapshotStatuses,
+        TransportVersion minTransportVersion,
         CancellableTask task,
         ActionListener<SnapshotsStatusResponse> listener
     ) {
@@ -196,13 +200,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,27 +232,34 @@ 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());
-                        shardStatus = new SnapshotIndexShardStatus(
-                            shardId,
-                            repositoriesService.repository(entry.repository())
-                                .getShardSnapshotStatus(
-                                    entry.snapshot().getSnapshotId(),
-                                    entry.indices().get(shardId.getIndexName()),
-                                    shardId
-                                )
-                        );
+                        // 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 once the snapshot has completed.
+                        if (minTransportVersion.onOrAfter(TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS)) {
+                            shardStatus = SnapshotIndexShardStatus.forDoneButMissingStats(shardId, """
+                                Snapshot shard stats missing from a currently running snapshot due to a node leaving the cluster after \
+                                completing the shard snapshot; retry once the snapshot has completed to load all shard stats from the \
+                                repository.""");
+                        } else {
+                            // BWC behavior, load the stats directly from the repository.
+                            shardStatus = new SnapshotIndexShardStatus(
+                                shardId,
+                                repositoriesService.repository(entry.repository())
+                                    .getShardSnapshotStatus(
+                                        entry.snapshot().getSnapshotId(),
+                                        entry.indices().get(shardId.getIndexName()),
+                                        shardId
+                                    )
+                            );
+                        }
                     } else {
                         shardStatus = new SnapshotIndexShardStatus(entry.shardId(shardEntry.getKey()), stage);
                     }