HDDS-7507. [Snapshot] Implement List Snapshot API Pagination

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

@@ -551,14 +551,81 @@ public void deleteSnapshot(String volumeName,
 
   /**
    * List snapshots in a volume/bucket.
-   * @param volumeName volume name
-   * @param bucketName bucket name
-   * @return list of snapshots for volume/bucket snapshotpath.
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   snapshots will be listed after this snapshot name
    * @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) {
+      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) {
+        currentIterator = getNextListOfSnapshots(currentValue.getName())
+            .iterator();
+      }
+      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) {
+      try {
+        return proxy.listSnapshot(volumeName, bucketName, snapshotPrefix,
+            prevSnapshot, listCacheSize);
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
   }
 
   public SnapshotDiffReport snapshotDiff(String volumeName, String bucketName,

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

@@ -1042,13 +1042,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

@@ -985,19 +985,24 @@ public SnapshotDiffReport snapshotDiff(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 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

@@ -664,13 +664,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

@@ -1153,13 +1153,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/om/TestOmMetrics.java

@@ -430,7 +430,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/interface-client/src/main/proto/OmClientProtocol.proto

@@ -1674,6 +1674,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

@@ -224,12 +224,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

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OmMetadataManagerImpl.java

@@ -1181,8 +1181,9 @@ public List<RepeatedOmKeyInfo> listTrash(String volumeName, String bucketName,
   }
 
   @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 {
     if (Strings.isNullOrEmpty(volumeName)) {
       throw new OMException("Volume name is required.", VOLUME_NOT_FOUND);
     }
@@ -1197,39 +1198,69 @@ public List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
           BUCKET_NOT_FOUND);
     }
 
-    String prefix = getBucketKey(volumeName, bucketName + OM_KEY_PREFIX);
+    String prefix;
+    if (StringUtil.isNotBlank(snapshotPrefix)) {
+      prefix = getOzoneKey(volumeName, bucketName, snapshotPrefix);
+    } else {
+      prefix = getBucketKey(volumeName, bucketName + OM_KEY_PREFIX);
+    }
+
+    String seek;
+    if (StringUtil.isNotBlank(prevSnapshot)) {
+      // Seek to the specified snapshot.
+      seek = getOzoneKey(volumeName, bucketName, prevSnapshot);
+    } else {
+      // This allows us to seek directly to the first key with the right prefix.
+      seek = getOzoneKey(volumeName, bucketName,
+          StringUtil.isNotBlank(
+              snapshotPrefix) ? snapshotPrefix : OM_KEY_PREFIX);
+    }
+
     TreeMap<String, SnapshotInfo> snapshotInfoMap = new TreeMap<>();
 
-    appendSnapshotFromCacheToMap(snapshotInfoMap, prefix);
-    appendSnapshotFromDBToMap(snapshotInfoMap, prefix);
+    int count = appendSnapshotFromCacheToMap(
+        snapshotInfoMap, prefix, seek, maxListResult);
+    appendSnapshotFromDBToMap(
+        snapshotInfoMap, prefix, seek, count, maxListResult);
 
     return new ArrayList<>(snapshotInfoMap.values());
   }
 
-  private void appendSnapshotFromCacheToMap(
-      TreeMap snapshotInfoMap, String prefix) {
+  private int appendSnapshotFromCacheToMap(
+      TreeMap snapshotInfoMap, String prefix,
+      String previous, int maxListResult) {
+    int count = 0;
     Iterator<Map.Entry<CacheKey<String>, CacheValue<SnapshotInfo>>> iterator =
         snapshotInfoTable.cacheIterator();
-    while (iterator.hasNext()) {
+    while (iterator.hasNext() && count < maxListResult) {
       Map.Entry<CacheKey<String>, CacheValue<SnapshotInfo>> entry =
           iterator.next();
       String snapshotKey = entry.getKey().getCacheKey();
       SnapshotInfo snapshotInfo = entry.getValue().getCacheValue();
-      if (snapshotInfo != null && snapshotKey.startsWith(prefix)) {
+      if (snapshotInfo != null && snapshotKey.startsWith(prefix) &&
+          snapshotKey.compareTo(previous) > 0) {
         snapshotInfoMap.put(snapshotKey, snapshotInfo);
+        count++;
       }
     }
+    return count;
   }
 
-  private void appendSnapshotFromDBToMap(TreeMap snapshotInfoMap, String prefix)
+  private void appendSnapshotFromDBToMap(TreeMap snapshotInfoMap,
+                                         String prefix, String previous,
+                                         int count, int maxListResult)
       throws IOException {
     try (TableIterator<String, ? extends KeyValue<String, SnapshotInfo>>
              snapshotIter = snapshotInfoTable.iterator()) {
       KeyValue< String, SnapshotInfo> snapshotinfo;
-      snapshotIter.seek(prefix);
-      while (snapshotIter.hasNext()) {
+      snapshotIter.seek(previous);
+      while (snapshotIter.hasNext() && count < maxListResult) {
         snapshotinfo = snapshotIter.next();
-        if (snapshotinfo != null && snapshotinfo.getKey().startsWith(prefix)) {
+        if (snapshotinfo != null &&
+            snapshotinfo.getKey().compareTo(previous) == 0) {
+          continue;
+        }
+        if (snapshotinfo != null && snapshotinfo.getKey().startsWith(prefix))  {
           CacheValue<SnapshotInfo> cacheValue =
               snapshotInfoTable.getCacheValue(
                   new CacheKey<>(snapshotinfo.getKey()));
@@ -1238,6 +1269,7 @@ private void appendSnapshotFromDBToMap(TreeMap snapshotInfoMap, String prefix)
           // in cache.
           if (cacheValue == null) {
             snapshotInfoMap.put(snapshotinfo.getKey(), snapshotinfo.getValue());
+            count++;
           }
         } else {
           break;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java

@@ -2777,8 +2777,9 @@ public List<RepeatedOmKeyInfo> listTrash(String volumeName,
   }
 
   @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 {
     if (isAclEnabled) {
       omMetadataReader.checkAcls(ResourceType.BUCKET, StoreType.OZONE,
           ACLType.LIST, volumeName, bucketName, null);
@@ -2788,7 +2789,8 @@ public List<SnapshotInfo> listSnapshot(String volumeName, String bucketName)
     auditMap.put(OzoneConsts.BUCKET, bucketName);
     try {
       metrics.incNumSnapshotLists();
-      return metadataManager.listSnapshot(volumeName, bucketName);
+      return metadataManager.listSnapshot(volumeName, bucketName, prevSnapshot,
+          prevSnapshot, maxListResult);
     } catch (Exception ex) {
       metrics.incNumSnapshotListFails();
       auditSuccess = false;

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/protocolPB/OzoneManagerRequestHandler.java

@@ -1243,7 +1243,8 @@ private OzoneManagerProtocolProtos.ListSnapshotResponse getSnapshots(
       OzoneManagerProtocolProtos.ListSnapshotRequest request)
       throws IOException {
     List<SnapshotInfo> snapshotInfos = impl.listSnapshot(
-        request.getVolumeName(), request.getBucketName());
+        request.getVolumeName(), request.getBucketName(), request.getPrefix(),
+        request.getPrevSnapshot(), request.getMaxListResult());
     List<OzoneManagerProtocolProtos.SnapshotInfo> snapshotInfoList =
         snapshotInfos.stream().map(SnapshotInfo::getProtobuf)
             .collect(Collectors.toList());