YARN-8449. RM HA for AM web server HTTPS Support. (Contributed by Robert Kanter)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java

@@ -1510,6 +1510,9 @@ public class ResourceManager extends CompositeService
     // recover applications
     rmAppManager.recover(state);
 
+    // recover ProxyCA
+    rmContext.getProxyCAManager().recover(state);
+
     setSchedulerRecoveryStartAndWaitTime(state, conf);
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/FileSystemRMStateStore.java

@@ -24,6 +24,8 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.HashMap;
@@ -114,6 +116,7 @@ public class FileSystemRMStateStore extends RMStateStore {
 
   Path amrmTokenSecretManagerRoot;
   private Path reservationRoot;
+  private Path proxyCARoot;
 
   @Override
   public synchronized void initInternal(Configuration conf)
@@ -125,6 +128,7 @@ public class FileSystemRMStateStore extends RMStateStore {
     amrmTokenSecretManagerRoot =
         new Path(rootDirPath, AMRMTOKEN_SECRET_MANAGER_ROOT);
     reservationRoot = new Path(rootDirPath, RESERVATION_SYSTEM_ROOT);
+    proxyCARoot = new Path(rootDirPath, PROXY_CA_ROOT);
     fsNumRetries =
         conf.getInt(YarnConfiguration.FS_RM_STATE_STORE_NUM_RETRIES,
             YarnConfiguration.DEFAULT_FS_RM_STATE_STORE_NUM_RETRIES);
@@ -157,6 +161,7 @@ public class FileSystemRMStateStore extends RMStateStore {
     mkdirsWithRetries(rmAppRoot);
     mkdirsWithRetries(amrmTokenSecretManagerRoot);
     mkdirsWithRetries(reservationRoot);
+    mkdirsWithRetries(proxyCARoot);
   }
 
   @Override
@@ -228,6 +233,8 @@ public class FileSystemRMStateStore extends RMStateStore {
     loadAMRMTokenSecretManagerState(rmState);
     // recover reservation state
     loadReservationSystemState(rmState);
+    // recover ProxyCAManager state
+    loadProxyCAManagerState(rmState);
     return rmState;
   }
 
@@ -395,6 +402,30 @@ public class FileSystemRMStateStore extends RMStateStore {
     }
   }
 
+  private void loadProxyCAManagerState(RMState rmState) throws Exception {
+    checkAndResumeUpdateOperation(proxyCARoot);
+
+    Path caCertPath = getNodePath(proxyCARoot, PROXY_CA_CERT_NODE);
+    Path caPrivateKeyPath = getNodePath(proxyCARoot, PROXY_CA_PRIVATE_KEY_NODE);
+
+    if (!existsWithRetries(caCertPath)
+        || !existsWithRetries(caPrivateKeyPath)) {
+      LOG.warn("Couldn't find Proxy CA data");
+      return;
+    }
+
+    FileStatus caCertFileStatus = getFileStatus(caCertPath);
+    byte[] caCertData = readFileWithRetries(caCertPath,
+        caCertFileStatus.getLen());
+
+    FileStatus caPrivateKeyFileStatus = getFileStatus(caPrivateKeyPath);
+    byte[] caPrivateKeyData = readFileWithRetries(caPrivateKeyPath,
+        caPrivateKeyFileStatus.getLen());
+
+    rmState.getProxyCAState().setCaCert(caCertData);
+    rmState.getProxyCAState().setCaPrivateKey(caPrivateKeyData);
+  }
+
   @Override
   public synchronized void storeApplicationStateInternal(ApplicationId appId,
       ApplicationStateData appStateDataPB) throws Exception {
@@ -593,6 +624,28 @@ public class FileSystemRMStateStore extends RMStateStore {
     }
   }
 
+  @Override
+  synchronized protected void storeProxyCACertState(
+      X509Certificate caCert, PrivateKey caPrivateKey) throws Exception {
+    byte[] caCertData = caCert.getEncoded();
+    byte[] caPrivateKeyData = caPrivateKey.getEncoded();
+
+    Path caCertPath = getNodePath(proxyCARoot, PROXY_CA_CERT_NODE);
+    Path caPrivateKeyPath = getNodePath(proxyCARoot, PROXY_CA_PRIVATE_KEY_NODE);
+
+    if (existsWithRetries(caCertPath)) {
+      updateFile(caCertPath, caCertData, true);
+    } else {
+      writeFileWithRetries(caCertPath, caCertData, true);
+    }
+
+    if (existsWithRetries(caPrivateKeyPath)) {
+      updateFile(caPrivateKeyPath, caPrivateKeyData, true);
+    } else {
+      writeFileWithRetries(caPrivateKeyPath, caPrivateKeyData, true);
+    }
+  }
+
   private Path getAppDir(Path root, ApplicationId appId) {
     return getNodePath(root, appId.toString());
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/LeveldbRMStateStore.java

@@ -27,6 +27,8 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.File;
 import java.io.IOException;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
 import java.util.HashMap;
 import java.util.Map.Entry;
 import java.util.Timer;
@@ -129,6 +131,14 @@ public class LeveldbRMStateStore extends RMStateStore {
         + reservationId;
   }
 
+  private String getProxyCACertNodeKey() {
+    return PROXY_CA_ROOT + SEPARATOR + PROXY_CA_CERT_NODE;
+  }
+
+  private String getProxyCAPrivateKeyNodeKey() {
+    return PROXY_CA_ROOT + SEPARATOR + PROXY_CA_PRIVATE_KEY_NODE;
+  }
+
   @Override
   protected void initInternal(Configuration conf) throws Exception {
     compactionIntervalMsec = conf.getLong(
@@ -274,6 +284,7 @@ public class LeveldbRMStateStore extends RMStateStore {
      loadRMApps(rmState);
      loadAMRMTokenSecretManagerState(rmState);
     loadReservationState(rmState);
+    loadProxyCAManagerState(rmState);
     return rmState;
    }
 
@@ -578,6 +589,34 @@ public class LeveldbRMStateStore extends RMStateStore {
     }
   }
 
+  private void loadProxyCAManagerState(RMState rmState) throws Exception {
+    byte[] caCertData;
+    byte[] caPrivateKeyData;
+
+    String caCertKey = getProxyCACertNodeKey();
+    String caPrivateKeyKey = getProxyCAPrivateKeyNodeKey();
+
+    try {
+      caCertData = db.get(bytes(caCertKey));
+    } catch (DBException e) {
+      throw new IOException(e);
+    }
+
+    try {
+      caPrivateKeyData = db.get(bytes(caPrivateKeyKey));
+    } catch (DBException e) {
+      throw new IOException(e);
+    }
+
+    if (caCertData == null || caPrivateKeyData == null) {
+      LOG.warn("Couldn't find Proxy CA data");
+      return;
+    }
+
+    rmState.proxyCAState.setCaCert(caCertData);
+    rmState.proxyCAState.setCaPrivateKey(caPrivateKeyData);
+  }
+
   @Override
   protected void storeApplicationStateInternal(ApplicationId appId,
       ApplicationStateData appStateData) throws IOException {
@@ -811,6 +850,29 @@ public class LeveldbRMStateStore extends RMStateStore {
     db.put(bytes(AMRMTOKEN_SECRET_MANAGER_ROOT), stateData);
   }
 
+  @Override
+  protected void storeProxyCACertState(
+      X509Certificate caCert, PrivateKey caPrivateKey) throws Exception {
+    byte[] caCertData = caCert.getEncoded();
+    byte[] caPrivateKeyData = caPrivateKey.getEncoded();
+
+    String caCertKey = getProxyCACertNodeKey();
+    String caPrivateKeyKey = getProxyCAPrivateKeyNodeKey();
+
+    try {
+      WriteBatch batch = db.createWriteBatch();
+      try {
+        batch.put(bytes(caCertKey), caCertData);
+        batch.put(bytes(caPrivateKeyKey), caPrivateKeyData);
+        db.write(batch);
+      } finally {
+        batch.close();
+      }
+    } catch (DBException e) {
+      throw new IOException(e);
+    }
+  }
+
   @Override
   public void deleteStore() throws IOException {
     Path root = getStorageDir();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/MemoryRMStateStore.java

@@ -19,6 +19,8 @@
 package org.apache.hadoop.yarn.server.resourcemanager.recovery;
 
 import java.io.IOException;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
@@ -78,6 +80,15 @@ public class MemoryRMStateStore extends RMStateStore {
         state.amrmTokenSecretManagerState == null ? null
             : AMRMTokenSecretManagerState
               .newInstance(state.amrmTokenSecretManagerState);
+    if (state.proxyCAState.getCaCert() != null) {
+      byte[] caCertData = state.proxyCAState.getCaCert().getEncoded();
+      returnState.proxyCAState.setCaCert(caCertData);
+    }
+    if (state.proxyCAState.getCaPrivateKey() != null) {
+      byte[] caPrivateKeyData
+          = state.proxyCAState.getCaPrivateKey().getEncoded();
+      returnState.proxyCAState.setCaPrivateKey(caPrivateKeyData);
+    }
     return returnState;
   }
   
@@ -277,6 +288,13 @@ public class MemoryRMStateStore extends RMStateStore {
     }
   }
 
+  @Override
+  protected void storeProxyCACertState(
+      X509Certificate caCert, PrivateKey caPrivateKey) throws Exception {
+    state.getProxyCAState().setCaCert(caCert);
+    state.getProxyCAState().setCaPrivateKey(caPrivateKey);
+  }
+
   @Override
   protected Version loadVersion() throws Exception {
     return null;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/NullRMStateStore.java

@@ -31,6 +31,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.AMRMTokenS
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationAttemptStateData;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.records.ApplicationStateData;
 
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+
 @Unstable
 public class NullRMStateStore extends RMStateStore {
 
@@ -174,4 +177,10 @@ public class NullRMStateStore extends RMStateStore {
   public void removeApplication(ApplicationId removeAppId) throws Exception {
     // Do nothing
   }
+
+  @Override
+  protected void storeProxyCACertState(
+      X509Certificate caCert, PrivateKey caPrivateKey) throws Exception {
+    // Do nothing
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStore.java

@@ -18,6 +18,15 @@
 
 package org.apache.hadoop.yarn.server.resourcemanager.recovery;
 
+import java.io.ByteArrayInputStream;
+import java.security.KeyFactory;
+import java.security.NoSuchAlgorithmException;
+import java.security.PrivateKey;
+import java.security.cert.CertificateException;
+import java.security.cert.CertificateFactory;
+import java.security.cert.X509Certificate;
+import java.security.spec.InvalidKeySpecException;
+import java.security.spec.PKCS8EncodedKeySpec;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -101,6 +110,9 @@ public abstract class RMStateStore extends AbstractService {
       "AMRMTokenSecretManagerRoot";
   protected static final String RESERVATION_SYSTEM_ROOT =
       "ReservationSystemRoot";
+  protected static final String PROXY_CA_ROOT = "ProxyCARoot";
+  protected static final String PROXY_CA_CERT_NODE = "caCert";
+  protected static final String PROXY_CA_PRIVATE_KEY_NODE = "caPrivateKey";
   protected static final String VERSION_NODE = "RMVersionNode";
   protected static final String EPOCH_NODE = "EpochNode";
   protected long baseEpoch;
@@ -183,6 +195,10 @@ public abstract class RMStateStore extends AbstractService {
           new RemoveReservationAllocationTransition())
       .addTransition(RMStateStoreState.ACTIVE, RMStateStoreState.FENCED,
           RMStateStoreEventType.FENCED)
+      .addTransition(RMStateStoreState.ACTIVE,
+          EnumSet.of(RMStateStoreState.ACTIVE, RMStateStoreState.FENCED),
+          RMStateStoreEventType.STORE_PROXY_CA_CERT,
+          new StoreProxyCACertTransition())
       .addTransition(RMStateStoreState.FENCED, RMStateStoreState.FENCED,
           EnumSet.of(
           RMStateStoreEventType.STORE_APP,
@@ -198,7 +214,8 @@ public abstract class RMStateStore extends AbstractService {
           RMStateStoreEventType.UPDATE_DELEGATION_TOKEN,
           RMStateStoreEventType.UPDATE_AMRM_TOKEN,
           RMStateStoreEventType.STORE_RESERVATION,
-          RMStateStoreEventType.REMOVE_RESERVATION));
+          RMStateStoreEventType.REMOVE_RESERVATION,
+          RMStateStoreEventType.STORE_PROXY_CA_CERT));
 
   private final StateMachine<RMStateStoreState,
                              RMStateStoreEventType,
@@ -615,6 +632,31 @@ public abstract class RMStateStore extends AbstractService {
     }
   }
 
+  private static class StoreProxyCACertTransition implements
+      MultipleArcTransition<RMStateStore, RMStateStoreEvent,
+          RMStateStoreState> {
+    @Override
+    public RMStateStoreState transition(RMStateStore store,
+        RMStateStoreEvent event) {
+      if (!(event instanceof RMStateStoreProxyCAEvent)) {
+        // should never happen
+        LOG.error("Illegal event type: " + event.getClass());
+        return RMStateStoreState.ACTIVE;
+      }
+      boolean isFenced = false;
+      RMStateStoreProxyCAEvent caEvent = (RMStateStoreProxyCAEvent) event;
+      try {
+        LOG.info("Storing CA Certificate and Private Key");
+        store.storeProxyCACertState(
+            caEvent.getCaCert(), caEvent.getCaPrivateKey());
+      } catch (Exception e) {
+        LOG.error("Error While Storing CA Certificate and Private Key", e);
+        isFenced = store.notifyStoreOperationFailedInternal(e);
+      }
+      return finalState(isFenced);
+    }
+  }
+
   private static RMStateStoreState finalState(boolean isFenced) {
     return isFenced ? RMStateStoreState.FENCED : RMStateStoreState.ACTIVE;
   }
@@ -676,6 +718,39 @@ public abstract class RMStateStore extends AbstractService {
     }
   }
 
+  public static class ProxyCAState {
+    private X509Certificate caCert;
+    private PrivateKey caPrivateKey;
+
+    public X509Certificate getCaCert() {
+      return caCert;
+    }
+
+    public PrivateKey getCaPrivateKey() {
+      return caPrivateKey;
+    }
+
+    public void setCaCert(X509Certificate caCert) {
+      this.caCert = caCert;
+    }
+
+    public void setCaPrivateKey(PrivateKey caPrivateKey) {
+      this.caPrivateKey = caPrivateKey;
+    }
+
+    public void setCaCert(byte[] caCertData) throws CertificateException {
+      ByteArrayInputStream bais = new ByteArrayInputStream(caCertData);
+      caCert = (X509Certificate)
+          CertificateFactory.getInstance("X.509").generateCertificate(bais);
+    }
+
+    public void setCaPrivateKey(byte[] caPrivateKeyData)
+        throws NoSuchAlgorithmException, InvalidKeySpecException {
+      caPrivateKey = KeyFactory.getInstance("RSA").generatePrivate(
+          new PKCS8EncodedKeySpec(caPrivateKeyData));
+    }
+  }
+
   /**
    * State of the ResourceManager
    */
@@ -690,6 +765,8 @@ public abstract class RMStateStore extends AbstractService {
     private Map<String, Map<ReservationId, ReservationAllocationStateProto>>
         reservationState = new TreeMap<>();
 
+    ProxyCAState proxyCAState = new ProxyCAState();
+
     public Map<ApplicationId, ApplicationStateData> getApplicationState() {
       return appState;
     }
@@ -706,6 +783,10 @@ public abstract class RMStateStore extends AbstractService {
         getReservationState() {
       return reservationState;
     }
+
+    public ProxyCAState getProxyCAState() {
+      return proxyCAState;
+    }
   }
     
   private Dispatcher rmDispatcher;
@@ -1273,4 +1354,21 @@ public abstract class RMStateStore extends AbstractService {
   protected EventHandler getRMStateStoreEventHandler() {
     return dispatcher.getEventHandler();
   }
+
+  /**
+   * ProxyCAManager calls this to store the CA Certificate and Private Key.
+   */
+  public void storeProxyCACert(X509Certificate caCert,
+      PrivateKey caPrivateKey) {
+    handleStoreEvent(new RMStateStoreProxyCAEvent(caCert, caPrivateKey,
+        RMStateStoreEventType.STORE_PROXY_CA_CERT));
+  }
+
+  /**
+   * Blocking API
+   * Derived classes must implement this method to store the CA Certificate
+   * and Private Key
+   */
+  protected abstract void storeProxyCACertState(
+      X509Certificate caCert, PrivateKey caPrivateKey) throws Exception;
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreEventType.java

@@ -36,4 +36,5 @@ public enum RMStateStoreEventType {
   UPDATE_AMRM_TOKEN,
   STORE_RESERVATION,
   REMOVE_RESERVATION,
+  STORE_PROXY_CA_CERT,
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreProxyCAEvent.java

@@ -0,0 +1,49 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.resourcemanager.recovery;
+
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+
+/**
+ * A event used to store ProxyCA information.
+ */
+public class RMStateStoreProxyCAEvent extends RMStateStoreEvent {
+  private X509Certificate caCert;
+  private PrivateKey caPrivateKey;
+
+  public RMStateStoreProxyCAEvent(RMStateStoreEventType type) {
+    super(type);
+  }
+
+  public RMStateStoreProxyCAEvent(X509Certificate caCert,
+      PrivateKey caPrivateKey, RMStateStoreEventType type) {
+    this(type);
+    this.caCert = caCert;
+    this.caPrivateKey = caPrivateKey;
+  }
+
+  public X509Certificate getCaCert() {
+    return caCert;
+  }
+
+  public PrivateKey getCaPrivateKey() {
+    return caPrivateKey;
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/ZKRMStateStore.java

@@ -68,6 +68,8 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.security.NoSuchAlgorithmException;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -153,6 +155,10 @@ import java.util.Set;
  *        |      ....
  *        |------PLAN_2
  *        ....
+ * |-- PROXY_CA_ROOT
+ *        |----- caCert
+ *        |----- caPrivateKey
+ *
  * Note: Changes from 1.1 to 1.2 - AMRMTokenSecretManager state has been saved
  * separately. The currentMasterkey and nextMasterkey have been stored.
  * Also, AMRMToken has been removed from ApplicationAttemptState.
@@ -198,6 +204,7 @@ public class ZKRMStateStore extends RMStateStore {
   private String dtSequenceNumberPath;
   private String amrmTokenSecretManagerRoot;
   private String reservationRoot;
+  private String proxyCARoot;
 
   @VisibleForTesting
   protected String znodeWorkingPath;
@@ -357,6 +364,7 @@ public class ZKRMStateStore extends RMStateStore {
         RM_DT_SEQUENTIAL_NUMBER_ZNODE_NAME);
     amrmTokenSecretManagerRoot =
         getNodePath(zkRootNodePath, AMRMTOKEN_SECRET_MANAGER_ROOT);
+    proxyCARoot = getNodePath(zkRootNodePath, PROXY_CA_ROOT);
     reservationRoot = getNodePath(zkRootNodePath, RESERVATION_SYSTEM_ROOT);
     zkManager = resourceManager.getZKManager();
     if(zkManager==null) {
@@ -402,6 +410,7 @@ public class ZKRMStateStore extends RMStateStore {
     create(dtSequenceNumberPath);
     create(amrmTokenSecretManagerRoot);
     create(reservationRoot);
+    create(proxyCARoot);
   }
 
   private void logRootNodeAcls(String prefix) throws Exception {
@@ -517,7 +526,8 @@ public class ZKRMStateStore extends RMStateStore {
     loadAMRMTokenSecretManagerState(rmState);
     // recover reservation state
     loadReservationSystemState(rmState);
-
+    // recover ProxyCAManager state
+    loadProxyCAManagerState(rmState);
     return rmState;
   }
 
@@ -813,6 +823,28 @@ public class ZKRMStateStore extends RMStateStore {
     }
   }
 
+  private void loadProxyCAManagerState(RMState rmState) throws Exception {
+    String caCertPath = getNodePath(proxyCARoot, PROXY_CA_CERT_NODE);
+    String caPrivateKeyPath = getNodePath(proxyCARoot,
+        PROXY_CA_PRIVATE_KEY_NODE);
+
+    if (!exists(caCertPath) || !exists(caPrivateKeyPath)) {
+      LOG.warn("Couldn't find Proxy CA data");
+      return;
+    }
+
+    byte[] caCertData = getData(caCertPath);
+    byte[] caPrivateKeyData = getData(caPrivateKeyPath);
+
+    if (caCertData == null || caPrivateKeyData == null) {
+      LOG.warn("Couldn't recover Proxy CA data");
+      return;
+    }
+
+    rmState.getProxyCAState().setCaCert(caCertData);
+    rmState.getProxyCAState().setCaPrivateKey(caPrivateKeyData);
+  }
+
   @Override
   public synchronized void storeApplicationStateInternal(ApplicationId appId,
       ApplicationStateData appStateDataPB) throws Exception {
@@ -1243,6 +1275,32 @@ public class ZKRMStateStore extends RMStateStore {
     }
   }
 
+  @Override
+  protected void storeProxyCACertState(
+      X509Certificate caCert, PrivateKey caPrivateKey) throws Exception {
+    byte[] caCertData = caCert.getEncoded();
+    byte[] caPrivateKeyData = caPrivateKey.getEncoded();
+
+    String caCertPath = getNodePath(proxyCARoot, PROXY_CA_CERT_NODE);
+    String caPrivateKeyPath = getNodePath(proxyCARoot,
+        PROXY_CA_PRIVATE_KEY_NODE);
+
+    if (exists(caCertPath)) {
+      zkManager.safeSetData(caCertPath, caCertData, -1, zkAcl,
+          fencingNodePath);
+    } else {
+      zkManager.safeCreate(caCertPath, caCertData, zkAcl,
+          CreateMode.PERSISTENT, zkAcl, fencingNodePath);
+    }
+    if (exists(caPrivateKeyPath)) {
+      zkManager.safeSetData(caPrivateKeyPath, caPrivateKeyData, -1, zkAcl,
+          fencingNodePath);
+    } else {
+      zkManager.safeCreate(caPrivateKeyPath, caPrivateKeyData, zkAcl,
+          CreateMode.PERSISTENT, zkAcl, fencingNodePath);
+    }
+  }
+
   /**
    * Get alternate path for app id if path according to configured split index
    * does not exist. We look for path based on all possible split indices.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/security/ProxyCAManager.java

@@ -28,6 +28,11 @@ import org.apache.hadoop.yarn.server.webproxy.ProxyCA;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+
 /**
  * Manager for {@link ProxyCA}, which contains the Certificate Authority for
  * AMs to have certificates for HTTPS communication with the RM Proxy.
@@ -40,16 +45,23 @@ public class ProxyCAManager extends AbstractService implements Recoverable {
 
   private ProxyCA proxyCA;
   private RMContext rmContext;
+  private boolean wasRecovered;
 
   public ProxyCAManager(ProxyCA proxyCA, RMContext rmContext) {
     super(ProxyCAManager.class.getName());
     this.proxyCA = proxyCA;
     this.rmContext = rmContext;
+    wasRecovered = false;
   }
 
   @Override
   protected void serviceStart() throws Exception {
-    proxyCA.init();
+    if (!wasRecovered) {
+      proxyCA.init();
+    }
+    wasRecovered = false;
+    rmContext.getStateStore().storeProxyCACert(
+        proxyCA.getCaCert(), proxyCA.getCaKeyPair().getPrivate());
     super.serviceStart();
   }
 
@@ -62,7 +74,12 @@ public class ProxyCAManager extends AbstractService implements Recoverable {
     return proxyCA;
   }
 
-  public void recover(RMState state) {
-    // TODO: RM HA YARN-8449
+  public void recover(RMState state)
+      throws GeneralSecurityException, IOException {
+    LOG.info("Recovering CA Certificate and Private Key");
+    X509Certificate caCert = state.getProxyCAState().getCaCert();
+    PrivateKey caPrivateKey = state.getProxyCAState().getCaPrivateKey();
+    proxyCA.init(caCert, caPrivateKey);
+    wasRecovered = true;
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/RMStateStoreTestBase.java

@@ -83,6 +83,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptS
 import org.apache.hadoop.yarn.server.resourcemanager.security.AMRMTokenSecretManager;
 import org.apache.hadoop.yarn.server.resourcemanager.security.ClientToAMTokenSecretManagerInRM;
 import org.apache.hadoop.yarn.server.security.MasterKeyData;
+import org.apache.hadoop.yarn.server.webproxy.ProxyCA;
 import org.apache.hadoop.yarn.util.ConverterUtils;
 import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
@@ -892,6 +893,38 @@ public class RMStateStoreTestBase {
     Assert.assertNull(reservations);
   }
 
+  public void testProxyCA(
+      RMStateStoreHelper stateStoreHelper) throws Exception {
+    RMStateStore store = stateStoreHelper.getRMStateStore();
+    TestDispatcher dispatcher = new TestDispatcher();
+    store.setRMDispatcher(dispatcher);
+
+    ProxyCA originalProxyCA = new ProxyCA();
+    originalProxyCA.init();
+    store.storeProxyCACert(originalProxyCA.getCaCert(),
+        originalProxyCA.getCaKeyPair().getPrivate());
+
+    RMStateStore.ProxyCAState proxyCAState =
+        store.loadState().getProxyCAState();
+    Assert.assertEquals(originalProxyCA.getCaCert(), proxyCAState.getCaCert());
+    Assert.assertEquals(originalProxyCA.getCaKeyPair().getPrivate(),
+        proxyCAState.getCaPrivateKey());
+
+    // Try replacing with a different ProxyCA
+    ProxyCA newProxyCA = new ProxyCA();
+    newProxyCA.init();
+    Assert.assertNotEquals(originalProxyCA.getCaCert(), newProxyCA.getCaCert());
+    Assert.assertNotEquals(originalProxyCA.getCaKeyPair().getPrivate(),
+        newProxyCA.getCaKeyPair().getPrivate());
+    store.storeProxyCACert(newProxyCA.getCaCert(),
+        newProxyCA.getCaKeyPair().getPrivate());
+
+    proxyCAState = store.loadState().getProxyCAState();
+    Assert.assertEquals(newProxyCA.getCaCert(), proxyCAState.getCaCert());
+    Assert.assertEquals(newProxyCA.getCaKeyPair().getPrivate(),
+        proxyCAState.getCaPrivateKey());
+  }
+
   private void validateStoredReservation(
       RMStateStoreHelper stateStoreHelper, TestDispatcher dispatcher,
       RMContext rmContext, ReservationId r1, String planName,

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/TestFSRMStateStore.java

@@ -205,6 +205,7 @@ public class TestFSRMStateStore extends RMStateStoreTestBase {
       testRemoveAttempt(fsTester);
       testAMRMTokenSecretManagerStateStore(fsTester);
       testReservationStateStore(fsTester);
+      testProxyCA(fsTester);
     } finally {
       cluster.shutdown();
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/TestLeveldbRMStateStore.java

@@ -124,6 +124,12 @@ public class TestLeveldbRMStateStore extends RMStateStoreTestBase {
     testReservationStateStore(tester);
   }
 
+  @Test(timeout = 60000)
+  public void testProxyCA() throws Exception {
+    LeveldbStateStoreTester tester = new LeveldbStateStoreTester();
+    testProxyCA(tester);
+  }
+
   @Test(timeout = 60000)
   public void testCompactionCycle() throws Exception {
     final DB mockdb = mock(DB.class);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/recovery/TestZKRMStateStore.java

@@ -289,6 +289,7 @@ public class TestZKRMStateStore extends RMStateStoreTestBase {
     testReservationStateStore(zkTester);
     ((TestZKRMStateStoreTester.TestZKRMStateStoreInternal)
         zkTester.getRMStateStore()).testRetryingCreateRootDir();
+    testProxyCA(zkTester);
   }
 
   @Test

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/security/TestProxyCAManager.java

@@ -21,14 +21,20 @@ package org.apache.hadoop.yarn.server.resourcemanager.security;
 
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
 import org.apache.hadoop.yarn.server.webproxy.ProxyCA;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+
+import static org.mockito.Matchers.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 public class TestProxyCAManager {
 
@@ -36,16 +42,64 @@ public class TestProxyCAManager {
   public void testBasics() throws Exception {
     ProxyCA proxyCA = spy(new ProxyCA());
     RMContext rmContext = mock(RMContext.class);
+    RMStateStore rmStateStore = mock(RMStateStore.class);
+    when(rmContext.getStateStore()).thenReturn(rmStateStore);
     ProxyCAManager proxyCAManager = new ProxyCAManager(proxyCA, rmContext);
     proxyCAManager.init(new YarnConfiguration());
     Assert.assertEquals(proxyCA, proxyCAManager.getProxyCA());
+    verify(rmContext, times(0)).getStateStore();
+    verify(rmStateStore, times(0)).storeProxyCACert(any(), any());
     verify(proxyCA, times(0)).init();
     Assert.assertNull(proxyCA.getCaCert());
     Assert.assertNull(proxyCA.getCaKeyPair());
 
     proxyCAManager.start();
+    verify(rmContext, times(1)).getStateStore();
+    verify(rmStateStore, times(1)).storeProxyCACert(proxyCA.getCaCert(),
+        proxyCA.getCaKeyPair().getPrivate());
     verify(proxyCA, times(1)).init();
     Assert.assertNotNull(proxyCA.getCaCert());
     Assert.assertNotNull(proxyCA.getCaKeyPair());
   }
+
+  @Test
+  public void testRecover() throws Exception {
+    ProxyCA proxyCA = spy(new ProxyCA());
+    RMContext rmContext = mock(RMContext.class);
+    RMStateStore rmStateStore = mock(RMStateStore.class);
+    when(rmContext.getStateStore()).thenReturn(rmStateStore);
+    ProxyCAManager proxyCAManager = new ProxyCAManager(proxyCA, rmContext);
+    proxyCAManager.init(new YarnConfiguration());
+    Assert.assertEquals(proxyCA, proxyCAManager.getProxyCA());
+    verify(rmContext, times(0)).getStateStore();
+    verify(rmStateStore, times(0)).storeProxyCACert(any(), any());
+    verify(proxyCA, times(0)).init();
+    Assert.assertNull(proxyCA.getCaCert());
+    Assert.assertNull(proxyCA.getCaKeyPair());
+
+    RMStateStore.RMState rmState = mock(RMStateStore.RMState.class);
+    RMStateStore.ProxyCAState proxyCAState =
+        mock(RMStateStore.ProxyCAState.class);
+    // We need to use a real certificate + private key because of validation
+    // so just grab them from another ProxyCA
+    ProxyCA otherProxyCA = new ProxyCA();
+    otherProxyCA.init();
+    X509Certificate certificate = otherProxyCA.getCaCert();
+    when(proxyCAState.getCaCert()).thenReturn(certificate);
+    PrivateKey privateKey = otherProxyCA.getCaKeyPair().getPrivate();
+    when(proxyCAState.getCaPrivateKey()).thenReturn(privateKey);
+    when(rmState.getProxyCAState()).thenReturn(proxyCAState);
+    proxyCAManager.recover(rmState);
+    verify(proxyCA, times(1)).init(certificate, privateKey);
+    Assert.assertEquals(certificate, proxyCA.getCaCert());
+    Assert.assertEquals(privateKey, proxyCA.getCaKeyPair().getPrivate());
+
+    proxyCAManager.start();
+    verify(rmContext, times(1)).getStateStore();
+    verify(rmStateStore, times(1)).storeProxyCACert(proxyCA.getCaCert(),
+        proxyCA.getCaKeyPair().getPrivate());
+    verify(proxyCA, times(0)).init();
+    Assert.assertEquals(certificate, proxyCA.getCaCert());
+    Assert.assertEquals(privateKey, proxyCA.getCaKeyPair().getPrivate());
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/main/java/org/apache/hadoop/yarn/server/webproxy/ProxyCA.java

@@ -69,6 +69,7 @@ import java.security.PrivateKey;
 import java.security.PublicKey;
 import java.security.SecureRandom;
 import java.security.Security;
+import java.security.Signature;
 import java.security.SignatureException;
 import java.security.cert.Certificate;
 import java.security.cert.CertificateException;
@@ -107,7 +108,24 @@ public class ProxyCA {
 
   public void init() throws GeneralSecurityException, IOException {
     createCACertAndKeyPair();
+    initInternal();
+  }
+
+  public void init(X509Certificate caCert, PrivateKey caPrivateKey)
+      throws GeneralSecurityException, IOException {
+    if (caCert == null || caPrivateKey == null
+        || !verifyCertAndKeys(caCert, caPrivateKey)) {
+      LOG.warn("Could not verify Certificate, Public Key, and Private Key: " +
+          "regenerating");
+      createCACertAndKeyPair();
+    } else {
+      this.caCert = caCert;
+      this.caKeyPair = new KeyPair(caCert.getPublicKey(), caPrivateKey);
+    }
+    initInternal();
+  }
 
+  private void initInternal() throws GeneralSecurityException, IOException {
     defaultTrustManager = null;
     TrustManagerFactory factory = TrustManagerFactory.getInstance(
         TrustManagerFactory.getDefaultAlgorithm());
@@ -405,4 +423,19 @@ public class ProxyCA {
   public KeyPair getCaKeyPair() {
     return caKeyPair;
   }
+
+  private boolean verifyCertAndKeys(X509Certificate cert,
+      PrivateKey privateKey) throws GeneralSecurityException {
+    PublicKey publicKey = cert.getPublicKey();
+    byte[] data = new byte[2000];
+    srand.nextBytes(data);
+    Signature signer = Signature.getInstance("SHA512withRSA");
+    signer.initSign(privateKey);
+    signer.update(data);
+    byte[] sig = signer.sign();
+    signer = Signature.getInstance("SHA512withRSA");
+    signer.initVerify(publicKey);
+    signer.update(data);
+    return signer.verify(sig);
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-web-proxy/src/test/java/org/apache/hadoop/yarn/server/webproxy/TestProxyCA.java

@@ -34,6 +34,7 @@ import javax.net.ssl.X509TrustManager;
 import javax.security.auth.x500.X500Principal;
 import java.security.InvalidKeyException;
 import java.security.Key;
+import java.security.KeyPair;
 import java.security.KeyStore;
 import java.security.NoSuchAlgorithmException;
 import java.security.NoSuchProviderException;
@@ -69,6 +70,89 @@ public class TestProxyCA {
     Assert.assertNotNull(proxyCA.getHostnameVerifier());
   }
 
+  @Test
+  public void testInit2Null() throws Exception {
+    ProxyCA proxyCA = new ProxyCA();
+    Assert.assertNull(proxyCA.getCaCert());
+    Assert.assertNull(proxyCA.getCaKeyPair());
+    Assert.assertNull(proxyCA.getX509KeyManager());
+    Assert.assertNull(proxyCA.getHostnameVerifier());
+
+    // null certificate and private key
+    proxyCA.init(null, null);
+    Assert.assertNotNull(proxyCA.getCaCert());
+    Assert.assertNotNull(proxyCA.getCaKeyPair());
+    Assert.assertNotNull(proxyCA.getX509KeyManager());
+    Assert.assertNotNull(proxyCA.getHostnameVerifier());
+  }
+
+  @Test
+  public void testInit2Mismatch() throws Exception {
+    ProxyCA proxyCA = new ProxyCA();
+    Assert.assertNull(proxyCA.getCaCert());
+    Assert.assertNull(proxyCA.getCaKeyPair());
+    Assert.assertNull(proxyCA.getX509KeyManager());
+    Assert.assertNull(proxyCA.getHostnameVerifier());
+
+    // certificate and private key don't match
+    CertKeyPair pair1 = createCertAndKeyPair();
+    CertKeyPair pair2 = createCertAndKeyPair();
+    Assert.assertNotEquals(pair1.getCert(), pair2.getCert());
+    Assert.assertNotEquals(pair1.getKeyPair().getPrivate(),
+        pair2.getKeyPair().getPrivate());
+    Assert.assertNotEquals(pair1.getKeyPair().getPublic(),
+        pair2.getKeyPair().getPublic());
+    proxyCA.init(pair1.getCert(), pair2.getKeyPair().getPrivate());
+    Assert.assertNotNull(proxyCA.getCaCert());
+    Assert.assertNotNull(proxyCA.getCaKeyPair());
+    Assert.assertNotNull(proxyCA.getX509KeyManager());
+    Assert.assertNotNull(proxyCA.getHostnameVerifier());
+    Assert.assertNotEquals(proxyCA.getCaCert(), pair1.getCert());
+    Assert.assertNotEquals(proxyCA.getCaKeyPair().getPrivate(),
+        pair2.getKeyPair().getPrivate());
+    Assert.assertNotEquals(proxyCA.getCaKeyPair().getPublic(),
+        pair2.getKeyPair().getPublic());
+  }
+
+  @Test
+  public void testInit2Invalid() throws Exception {
+    ProxyCA proxyCA = new ProxyCA();
+    Assert.assertNull(proxyCA.getCaCert());
+    Assert.assertNull(proxyCA.getCaKeyPair());
+    Assert.assertNull(proxyCA.getX509KeyManager());
+    Assert.assertNull(proxyCA.getHostnameVerifier());
+
+    // Invalid key - fail the verification
+    X509Certificate certificate = Mockito.mock(X509Certificate.class);
+    PrivateKey privateKey = Mockito.mock(PrivateKey.class);
+    try {
+      proxyCA.init(certificate, privateKey);
+      Assert.fail("Expected InvalidKeyException");
+    } catch (InvalidKeyException e) {
+      // expected
+    }
+  }
+
+  @Test
+  public void testInit2() throws Exception {
+    ProxyCA proxyCA = new ProxyCA();
+    Assert.assertNull(proxyCA.getCaCert());
+    Assert.assertNull(proxyCA.getCaKeyPair());
+    Assert.assertNull(proxyCA.getX509KeyManager());
+    Assert.assertNull(proxyCA.getHostnameVerifier());
+
+    // certificate and private key do match
+    CertKeyPair pair = createCertAndKeyPair();
+    proxyCA.init(pair.getCert(), pair.getKeyPair().getPrivate());
+    Assert.assertEquals(pair.getCert(), proxyCA.getCaCert());
+    Assert.assertEquals(pair.getKeyPair().getPrivate(),
+        proxyCA.getCaKeyPair().getPrivate());
+    Assert.assertEquals(pair.getKeyPair().getPublic(),
+        proxyCA.getCaKeyPair().getPublic());
+    Assert.assertNotNull(proxyCA.getX509KeyManager());
+    Assert.assertNotNull(proxyCA.getHostnameVerifier());
+  }
+
   @Test
   public void testCreateChildKeyStore() throws Exception {
     ProxyCA proxyCA = new ProxyCA();
@@ -515,4 +599,29 @@ public class TestProxyCA {
       Certificate[] certs) {
     return Arrays.copyOf(certs, certs.length, X509Certificate[].class);
   }
+
+  private static class CertKeyPair {
+    private X509Certificate cert;
+    private KeyPair keyPair;
+
+    public CertKeyPair(X509Certificate cert, KeyPair keyPair) {
+      this.cert = cert;
+      this.keyPair = keyPair;
+    }
+
+    public X509Certificate getCert() {
+      return cert;
+    }
+
+    public KeyPair getKeyPair() {
+      return keyPair;
+    }
+  }
+
+  private CertKeyPair createCertAndKeyPair() throws Exception {
+    // Re-use a ProxyCA to generate a valid Certificate and KeyPair
+    ProxyCA proxyCA = new ProxyCA();
+    proxyCA.init();
+    return new CertKeyPair(proxyCA.getCaCert(), proxyCA.getCaKeyPair());
+  }
 }