Implement Numeric Pagination in Get Snapshots API (#76532)

* Return Total Result Count and Remaining Count in Get Snapshots Response (#76150)

Add total result count and remaining count to get snapshots response.

* Implement Numeric Offset Parameter in Get Snapshots API (#76233)

Add numeric offset parameter to this API.

Relates #74350

Author: original-brownbear

docs/reference/snapshot-restore/apis/get-snapshot-api.asciidoc

@@ -134,12 +134,17 @@ Sort order. Valid values are `asc` for ascending and `desc` for descending order
 (Optional, string)
 Offset identifier to start pagination from as returned by the `next` field in the response body.
 
+`offset`::
+(Optional, integer)
+Numeric offset to start pagination from based on the snapshots matching this request. Using a non-zero value for this parameter is mutually
+exclusive with using the `after` parameter. Defaults to `0`.
+
 NOTE: The `after` parameter and `next` field allow for iterating through snapshots with some consistency guarantees regarding concurrent
 creation or deletion of snapshots. It is guaranteed that  any snapshot that exists at the beginning of the iteration and not concurrently
 deleted will be seen during the iteration. Snapshots concurrently created may be seen during an iteration.
 
-NOTE: The pagination parameters `size`, `order`, `after` and `sort` are not supported when using `verbose=false` and the sort order for
-requests with `verbose=false` is undefined.
+NOTE: The pagination parameters `size`, `order`, `after`, `offset` and `sort` are not supported when using `verbose=false` and the sort
+order for requests with `verbose=false` is undefined.
 
 [role="child_attributes"]
 [[get-snapshot-api-response-body]]
@@ -283,6 +288,15 @@ The snapshot `state` can be one of the following values:
 If the request contained a size limit and there might be more results, a `next` field will be added to the response and can be used as the
 `after` query parameter to fetch additional results.
 
+`total`::
+(integer)
+The total number of snapshots that match the request when ignoring size limit or `after` query parameter.
+
+`remaining`::
+(integer)
+The number of remaining snapshots that were not returned due to size limits and that can be fetched by additional requests using the `next`
+field value.
+
 [[get-snapshot-api-example]]
 ==== {api-examples-title}
 
@@ -322,7 +336,9 @@ The API returns the following response:
         "successful": 0
       }
     }
-  ]
+  ],
+  "total": 1,
+  "remaining": 0
 }
 ----
 // TESTRESPONSE[s/"uuid": "vdRctLCxSketdKb54xw67g"/"uuid": $body.snapshots.0.uuid/]
@@ -392,10 +408,12 @@ The API returns the following response:
         "total": 0,
         "failed": 0,
         "successful": 0
-      }
+      },
     }
   ],
-  "next": "c25hcHNob3RfMixteV9yZXBvc2l0b3J5LHNuYXBzaG90XzI="
+  "next": "c25hcHNob3RfMixteV9yZXBvc2l0b3J5LHNuYXBzaG90XzI=",
+  "total": 3,
+  "remaining": 1
 }
 ----
 // TESTRESPONSE[s/"uuid": "dKb54xw67gvdRctLCxSket"/"uuid": $body.snapshots.0.uuid/]
@@ -449,7 +467,59 @@ The API returns the following response:
         "successful": 0
       }
     }
-  ]
+  ],
+  "total": 3,
+  "remaining": 0
+}
+----
+// TESTRESPONSE[s/"uuid": "dRctdKb54xw67gvLCxSket"/"uuid": $body.snapshots.0.uuid/]
+// TESTRESPONSE[s/"version_id": <version_id>/"version_id": $body.snapshots.0.version_id/]
+// TESTRESPONSE[s/"version": <version>/"version": $body.snapshots.0.version/]
+// TESTRESPONSE[s/"start_time": "2020-07-06T21:55:18.129Z"/"start_time": $body.snapshots.0.start_time/]
+// TESTRESPONSE[s/"start_time_in_millis": 1593093628850/"start_time_in_millis": $body.snapshots.0.start_time_in_millis/]
+// TESTRESPONSE[s/"end_time": "2020-07-06T21:55:18.129Z"/"end_time": $body.snapshots.0.end_time/]
+// TESTRESPONSE[s/"end_time_in_millis": 1593094752018/"end_time_in_millis": $body.snapshots.0.end_time_in_millis/]
+// TESTRESPONSE[s/"duration_in_millis": 0/"duration_in_millis": $body.snapshots.0.duration_in_millis/]
+
+Alternatively, the same result could be retrieved by using an offset value of `2` to skip the two snapshot already seen.
+
+[source,console]
+----
+GET /_snapshot/my_repository/snapshot*?size=2&sort=name&offset=2
+----
+
+The API returns the following response:
+
+[source,console-result]
+----
+{
+  "snapshots": [
+    {
+      "snapshot": "snapshot_3",
+      "uuid": "dRctdKb54xw67gvLCxSket",
+      "repository": "my_repository",
+      "version_id": <version_id>,
+      "version": <version>,
+      "indices": [],
+      "data_streams": [],
+      "feature_states": [],
+      "include_global_state": true,
+      "state": "SUCCESS",
+      "start_time": "2020-07-06T21:55:18.129Z",
+      "start_time_in_millis": 1593093628850,
+      "end_time": "2020-07-06T21:55:18.129Z",
+      "end_time_in_millis": 1593094752018,
+      "duration_in_millis": 0,
+      "failures": [],
+      "shards": {
+        "total": 0,
+        "failed": 0,
+        "successful": 0
+      }
+    }
+  ],
+  "total": 3,
+  "remaining": 0
 }
 ----
 // TESTRESPONSE[s/"uuid": "dRctdKb54xw67gvLCxSket"/"uuid": $body.snapshots.0.uuid/]

qa/smoke-test-http/src/test/java/org/elasticsearch/http/snapshots/RestGetSnapshotsIT.java

@@ -20,7 +20,6 @@
 import org.elasticsearch.common.xcontent.NamedXContentRegistry;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.json.JsonXContent;
-import org.elasticsearch.core.Tuple;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.snapshots.AbstractSnapshotIntegTestCase;
 import org.elasticsearch.snapshots.SnapshotInfo;
@@ -106,31 +105,34 @@ private void doTestPagination(String repoName,
                                   GetSnapshotsRequest.SortBy sort,
                                   SortOrder order) throws IOException {
         final List<SnapshotInfo> allSnapshotsSorted = allSnapshotsSorted(names, repoName, sort, order);
-        final Tuple<String, List<SnapshotInfo>> batch1 = sortedWithLimit(repoName, sort, null, 2, order);
-        assertEquals(allSnapshotsSorted.subList(0, 2), batch1.v2());
-        final Tuple<String, List<SnapshotInfo>> batch2 = sortedWithLimit(repoName, sort, batch1.v1(), 2, order);
-        assertEquals(allSnapshotsSorted.subList(2, 4), batch2.v2());
-        final int lastBatch = names.size() - batch1.v2().size() - batch2.v2().size();
-        final Tuple<String, List<SnapshotInfo>> batch3 = sortedWithLimit(repoName, sort, batch2.v1(), lastBatch, order);
-        assertEquals(batch3.v2(), allSnapshotsSorted.subList(batch1.v2().size() + batch2.v2().size(), names.size()));
-        final Tuple<String, List<SnapshotInfo>> batch3NoLimit = sortedWithLimit(
+        final GetSnapshotsResponse batch1 = sortedWithLimit(repoName, sort, null, 2, order);
+        assertEquals(allSnapshotsSorted.subList(0, 2), batch1.getSnapshots());
+        final GetSnapshotsResponse batch2 = sortedWithLimit(repoName, sort, batch1.next(), 2, order);
+        assertEquals(allSnapshotsSorted.subList(2, 4), batch2.getSnapshots());
+        final int lastBatch = names.size() - batch1.getSnapshots().size() - batch2.getSnapshots().size();
+        final GetSnapshotsResponse batch3 = sortedWithLimit(repoName, sort, batch2.next(), lastBatch, order);
+        assertEquals(
+            batch3.getSnapshots(),
+            allSnapshotsSorted.subList(batch1.getSnapshots().size() + batch2.getSnapshots().size(), names.size())
+        );
+        final GetSnapshotsResponse batch3NoLimit = sortedWithLimit(
                 repoName,
                 sort,
-                batch2.v1(),
+                batch2.next(),
                 GetSnapshotsRequest.NO_LIMIT,
                 order
         );
-        assertNull(batch3NoLimit.v1());
-        assertEquals(batch3.v2(), batch3NoLimit.v2());
-        final Tuple<String, List<SnapshotInfo>> batch3LargeLimit = sortedWithLimit(
+        assertNull(batch3NoLimit.next());
+        assertEquals(batch3.getSnapshots(), batch3NoLimit.getSnapshots());
+        final GetSnapshotsResponse batch3LargeLimit = sortedWithLimit(
                 repoName,
                 sort,
-                batch2.v1(),
+                batch2.next(),
                 lastBatch + randomIntBetween(1, 100),
                 order
         );
-        assertEquals(batch3.v2(), batch3LargeLimit.v2());
-        assertNull(batch3LargeLimit.v1());
+        assertEquals(batch3.getSnapshots(), batch3LargeLimit.getSnapshots());
+        assertNull(batch3LargeLimit.next());
     }
 
     public void testSortAndPaginateWithInProgress() throws Exception {
@@ -180,16 +182,23 @@ private static void assertStablePagination(String repoName,
         final List<SnapshotInfo> allSorted = allSnapshotsSorted(allSnapshotNames, repoName, sort, order);
 
         for (int i = 1; i <= allSnapshotNames.size(); i++) {
-            final List<SnapshotInfo> subsetSorted = sortedWithLimit(repoName, sort, null, i, order).v2();
+            final List<SnapshotInfo> subsetSorted = sortedWithLimit(repoName, sort, null, i, order).getSnapshots();
             assertEquals(subsetSorted, allSorted.subList(0, i));
         }
 
         for (int j = 0; j < allSnapshotNames.size(); j++) {
             final SnapshotInfo after = allSorted.get(j);
             for (int i = 1; i < allSnapshotNames.size() - j; i++) {
-                final List<SnapshotInfo> subsetSorted = sortedWithLimit(
-                        repoName, sort, GetSnapshotsRequest.After.from(after, sort).asQueryParam(), i, order).v2();
+                final GetSnapshotsResponse getSnapshotsResponse =
+                    sortedWithLimit(repoName, sort, GetSnapshotsRequest.After.from(after, sort).asQueryParam(), i, order);
+                final GetSnapshotsResponse getSnapshotsResponseNumeric = sortedWithLimit(repoName, sort, j + 1, i, order);
+                final List<SnapshotInfo> subsetSorted = getSnapshotsResponse.getSnapshots();
+                assertEquals(subsetSorted, getSnapshotsResponseNumeric.getSnapshots());
                 assertEquals(subsetSorted, allSorted.subList(j + 1, j + i + 1));
+                assertEquals(allSnapshotNames.size(), getSnapshotsResponse.totalCount());
+                assertEquals(allSnapshotNames.size() - (j + i + 1), getSnapshotsResponse.remaining());
+                assertEquals(getSnapshotsResponseNumeric.totalCount(), getSnapshotsResponse.totalCount());
+                assertEquals(getSnapshotsResponseNumeric.remaining(), getSnapshotsResponse.remaining());
             }
         }
     }
@@ -203,9 +212,11 @@ private static List<SnapshotInfo> allSnapshotsSorted(Collection<String> allSnaps
         if (order == SortOrder.DESC || randomBoolean()) {
             request.addParameter("order", order.toString());
         }
-        final Response response = getRestClient().performRequest(request);
-        final List<SnapshotInfo> snapshotInfos = readSnapshotInfos(response).v2();
+        final GetSnapshotsResponse getSnapshotsResponse = readSnapshotInfos(getRestClient().performRequest(request));
+        final List<SnapshotInfo> snapshotInfos = getSnapshotsResponse.getSnapshots();
         assertEquals(snapshotInfos.size(), allSnapshotNames.size());
+        assertEquals(getSnapshotsResponse.totalCount(), allSnapshotNames.size());
+        assertEquals(0, getSnapshotsResponse.remaining());
         for (SnapshotInfo snapshotInfo : snapshotInfos) {
             assertThat(snapshotInfo.snapshotId().getName(), is(in(allSnapshotNames)));
         }
@@ -216,20 +227,19 @@ private static Request baseGetSnapshotsRequest(String repoName) {
         return new Request(HttpGet.METHOD_NAME, "/_snapshot/" + repoName + "/*");
     }
 
-    private static Tuple<String, List<SnapshotInfo>> readSnapshotInfos(Response response) throws IOException {
+    private static GetSnapshotsResponse readSnapshotInfos(Response response) throws IOException {
         try (InputStream input = response.getEntity().getContent();
              XContentParser parser = JsonXContent.jsonXContent.createParser(
                      NamedXContentRegistry.EMPTY, DeprecationHandler.THROW_UNSUPPORTED_OPERATION, input)) {
-            final GetSnapshotsResponse getSnapshotsResponse = GetSnapshotsResponse.fromXContent(parser);
-            return Tuple.tuple(getSnapshotsResponse.next(), getSnapshotsResponse.getSnapshots());
+            return GetSnapshotsResponse.fromXContent(parser);
         }
     }
 
-    private static Tuple<String, List<SnapshotInfo>> sortedWithLimit(String repoName,
-                                                                     GetSnapshotsRequest.SortBy sortBy,
-                                                                     String after,
-                                                                     int size,
-                                                                     SortOrder order) throws IOException {
+    private static GetSnapshotsResponse sortedWithLimit(String repoName,
+                                                        GetSnapshotsRequest.SortBy sortBy,
+                                                        String after,
+                                                        int size,
+                                                        SortOrder order) throws IOException {
         final Request request = baseGetSnapshotsRequest(repoName);
         request.addParameter("sort", sortBy.toString());
         if (size != GetSnapshotsRequest.NO_LIMIT || randomBoolean()) {
@@ -244,4 +254,22 @@ private static Tuple<String, List<SnapshotInfo>> sortedWithLimit(String repoName
         final Response response = getRestClient().performRequest(request);
         return readSnapshotInfos(response);
     }
+
+    private static GetSnapshotsResponse sortedWithLimit(String repoName,
+                                                        GetSnapshotsRequest.SortBy sortBy,
+                                                        int offset,
+                                                        int size,
+                                                        SortOrder order) throws IOException {
+        final Request request = baseGetSnapshotsRequest(repoName);
+        request.addParameter("sort", sortBy.toString());
+        if (size != GetSnapshotsRequest.NO_LIMIT || randomBoolean()) {
+            request.addParameter("size", String.valueOf(size));
+        }
+        request.addParameter("offset", String.valueOf(offset));
+        if (order == SortOrder.DESC || randomBoolean()) {
+            request.addParameter("order", order.toString());
+        }
+        final Response response = getRestClient().performRequest(request);
+        return readSnapshotInfos(response);
+    }
 }