HDDS-4876. [SCM HA Security] Add failover proxy to SCM Security Server Protocol

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/exception/SCMSecurityException.java

@@ -35,6 +35,16 @@ public SCMSecurityException(String message) {
     this.errorCode = ErrorCode.DEFAULT;
   }
 
+  /**
+   * Ctor.
+   * @param message - Error Message
+   * @param errorCode - Error code
+   */
+  public SCMSecurityException(String message, ErrorCode errorCode) {
+    super(message);
+    this.errorCode = errorCode;
+  }
+
   /**
    * Ctor.
    * @param message - Message.
@@ -47,11 +57,23 @@ public SCMSecurityException(String message, Throwable cause) {
 
   /**
    * Ctor.
-   * @param message - Message.
+   * @param message - Error Message
+   * @param cause - Actual cause.
+   * @param errorCode - Error code.
+   */
+  public SCMSecurityException(String message, Throwable cause,
+      ErrorCode errorCode) {
+    super(message, cause);
+    this.errorCode = errorCode;
+  }
+
+  /**
+   * Ctor.
+   * @param cause - Actual cause.
    * @param error   - error code.
    */
-  public SCMSecurityException(String message, ErrorCode error) {
-    super(message);
+  public SCMSecurityException(Exception cause, ErrorCode error) {
+    super(cause);
     this.errorCode = error;
   }
 
@@ -72,6 +94,17 @@ public ErrorCode getErrorCode() {
    * Error codes to make it easy to decode these exceptions.
    */
   public enum ErrorCode {
+    OK,
+    INVALID_CSR,
+    UNABLE_TO_ISSUE_CERTIFICATE,
+    GET_DN_CERTIFICATE_FAILED,
+    GET_OM_CERTIFICATE_FAILED,
+    GET_SCM_CERTIFICATE_FAILED,
+    GET_CERTIFICATE_FAILED,
+    GET_CA_CERT_FAILED,
+    CERTIFICATE_NOT_FOUND,
+    PEM_ENCODE_FAILED,
+    INTERNAL_ERROR,
     DEFAULT,
     MISSING_BLOCK_TOKEN,
     BLOCK_TOKEN_VERIFICATION_FAILED

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/security/x509/certificate/utils/CertificateCodec.java

@@ -50,6 +50,7 @@
 import static java.nio.file.attribute.PosixFilePermission.OWNER_EXECUTE;
 import static java.nio.file.attribute.PosixFilePermission.OWNER_READ;
 import static java.nio.file.attribute.PosixFilePermission.OWNER_WRITE;
+import static org.apache.hadoop.hdds.security.exception.SCMSecurityException.ErrorCode.PEM_ENCODE_FAILED;
 
 /**
  * A class used to read and write X.509 certificates  PEM encoded Streams.
@@ -125,7 +126,7 @@ public static String getPEMEncodedString(X509Certificate certificate)
       LOG.error("Error in encoding certificate." + certificate
           .getSubjectDN().toString(), e);
       throw new SCMSecurityException("PEM Encoding failed for certificate." +
-          certificate.getSubjectDN().toString(), e);
+          certificate.getSubjectDN().toString(), e, PEM_ENCODE_FAILED);
     }
   }
 

hadoop-hdds/framework/src/main/java/org/apache/hadoop/hdds/protocolPB/SCMSecurityProtocolClientSideTranslatorPB.java

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.function.Consumer;
 
+import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.protocol.SCMSecurityProtocol;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
@@ -37,7 +38,10 @@
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMSecurityRequest.Builder;
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.SCMSecurityResponse;
 import org.apache.hadoop.hdds.protocol.proto.SCMSecurityProtocolProtos.Type;
+import org.apache.hadoop.hdds.scm.proxy.SCMSecurityProtocolFailoverProxyProvider;
+import org.apache.hadoop.hdds.security.exception.SCMSecurityException;
 import org.apache.hadoop.hdds.tracing.TracingUtil;
+import org.apache.hadoop.io.retry.RetryProxy;
 import org.apache.hadoop.ipc.ProtobufHelper;
 import org.apache.hadoop.ipc.ProtocolTranslator;
 import org.apache.hadoop.ipc.RPC;
@@ -58,12 +62,22 @@ public class SCMSecurityProtocolClientSideTranslatorPB implements
    */
   private static final RpcController NULL_RPC_CONTROLLER = null;
   private final SCMSecurityProtocolPB rpcProxy;
+  private SCMSecurityProtocolFailoverProxyProvider failoverProxyProvider;
 
   public SCMSecurityProtocolClientSideTranslatorPB(
       SCMSecurityProtocolPB rpcProxy) {
     this.rpcProxy = rpcProxy;
   }
 
+  public SCMSecurityProtocolClientSideTranslatorPB(
+      SCMSecurityProtocolFailoverProxyProvider proxyProvider) {
+    Preconditions.checkState(proxyProvider != null);
+    this.failoverProxyProvider = proxyProvider;
+    this.rpcProxy = (SCMSecurityProtocolPB) RetryProxy.create(
+        SCMSecurityProtocolPB.class, failoverProxyProvider,
+        failoverProxyProvider.getRetryPolicy());
+  }
+
   /**
    * Helper method to wrap the request and send the message.
    */
@@ -80,12 +94,29 @@ private SCMSecurityResponse submitRequest(
       SCMSecurityRequest wrapper = builder.build();
 
       response = rpcProxy.submitRequest(NULL_RPC_CONTROLLER, wrapper);
+
+      handleError(response);
+
     } catch (ServiceException ex) {
       throw ProtobufHelper.getRemoteException(ex);
     }
     return response;
   }
 
+  /**
+   * If response is not successful, throw exception.
+   * @param resp - SCMSecurityResponse
+   * @return if response is success, return response, else throw exception.
+   * @throws SCMSecurityException
+   */
+  private SCMSecurityResponse handleError(SCMSecurityResponse resp)
+      throws SCMSecurityException {
+    if (resp.getStatus() != SCMSecurityProtocolProtos.Status.OK) {
+      throw new SCMSecurityException(resp.getMessage(),
+          SCMSecurityException.ErrorCode.values()[resp.getStatus().ordinal()]);
+    }
+    return resp;
+  }
   /**
    * Closes this stream and releases any system resources associated
    * with it. If the stream is already closed then invoking this