HDDS-14379. Implement basic Hadoop OM client proxy provider to read from followers

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/HddsUtils.java

@@ -41,6 +41,7 @@
 import java.io.File;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.lang.reflect.UndeclaredThrowableException;
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.nio.file.Path;
@@ -615,6 +616,9 @@ public static Throwable getUnwrappedException(Exception ex) {
     if (t instanceof RemoteException) {
       t = ((RemoteException) t).unwrapRemoteException();
     }
+    if (t instanceof UndeclaredThrowableException) {
+      t = ((UndeclaredThrowableException) t).getUndeclaredThrowable();
+    }
     while (t != null) {
       if (t instanceof RpcException ||
           t instanceof AccessControlException ||

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

@@ -488,6 +488,11 @@ public final class OzoneConfigKeys {
   public static final long OZONE_CLIENT_WAIT_BETWEEN_RETRIES_MILLIS_DEFAULT =
       2000;
 
+  // Ozone Client Follower Read
+  public static final String OZONE_CLIENT_FOLLOWER_READ_ENABLED_KEY =
+      "ozone.client.follower.read.enabled";
+  public static final boolean OZONE_CLIENT_FOLLOWER_READ_ENABLED_DEFAULT = false;
+
   public static final String OZONE_FREON_HTTP_ENABLED_KEY =
       "ozone.freon.http.enabled";
   public static final String OZONE_FREON_HTTP_BIND_HOST_KEY =

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -3141,6 +3141,13 @@
       wait time is introduced after all the OM proxies have been attempted once.
     </description>
   </property>
+  <property>
+    <name>ozone.client.follower.read.enabled</name>
+    <value>false</value>
+    <description>
+      Enable client to read from OM followers.
+    </description>
+  </property>
   <property>
     <name>ozone.om.admin.protocol.max.retries</name>
     <value>20</value>

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -78,6 +78,7 @@
 import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -228,8 +229,7 @@ public static int getOmRpcPort(ConfigurationSource conf) {
    * @param omRequest OMRequest proto
    * @return True if its readOnly, false otherwise.
    */
-  public static boolean isReadOnly(
-      OzoneManagerProtocolProtos.OMRequest omRequest) {
+  public static boolean isReadOnly(OMRequest omRequest) {
     OzoneManagerProtocolProtos.Type cmdType = omRequest.getCmdType();
     switch (cmdType) {
     case CheckVolumeAccess:
@@ -352,6 +352,135 @@ public static boolean isReadOnly(
     }
   }
 
+  /**
+   * Checks if the OM request should be sent to the follower or leader.
+   * <p>
+   * Note that this method is not equivalent to {@link OmUtils#isReadOnly(OMRequest)}
+   * since there are cases that a "read" requests (ones that do not go through Ratis) requires
+   * to be sent to the leader.
+   * @param omRequest OMRequest proto
+   * @return True if the request should be sent to the follower.
+   */
+  public static boolean shouldSendToFollower(OMRequest omRequest) {
+    OzoneManagerProtocolProtos.Type cmdType = omRequest.getCmdType();
+    switch (cmdType) {
+    case CheckVolumeAccess:
+    case InfoVolume:
+    case ListVolume:
+    case InfoBucket:
+    case ListBuckets:
+    case LookupKey:
+    case ListKeys:
+    case ListKeysLight:
+    case ListTrash:
+      // ListTrash is deprecated by HDDS-11251. Keeping this in here
+      // As protobuf currently doesn't support deprecating enum fields
+      // TODO: Remove once migrated to proto3 and mark fields in proto
+      // as deprecated
+    case ListOpenFiles:
+    case ListMultiPartUploadParts:
+    case GetFileStatus:
+    case LookupFile:
+    case ListStatus:
+    case ListStatusLight:
+    case GetAcl:
+    case ListMultipartUploads:
+    case FinalizeUpgradeProgress:
+    case PrepareStatus:
+    case GetS3VolumeContext:
+    case ListTenant:
+    case TenantGetUserInfo:
+    case TenantListUser:
+    case ListSnapshot:
+    case RefetchSecretKey:
+    case GetKeyInfo:
+    case GetSnapshotInfo:
+    case GetObjectTagging:
+      return true;
+    case CreateVolume:
+    case SetVolumeProperty:
+    case DeleteVolume:
+    case CreateBucket:
+    case SetBucketProperty:
+    case DeleteBucket:
+    case CreateKey:
+    case RenameKey:
+    case RenameKeys:
+    case DeleteKey:
+    case DeleteKeys:
+    case CommitKey:
+    case AllocateBlock:
+    case InitiateMultiPartUpload:
+    case CommitMultiPartUpload:
+    case CompleteMultiPartUpload:
+    case AbortMultiPartUpload:
+    case GetS3Secret:
+    case GetDelegationToken:
+    case RenewDelegationToken:
+    case CancelDelegationToken:
+    case CreateDirectory:
+    case CreateFile:
+    case RemoveAcl:
+    case SetAcl:
+    case AddAcl:
+    case PurgeKeys:
+    case RecoverTrash:
+      // RecoverTrash is deprecated by HDDS-11251. Keeping this in here
+      // As protobuf currently doesn't support deprecating enum fields
+      // TODO: Remove once migrated to proto3 and mark fields in proto
+      // as deprecated
+    case FinalizeUpgrade:
+    case Prepare:
+    case CancelPrepare:
+    case DeleteOpenKeys:
+    case SetS3Secret:
+    case RevokeS3Secret:
+    case PurgeDirectories:
+    case PurgePaths:
+    case CreateTenant:
+    case DeleteTenant:
+    case TenantAssignUserAccessId:
+    case TenantRevokeUserAccessId:
+    case TenantAssignAdmin:
+    case TenantRevokeAdmin:
+    case SetRangerServiceVersion:
+    case CreateSnapshot:
+    case DeleteSnapshot:
+    case RenameSnapshot:
+    case SnapshotMoveDeletedKeys:
+    case SnapshotMoveTableKeys:
+    case SnapshotPurge:
+    case RecoverLease:
+    case SetTimes:
+    case AbortExpiredMultiPartUploads:
+    case SetSnapshotProperty:
+    case QuotaRepair:
+    case PutObjectTagging:
+    case DeleteObjectTagging:
+    case ServiceList: // OM leader should have the most up-to-date OM service list info
+    case RangerBGSync: // Ranger Background Sync task is only run on leader
+    case SnapshotDiff:
+    case CancelSnapshotDiff:
+    case ListSnapshotDiffJobs:
+    case PrintCompactionLogDag:
+      // Snapshot diff is a local to a single OM node so we should not send it arbitrarily
+      // to any OM nodes
+    case TransferLeadership: // Transfer leadership should be initiated by the leader
+    case SetSafeMode: // SafeMode should be initiated by the leader
+    case StartQuotaRepair:
+    case GetQuotaRepairStatus:
+      // Quota repair lifecycle request should be initiated by the leader
+    case DBUpdates: // We are currently only interested on the leader DB info
+    case UnknownCommand:
+      return false;
+    case EchoRPC:
+      return omRequest.getEchoRPCRequest().getReadOnly();
+    default:
+      LOG.error("CmdType {} is not categorized to be sent to follower.", cmdType);
+      return false;
+    }
+  }
+
   public static byte[] getSHADigest() throws IOException {
     try {
       SRAND.nextBytes(randomBytes);

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/ha/HadoopRpcOMFailoverProxyProvider.java

@@ -45,7 +45,7 @@
 public class HadoopRpcOMFailoverProxyProvider<T> extends
       OMFailoverProxyProviderBase<T> {
 
-  private static final Logger LOG =
+  protected static final Logger LOG =
       LoggerFactory.getLogger(HadoopRpcOMFailoverProxyProvider.class);
 
   private final Text delegationTokenService;
@@ -123,7 +123,7 @@ public synchronized ProxyInfo<T> getProxy() {
   /**
    * Creates proxy object.
    */
-  protected ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
+  protected synchronized ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
     if (omProxyInfo.proxy == null) {
       try {
         omProxyInfo.proxy = createOMProxy(omProxyInfo.getAddress());
@@ -136,6 +136,11 @@ protected ProxyInfo<T> createOMProxyIfNeeded(OMProxyInfo<T> omProxyInfo) {
     return omProxyInfo;
   }
 
+  protected synchronized ProxyInfo<T> createOMProxyIfNeeded(String omNodeId) {
+    OMProxyInfo<T> omProxyInfo = getOmProxy(omNodeId);
+    return createOMProxyIfNeeded(omProxyInfo);
+  }
+
   public Text getCurrentProxyDelegationToken() {
     return delegationTokenService;
   }