MAPREDUCE-3940. ContainerTokens should have an expiry interval. Contributed by Siddharth Seth and Vinod Kumar Vavilapalli.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1359910 13f79535-47bb-0310-9956-ffa450edef68

Author: vinoduec

hadoop-mapreduce-project/CHANGES.txt

@@ -670,6 +670,9 @@ Release 0.23.3 - UNRELEASED
     MAPREDUCE-4252. MR2 job never completes with 1 pending task (Tom White via
     bobby)
 
+    MAPREDUCE-3940. ContainerTokens should have an expiry interval. (Siddharth
+    Seth and Vinod Kumar Vavilapalli via vinodkv)
+
 Release 0.23.2 - UNRELEASED
 
   INCOMPATIBLE CHANGES

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestRMContainerAllocator.java

@@ -29,6 +29,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
@@ -86,11 +87,13 @@
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.server.resourcemanager.MockNM;
 import org.apache.hadoop.yarn.server.resourcemanager.MockRM;
+import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.Allocation;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.SchedulerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
+import org.apache.hadoop.yarn.server.security.ContainerTokenSecretManager;
 import org.apache.hadoop.yarn.util.BuilderUtils;
 import org.junit.After;
 import org.junit.Test;
@@ -352,7 +355,7 @@ public void handle(SchedulerEvent event) {
     }
     @Override
     protected ResourceScheduler createScheduler() {
-      return new MyFifoScheduler();
+      return new MyFifoScheduler(this.getRMContext());
     }
   }
 
@@ -1091,6 +1094,19 @@ public void testBlackListedNodesWithSchedulingToThatNode() throws Exception {
   }
 
   private static class MyFifoScheduler extends FifoScheduler {
+
+    public MyFifoScheduler(RMContext rmContext) {
+      super();
+      try {
+        Configuration conf = new Configuration();
+        reinitialize(conf, new ContainerTokenSecretManager(conf),
+            rmContext);
+      } catch (IOException ie) {
+        LOG.info("add application failed with ", ie);
+        assert (false);
+      }
+    }
+
     // override this to copy the objects otherwise FifoScheduler updates the
     // numContainers in same objects as kept by RMContainerAllocator
     @Override

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api/records/ContainerToken.java

@@ -50,46 +50,46 @@ public interface ContainerToken extends DelegationToken {
    */
   @Public
   @Stable
-  public abstract ByteBuffer getIdentifier();
+  ByteBuffer getIdentifier();
 
   @Private
   @Stable
-  public abstract void setIdentifier(ByteBuffer identifier);
+  void setIdentifier(ByteBuffer identifier);
 
   /**
    * Get the token password
    * @return token password
    */
   @Public
   @Stable
-  public abstract ByteBuffer getPassword();
+  ByteBuffer getPassword();
 
   @Private
   @Stable
-  public abstract void setPassword(ByteBuffer password);
+  void setPassword(ByteBuffer password);
 
   /**
    * Get the token kind.
    * @return token kind
    */
   @Public
   @Stable
-  public abstract String getKind();
+  String getKind();
 
   @Private
   @Stable
-  public abstract void setKind(String kind);
+  void setKind(String kind);
 
   /**
    * Get the service to which the token is allocated.
    * @return service to which the token is allocated
    */
   @Public
   @Stable
-  public abstract String getService();
+  String getService();
 
   @Private
   @Stable
-  public abstract void setService(String service);
+  void setService(String service);
 
 }

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/api/impl/pb/client/ContainerManagerPBClientImpl.java

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.io.Closeable;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.ipc.ProtobufRpcEngine;
@@ -49,7 +50,8 @@
 
 import com.google.protobuf.ServiceException;
 
-public class ContainerManagerPBClientImpl implements ContainerManager {
+public class ContainerManagerPBClientImpl implements ContainerManager,
+    Closeable {
 
   // Not a documented config. Only used for tests
   static final String NM_COMMAND_TIMEOUT = YarnConfiguration.YARN_PREFIX

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/security/ContainerTokenIdentifier.java

@@ -35,6 +35,11 @@
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.util.BuilderUtils;
 
+/**
+ * TokenIdentifier for a container. Encodes {@link ContainerId},
+ * {@link Resource} needed by the container and the target NMs host-address.
+ * 
+ */
 public class ContainerTokenIdentifier extends TokenIdentifier {
 
   private static Log LOG = LogFactory.getLog(ContainerTokenIdentifier.class);
@@ -44,14 +49,19 @@ public class ContainerTokenIdentifier extends TokenIdentifier {
   private ContainerId containerId;
   private String nmHostAddr;
   private Resource resource;
+  private long expiryTimeStamp;
 
   public ContainerTokenIdentifier(ContainerId containerID, String hostName,
-      Resource r) {
+      Resource r, long expiryTimeStamp) {
     this.containerId = containerID;
     this.nmHostAddr = hostName;
     this.resource = r;
+    this.expiryTimeStamp = expiryTimeStamp;
   }
 
+  /**
+   * Default constructor needed by RPC layer/SecretManager.
+   */
   public ContainerTokenIdentifier() {
   }
 
@@ -67,6 +77,10 @@ public Resource getResource() {
     return this.resource;
   }
 
+  public long getExpiryTimeStamp() {
+    return this.expiryTimeStamp;
+  }
+
   @Override
   public void write(DataOutput out) throws IOException {
     LOG.debug("Writing ContainerTokenIdentifier to RPC layer: " + this);
@@ -79,6 +93,7 @@ public void write(DataOutput out) throws IOException {
     out.writeInt(this.containerId.getId());
     out.writeUTF(this.nmHostAddr);
     out.writeInt(this.resource.getMemory());
+    out.writeLong(this.expiryTimeStamp);
   }
 
   @Override
@@ -91,6 +106,7 @@ public void readFields(DataInput in) throws IOException {
         .readInt());
     this.nmHostAddr = in.readUTF();
     this.resource = BuilderUtils.newResource(in.readInt());
+    this.expiryTimeStamp = in.readLong();
   }
 
   @Override
@@ -103,6 +119,7 @@ public UserGroupInformation getUser() {
     return UserGroupInformation.createRemoteUser(this.containerId.toString());
   }
 
+  // TODO: Needed?
   @InterfaceAudience.Private
   public static class Renewer extends Token.TrivialRenewer {
     @Override

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/security/ContainerTokenSecretManager.java

@@ -18,16 +18,29 @@
 
 package org.apache.hadoop.yarn.server.security;
 
+import java.nio.ByteBuffer;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
 import javax.crypto.SecretKey;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.security.token.SecretManager;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.api.records.ContainerToken;
+import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.security.ContainerTokenIdentifier;
+import org.apache.hadoop.yarn.util.BuilderUtils;
 
+/**
+ * SecretManager for ContainerTokens. Used by both RM and NM and hence is
+ * present in yarn-server-common package.
+ * 
+ */
 public class ContainerTokenSecretManager extends
     SecretManager<ContainerTokenIdentifier> {
 
@@ -36,7 +49,34 @@ public class ContainerTokenSecretManager extends
 
   Map<String, SecretKey> secretkeys =
     new ConcurrentHashMap<String, SecretKey>();
-  
+
+  private final long containerTokenExpiryInterval;
+
+  public ContainerTokenSecretManager(Configuration conf) {
+    this.containerTokenExpiryInterval =
+        conf.getInt(YarnConfiguration.RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS,
+          YarnConfiguration.DEFAULT_RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS);
+  }
+
+  public ContainerToken createContainerToken(ContainerId containerId,
+      NodeId nodeId, Resource capability) {
+    try {
+      long expiryTimeStamp =
+          System.currentTimeMillis() + containerTokenExpiryInterval;
+      ContainerTokenIdentifier tokenIdentifier =
+          new ContainerTokenIdentifier(containerId, nodeId.toString(),
+            capability, expiryTimeStamp);
+      return BuilderUtils.newContainerToken(nodeId,
+        ByteBuffer.wrap(this.createPassword(tokenIdentifier)), tokenIdentifier);
+    } catch (IllegalArgumentException e) {
+      // this could be because DNS is down - in which case we just want
+      // to retry and not bring RM down. Caller should note and act on the fact
+      // that container is not creatable.
+      LOG.error("Error trying to create new container", e);
+      return null;
+    }
+  }
+
   // Used by master for generation of secretyKey per host
   public SecretKey createAndGetSecretKey(CharSequence hostName) {
     String hostNameStr = hostName.toString();

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/NodeManager.java

@@ -116,7 +116,7 @@ public void init(Configuration conf) {
     if (UserGroupInformation.isSecurityEnabled()) {
       LOG.info("Security is enabled on NodeManager. "
           + "Creating ContainerTokenSecretManager");
-      this.containerTokenSecretManager = new ContainerTokenSecretManager();
+      this.containerTokenSecretManager = new ContainerTokenSecretManager(conf);
     }
 
     this.aclsManager = new ApplicationACLsManager(conf);

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java

@@ -324,6 +324,15 @@ private void authorizeRequest(ContainerId containerID,
                 + containerIDStr);
       } else {
 
+        // Ensure the token is not expired. 
+        // Token expiry is not checked for stopContainer/getContainerStatus
+        if (tokenId.getExpiryTimeStamp() < System.currentTimeMillis()) {
+          unauthorized = true;
+          messageBuilder.append("\nThis token is expired. current time is "
+              + System.currentTimeMillis() + " found "
+              + tokenId.getExpiryTimeStamp());
+        }
+        
         Resource resource = tokenId.getResource();
         if (!resource.equals(launchContext.getResource())) {
           unauthorized = true;

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestEventFlow.java

@@ -86,7 +86,8 @@ public void testSuccessfulContainerLaunch() throws InterruptedException,
     healthChecker.init(conf);
     LocalDirsHandlerService dirsHandler = healthChecker.getDiskHandler();
     NodeManagerMetrics metrics = NodeManagerMetrics.create();
-    ContainerTokenSecretManager containerTokenSecretManager =  new ContainerTokenSecretManager();
+    ContainerTokenSecretManager containerTokenSecretManager =
+        new ContainerTokenSecretManager(conf);
     NodeStatusUpdater nodeStatusUpdater =
         new NodeStatusUpdaterImpl(context, dispatcher, healthChecker, metrics, containerTokenSecretManager) {
       @Override

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/BaseContainerManagerTest.java

@@ -70,7 +70,8 @@ public abstract class BaseContainerManagerTest {
   protected static File localLogDir;
   protected static File remoteLogDir;
   protected static File tmpDir;
-  protected ContainerTokenSecretManager containerTokenSecretManager = new ContainerTokenSecretManager();
+  protected ContainerTokenSecretManager containerTokenSecretManager =
+      new ContainerTokenSecretManager(new Configuration());
 
   protected final NodeManagerMetrics metrics = NodeManagerMetrics.create();