Add state query param to Get snapshots API

rest-api-spec/src/main/resources/rest-api-spec/api/snapshot.get.json

@@ -85,6 +85,10 @@
       "verbose":{
         "type":"boolean",
         "description":"Whether to show verbose snapshot info or only show the basic info found in the repository index blob"
+      },
+      "state": {
+        "type": "string",
+        "description": "Filter snapshots by state. Valid values are 'SUCCESS', 'IN_PROGRESS', 'FAILED', 'PARTIAL', or 'INCOMPATIBLE'."
       }
     }
   }

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

@@ -69,6 +69,7 @@
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.in;
 import static org.hamcrest.Matchers.is;
+import static org.hamcrest.core.StringContains.containsString;
 
 public class GetSnapshotsIT extends AbstractSnapshotIntegTestCase {
 
@@ -633,6 +634,57 @@ public void testRetrievingSnapshotsWhenRepositoryIsMissing() throws Exception {
         expectThrows(RepositoryMissingException.class, multiRepoFuture::actionGet);
     }
 
+    public void testFilterByState() throws Exception {
+        final String repoName = "test-repo";
+        final Path repoPath = randomRepoPath();
+        createRepository(repoName, "mock", repoPath);
+        maybeInitWithOldSnapshotVersion(repoName, repoPath);
+
+        // Create a successful snapshot
+        String successSnapshot = "snapshot-success";
+        createFullSnapshot(repoName, successSnapshot);
+
+        // Create a snapshot in progress
+        String inProgressSnapshot = "snapshot-in-progress";
+        blockAllDataNodes(repoName);
+        startFullSnapshot(repoName, inProgressSnapshot);
+        awaitNumberOfSnapshotsInProgress(1);
+        awaitClusterState(
+            state -> SnapshotsInProgress.get(state)
+                .asStream()
+                .flatMap(s -> s.shards().entrySet().stream())
+                .allMatch(
+                    e -> e.getKey().getIndexName().equals("test-index-1") == false
+                        || e.getValue().state() == SnapshotsInProgress.ShardState.SUCCESS
+                )
+        );
+
+        // Fetch all snapshots
+        GetSnapshotsResponse responseAll = clusterAdmin().prepareGetSnapshots(TEST_REQUEST_TIMEOUT, repoName).get();
+        assertThat(responseAll.getSnapshots(), hasSize(2));
+
+        // Fetch snapshots with state=SUCCESS
+        GetSnapshotsResponse responseSuccess = clusterAdmin().prepareGetSnapshots(TEST_REQUEST_TIMEOUT, repoName)
+            .setState(String.valueOf(SnapshotState.SUCCESS))
+            .get();
+        assertThat(responseSuccess.getSnapshots(), hasSize(1));
+        assertThat(responseSuccess.getSnapshots().get(0).state(), is(SnapshotState.SUCCESS));
+
+        // Fetch snapshots with state=IN_PROGRESS
+        GetSnapshotsResponse responseInProgress = clusterAdmin().prepareGetSnapshots(TEST_REQUEST_TIMEOUT, repoName)
+            .setState(String.valueOf(SnapshotState.IN_PROGRESS))
+            .get();
+        assertThat(responseInProgress.getSnapshots(), hasSize(1));
+        assertThat(responseInProgress.getSnapshots().get(0).state(), is(SnapshotState.IN_PROGRESS));
+
+        // Fetch snapshots with invalid state
+        IllegalArgumentException e = expectThrows(
+            IllegalArgumentException.class,
+            () -> clusterAdmin().prepareGetSnapshots(TEST_REQUEST_TIMEOUT, repoName).setState("INVALID_STATE").get()
+        );
+        assertThat(e.getMessage(), containsString("state must be SUCCESS, IN_PROGRESS, FAILED, PARTIAL, or INCOMPATIBLE"));
+    }
+
     // Create a snapshot that is guaranteed to have a unique start time and duration for tests around ordering by either.
     // Don't use this with more than 3 snapshots on platforms with low-resolution clocks as the durations could always collide there
     // causing an infinite loop

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

@@ -206,6 +206,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS = def(9_016_0_00);
     public static final TransportVersion ESQL_DRIVER_NODE_DESCRIPTION = def(9_017_0_00);
     public static final TransportVersion MULTI_PROJECT = def(9_018_0_00);
+    public static final TransportVersion STATE_PARAM_GET_SNAPSHOT = def(9_019_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/GetSnapshotsRequest.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.search.sort.SortOrder;
+import org.elasticsearch.snapshots.SnapshotState;
 import org.elasticsearch.tasks.CancellableTask;
 import org.elasticsearch.tasks.Task;
 import org.elasticsearch.tasks.TaskId;
@@ -39,6 +40,7 @@ public class GetSnapshotsRequest extends MasterNodeRequest<GetSnapshotsRequest>
     public static final boolean DEFAULT_VERBOSE_MODE = true;
 
     private static final TransportVersion INDICES_FLAG_VERSION = TransportVersions.V_8_3_0;
+    private static final TransportVersion STATE_FLAG_VERSION = TransportVersions.STATE_PARAM_GET_SNAPSHOT;
 
     public static final int NO_LIMIT = -1;
 
@@ -77,6 +79,8 @@ public class GetSnapshotsRequest extends MasterNodeRequest<GetSnapshotsRequest>
 
     private boolean includeIndexNames = true;
 
+    private String state;
+
     public GetSnapshotsRequest(TimeValue masterNodeTimeout) {
         super(masterNodeTimeout);
     }
@@ -118,6 +122,9 @@ public GetSnapshotsRequest(StreamInput in) throws IOException {
         if (in.getTransportVersion().onOrAfter(INDICES_FLAG_VERSION)) {
             includeIndexNames = in.readBoolean();
         }
+        if (in.getTransportVersion().onOrAfter(STATE_FLAG_VERSION)) {
+            state = in.readOptionalString();
+        }
     }
 
     @Override
@@ -137,6 +144,9 @@ public void writeTo(StreamOutput out) throws IOException {
         if (out.getTransportVersion().onOrAfter(INDICES_FLAG_VERSION)) {
             out.writeBoolean(includeIndexNames);
         }
+        if (out.getTransportVersion().onOrAfter(STATE_FLAG_VERSION)) {
+            out.writeOptionalString(state);
+        }
     }
 
     @Override
@@ -177,6 +187,12 @@ public ActionRequestValidationException validate() {
         } else if (after != null && fromSortValue != null) {
             validationException = addValidationError("can't use after and from_sort_value simultaneously", validationException);
         }
+        if (state != null && Arrays.stream(SnapshotState.values()).noneMatch(s -> s.name().equalsIgnoreCase(state))) {
+            validationException = addValidationError(
+                "state must be SUCCESS, IN_PROGRESS, FAILED, PARTIAL, or INCOMPATIBLE",
+                validationException
+            );
+        }
         return validationException;
     }
 
@@ -342,6 +358,15 @@ public boolean verbose() {
         return verbose;
     }
 
+    public String state() {
+        return state;
+    }
+
+    public GetSnapshotsRequest state(String state) {
+        this.state = state;
+        return this;
+    }
+
     @Override
     public Task createTask(long id, String type, String action, TaskId parentTaskId, Map<String, String> headers) {
         return new CancellableTask(id, type, action, getDescription(), parentTaskId, headers);

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/GetSnapshotsRequestBuilder.java

@@ -150,4 +150,8 @@ public GetSnapshotsRequestBuilder setIncludeIndexNames(boolean indices) {
 
     }
 
+    public GetSnapshotsRequestBuilder setState(String state) {
+        request.state(state);
+        return this;
+    }
 }

server/src/main/java/org/elasticsearch/action/admin/cluster/snapshots/get/TransportGetSnapshotsAction.java

@@ -160,7 +160,8 @@ protected void masterOperation(
             request.size(),
             SnapshotsInProgress.get(state),
             request.verbose(),
-            request.includeIndexNames()
+            request.includeIndexNames(),
+            request.state()
         ).runOperation(listener);
     }
 
@@ -181,6 +182,7 @@ private class GetSnapshotsOperation {
         private final SnapshotNamePredicate snapshotNamePredicate;
         private final SnapshotPredicates fromSortValuePredicates;
         private final Predicate<String> slmPolicyPredicate;
+        private final String state;
 
         // snapshot ordering/pagination
         private final SnapshotSortKey sortBy;
@@ -224,7 +226,8 @@ private class GetSnapshotsOperation {
             int size,
             SnapshotsInProgress snapshotsInProgress,
             boolean verbose,
-            boolean indices
+            boolean indices,
+            String state
         ) {
             this.cancellableTask = cancellableTask;
             this.repositories = repositories;
@@ -237,6 +240,7 @@ private class GetSnapshotsOperation {
             this.snapshotsInProgress = snapshotsInProgress;
             this.verbose = verbose;
             this.indices = indices;
+            this.state = state;
 
             this.snapshotNamePredicate = SnapshotNamePredicate.forSnapshots(ignoreUnavailable, snapshots);
             this.fromSortValuePredicates = SnapshotPredicates.forFromSortValue(fromSortValue, sortBy, order);
@@ -571,6 +575,10 @@ private boolean matchesPredicates(SnapshotInfo snapshotInfo) {
                 return false;
             }
 
+            if (state != null) {
+                return state.equalsIgnoreCase(snapshotInfo.state().toString());
+            }
+
             if (slmPolicyPredicate == SlmPolicyPredicate.MATCH_ALL_POLICIES) {
                 return true;
             }

server/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestGetSnapshotsAction.java

@@ -82,6 +82,7 @@ public RestChannelConsumer prepareRequest(final RestRequest request, final NodeC
         final SortOrder order = SortOrder.fromString(request.param("order", getSnapshotsRequest.order().toString()));
         getSnapshotsRequest.order(order);
         getSnapshotsRequest.includeIndexNames(request.paramAsBoolean(INDEX_NAMES_XCONTENT_PARAM, getSnapshotsRequest.includeIndexNames()));
+        getSnapshotsRequest.state(request.param("state"));
         return channel -> new RestCancellableNodeClient(client, request.getHttpChannel()).admin()
             .cluster()
             .getSnapshots(getSnapshotsRequest, new RestRefCountedChunkedToXContentListener<>(channel));

server/src/test/java/org/elasticsearch/action/admin/cluster/snapshots/get/GetSnapshotsRequestTests.java

@@ -86,6 +86,12 @@ public void testValidateParameters() {
             final ActionRequestValidationException e = request.validate();
             assertThat(e.getMessage(), containsString("can't use slm policy filter with verbose=false"));
         }
+        {
+            final GetSnapshotsRequest request = new GetSnapshotsRequest(TEST_REQUEST_TIMEOUT, "repo", "snapshot").state("foo")
+                .verbose(false);
+            final ActionRequestValidationException e = request.validate();
+            assertThat(e.getMessage(), containsString("state must be SUCCESS, IN_PROGRESS, FAILED, PARTIAL, or INCOMPATIBLE"));
+        }
     }
 
     public void testGetDescription() {