HDDS-6856. HA aware reads from Snapshots

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneManagerVersion.java

@@ -54,6 +54,9 @@ public enum OzoneManagerVersion implements ComponentVersion {
 
   S3_LIST_MULTIPART_UPLOADS_PAGINATION(11,
       "OzoneManager version that supports S3 list multipart uploads API with pagination"),
+
+  SNAPSHOT_READ_FROM_NON_LEADER(12, "OzoneManager version that supports " +
+      "snapshot read operations from non-leader"),
 
   FUTURE_VERSION(-1, "Used internally in the client when the server side is "
       + " newer and an unknown server version has arrived to the client.");

hadoop-ozone/cli-shell/src/main/java/org/apache/hadoop/ozone/shell/snapshot/InfoSnapshotHandler.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.shell.snapshot;
 
 import java.io.IOException;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.ozone.client.OzoneClient;
 import org.apache.hadoop.ozone.client.OzoneSnapshot;
 import org.apache.hadoop.ozone.shell.Handler;
@@ -39,6 +40,11 @@ public class InfoSnapshotHandler extends Handler {
       index = "1", arity = "1")
   private String snapshotName;
 
+  @CommandLine.Option(
+      names = {"-n", "--om-node-id"},
+      description = "The id of OM node to get the snapshot information from")
+  private String omNodeId;
+
   @Override
   protected OzoneAddress getAddress() {
     return snapshotPath.getValue();
@@ -50,8 +56,14 @@ protected void execute(OzoneClient client, OzoneAddress address)
     String volumeName = snapshotPath.getValue().getVolumeName();
     String bucketName = snapshotPath.getValue().getBucketName();
 
-    OzoneSnapshot ozoneSnapshot = client.getObjectStore()
-        .getSnapshotInfo(volumeName, bucketName, snapshotName);
+    OzoneSnapshot ozoneSnapshot;
+    if (StringUtils.isEmpty(omNodeId)) {
+      ozoneSnapshot = client.getObjectStore()
+          .getSnapshotInfo(volumeName, bucketName, snapshotName);
+    } else {
+      ozoneSnapshot = client.getObjectStore()
+          .getSnapshotInfo(volumeName, bucketName, snapshotName, omNodeId);
+    }
 
     if (isVerbose()) {
       err().printf("Snapshot info for snapshot: %s under o3://%s/%s %n ",

hadoop-ozone/cli-shell/src/main/java/org/apache/hadoop/ozone/shell/snapshot/ListSnapshotDiffHandler.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.ozone.client.OzoneClient;
 import org.apache.hadoop.ozone.client.OzoneSnapshotDiff;
 import org.apache.hadoop.ozone.shell.Handler;
@@ -48,6 +49,11 @@ public class ListSnapshotDiffHandler extends Handler {
       defaultValue = "false")
   private boolean listAll;
 
+  @CommandLine.Option(
+      names = {"-n", "--om-node-id"},
+      description = "The id of OM node to list snapshotDiff jobs from")
+  private String omNodeId;
+
   @Override
   protected OzoneAddress getAddress() {
     return snapshotPath.getValue();
@@ -60,9 +66,14 @@ protected void execute(OzoneClient client, OzoneAddress address)
     String volumeName = snapshotPath.getValue().getVolumeName();
     String bucketName = snapshotPath.getValue().getBucketName();
 
-    List<OzoneSnapshotDiff> jobList =
-        client.getObjectStore().listSnapshotDiffJobs(
-            volumeName, bucketName, jobStatus, listAll);
+    List<OzoneSnapshotDiff> jobList;
+    if (StringUtils.isEmpty(omNodeId)) {
+      jobList = client.getObjectStore().listSnapshotDiffJobs(
+          volumeName, bucketName, jobStatus, listAll);
+    } else {
+      jobList = client.getObjectStore().listSnapshotDiffJobs(
+          volumeName, bucketName, jobStatus, listAll, omNodeId);
+    }
 
     int counter = printAsJsonArray(jobList.iterator(),
         jobList.size());

hadoop-ozone/cli-shell/src/main/java/org/apache/hadoop/ozone/shell/snapshot/ListSnapshotHandler.java

@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 import java.util.Iterator;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.ozone.client.OzoneClient;
 import org.apache.hadoop.ozone.client.OzoneSnapshot;
 import org.apache.hadoop.ozone.shell.Handler;
@@ -42,6 +43,11 @@ public class ListSnapshotHandler extends Handler {
   @CommandLine.Mixin
   private ListOptions listOptions;
 
+  @CommandLine.Option(
+      names = {"-n", "--om-node-id"},
+      description = "The id of OM node to ist the snapshots from")
+  private String omNodeId;
+
   @Override
   protected OzoneAddress getAddress() {
     return snapshotPath.getValue();
@@ -53,9 +59,17 @@ protected void execute(OzoneClient client, OzoneAddress address)
     String volumeName = snapshotPath.getValue().getVolumeName();
     String bucketName = snapshotPath.getValue().getBucketName();
 
-    Iterator<OzoneSnapshot> snapshotInfos = client.getObjectStore()
-        .listSnapshot(volumeName, bucketName, listOptions.getPrefix(),
-            listOptions.getStartItem());
+    Iterator<OzoneSnapshot> snapshotInfos;
+    if (StringUtils.isEmpty(omNodeId)) {
+      snapshotInfos = client.getObjectStore()
+          .listSnapshot(volumeName, bucketName, listOptions.getPrefix(),
+              listOptions.getStartItem());
+    } else {
+      snapshotInfos = client.getObjectStore()
+          .listSnapshot(volumeName, bucketName, listOptions.getPrefix(),
+              listOptions.getStartItem(), omNodeId);
+    }
+
     int counter = printAsJsonArray(snapshotInfos, listOptions.getLimit());
     if (isVerbose()) {
       err().printf("Found : %d snapshots for o3://%s/%s %n", counter,

hadoop-ozone/cli-shell/src/main/java/org/apache/hadoop/ozone/shell/snapshot/SnapshotDiffHandler.java

@@ -34,6 +34,7 @@
 import org.apache.hadoop.ozone.shell.Handler;
 import org.apache.hadoop.ozone.shell.OzoneAddress;
 import org.apache.hadoop.ozone.shell.bucket.BucketUri;
+import org.apache.hadoop.ozone.snapshot.CancelSnapshotDiffResponse;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffReportOzone;
 import org.apache.hadoop.ozone.snapshot.SnapshotDiffResponse;
 import picocli.CommandLine;
@@ -91,6 +92,11 @@ public class SnapshotDiffHandler extends Handler {
       description = "Format output as JSON")
   private boolean json;
 
+  @CommandLine.Option(
+      names = {"-n", "--om-node-id"},
+      description = "The id of OM node to get the snapshot diff from")
+  private String omNodeId;
+
   @Override
   protected OzoneAddress getAddress() {
     return snapshotPath.getValue();
@@ -106,16 +112,22 @@ protected void execute(OzoneClient client, OzoneAddress address)
     OmUtils.validateSnapshotName(toSnapshot);
 
     if (cancel) {
-      cancelSnapshotDiff(client.getObjectStore(), volumeName, bucketName);
+      cancelSnapshotDiff(client.getObjectStore(), volumeName, bucketName, omNodeId);
     } else {
-      getSnapshotDiff(client.getObjectStore(), volumeName, bucketName);
+      getSnapshotDiff(client.getObjectStore(), volumeName, bucketName, omNodeId);
     }
   }
 
   private void getSnapshotDiff(ObjectStore store, String volumeName,
-                               String bucketName) throws IOException {
-    SnapshotDiffResponse diffResponse = store.snapshotDiff(volumeName, bucketName, fromSnapshot, toSnapshot,
-        token, pageSize, forceFullDiff, diffDisableNativeLibs);
+                               String bucketName, String nodeId) throws IOException {
+    SnapshotDiffResponse diffResponse;
+    if (StringUtils.isEmpty(nodeId)) {
+      diffResponse = store.snapshotDiff(volumeName, bucketName, fromSnapshot, toSnapshot,
+          token, pageSize, forceFullDiff, diffDisableNativeLibs);
+    } else {
+      diffResponse = store.snapshotDiff(volumeName, bucketName, fromSnapshot, toSnapshot,
+          token, pageSize, forceFullDiff, diffDisableNativeLibs, nodeId);
+    }
     try (PrintWriter writer = out()) {
       if (json) {
         writer.println(toJsonStringWithDefaultPrettyPrinter(getJsonObject(diffResponse)));
@@ -126,9 +138,15 @@ private void getSnapshotDiff(ObjectStore store, String volumeName,
   }
 
   private void cancelSnapshotDiff(ObjectStore store, String volumeName,
-                                  String bucketName) throws IOException {
+                                  String bucketName, String nodeId) throws IOException {
     try (PrintWriter writer = out()) {
-      writer.println(store.cancelSnapshotDiff(volumeName, bucketName, fromSnapshot, toSnapshot));
+      CancelSnapshotDiffResponse cancelResponse;
+      if (StringUtils.isEmpty(nodeId)) {
+        cancelResponse = store.cancelSnapshotDiff(volumeName, bucketName, fromSnapshot, toSnapshot);
+      } else {
+        cancelResponse = store.cancelSnapshotDiff(volumeName, bucketName, fromSnapshot, toSnapshot, nodeId);
+      }
+      writer.println(cancelResponse);
     }
   }
 

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

@@ -603,6 +603,22 @@ public OzoneSnapshot getSnapshotInfo(String volumeName,
     return proxy.getSnapshotInfo(volumeName, bucketName, snapshotName);
   }
 
+  /**
+   * Returns snapshot info for volume/bucket snapshot path.
+   * @param volumeName volume name
+   * @param bucketName bucket name
+   * @param snapshotName snapshot name
+   * @param omNodeId OM node ID to get the snapshot info
+   * @return snapshot info for volume/bucket snapshot path.
+   * @throws IOException
+   */
+  public OzoneSnapshot getSnapshotInfo(String volumeName,
+                                       String bucketName,
+                                       String snapshotName,
+                                       String omNodeId) throws IOException {
+    return proxy.getSnapshotInfo(volumeName, bucketName, snapshotName, omNodeId);
+  }
+
   /**
    * List snapshots in a volume/bucket.
    * @param volumeName     volume name
@@ -619,6 +635,24 @@ public Iterator<OzoneSnapshot> listSnapshot(String volumeName,
     return new SnapshotIterator(volumeName, bucketName, snapshotPrefix, prevSnapshot);
   }
 
+  /**
+   * List snapshots in a volume/bucket.
+   * @param volumeName     volume name
+   * @param bucketName     bucket name
+   * @param snapshotPrefix snapshot prefix to match
+   * @param prevSnapshot   snapshots will be listed after this snapshot name
+   * @param omNodeId       OM node ID to list the snapshots
+   * @return an iterator of snapshots for volume/bucket snapshot path.
+   * @throws IOException
+   */
+  public Iterator<OzoneSnapshot> listSnapshot(String volumeName,
+                                              String bucketName,
+                                              String snapshotPrefix,
+                                              String prevSnapshot,
+                                              String omNodeId) throws IOException {
+    return new SnapshotIterator(volumeName, bucketName, snapshotPrefix, prevSnapshot, omNodeId);
+  }
+
   /**
    * An Iterator to iterate over {@link OzoneSnapshot} list.
    */
@@ -627,6 +661,7 @@ private final class SnapshotIterator implements Iterator<OzoneSnapshot> {
     private final String volumeName;
     private final String bucketName;
     private final String snapshotPrefix;
+    private final String omNodeId;
     private String lastSnapshot = null;
 
     private Iterator<OzoneSnapshot> currentIterator;
@@ -638,16 +673,34 @@ private final class SnapshotIterator implements Iterator<OzoneSnapshot> {
       this.volumeName = volumeName;
       this.bucketName = bucketName;
       this.snapshotPrefix = snapshotPrefix;
+      this.omNodeId = null;
       // Initialized the currentIterator and continuationToken.
       getNextListOfSnapshots(prevSnapshot);
     }
 
+    SnapshotIterator(String volumeName,
+                     String bucketName,
+                     String snapshotPrefix,
+                     String prevSnapshot,
+                     String omNodeId) throws IOException {
+      this.volumeName = volumeName;
+      this.bucketName = bucketName;
+      this.snapshotPrefix = snapshotPrefix;
+      this.omNodeId = omNodeId;
+      // Initialized the currentIterator and continuationToken
+      getNextListOfSnapshots(prevSnapshot, omNodeId);
+    }
+
     @Override
     public boolean hasNext() {
       if (!currentIterator.hasNext() && StringUtils.isNotEmpty(lastSnapshot)) {
         try {
           // fetch the next page if lastSnapshot is not null.
-          getNextListOfSnapshots(lastSnapshot);
+          if (!StringUtils.isBlank(omNodeId)) {
+            getNextListOfSnapshots(lastSnapshot, omNodeId);
+          } else {
+            getNextListOfSnapshots(lastSnapshot);
+          }
         } catch (IOException e) {
           LOG.error("Error retrieving next batch of list results", e);
         }
@@ -669,6 +722,13 @@ private void getNextListOfSnapshots(String startSnapshot) throws IOException {
       currentIterator = response.getSnapshotInfos().stream().map(OzoneSnapshot::fromSnapshotInfo).iterator();
       lastSnapshot = response.getLastSnapshot();
     }
+
+    private void getNextListOfSnapshots(String startSnapshot, String nodeId) throws IOException {
+      ListSnapshotResponse response =
+          proxy.listSnapshot(volumeName, bucketName, snapshotPrefix, startSnapshot, listCacheSize, nodeId);
+      currentIterator = response.getSnapshotInfos().stream().map(OzoneSnapshot::fromSnapshotInfo).iterator();
+      lastSnapshot = response.getLastSnapshot();
+    }
   }
 
   /**
@@ -711,6 +771,36 @@ public SnapshotDiffResponse snapshotDiff(String volumeName,
         token, pageSize, forceFullDiff, disableNativeDiff);
   }
 
+  /**
+   * Get the differences between two snapshots.
+   * @param volumeName Name of the volume to which the snapshot bucket belong
+   * @param bucketName Name of the bucket to which the snapshots belong
+   * @param fromSnapshot The name of the starting snapshot
+   * @param toSnapshot The name of the ending snapshot
+   * @param token to get the index to return diff report from.
+   * @param pageSize maximum entries returned to the report.
+   * @param forceFullDiff request to force full diff, skipping DAG optimization
+   * @param disableNativeDiff request to force diff to perform diffs without
+   *                           native lib
+   * @param omNodeId OM node ID to send the snapshot diff request.
+   * @return the difference report between two snapshots
+   * @throws IOException in case of any exception while generating snapshot diff
+   */
+  @SuppressWarnings("parameternumber")
+  public SnapshotDiffResponse snapshotDiff(String volumeName,
+                                           String bucketName,
+                                           String fromSnapshot,
+                                           String toSnapshot,
+                                           String token,
+                                           int pageSize,
+                                           boolean forceFullDiff,
+                                           boolean disableNativeDiff,
+                                           String omNodeId)
+      throws IOException {
+    return proxy.snapshotDiff(volumeName, bucketName, fromSnapshot, toSnapshot,
+        token, pageSize, forceFullDiff, disableNativeDiff, omNodeId);
+  }
+
   /**
    * Cancel the snap diff jobs.
    * @param volumeName Name of the volume to which the snapshot bucket belong
@@ -729,6 +819,26 @@ public CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
         toSnapshot);
   }
 
+  /**
+   * Cancel the snap diff jobs.
+   * @param volumeName Name of the volume to which the snapshot bucket belong
+   * @param bucketName Name of the bucket to which the snapshots belong
+   * @param fromSnapshot The name of the starting snapshot
+   * @param toSnapshot The name of the ending snapshot
+   * @param omNodeId OM node ID to send the cancel snapshot diff jobs request.
+   * @return the success if cancel succeeds.
+   * @throws IOException in case of any exception while generating snapshot diff
+   */
+  public CancelSnapshotDiffResponse cancelSnapshotDiff(String volumeName,
+                                                       String bucketName,
+                                                       String fromSnapshot,
+                                                       String toSnapshot,
+                                                       String omNodeId)
+      throws IOException {
+    return proxy.cancelSnapshotDiff(volumeName, bucketName, fromSnapshot,
+        toSnapshot, omNodeId);
+  }
+
   /**
    * Get a list of the SnapshotDiff jobs for a bucket based on the JobStatus.
    * @param volumeName Name of the volume to which the snapshotted bucket belong
@@ -746,4 +856,24 @@ public List<OzoneSnapshotDiff> listSnapshotDiffJobs(String volumeName,
     return proxy.listSnapshotDiffJobs(volumeName,
         bucketName, jobStatus, listAll);
   }
+
+  /**
+   * Get a list of the SnapshotDiff jobs for a bucket based on the JobStatus.
+   * @param volumeName Name of the volume to which the snapshotted bucket belong
+   * @param bucketName Name of the bucket to which the snapshots belong
+   * @param jobStatus JobStatus to be used to filter the snapshot diff jobs
+   * @param listAll Option to specify whether to list all jobs or not
+   * @param omNodeId OM node ID to get the list of SnapshotDiff jobs from
+   * @return a list of SnapshotDiffJob objects from the specified OM node
+   * @throws IOException in case there is a failure while getting a response.
+   */
+  public List<OzoneSnapshotDiff> listSnapshotDiffJobs(String volumeName,
+                                                      String bucketName,
+                                                      String jobStatus,
+                                                      boolean listAll,
+                                                      String omNodeId)
+      throws IOException {
+    return proxy.listSnapshotDiffJobs(volumeName,
+        bucketName, jobStatus, listAll, omNodeId);
+  }
 }