HADOOP-6573. Support for persistent delegation tokens. Contributed by Jitendra Pandey.

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

CHANGES.txt

@@ -170,6 +170,9 @@ Trunk (unreleased changes)
     HADOOP-6596. Add a version field to the AbstractDelegationTokenIdentifier's
     serialized value. (omalley)
 
+    HADOOP-6573. Support for persistent delegation tokens.
+    (Jitendra Pandey via shv)
+
   OPTIMIZATIONS
 
     HADOOP-6467. Improve the performance on HarFileSystem.listStatus(..).

src/java/org/apache/hadoop/security/token/delegation/AbstractDelegationTokenSecretManager.java

@@ -52,23 +52,30 @@ class AbstractDelegationTokenSecretManager<TokenIdent
 
   /** 
    * Cache of currently valid tokens, mapping from DelegationTokenIdentifier 
-   * to DelegationTokenInformation. Protected by its own lock.
+   * to DelegationTokenInformation. Protected by this object lock.
    */
-  private final Map<TokenIdent, DelegationTokenInformation> currentTokens 
+  protected final Map<TokenIdent, DelegationTokenInformation> currentTokens 
       = new HashMap<TokenIdent, DelegationTokenInformation>();
   
   /**
-   * Sequence number to create DelegationTokenIdentifier
+   * Sequence number to create DelegationTokenIdentifier.
+   * Protected by this object lock.
    */
-  private int delegationTokenSequenceNumber = 0;
+  protected int delegationTokenSequenceNumber = 0;
   
-  private final Map<Integer, DelegationKey> allKeys 
+  /**
+   * Access to allKeys is protected by this object lock
+   */
+  protected final Map<Integer, DelegationKey> allKeys 
       = new HashMap<Integer, DelegationKey>();
   
   /**
-   * Access to currentId and currentKey is protected by this object lock.
+   * Access to currentId is protected by this object lock.
+   */
+  protected int currentId = 0;
+  /**
+   * Access to currentKey is protected by this object lock
    */
-  private int currentId = 0;
   private DelegationKey currentKey;
   
   private long keyUpdateInterval;
@@ -76,7 +83,7 @@ class AbstractDelegationTokenSecretManager<TokenIdent
   private long tokenRemoverScanInterval;
   private long tokenRenewInterval;
   private Thread tokenRemoverThread;
-  private volatile boolean running;
+  protected volatile boolean running;
 
   public AbstractDelegationTokenSecretManager(long delegationKeyUpdateInterval,
       long delegationTokenMaxLifetime, long delegationTokenRenewInterval,
@@ -112,18 +119,40 @@ public synchronized DelegationKey[] getAllKeys() {
     return allKeys.values().toArray(new DelegationKey[0]);
   }
   
-  /** Update the current master key */
+  protected void logUpdateMasterKey(DelegationKey key) throws IOException {
-  private synchronized void updateCurrentKey() throws IOException {
+    return;
-    LOG.info("Updating the current master key for generating delegation tokens");
-    /* Create a new currentKey with an estimated expiry date. */
-    currentId++;
-    currentKey = new DelegationKey(currentId, System.currentTimeMillis()
-        + keyUpdateInterval + tokenMaxLifetime, generateSecret());
-    allKeys.put(currentKey.getKeyId(), currentKey);
   }
   
-  /** Update the current master key for generating delegation tokens */
+  /** 
-  public synchronized void rollMasterKey() throws IOException {
+   * Update the current master key 
+   * This is called once by startThreads before tokenRemoverThread is created, 
+   * and only by tokenRemoverThread afterwards.
+   */
+  private void updateCurrentKey() throws IOException {
+    LOG.info("Updating the current master key for generating delegation tokens");
+    /* Create a new currentKey with an estimated expiry date. */
+    int newCurrentId;
+    synchronized (this) {
+      newCurrentId = currentId+1;
+    }
+    DelegationKey newKey = new DelegationKey(newCurrentId, System
+        .currentTimeMillis()
+        + keyUpdateInterval + tokenMaxLifetime, generateSecret());
+    //Log must be invoked outside the lock on 'this'
+    logUpdateMasterKey(newKey);
+    synchronized (this) {
+      currentId = newKey.getKeyId();
+      currentKey = newKey;
+      allKeys.put(currentKey.getKeyId(), currentKey);
+    }
+  }
+  
+  /** 
+   * Update the current master key for generating delegation tokens 
+   * It should be called only by tokenRemoverThread.
+   */
+  void rollMasterKey() throws IOException {
+    synchronized (this) {
       removeExpiredKeys();
       /* set final expiry date for retiring currentKey */
       currentKey.setExpiryDate(System.currentTimeMillis() + tokenMaxLifetime);
@@ -133,6 +162,7 @@ public synchronized void rollMasterKey() throws IOException {
        * allKeys just in case.
        */
       allKeys.put(currentKey.getKeyId(), currentKey);
+    }
     updateCurrentKey();
   }
 
@@ -148,35 +178,24 @@ private synchronized void removeExpiredKeys() {
   }
   
   @Override
-  protected byte[] createPassword(TokenIdent identifier) {
+  protected synchronized byte[] createPassword(TokenIdent identifier) {
     int sequenceNum;
-    int id;
-    DelegationKey key;
     long now = System.currentTimeMillis();
-    synchronized (this) {
-      id = currentId;
-      key = currentKey;
     sequenceNum = ++delegationTokenSequenceNumber;
-    }
     identifier.setIssueDate(now);
     identifier.setMaxDate(now + tokenMaxLifetime);
-    identifier.setMasterKeyId(id);
+    identifier.setMasterKeyId(currentId);
     identifier.setSequenceNumber(sequenceNum);
-    byte[] password = createPassword(identifier.getBytes(), key.getKey());
+    byte[] password = createPassword(identifier.getBytes(), currentKey.getKey());
-    synchronized (currentTokens) {
     currentTokens.put(identifier, new DelegationTokenInformation(now
         + tokenRenewInterval, password));
-    }
     return password;
   }
 
   @Override
-  public byte[] retrievePassword(TokenIdent identifier
+  public synchronized byte[] retrievePassword(TokenIdent identifier)
-                                 ) throws InvalidToken {
+      throws InvalidToken {
-    DelegationTokenInformation info = null;
+    DelegationTokenInformation info = currentTokens.get(identifier);
-    synchronized (currentTokens) {
-      info = currentTokens.get(identifier);
-    }
     if (info == null) {
       throw new InvalidToken("token is expired or doesn't exist");
     }
@@ -195,18 +214,14 @@ public byte[] retrievePassword(TokenIdent identifier
    * @throws InvalidToken if the token is invalid
    * @throws AccessControlException if the user can't renew token
    */
-  public long renewToken(Token<TokenIdent> token,
+  public synchronized long renewToken(Token<TokenIdent> token,
                          String renewer) throws InvalidToken, IOException {
     long now = System.currentTimeMillis();
     ByteArrayInputStream buf = new ByteArrayInputStream(token.getIdentifier());
     DataInputStream in = new DataInputStream(buf);
     TokenIdent id = createIdentifier();
     id.readFields(in);
-    synchronized (currentTokens) {
+
-      if (currentTokens.get(id) == null) {
-        throw new InvalidToken("Renewal request for unknown token");
-      }
-    }
     if (id.getMaxDate() < now) {
       throw new InvalidToken("User " + renewer + 
                              " tried to renew an expired token");
@@ -222,36 +237,36 @@ public long renewToken(Token<TokenIdent> token,
                                        "renewer specified as " + 
                                        id.getRenewer());
     }
-    DelegationKey key = null;
+    DelegationKey key = allKeys.get(id.getMasterKeyId());
-    synchronized (this) {
-      key = allKeys.get(id.getMasterKeyId());
-    }
     if (key == null) {
-      throw new InvalidToken("Unable to find master key for keyId=" + 
+      throw new InvalidToken("Unable to find master key for keyId="
-                             id.getMasterKeyId() +
+          + id.getMasterKeyId()
-                             " from cache. Failed to renew an unexpired token"+
+          + " from cache. Failed to renew an unexpired token"
-                             " with sequenceNumber=" + id.getSequenceNumber());
+          + " with sequenceNumber=" + id.getSequenceNumber());
     }
     byte[] password = createPassword(token.getIdentifier(), key.getKey());
     if (!Arrays.equals(password, token.getPassword())) {
-      throw new AccessControlException("Client " + renewer + 
+      throw new AccessControlException("Client " + renewer
-                                       " is trying to renew a token with " +
+          + " is trying to renew a token with " + "wrong password");
-                                       "wrong password");
+    }
+    long renewTime = Math.min(id.getMaxDate(), now + tokenRenewInterval);
+    DelegationTokenInformation info = new DelegationTokenInformation(renewTime,
+        password);
+
+    if (currentTokens.get(id) == null) {
+      throw new InvalidToken("Renewal request for unknown token");
     }
-    DelegationTokenInformation info = new DelegationTokenInformation(
-        Math.min(id.getMaxDate(), now + tokenRenewInterval), password);
-    synchronized (currentTokens) {
     currentTokens.put(id, info);
-    }
+    return renewTime;
-    return info.getRenewDate();
   }
   
   /**
    * Cancel a token by removing it from cache.
+   * @return Identifier of the canceled token
    * @throws InvalidToken for invalid token
    * @throws AccessControlException if the user isn't allowed to cancel
    */
-  public void cancelToken(Token<TokenIdent> token,
+  public synchronized TokenIdent cancelToken(Token<TokenIdent> token,
       String canceller) throws IOException {
     ByteArrayInputStream buf = new ByteArrayInputStream(token.getIdentifier());
     DataInputStream in = new DataInputStream(buf);
@@ -262,18 +277,17 @@ public void cancelToken(Token<TokenIdent> token,
     }
     String owner = id.getUser().getUserName();
     Text renewer = id.getRenewer();
-    if (!canceller.equals(owner) && 
+    if (!canceller.equals(owner)
-        (renewer == null || !canceller.equals(renewer.toString()))) {
+        && (renewer == null || !canceller.equals(renewer.toString()))) {
-      throw new AccessControlException(canceller + 
+      throw new AccessControlException(canceller
-                                      " is not authorized to cancel the token");
+          + " is not authorized to cancel the token");
     }
     DelegationTokenInformation info = null;
-    synchronized (currentTokens) {
     info = currentTokens.remove(id);
-    }
     if (info == null) {
       throw new InvalidToken("Token not found");
     }
+    return id;
   }
   
   /**
@@ -285,16 +299,16 @@ public static SecretKey createSecretKey(byte[] key) {
     return SecretManager.createSecretKey(key);
   }
 
-  /** Utility class to encapsulate a token's renew date and password. */
+  /** Class to encapsulate a token's renew date and password. */
-  private static class DelegationTokenInformation {
+  public static class DelegationTokenInformation {
     long renewDate;
     byte[] password;
-    DelegationTokenInformation(long renewDate, byte[] password) {
+    public DelegationTokenInformation(long renewDate, byte[] password) {
       this.renewDate = renewDate;
       this.password = password;
     }
     /** returns renew date */
-    long getRenewDate() {
+    public long getRenewDate() {
       return renewDate;
     }
     /** returns password */
@@ -304,9 +318,8 @@ byte[] getPassword() {
   }
   
   /** Remove expired delegation tokens from cache */
-  private void removeExpiredToken() {
+  private synchronized void removeExpiredToken() {
     long now = System.currentTimeMillis();
-    synchronized (currentTokens) {
     Iterator<DelegationTokenInformation> i = currentTokens.values().iterator();
     while (i.hasNext()) {
       long renewDate = i.next().getRenewDate();
@@ -315,14 +328,15 @@ private void removeExpiredToken() {
       }
     }
   }
-  }
 
   public synchronized void stopThreads() {
     if (LOG.isDebugEnabled())
       LOG.debug("Stopping expired delegation token remover thread");
     running = false;
+    if (tokenRemoverThread != null) {
       tokenRemoverThread.interrupt();
     }
+  }
   
   private class ExpiredTokenRemover extends Thread {
     private long lastMasterKeyUpdate;

src/test/core/org/apache/hadoop/security/token/delegation/TestDelegationToken.java

@@ -25,7 +25,10 @@
 import java.io.IOException;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 
 import junit.framework.Assert;
 
@@ -36,8 +39,11 @@
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.token.SecretManager;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.SecretManager.InvalidToken;
+import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSecretManager.DelegationTokenInformation;
+import org.apache.hadoop.util.Daemon;
 import org.apache.hadoop.util.StringUtils;
 import org.junit.Test;
 
@@ -91,6 +97,18 @@ public TestDelegationTokenIdentifier createIdentifier() {
     protected byte[] createPassword(TestDelegationTokenIdentifier t) {
       return super.createPassword(t);
     }
+    
+    public byte[] createPassword(TestDelegationTokenIdentifier t, DelegationKey key) {
+      return SecretManager.createPassword(t.getBytes(), key.getKey());
+    }
+    
+    public Map<TestDelegationTokenIdentifier, DelegationTokenInformation> getAllTokens() {
+      return currentTokens;
+    }
+    
+    public DelegationKey getKey(TestDelegationTokenIdentifier id) {
+      return allKeys.get(id.getMasterKeyId());
+    }
   }
   
   public static class TokenSelector extends 
@@ -299,4 +317,52 @@ public void testDelegationTokenSelector() throws Exception {
       dtSecretManager.stopThreads();
     }
   }
+  
+  @Test
+  public void testParallelDelegationTokenCreation() throws Exception {
+    final TestDelegationTokenSecretManager dtSecretManager = 
+        new TestDelegationTokenSecretManager(2000, 24 * 60 * 60 * 1000, 
+            7 * 24 * 60 * 60 * 1000, 2000);
+    try {
+      dtSecretManager.startThreads();
+      int numThreads = 100;
+      final int numTokensPerThread = 100;
+      class tokenIssuerThread implements Runnable {
+
+        public void run() {
+          for(int i =0;i <numTokensPerThread; i++) {
+            generateDelegationToken(dtSecretManager, "auser", "arenewer");
+            try {
+              Thread.sleep(250); 
+            } catch (Exception e) {
+            }
+          }
+        }
+      }
+      Thread[] issuers = new Thread[numThreads];
+      for (int i =0; i <numThreads; i++) {
+        issuers[i] = new Daemon(new tokenIssuerThread());
+        issuers[i].start();
+      }
+      for (int i =0; i <numThreads; i++) {
+        issuers[i].join();
+      }
+      Map<TestDelegationTokenIdentifier, DelegationTokenInformation> tokenCache = dtSecretManager
+          .getAllTokens();
+      Assert.assertEquals(numTokensPerThread*numThreads, tokenCache.size());
+      Iterator<TestDelegationTokenIdentifier> iter = tokenCache.keySet().iterator();
+      while (iter.hasNext()) {
+        TestDelegationTokenIdentifier id = iter.next();
+        DelegationTokenInformation info = tokenCache.get(id);
+        Assert.assertTrue(info != null);
+        DelegationKey key = dtSecretManager.getKey(id);
+        Assert.assertTrue(key != null);
+        byte[] storedPassword = dtSecretManager.retrievePassword(id);
+        byte[] password = dtSecretManager.createPassword(id, key);
+        Assert.assertTrue(Arrays.equals(password, storedPassword));
+      }
+    } finally {
+      dtSecretManager.stopThreads();
+    }
+  }
 }