YARN-11295. [Federation] Router Support DelegationToken in MemoryStore mode. (#5032)

Author: slfan1989

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/AbstractClientRequestInterceptor.java

@@ -23,6 +23,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
+import org.apache.hadoop.yarn.server.router.security.RouterDelegationTokenSecretManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,6 +45,8 @@ public abstract class AbstractClientRequestInterceptor
   @SuppressWarnings("checkstyle:visibilitymodifier")
   protected UserGroupInformation user = null;
 
+  private RouterDelegationTokenSecretManager tokenSecretManager = null;
+
   /**
    * Sets the {@link ClientRequestInterceptor} in the chain.
    */
@@ -125,4 +128,13 @@ private void setupUser(String userName) {
     }
   }
 
+  @Override
+  public RouterDelegationTokenSecretManager getTokenSecretManager() {
+    return tokenSecretManager;
+  }
+
+  @Override
+  public void setTokenSecretManager(RouterDelegationTokenSecretManager tokenSecretManager) {
+    this.tokenSecretManager = tokenSecretManager;
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/ClientRequestInterceptor.java

@@ -20,6 +20,7 @@
 
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
+import org.apache.hadoop.yarn.server.router.security.RouterDelegationTokenSecretManager;
 
 /**
  * Defines the contract to be implemented by the request interceptor classes,
@@ -62,4 +63,18 @@ public interface ClientRequestInterceptor
    */
   ClientRequestInterceptor getNextInterceptor();
 
+  /**
+   * Set RouterDelegationTokenSecretManager for specific interceptor to support Token operations,
+   * including create Token, update Token, and delete Token.
+   *
+   * @param tokenSecretManager Router DelegationTokenSecretManager
+   */
+  void setTokenSecretManager(RouterDelegationTokenSecretManager tokenSecretManager);
+
+  /**
+   * Get RouterDelegationTokenSecretManager.
+   *
+   * @return Router DelegationTokenSecretManager.
+   */
+  RouterDelegationTokenSecretManager getTokenSecretManager();
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterClientRMService.java

@@ -22,6 +22,7 @@
 import java.net.InetSocketAddress;
 import java.util.Collections;
 import java.util.Map;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.conf.Configuration;
@@ -105,6 +106,7 @@
 import org.apache.hadoop.yarn.exceptions.YarnException;
 import org.apache.hadoop.yarn.ipc.YarnRPC;
 import org.apache.hadoop.yarn.server.router.RouterServerUtil;
+import org.apache.hadoop.yarn.server.router.security.RouterDelegationTokenSecretManager;
 import org.apache.hadoop.yarn.server.router.security.authorize.RouterPolicyProvider;
 import org.apache.hadoop.yarn.util.LRUCacheHashMap;
 import org.slf4j.Logger;
@@ -136,6 +138,8 @@ public class RouterClientRMService extends AbstractService
   // and remove the oldest used ones.
   private Map<String, RequestInterceptorChainWrapper> userPipelineMap;
 
+  private RouterDelegationTokenSecretManager routerDTSecretManager;
+
   public RouterClientRMService() {
     super(RouterClientRMService.class.getName());
   }
@@ -164,8 +168,12 @@ protected void serviceStart() throws Exception {
         serverConf.getInt(YarnConfiguration.RM_CLIENT_THREAD_COUNT,
             YarnConfiguration.DEFAULT_RM_CLIENT_THREAD_COUNT);
 
+    // Initialize RouterRMDelegationTokenSecretManager.
+    routerDTSecretManager = createRouterRMDelegationTokenSecretManager(conf);
+    routerDTSecretManager.startThreads();
+
     this.server = rpc.getServer(ApplicationClientProtocol.class, this,
-        listenerEndpoint, serverConf, null, numWorkerThreads);
+        listenerEndpoint, serverConf, routerDTSecretManager, numWorkerThreads);
 
     // Enable service authorization?
     if (conf.getBoolean(
@@ -508,6 +516,13 @@ private RequestInterceptorChainWrapper initializePipeline(String user) {
         ClientRequestInterceptor interceptorChain =
             this.createRequestInterceptorChain();
         interceptorChain.init(user);
+
+        // We set the RouterDelegationTokenSecretManager instance to the interceptorChain
+        // and let the interceptor use it.
+        if (routerDTSecretManager != null) {
+          interceptorChain.setTokenSecretManager(routerDTSecretManager);
+        }
+
         chainWrapper.init(interceptorChain);
       } catch (Exception e) {
         LOG.error("Init ClientRequestInterceptor error for user: {}.", user, e);
@@ -558,4 +573,42 @@ protected void finalize() {
   public Map<String, RequestInterceptorChainWrapper> getUserPipelineMap() {
     return userPipelineMap;
   }
+
+  /**
+   * Create RouterRMDelegationTokenSecretManager.
+   * In the YARN federation, the Router will replace the RM to
+   * manage the RMDelegationToken (generate, update, cancel),
+   * so the relevant configuration parameters still obtain the configuration parameters of the RM.
+   *
+   * @param conf Configuration
+   * @return RouterDelegationTokenSecretManager.
+   */
+  protected RouterDelegationTokenSecretManager createRouterRMDelegationTokenSecretManager(
+      Configuration conf) {
+
+    long secretKeyInterval = conf.getLong(
+        YarnConfiguration.RM_DELEGATION_KEY_UPDATE_INTERVAL_KEY,
+        YarnConfiguration.RM_DELEGATION_KEY_UPDATE_INTERVAL_DEFAULT);
+
+    long tokenMaxLifetime = conf.getLong(
+        YarnConfiguration.RM_DELEGATION_TOKEN_MAX_LIFETIME_KEY,
+        YarnConfiguration.RM_DELEGATION_TOKEN_MAX_LIFETIME_DEFAULT);
+
+    long tokenRenewInterval = conf.getLong(
+        YarnConfiguration.RM_DELEGATION_TOKEN_RENEW_INTERVAL_KEY,
+        YarnConfiguration.RM_DELEGATION_TOKEN_RENEW_INTERVAL_DEFAULT);
+
+    long removeScanInterval = conf.getTimeDuration(
+        YarnConfiguration.RM_DELEGATION_TOKEN_REMOVE_SCAN_INTERVAL_KEY,
+        YarnConfiguration.RM_DELEGATION_TOKEN_REMOVE_SCAN_INTERVAL_DEFAULT,
+        TimeUnit.MILLISECONDS);
+
+    return new RouterDelegationTokenSecretManager(secretKeyInterval,
+        tokenMaxLifetime, tokenRenewInterval, removeScanInterval);
+  }
+
+  @VisibleForTesting
+  public RouterDelegationTokenSecretManager getRouterDTSecretManager() {
+    return routerDTSecretManager;
+  }
 }