Allow missing shard stats for restarted nodes for _snapshot/_status

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;
@@ -249,22 +248,23 @@ 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
-        );
+        final var snapshotStatusAfterRestart = getSnapshotStatus(repoName, snapshotOne);
+        final var snapshotShardStateAfterNodeRestart = stateFirstShard(snapshotStatusAfterRestart, indexTwo);
         assertThat(snapshotShardStateAfterNodeRestart.getStage(), is(SnapshotIndexShardStage.DONE));
-        assertNotNull("expected a description string for missing stats", snapshotShardStateAfterNodeRestart.getDescription());
-        assertThat(snapshotShardStateAfterNodeRestart.getStats().getTotalFileCount(), equalTo(0));
-        assertThat(snapshotShardStateAfterNodeRestart.getStats().getTotalSize(), equalTo(0L));
+        assertNotNull("expected a non-null description string for missing stats", snapshotShardStateAfterNodeRestart.getDescription());
+        final var missingStats = snapshotShardStateAfterNodeRestart.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

@@ -273,7 +273,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion INFERENCE_CUSTOM_SERVICE_ADDED = def(9_084_0_00);
     public static final TransportVersion ESQL_LIMIT_ROW_SIZE = def(9_085_0_00);
     public static final TransportVersion ESQL_REGEX_MATCH_WITH_CASE_INSENSITIVITY = def(9_086_0_00);
-    public static final TransportVersion SNAPSHOT_INDEX_SHARD_STATUS_DESCRIPTION_ADDED = def(9_087_0_00);
+    public static final TransportVersion SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS = def(9_087_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -21,7 +21,7 @@
 import java.io.IOException;
 import java.util.Objects;
 
-import static org.elasticsearch.TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_DESCRIPTION_ADDED;
+import static org.elasticsearch.TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS;
 
 public class SnapshotIndexShardStatus extends BroadcastShardResponse implements ToXContentFragment {
 
@@ -41,7 +41,7 @@ public SnapshotIndexShardStatus(StreamInput in) throws IOException {
         stats = new SnapshotStats(in);
         nodeId = in.readOptionalString();
         failure = in.readOptionalString();
-        if (in.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_DESCRIPTION_ADDED)) {
+        if (in.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS)) {
             description = in.readOptionalString();
         }
     }
@@ -100,6 +100,20 @@ public SnapshotIndexShardStatus(StreamInput in) throws IOException {
         this.description = description;
     }
 
+    /**
+     * Creates an instance for scenarios where the snapshot stats are unavailable, with a non-null description of why the stats are missing.
+     */
+    public static SnapshotIndexShardStatus forMissingStats(ShardId shardId, SnapshotIndexShardStage stage, String description) {
+        return new SnapshotIndexShardStatus(
+            shardId,
+            stage,
+            SnapshotStats.forMissingStats(),
+            null,
+            null,
+            Objects.requireNonNull(description)
+        );
+    }
+
     /**
      * Returns snapshot stage
      */
@@ -142,7 +156,7 @@ public void writeTo(StreamOutput out) throws IOException {
         stats.writeTo(out);
         out.writeOptionalString(nodeId);
         out.writeOptionalString(failure);
-        if (out.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_DESCRIPTION_ADDED)) {
+        if (out.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS)) {
             out.writeOptionalString(description);
         }
     }

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

@@ -23,6 +23,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;
@@ -37,6 +39,18 @@ public class SnapshotStats implements Writeable, ToXContentObject {
     SnapshotStats() {}
 
     SnapshotStats(StreamInput in) throws IOException {
+        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();
 
@@ -71,6 +85,20 @@ public class SnapshotStats implements Writeable, ToXContentObject {
         this.processedSize = processedSize;
     }
 
+    /**
+     * Returns a stats instance with -1 for all non-time 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
      */
@@ -129,6 +157,22 @@ public long getProcessedSize() {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        if (out.getTransportVersion().onOrAfter(SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS)) {
+            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);
 
@@ -300,6 +344,10 @@ public static SnapshotStats fromXContent(XContentParser parser) throws IOExcepti
      * @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
     ) {
@@ -235,19 +239,27 @@ void buildResponse(
                     };
                     final SnapshotIndexShardStatus shardStatus;
                     if (stage == SnapshotIndexShardStage.DONE) {
+                        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(
-                            entry.shardId(shardEntry.getKey()),
-                            stage,
-                            new SnapshotStats(),
-                            null,
-                            null,
-                            """
+                        // that can be used to load them once the snapshot has completed.
+                        if (minTransportVersion.onOrAfter(TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS)) {
+                            shardStatus = SnapshotIndexShardStatus.forMissingStats(shardId, stage, """
                                 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."""
-                        );
+                                completing the shard snapshot; use /_snapshot/<repository>/<snapshot>/_status to load from the repository \
+                                once the snapshot has completed.""");
+                        } 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);
                     }

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

@@ -9,10 +9,16 @@
 
 package org.elasticsearch.action.admin.cluster.snapshots.status;
 
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.TransportVersions;
+import org.elasticsearch.common.io.stream.ByteArrayStreamInput;
+import org.elasticsearch.common.io.stream.OutputStreamStreamOutput;
 import org.elasticsearch.test.AbstractXContentTestCase;
 import org.elasticsearch.xcontent.XContentParser;
 
+import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.List;
 
 public class SnapshotStatsTests extends AbstractXContentTestCase<SnapshotStats> {
 
@@ -48,4 +54,46 @@ protected SnapshotStats doParseInstance(XContentParser parser) throws IOExceptio
     protected boolean supportsUnknownFields() {
         return true;
     }
+
+    public void testMissingStats() throws IOException {
+        final var populatedStats = createTestInstance();
+        final var missingStats = SnapshotStats.forMissingStats();
+        assertEquals(0L, missingStats.getStartTime());
+        assertEquals(0L, missingStats.getTime());
+        assertEquals(-1, missingStats.getTotalFileCount());
+        assertEquals(-1, missingStats.getIncrementalFileCount());
+        assertEquals(-1, missingStats.getProcessedFileCount());
+        assertEquals(-1L, missingStats.getTotalSize());
+        assertEquals(-1L, missingStats.getIncrementalSize());
+        assertEquals(-1L, missingStats.getProcessedSize());
+
+        // Verify round trip serialization.
+        for (var transportVersion : List.of(
+            TransportVersions.MINIMUM_COMPATIBLE,
+            TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS,
+            TransportVersion.current()
+        )) {
+
+            for (var stats : List.of(populatedStats, missingStats)) {
+                final var bytesOut = new ByteArrayOutputStream();
+
+                try (var streamOut = new OutputStreamStreamOutput(bytesOut)) {
+                    streamOut.setTransportVersion(transportVersion);
+
+                    if (transportVersion.onOrAfter(TransportVersions.SNAPSHOT_INDEX_SHARD_STATUS_MISSING_STATS) || stats != missingStats) {
+                        stats.writeTo(streamOut);
+                    } else {
+                        assertThrows(IllegalStateException.class, () -> stats.writeTo(streamOut));
+                        continue;
+                    }
+                }
+
+                try (var streamIn = new ByteArrayStreamInput(bytesOut.toByteArray())) {
+                    streamIn.setTransportVersion(transportVersion);
+                    final var statsRead = new SnapshotStats(streamIn);
+                    assertEquals(stats, statsRead);
+                }
+            }
+        }
+    }
 }