HDDS-7507. [Snapshot] Implement List Snapshot API Pagination (#4065)

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/ObjectStore.java

@@ -578,14 +578,84 @@ public void deleteSnapshot(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   snapshots will be listed after this snapshot name
    * @return list of snapshots for volume/bucket snapshot path.
    * @throws IOException
    */
-  public List<OzoneSnapshot> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
-    return proxy.listSnapshot(volumeName, bucketName);
+  public Iterator<? extends OzoneSnapshot> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot) throws IOException {
+    return new SnapshotIterator(
+        volumeName, bucketName, snapshotPrefix, prevSnapshot);
+  }
+
+  /**
+   * An Iterator to iterate over {@link OzoneSnapshot} list.
+   */
+  private class SnapshotIterator implements Iterator<OzoneSnapshot> {
+
+    private String volumeName = null;
+    private String bucketName = null;
+    private String snapshotPrefix = null;
+
+    private Iterator<OzoneSnapshot> currentIterator;
+    private OzoneSnapshot currentValue;
+
+    /**
+     * Creates an Iterator to iterate over all snapshots after
+     * prevSnapshot of specified bucket. If prevSnapshot is null it iterates
+     * from the first snapshot. The returned snapshots match snapshot prefix.
+     * @param snapshotPrefix snapshot prefix to match
+     * @param prevSnapshot snapshots will be listed after this snapshot name
+     */
+    SnapshotIterator(String volumeName, String bucketName,
+                     String snapshotPrefix, String prevSnapshot)
+        throws IOException {
+      this.volumeName = volumeName;
+      this.bucketName = bucketName;
+      this.snapshotPrefix = snapshotPrefix;
+      this.currentValue = null;
+      this.currentIterator = getNextListOfSnapshots(prevSnapshot).iterator();
+    }
+
+    @Override
+    public boolean hasNext() {
+      // IMPORTANT: Without this logic, remote iteration will not work.
+      // Removing this will break the listSnapshot call if we try to
+      // list more than 1000 (ozone.client.list.cache ) snapshots.
+      if (!currentIterator.hasNext() && currentValue != null) {
+        try {
+          currentIterator = getNextListOfSnapshots(currentValue.getName())
+              .iterator();
+        } catch (IOException e) {
+          LOG.error("Error retrieving next batch of list results", e);
+        }
+      }
+      return currentIterator.hasNext();
+    }
+
+    @Override
+    public OzoneSnapshot next() {
+      if (hasNext()) {
+        currentValue = currentIterator.next();
+        return currentValue;
+      }
+      throw new NoSuchElementException();
+    }
+
+    /**
+     * Returns the next set of snapshot list using proxy.
+     * @param prevSnapshot previous snapshot, this will be excluded from result
+     * @return {@code List<OzoneSnapshot>}
+     */
+    private List<OzoneSnapshot> getNextListOfSnapshots(String prevSnapshot)
+        throws IOException {
+      return proxy.listSnapshot(volumeName, bucketName, snapshotPrefix,
+          prevSnapshot, listCacheSize);
+    }
   }
 
   /**

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/protocol/ClientProtocol.java

@@ -1045,13 +1045,17 @@ void deleteSnapshot(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   start of the list, this snapshot is excluded
+   * @param maxListResult  max numbet of snapshots to return
    * @return list of snapshots for volume/bucket snapshotpath.
    * @throws IOException
    */
-  List<OzoneSnapshot> listSnapshot(String volumeName, String bucketName)
-      throws IOException;
+  List<OzoneSnapshot> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException;
 
 
   /**

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -998,19 +998,24 @@ public SnapshotDiffResponse snapshotDiff(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   start of the list, this snapshot is excluded
+   * @param maxListResult  max numbet of snapshots to return
    * @return list of snapshots for volume/bucket snapshotpath.
    * @throws IOException
    */
   @Override
-  public List<OzoneSnapshot> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
+  public List<OzoneSnapshot> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
     Preconditions.checkArgument(Strings.isNotBlank(volumeName),
         "volume can't be null or empty.");
     Preconditions.checkArgument(Strings.isNotBlank(bucketName),
         "bucket can't be null or empty.");
-    return ozoneManagerClient.listSnapshot(volumeName, bucketName).stream()
+    return ozoneManagerClient.listSnapshot(volumeName, bucketName,
+            snapshotPrefix, prevSnapshot, maxListResult).stream()
         .map(snapshotInfo -> OzoneSnapshot.fromSnapshotInfo(snapshotInfo))
         .collect(Collectors.toList());
   }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocol/OzoneManagerProtocol.java

@@ -689,13 +689,17 @@ default void deleteSnapshot(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   start of the list, this snapshot is excluded
+   * @param maxListResult  max numbet of snapshots to return
    * @return list of snapshots for volume/bucket snapshotpath.
    * @throws IOException
    */
-  default List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
+  default List<SnapshotInfo> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
     throw new UnsupportedOperationException("OzoneManager does not require " +
         "this to be implemented");
   }

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -1171,13 +1171,23 @@ public void deleteSnapshot(String volumeName,
    * {@inheritDoc}
    */
   @Override
-  public List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
-      throws IOException {
+  public List<SnapshotInfo> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException {
     final OzoneManagerProtocolProtos.ListSnapshotRequest.Builder
         requestBuilder =
         OzoneManagerProtocolProtos.ListSnapshotRequest.newBuilder()
             .setVolumeName(volumeName)
-            .setBucketName(bucketName);
+            .setBucketName(bucketName)
+            .setMaxListResult(maxListResult);
+
+    if (prevSnapshot != null) {
+      requestBuilder.setPrevSnapshot(prevSnapshot);
+    }
+
+    if (snapshotPrefix != null) {
+      requestBuilder.setPrefix(snapshotPrefix);
+    }
 
     final OMRequest omRequest = createOMRequest(Type.ListSnapshot)
         .setListSnapshotRequest(requestBuilder)

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java

@@ -78,6 +78,7 @@
 import org.apache.hadoop.ozone.client.OzoneKeyDetails;
 import org.apache.hadoop.ozone.client.OzoneKeyLocation;
 import org.apache.hadoop.ozone.client.OzoneMultipartUploadPartListParts;
+import org.apache.hadoop.ozone.client.OzoneSnapshot;
 import org.apache.hadoop.ozone.client.OzoneVolume;
 import org.apache.hadoop.ozone.client.VolumeArgs;
 import org.apache.hadoop.ozone.client.io.OzoneInputStream;
@@ -4159,4 +4160,85 @@ private void assertBucketCount(OzoneVolume volume,
     Assert.assertEquals(expectedBucketCount, bucketCount);
   }
 
+  @Test
+  public void testListSnapshot() throws IOException {
+    String volumeA = "vol-a-" + RandomStringUtils.randomNumeric(5);
+    String volumeB = "vol-b-" + RandomStringUtils.randomNumeric(5);
+    String bucketA = "buc-a-" + RandomStringUtils.randomNumeric(5);
+    String bucketB = "buc-b-" + RandomStringUtils.randomNumeric(5);
+    store.createVolume(volumeA);
+    store.createVolume(volumeB);
+    OzoneVolume volA = store.getVolume(volumeA);
+    OzoneVolume volB = store.getVolume(volumeB);
+    volA.createBucket(bucketA);
+    volA.createBucket(bucketB);
+    volB.createBucket(bucketA);
+    volB.createBucket(bucketB);
+    String snapshotPrefixA = "snapshot-a-";
+    String snapshotPrefixB = "snapshot-b-";
+    for (int i = 0; i < 10; i++) {
+      store.createSnapshot(volumeA, bucketA,
+          snapshotPrefixA + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeA, bucketB,
+          snapshotPrefixA + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeB, bucketA,
+          snapshotPrefixA + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeB, bucketB,
+          snapshotPrefixA + i + "-" + RandomStringUtils.randomNumeric(5));
+    }
+    for (int i = 0; i < 10; i++) {
+      store.createSnapshot(volumeA, bucketA,
+          snapshotPrefixB + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeA, bucketB,
+          snapshotPrefixB + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeB, bucketA,
+          snapshotPrefixB + i + "-" + RandomStringUtils.randomNumeric(5));
+      store.createSnapshot(volumeB, bucketB,
+          snapshotPrefixB + i + "-" + RandomStringUtils.randomNumeric(5));
+    }
+
+    Iterator<? extends OzoneSnapshot> snapshotIter =
+        store.listSnapshot(volumeA, bucketA, null, null);
+    int volABucketASnapshotCount = 0;
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      volABucketASnapshotCount++;
+    }
+    Assert.assertEquals(20, volABucketASnapshotCount);
+
+    snapshotIter = store.listSnapshot(volumeA, bucketB, null, null);
+    int volABucketBSnapshotCount = 0;
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      volABucketBSnapshotCount++;
+    }
+    Assert.assertEquals(20, volABucketASnapshotCount);
+
+    snapshotIter = store.listSnapshot(volumeB, bucketA, null, null);
+    int volBBucketASnapshotCount = 0;
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      volBBucketASnapshotCount++;
+    }
+    Assert.assertEquals(20, volABucketASnapshotCount);
+
+    snapshotIter = store.listSnapshot(volumeB, bucketB, null, null);
+    int volBBucketBSnapshotCount = 0;
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      volBBucketBSnapshotCount++;
+    }
+    Assert.assertEquals(20, volABucketASnapshotCount);
+
+    int volABucketASnapshotACount = 0;
+    snapshotIter = store.listSnapshot(volumeA, bucketA, snapshotPrefixA, null);
+    while (snapshotIter.hasNext()) {
+      OzoneSnapshot snapshot = snapshotIter.next();
+      Assert.assertTrue(snapshot.getName().startsWith(snapshotPrefixA));
+      volABucketASnapshotACount++;
+    }
+    Assert.assertEquals(10, volABucketASnapshotACount);
+    Assert.assertFalse(snapshotIter.hasNext());
+
+  }
 }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmMetrics.java

@@ -436,7 +436,8 @@ public void testSnapshotOps() throws Exception {
     writeClient.createSnapshot(volumeName, bucketName, snapshot2);
 
     // List snapshots
-    writeClient.listSnapshot(volumeName, bucketName);
+    writeClient.listSnapshot(
+        volumeName, bucketName, null, null, Integer.MAX_VALUE);
 
     omMetrics = getMetrics("OMMetrics");
     assertCounter("NumSnapshotActive", 2L, omMetrics);

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOmSnapshot.java

@@ -225,7 +225,7 @@ private static void preFinalizationChecks() throws Exception {
         store.createSnapshot(volumeName, bucketName,
             UUID.randomUUID().toString()));
     expectFailurePreFinalization(() ->
-        store.listSnapshot(volumeName, bucketName));
+        store.listSnapshot(volumeName, bucketName, null, null));
     expectFailurePreFinalization(() ->
         store.snapshotDiff(volumeName, bucketName,
             UUID.randomUUID().toString(),

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -1697,6 +1697,9 @@ message CreateSnapshotRequest {
 message ListSnapshotRequest {
   optional string volumeName = 1;
   optional string bucketName = 2;
+  optional string prefix = 3;
+  optional string prevSnapshot = 4;
+  optional uint32 maxListResult = 5;
 }
 
 message SnapshotDiffRequest {

hadoop-ozone/interface-storage/src/main/java/org/apache/hadoop/ozone/om/OMMetadataManager.java

@@ -225,12 +225,16 @@ List<RepeatedOmKeyInfo> listTrash(String volumeName, String bucketName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   start of the list, this snapshot is excluded
+   * @param maxListResult  max numbet of snapshots to return
    * @return list of snapshot
    */
-  List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
-      throws IOException;
+  List<SnapshotInfo> listSnapshot(
+      String volumeName, String bucketName, String snapshotPrefix,
+      String prevSnapshot, int maxListResult) throws IOException;
 
   /**
    * Recover trash allows the user to recover the keys