HDFS-984. Persistent delegation tokens. Contributed by Jitendra Pandey.

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

CHANGES.txt

@@ -7,6 +7,8 @@ Trunk (unreleased changes)
     HDFS-905. Use the new UserGroupInformation from HDFS-6299. 
     (jghoman via omalley)
 
+    HDFS-984. Persistent delegation tokens. (Jitendra Pandey via shv)
+
   NEW FEATURES
 
     HDFS-654. Add support new atomic rename functionality in HDFS for 

src/java/org/apache/hadoop/hdfs/protocol/FSConstants.java

@@ -17,7 +17,6 @@
  */
 package org.apache.hadoop.hdfs.protocol;
 
-import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 
 /************************************
@@ -91,7 +90,8 @@ public interface FSConstants {
   // Version is reflected in the data storage file.
   // Versions are negative.
   // Decrement LAYOUT_VERSION to define a new version.
-  public static final int LAYOUT_VERSION = -23;
+  public static final int LAYOUT_VERSION = -24;
   // Current version: 
-  // -23: Symbolic links
+  // -24: added new OP_[GET|RENEW|CANCEL]_DELEGATION_TOKEN and
+  // OP_UPDATE_MASTER_KEY.
 }

src/java/org/apache/hadoop/hdfs/security/token/delegation/DelegationTokenSecretManager.java

@@ -18,8 +18,17 @@
 
 package org.apache.hadoop.hdfs.security.token.delegation;
 
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.util.Iterator;
+
 import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSecretManager;
+import org.apache.hadoop.security.token.delegation.DelegationKey;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 /**
  * A HDFS specific delegation token secret manager.
@@ -30,6 +39,10 @@ import org.apache.hadoop.security.token.delegation.AbstractDelegationTokenSecret
 public class DelegationTokenSecretManager
     extends AbstractDelegationTokenSecretManager<DelegationTokenIdentifier> {
 
+  private static final Log LOG = LogFactory
+      .getLog(DelegationTokenSecretManager.class);
+  
+  private final FSNamesystem namesystem;
   /**
    * Create a secret manager
    * @param delegationKeyUpdateInterval the number of seconds for rolling new
@@ -41,16 +54,220 @@ public class DelegationTokenSecretManager
    *        for expired tokens
    */
   public DelegationTokenSecretManager(long delegationKeyUpdateInterval,
-                                      long delegationTokenMaxLifetime, 
-                                      long delegationTokenRenewInterval,
-                                      long delegationTokenRemoverScanInterval) {
+      long delegationTokenMaxLifetime, long delegationTokenRenewInterval,
+      long delegationTokenRemoverScanInterval, FSNamesystem namesystem) {
     super(delegationKeyUpdateInterval, delegationTokenMaxLifetime,
-          delegationTokenRenewInterval, delegationTokenRemoverScanInterval);
+        delegationTokenRenewInterval, delegationTokenRemoverScanInterval);
+    this.namesystem = namesystem;
   }
 
-  @Override
+  @Override //SecretManager
   public DelegationTokenIdentifier createIdentifier() {
     return new DelegationTokenIdentifier();
   }
 
+  /**
+   * Returns expiry time of a token given its identifier.
+   * 
+   * @param dtId DelegationTokenIdentifier of a token
+   * @return Expiry time of the token
+   * @throws IOException
+   */
+  public synchronized long getTokenExpiryTime(
+      DelegationTokenIdentifier dtId) throws IOException {
+    DelegationTokenInformation info = currentTokens.get(dtId);
+    if (info != null) {
+      return info.getRenewDate();
+    } else {
+      throw new IOException("No delegation token found for this identifier");
+    }
+  }
+  
+  /**
+   * Load SecretManager state from fsimage.
+   * 
+   * @param in input stream to read fsimage
+   * @throws IOException
+   */
+  public synchronized void loadSecretManagerState(DataInputStream in)
+      throws IOException {
+    if (running) {
+      // a safety check
+      throw new IOException(
+          "Can't load state from image in a running SecretManager.");
+    }
+    currentId = in.readInt();
+    loadAllKeys(in);
+    delegationTokenSequenceNumber = in.readInt();
+    loadCurrentTokens(in);
+  }
+  
+  /**
+   * Store the current state of the SecretManager for persistence
+   * 
+   * @param out Output stream for writing into fsimage.
+   * @throws IOException
+   */
+  public synchronized void saveSecretManagerState(DataOutputStream out)
+      throws IOException {
+    out.writeInt(currentId);
+    saveAllKeys(out);
+    out.writeInt(delegationTokenSequenceNumber);
+    saveCurrentTokens(out);
+  }
+  
+  /**
+   * This method is intended to be used only while reading edit logs.
+   * 
+   * @param identifier DelegationTokenIdentifier read from the edit logs or
+   * fsimage
+   * 
+   * @param expiryTime token expiry time
+   * @throws IOException
+   */
+  public synchronized void addPersistedDelegationToken(
+      DelegationTokenIdentifier identifier, long expiryTime) throws IOException {
+    if (running) {
+      // a safety check
+      throw new IOException(
+          "Can't add persisted delegation token to a running SecretManager.");
+    }
+    int keyId = identifier.getMasterKeyId();
+    DelegationKey dKey = allKeys.get(keyId);
+    if (dKey == null) {
+      LOG
+          .warn("No KEY found for persisted identifier "
+              + identifier.toString());
+      return;
+    }
+    byte[] password = createPassword(identifier.getBytes(), dKey.getKey());
+    if (identifier.getSequenceNumber() > this.delegationTokenSequenceNumber) {
+      this.delegationTokenSequenceNumber = identifier.getSequenceNumber();
+    }
+    if (currentTokens.get(identifier) == null) {
+      currentTokens.put(identifier, new DelegationTokenInformation(expiryTime,
+          password));
+    } else {
+      throw new IOException(
+          "Same delegation token being added twice; invalid entry in fsimage or editlogs");
+    }
+  }
+
+  /**
+   * Add a MasterKey to the list of keys.
+   * 
+   * @param key DelegationKey
+   * @throws IOException
+   */
+  public synchronized void updatePersistedMasterKey(DelegationKey key)
+      throws IOException {
+    addKey(key);
+  }
+  
+  /**
+   * Update the token cache with renewal record in edit logs.
+   * 
+   * @param identifier DelegationTokenIdentifier of the renewed token
+   * @param expiryTime
+   * @throws IOException
+   */
+  public synchronized void updatePersistedTokenRenewal(
+      DelegationTokenIdentifier identifier, long expiryTime) throws IOException {
+    if (running) {
+      // a safety check
+      throw new IOException(
+          "Can't update persisted delegation token renewal to a running SecretManager.");
+    }
+    DelegationTokenInformation info = null;
+    info = currentTokens.get(identifier);
+    if (info != null) {
+      int keyId = identifier.getMasterKeyId();
+      byte[] password = createPassword(identifier.getBytes(), allKeys
+          .get(keyId).getKey());
+      currentTokens.put(identifier, new DelegationTokenInformation(expiryTime,
+          password));
+    }
+  }
+
+  /**
+   *  Update the token cache with the cancel record in edit logs
+   *  
+   *  @param identifier DelegationTokenIdentifier of the canceled token
+   *  @throws IOException
+   */
+  public synchronized void updatePersistedTokenCancellation(
+      DelegationTokenIdentifier identifier) throws IOException {
+    if (running) {
+      // a safety check
+      throw new IOException(
+          "Can't update persisted delegation token renewal to a running SecretManager.");
+    }
+    currentTokens.remove(identifier);
+  }
+
+  /**
+   * Private helper methods to save delegation keys and tokens in fsimage
+   */
+  private synchronized void saveCurrentTokens(DataOutputStream out)
+      throws IOException {
+    out.writeInt(currentTokens.size());
+    Iterator<DelegationTokenIdentifier> iter = currentTokens.keySet()
+        .iterator();
+    while (iter.hasNext()) {
+      DelegationTokenIdentifier id = iter.next();
+      id.write(out);
+      DelegationTokenInformation info = currentTokens.get(id);
+      out.writeLong(info.getRenewDate());
+    }
+  }
+  
+  /*
+   * Save the current state of allKeys
+   */
+  private synchronized void saveAllKeys(DataOutputStream out)
+      throws IOException {
+    out.writeInt(allKeys.size());
+    Iterator<Integer> iter = allKeys.keySet().iterator();
+    while (iter.hasNext()) {
+      Integer key = iter.next();
+      allKeys.get(key).write(out);
+    }
+  }
+  
+  /**
+   * Private helper methods to load Delegation tokens from fsimage
+   */
+  private synchronized void loadCurrentTokens(DataInputStream in)
+      throws IOException {
+    int numberOfTokens = in.readInt();
+    for (int i = 0; i < numberOfTokens; i++) {
+      DelegationTokenIdentifier id = new DelegationTokenIdentifier();
+      id.readFields(in);
+      long expiryTime = in.readLong();
+      addPersistedDelegationToken(id, expiryTime);
+    }
+  }
+
+  /**
+   * Private helper method to load delegation keys from fsimage.
+   * @param in
+   * @throws IOException
+   */
+  private synchronized void loadAllKeys(DataInputStream in) throws IOException {
+    int numberOfKeys = in.readInt();
+    for (int i = 0; i < numberOfKeys; i++) {
+      DelegationKey value = new DelegationKey();
+      value.readFields(in);
+      addKey(value);
+    }
+  }
+
+  /**
+   * Call namesystem to update editlogs for new master key.
+   */
+  @Override //AbstractDelegationTokenManager
+  protected void logUpdateMasterKey(DelegationKey key)
+      throws IOException {
+    namesystem.logUpdateMasterKey(key);
+  }
 }

src/java/org/apache/hadoop/hdfs/server/namenode/FSEditLog.java

@@ -34,6 +34,7 @@ import org.apache.hadoop.hdfs.DeprecatedUTF8;
 import org.apache.hadoop.hdfs.protocol.Block;
 import org.apache.hadoop.hdfs.protocol.DatanodeID;
 import org.apache.hadoop.hdfs.protocol.FSConstants;
+import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
 import org.apache.hadoop.hdfs.server.common.GenerationStamp;
 import org.apache.hadoop.hdfs.server.common.Storage;
@@ -51,6 +52,7 @@ import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableFactories;
 import org.apache.hadoop.io.WritableFactory;
+import org.apache.hadoop.security.token.delegation.DelegationKey;
 
 /**
  * FSEditLog maintains a log of the namespace modifications.
@@ -79,6 +81,10 @@ public class FSEditLog {
   private static final byte OP_RENAME = 15;  // new rename
   private static final byte OP_CONCAT_DELETE = 16; // concat files.
   private static final byte OP_SYMLINK = 17; // a symbolic link
+  private static final byte OP_GET_DELEGATION_TOKEN = 18; //new delegation token
+  private static final byte OP_RENEW_DELEGATION_TOKEN = 19; //renew delegation token
+  private static final byte OP_CANCEL_DELEGATION_TOKEN = 20; //cancel delegation token
+  private static final byte OP_UPDATE_MASTER_KEY = 21; //update master key
 
   /* 
    * The following operations are used to control remote edit log streams,
@@ -427,7 +433,13 @@ public class FSEditLog {
         numOpRenameOld = 0, numOpSetRepl = 0, numOpMkDir = 0,
         numOpSetPerm = 0, numOpSetOwner = 0, numOpSetGenStamp = 0,
         numOpTimes = 0, numOpRename = 0, numOpConcatDelete = 0, 
-        numOpSymlink = 0, numOpOther = 0;
+        numOpSymlink = 0, numOpGetDelegationToken = 0,
+        numOpRenewDelegationToken = 0, numOpCancelDelegationToken = 0, 
+        numOpUpdateMasterKey = 0, numOpOther = 0;
+
+    DelegationTokenIdentifier delegationTokenId = new DelegationTokenIdentifier();
+    DelegationKey delegationKey = new DelegationKey();
+
     try {
       while (true) {
         long timestamp = 0;
@@ -736,6 +748,52 @@ public class FSEditLog {
           fsNamesys.changeLease(s, d, dinfo);
           break;
         }
+        case OP_GET_DELEGATION_TOKEN: {
+          if (logVersion > -24) {
+            throw new IOException("Unexpected opcode " + opcode
+                + " for version " + logVersion);
+          }
+          numOpGetDelegationToken++;
+          delegationTokenId.readFields(in);
+          long expiryTime = readLong(in);
+          fsNamesys.getDelegationTokenSecretManager()
+              .addPersistedDelegationToken(delegationTokenId, expiryTime);
+          break;
+        }
+        case OP_RENEW_DELEGATION_TOKEN: {
+          if (logVersion > -24) {
+            throw new IOException("Unexpected opcode " + opcode
+                + " for version " + logVersion);
+          }
+          numOpRenewDelegationToken++;
+          delegationTokenId.readFields(in);
+          long expiryTime = readLong(in);
+          fsNamesys.getDelegationTokenSecretManager()
+              .updatePersistedTokenRenewal(delegationTokenId, expiryTime);
+          break;
+        }
+        case OP_CANCEL_DELEGATION_TOKEN: {
+          if (logVersion > -24) {
+            throw new IOException("Unexpected opcode " + opcode
+                + " for version " + logVersion);
+          }
+          numOpCancelDelegationToken++;
+          delegationTokenId.readFields(in);
+          fsNamesys.getDelegationTokenSecretManager()
+              .updatePersistedTokenCancellation(delegationTokenId);
+          break;
+        }
+        case OP_UPDATE_MASTER_KEY: {
+          if (logVersion > -24) {
+            throw new IOException("Unexpected opcode " + opcode
+                + " for version " + logVersion);
+          }
+          numOpUpdateMasterKey++;
+          delegationKey.readFields(in);
+          fsNamesys.getDelegationTokenSecretManager().updatePersistedMasterKey(
+              delegationKey);
+          break;
+        }
         default: {
           throw new IOException("Never seen opcode " + opcode);
         }
@@ -756,6 +814,10 @@ public class FSEditLog {
           + " numOpTimes = " + numOpTimes
           + " numOpConcatDelete  = " + numOpConcatDelete
           + " numOpRename = " + numOpRename
+          + " numOpGetDelegationToken = " + numOpGetDelegationToken
+          + " numOpRenewDelegationToken = " + numOpRenewDelegationToken
+          + " numOpCancelDelegationToken = " + numOpCancelDelegationToken
+          + " numOpUpdateMasterKey = " + numOpUpdateMasterKey
           + " numOpOther = " + numOpOther);
     }
     return numEdits;
@@ -1103,6 +1165,30 @@ public class FSEditLog {
             node.getPermissionStatus());
   }
   
+  /**
+   * log delegation token to edit log
+   * @param id DelegationTokenIdentifier
+   * @param expiryTime of the token
+   * @return
+   */
+  void logGetDelegationToken(DelegationTokenIdentifier id,
+      long expiryTime) {
+    logEdit(OP_GET_DELEGATION_TOKEN, id, FSEditLog.toLogLong(expiryTime));
+  }
+  
+  void logRenewDelegationToken(DelegationTokenIdentifier id,
+      long expiryTime) {
+    logEdit(OP_RENEW_DELEGATION_TOKEN, id, FSEditLog.toLogLong(expiryTime));
+  }
+  
+  void logCancelDelegationToken(DelegationTokenIdentifier id) {
+    logEdit(OP_CANCEL_DELEGATION_TOKEN, id);
+  }
+  
+  void logUpdateMasterKey(DelegationKey key) {
+    logEdit(OP_UPDATE_MASTER_KEY, key);
+  }
+  
   static private DeprecatedUTF8 toLogReplication(short replication) {
     return new DeprecatedUTF8(Short.toString(replication));
   }

src/java/org/apache/hadoop/hdfs/server/namenode/FSImage.java

@@ -1146,6 +1146,8 @@ public class FSImage extends Storage {
       // load Files Under Construction
       this.loadFilesUnderConstruction(imgVersion, in, fsNamesys);
       
+      this.loadSecretManagerState(imgVersion, in, fsNamesys);
+      
     } finally {
       in.close();
     }
@@ -1219,6 +1221,7 @@ public class FSImage extends Storage {
       // save the rest of the nodes
       saveImage(strbuf, 0, fsDir.rootDir, out);
       fsNamesys.saveFilesUnderConstruction(out);
+      fsNamesys.saveSecretManagerState(out);
       strbuf = null;
     } finally {
       out.close();
@@ -1430,6 +1433,16 @@ public class FSImage extends Storage {
     }
   }
 
+  private void loadSecretManagerState(int version,  DataInputStream in, 
+      FSNamesystem fs) throws IOException {
+    if (version > -23) {
+      //SecretManagerState is not available.
+      //This must not happen if security is turned on.
+      return; 
+    }
+    fs.loadSecretManagerState(in);
+  }
+  
   // Helper function that reads in an INodeUnderConstruction
   // from the input stream
   //

src/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -36,6 +36,7 @@ import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.SecretManager.InvalidToken;
+import org.apache.hadoop.security.token.delegation.DelegationKey;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenSecretManager;
 import org.apache.hadoop.util.*;
@@ -73,6 +74,8 @@ import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.Text;
 
 import java.io.BufferedWriter;
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
 import java.io.File;
 import java.io.FileWriter;
 import java.io.FileNotFoundException;
@@ -286,11 +289,16 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
     }
   }
 
+  void activateSecretManager() throws IOException {
+    if (dtSecretManager != null) {
+      dtSecretManager.startThreads();
+    }
+  }
+  
   /**
    * Activate FSNamesystem daemons.
    */
   void activate(Configuration conf) throws IOException {
-    dtSecretManager.startThreads();
     setBlockTotal();
     blockManager.activate();
     this.hbthread = new Daemon(new HeartbeatMonitor());
@@ -365,6 +373,7 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
     this.blockManager = new BlockManager(this, conf);
     setConfigurationParameters(conf);
     this.dir = new FSDirectory(fsImage, this, conf);
+    dtSecretManager = createDelegationTokenSecretManager(conf);
   }
 
   /**
@@ -4407,15 +4416,23 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
         conf.getLong(
             DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_RENEW_INTERVAL_KEY,
             DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_RENEW_INTERVAL_DEFAULT),
-        DELEGATION_TOKEN_REMOVER_SCAN_INTERVAL);
+        DELEGATION_TOKEN_REMOVER_SCAN_INTERVAL, this);
   }
 
   public DelegationTokenSecretManager getDelegationTokenSecretManager() {
     return dtSecretManager;
   }
 
+  /**
+   * @param renewer
+   * @return Token<DelegationTokenIdentifier>
+   * @throws IOException
+   */
   public Token<DelegationTokenIdentifier> getDelegationToken(Text renewer)
       throws IOException {
+    if (isInSafeMode()) {
+      throw new SafeModeException("Cannot issue delegation token", safeMode);
+    }
     UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
     String user = ugi.getUserName();
     Text owner = new Text(user);
@@ -4425,18 +4442,116 @@ public class FSNamesystem implements FSConstants, FSNamesystemMBean, FSClusterSt
     }
     DelegationTokenIdentifier dtId = new DelegationTokenIdentifier(owner,
         renewer, realUser);
-    return new Token<DelegationTokenIdentifier>(dtId, dtSecretManager);
+    Token<DelegationTokenIdentifier> token = new Token<DelegationTokenIdentifier>(
+        dtId, dtSecretManager);
+    long expiryTime = dtSecretManager.getTokenExpiryTime(dtId);
+    logGetDelegationToken(dtId, expiryTime);
+    return token;
   }
 
+  /**
+   * 
+   * @param token
+   * @return New expiryTime of the token
+   * @throws InvalidToken
+   * @throws IOException
+   */
   public long renewDelegationToken(Token<DelegationTokenIdentifier> token)
       throws InvalidToken, IOException {
+    if (isInSafeMode()) {
+      throw new SafeModeException("Cannot renew delegation token", safeMode);
+    }
     String renewer = UserGroupInformation.getCurrentUser().getShortUserName();
-    return dtSecretManager.renewToken(token, renewer);
+    long expiryTime = dtSecretManager.renewToken(token, renewer);
+    DelegationTokenIdentifier id = new DelegationTokenIdentifier();
+    ByteArrayInputStream buf = new ByteArrayInputStream(token.getIdentifier());
+    DataInputStream in = new DataInputStream(buf);
+    id.readFields(in);
+    logRenewDelegationToken(id, expiryTime);
+    return expiryTime;
   }
 
+  /**
+   * 
+   * @param token
+   * @throws IOException
+   */
   public void cancelDelegationToken(Token<DelegationTokenIdentifier> token)
       throws IOException {
+    if (isInSafeMode()) {
+      throw new SafeModeException("Cannot cancel delegation token", safeMode);
+    }
     String canceller = UserGroupInformation.getCurrentUser().getShortUserName();
-    dtSecretManager.cancelToken(token, canceller);
+    DelegationTokenIdentifier id = dtSecretManager
+        .cancelToken(token, canceller);
+    logCancelDelegationToken(id);
+  }
+  
+  /**
+   * @param out save state of the secret manager
+   */
+  void saveSecretManagerState(DataOutputStream out) throws IOException {
+    dtSecretManager.saveSecretManagerState(out);
+  }
+
+  /**
+   * @param in load the state of secret manager from input stream
+   */
+  void loadSecretManagerState(DataInputStream in) throws IOException {
+    dtSecretManager.loadSecretManagerState(in);
+  }
+
+  /**
+   * Log the getDelegationToken operation to edit logs
+   * 
+   * @param id identifer of the new delegation token
+   * @param expiryTime when delegation token expires
+   */
+  private void logGetDelegationToken(DelegationTokenIdentifier id,
+      long expiryTime) throws IOException {
+    synchronized (this) {
+      getEditLog().logGetDelegationToken(id, expiryTime);
+    }
+    getEditLog().logSync();
+  }
+
+  /**
+   * Log the renewDelegationToken operation to edit logs
+   * 
+   * @param id identifer of the delegation token being renewed
+   * @param expiryTime when delegation token expires
+   */
+  private void logRenewDelegationToken(DelegationTokenIdentifier id,
+      long expiryTime) throws IOException {
+    synchronized (this) {
+      getEditLog().logRenewDelegationToken(id, expiryTime);
+    }
+    getEditLog().logSync();
+  }
+
+  
+  /**
+   * Log the cancelDelegationToken operation to edit logs
+   * 
+   * @param id identifer of the delegation token being cancelled
+   */
+  private void logCancelDelegationToken(DelegationTokenIdentifier id)
+      throws IOException {
+    synchronized (this) {
+      getEditLog().logCancelDelegationToken(id);
+    }
+    getEditLog().logSync();
+  }
+
+  /**
+   * Log the updateMasterKey operation to edit logs
+   * 
+   * @param key new delegation key.
+   */
+  public void logUpdateMasterKey(DelegationKey key) throws IOException {
+    synchronized (this) {
+      getEditLog().logUpdateMasterKey(key);
+    }
+    getEditLog().logSync();
   }
 }

src/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java

@@ -315,6 +315,10 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
    * Activate name-node servers and threads.
    */
   void activate(Configuration conf) throws IOException {
+    if ((isRole(NamenodeRole.ACTIVE))
+        && (UserGroupInformation.isSecurityEnabled())) {
+      namesystem.activateSecretManager();
+    }
     namesystem.activate(conf);
     startHttpServer(conf);
     server.start();  //start RPC server

src/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/ImageLoaderCurrent.java

@@ -25,10 +25,12 @@ import java.util.Date;
 
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.hdfs.protocol.DatanodeInfo.AdminStates;
+import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.hdfs.server.namenode.FSImage;
 import org.apache.hadoop.hdfs.tools.offlineImageViewer.ImageVisitor.ImageElement;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.security.token.delegation.DelegationKey;
 
 /**
  * ImageLoaderCurrent processes Hadoop FSImage files and walks over
@@ -92,12 +94,29 @@ import org.apache.hadoop.io.WritableUtils;
  *      string
  *      string
  *      enum
+ *    CurrentDelegationKeyId (int)
+ *    NumDelegationKeys (int)
+ *      DelegationKeys (count = NumDelegationKeys)
+ *        DelegationKeyLength (vint)
+ *        DelegationKey (bytes)
+ *    DelegationTokenSequenceNumber (int)
+ *    NumDelegationTokens (int)
+ *    DelegationTokens (count = NumDelegationTokens)
+ *      DelegationTokenIdentifier
+ *        owner (String)
+ *        renewer (String)
+ *        realUser (String)
+ *        issueDate (vlong)
+ *        maxDate (vlong)
+ *        sequenceNumber (vint)
+ *        masterKeyId (vint)
+ *      expiryTime (long)     
  *
  */
 class ImageLoaderCurrent implements ImageLoader {
   protected final DateFormat dateFormat = 
                                       new SimpleDateFormat("yyyy-MM-dd HH:mm");
-  private static int [] versions = {-16, -17, -18, -19, -20, -21, -22, -23};
+  private static int [] versions = {-16, -17, -18, -19, -20, -21, -22, -23, -24};
   private int imageVersion = 0;
 
   /* (non-Javadoc)
@@ -136,6 +155,10 @@ class ImageLoaderCurrent implements ImageLoader {
 
       processINodesUC(in, v, skipBlocks);
 
+      if (imageVersion <= -24) {
+        processDelegationTokens(in, v);
+      }
+      
       v.leaveEnclosingElement(); // FSImage
       v.finish();
     } catch(IOException e) {
@@ -145,6 +168,36 @@ class ImageLoaderCurrent implements ImageLoader {
     }
   }
 
+  /**
+   * Process the Delegation Token related section in fsimage.
+   * 
+   * @param in DataInputStream to process
+   * @param v Visitor to walk over records
+   */
+  private void processDelegationTokens(DataInputStream in, ImageVisitor v)
+      throws IOException {
+    v.visit(ImageElement.CURRENT_DELEGATION_KEY_ID, in.readInt());
+    int numDKeys = in.readInt();
+    v.visitEnclosingElement(ImageElement.DELEGATION_KEYS,
+        ImageElement.NUM_DELEGATION_KEYS, numDKeys);
+    for(int i =0; i < numDKeys; i++) {
+      DelegationKey key = new DelegationKey();
+      key.readFields(in);
+      v.visit(ImageElement.DELEGATION_KEY, key.toString());
+    }
+    v.leaveEnclosingElement();
+    v.visit(ImageElement.DELEGATION_TOKEN_SEQUENCE_NUMBER, in.readInt());
+    int numDTokens = in.readInt();
+    v.visitEnclosingElement(ImageElement.DELEGATION_TOKENS,
+        ImageElement.NUM_DELEGATION_TOKENS, numDTokens);
+    for(int i=0; i<numDTokens; i++){
+      DelegationTokenIdentifier id = new  DelegationTokenIdentifier();
+      id.readFields(in);
+      v.visit(ImageElement.DELEGATION_TOKEN_IDENTIFIER, id.toString());
+    }
+    v.leaveEnclosingElement();
+  }
+
   /**
    * Process the INodes under construction section of the fsimage.
    *

src/java/org/apache/hadoop/hdfs/tools/offlineImageViewer/ImageVisitor.java

@@ -60,7 +60,16 @@ abstract class ImageVisitor {
     CLIENT_MACHINE,
     USER_NAME,
     GROUP_NAME,
-    PERMISSION_STRING
+    PERMISSION_STRING,
+    CURRENT_DELEGATION_KEY_ID,
+    NUM_DELEGATION_KEYS,
+    DELEGATION_KEYS,
+    DELEGATION_KEY,
+    DELEGATION_TOKEN_SEQUENCE_NUMBER,
+    NUM_DELEGATION_TOKENS,
+    DELEGATION_TOKENS,
+    DELEGATION_TOKEN_IDENTIFIER,
+    DELEGATION_TOKEN_EXPIRY_TIME
   }
   
   /**

src/test/hdfs/org/apache/hadoop/hdfs/security/TestClientProtocolWithDelegationToken.java

@@ -42,6 +42,7 @@ import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenSecretManager;
+import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
 import org.apache.hadoop.security.SaslInputStream;
 import org.apache.hadoop.security.SaslRpcClient;
 import org.apache.hadoop.security.SaslRpcServer;
@@ -75,13 +76,14 @@ public class TestClientProtocolWithDelegationToken {
   @Test
   public void testDelegationTokenRpc() throws Exception {
     ClientProtocol mockNN = mock(ClientProtocol.class);
+    FSNamesystem mockNameSys = mock(FSNamesystem.class);
     when(mockNN.getProtocolVersion(anyString(), anyLong())).thenReturn(
         ClientProtocol.versionID);
     DelegationTokenSecretManager sm = new DelegationTokenSecretManager(
         DFSConfigKeys.DFS_NAMENODE_DELEGATION_KEY_UPDATE_INTERVAL_DEFAULT,
         DFSConfigKeys.DFS_NAMENODE_DELEGATION_KEY_UPDATE_INTERVAL_DEFAULT,
         DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_MAX_LIFETIME_DEFAULT,
-        3600000);
+        3600000, mockNameSys);
     sm.startThreads();
     final Server server = RPC.getServer(ClientProtocol.class, mockNN, ADDRESS,
         0, 5, true, conf, sm);

src/test/hdfs/org/apache/hadoop/hdfs/security/TestDelegationToken.java

@@ -22,8 +22,6 @@ package org.apache.hadoop.hdfs.security;
 
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
-import java.io.IOException;
-import java.security.PrivilegedExceptionAction;
 
 import junit.framework.Assert;
 
@@ -35,7 +33,6 @@ import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.security.AccessControlException;
-import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.SecretManager.InvalidToken;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
@@ -57,6 +54,7 @@ public class TestDelegationToken {
     FileSystem.setDefaultUri(config, "hdfs://localhost:" + "0");
     cluster = new MiniDFSCluster(0, config, 1, true, true, true,  null, null, null, null);
     cluster.waitActive();
+    cluster.getNamesystem().getDelegationTokenSecretManager().startThreads();
   }
 
   @After

src/test/hdfs/org/apache/hadoop/hdfs/security/TestDelegationTokenForProxyUser.java

@@ -99,6 +99,7 @@ public class TestDelegationTokenForProxyUser {
     cluster = new MiniDFSCluster(0, config, 1, true, true, true, null, null,
         null, null);
     cluster.waitActive();
+    cluster.getNamesystem().getDelegationTokenSecretManager().startThreads();
   }
 
   @After

src/test/hdfs/org/apache/hadoop/hdfs/server/namenode/TestCheckPointForSecurityTokens.java

@@ -0,0 +1,123 @@
+/**
+ * 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.hdfs.server.namenode;
+
+import junit.framework.Assert;
+import java.io.*;
+import java.net.URI;
+import java.util.Collection;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.FSConstants.SafeModeAction;
+import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.hdfs.server.common.HdfsConstants.StartupOption;
+import org.apache.hadoop.hdfs.tools.DFSAdmin;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+import org.junit.Test;
+
+/**
+ * This class tests the creation and validation of a checkpoint.
+ */
+public class TestCheckPointForSecurityTokens {
+  static final long seed = 0xDEADBEEFL;
+  static final int blockSize = 4096;
+  static final int fileSize = 8192;
+  static final int numDatanodes = 3;
+  short replication = 3;
+
+  NameNode startNameNode( Configuration conf,
+                          String imageDirs,
+                          String editsDirs,
+                          StartupOption start) throws IOException {
+    conf.set(DFSConfigKeys.FS_DEFAULT_NAME_KEY, "hdfs://localhost:0");
+    conf.set(DFSConfigKeys.DFS_NAMENODE_HTTP_ADDRESS_KEY, "0.0.0.0:0");  
+    conf.set(DFSConfigKeys.DFS_NAMENODE_NAME_DIR_KEY, imageDirs);
+    conf.set(DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_KEY, editsDirs);
+    String[] args = new String[]{start.getName()};
+    NameNode nn = NameNode.createNameNode(args, conf);
+    Assert.assertTrue(nn.isInSafeMode());
+    return nn;
+  }
+
+  /**
+   * Tests save namepsace.
+   */
+  @Test
+  public void testSaveNamespace() throws IOException {
+    MiniDFSCluster cluster = null;
+    DistributedFileSystem fs = null;
+    try {
+      Configuration conf = new HdfsConfiguration();
+      cluster = new MiniDFSCluster(conf, numDatanodes, true, null);
+      cluster.waitActive();
+      fs = (DistributedFileSystem)(cluster.getFileSystem());
+      FSNamesystem namesystem = cluster.getNamesystem();
+      namesystem.getDelegationTokenSecretManager().startThreads();
+      String renewer = UserGroupInformation.getLoginUser().getUserName();
+      Token<DelegationTokenIdentifier> token = namesystem
+          .getDelegationToken(new Text(renewer)); 
+      
+      // Saving image without safe mode should fail
+      DFSAdmin admin = new DFSAdmin(conf);
+      String[] args = new String[]{"-saveNamespace"};
+
+      // verify that the edits file is NOT empty
+      Collection<URI> editsDirs = cluster.getNameEditsDirs();
+      for(URI uri : editsDirs) {
+        File ed = new File(uri.getPath());
+        Assert.assertTrue(new File(ed, "current/edits").length() > Integer.SIZE/Byte.SIZE);
+      }
+
+      // Saving image in safe mode should succeed
+      fs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+      try {
+        admin.run(args);
+      } catch(Exception e) {
+        throw new IOException(e.getMessage());
+      }
+      // verify that the edits file is empty
+      for(URI uri : editsDirs) {
+        File ed = new File(uri.getPath());
+        Assert.assertTrue(new File(ed, "current/edits").length() == Integer.SIZE/Byte.SIZE);
+      }
+
+      // restart cluster and verify file exists
+      cluster.shutdown();
+      cluster = null;
+
+      cluster = new MiniDFSCluster(conf, numDatanodes, false, null);
+      cluster.waitActive();
+      //Should be able to renew & cancel the delegation token after cluster restart
+      try {
+        cluster.getNamesystem().renewDelegationToken(token);
+        cluster.getNamesystem().cancelDelegationToken(token);
+      } catch (IOException e) {
+        Assert.fail("Could not renew or cancel the token");
+      }
+    } finally {
+      if(fs != null) fs.close();
+      if(cluster!= null) cluster.shutdown();
+    }
+  }
+}

src/test/hdfs/org/apache/hadoop/hdfs/server/namenode/TestSecurityTokenEditLog.java

@@ -0,0 +1,156 @@
+/**
+ * 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.hdfs.server.namenode;
+
+import junit.framework.TestCase;
+import java.io.*;
+import java.net.URI;
+import java.util.Iterator;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.hdfs.server.namenode.EditLogFileInputStream;
+import org.apache.hadoop.hdfs.server.common.Storage.StorageDirectory;
+import org.apache.hadoop.hdfs.server.namenode.FSImage.NameNodeDirType;
+import org.apache.hadoop.hdfs.server.namenode.FSImage.NameNodeFile;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
+
+/**
+ * This class tests the creation and validation of a checkpoint.
+ */
+public class TestSecurityTokenEditLog extends TestCase {
+  static final int NUM_DATA_NODES = 1;
+
+  // This test creates NUM_THREADS threads and each thread does
+  // 2 * NUM_TRANSACTIONS Transactions concurrently.
+  static final int NUM_TRANSACTIONS = 100;
+  static final int NUM_THREADS = 100;
+  static final int opsPerTrans = 3;
+
+  //
+  // an object that does a bunch of transactions
+  //
+  static class Transactions implements Runnable {
+    FSNamesystem namesystem;
+    int numTransactions;
+    short replication = 3;
+    long blockSize = 64;
+
+    Transactions(FSNamesystem ns, int num) {
+      namesystem = ns;
+      numTransactions = num;
+    }
+
+    // add a bunch of transactions.
+    public void run() {
+      FSEditLog editLog = namesystem.getEditLog();
+
+      for (int i = 0; i < numTransactions; i++) {
+        try {
+          String renewer = UserGroupInformation.getLoginUser().getUserName();
+          Token<DelegationTokenIdentifier> token = namesystem
+              .getDelegationToken(new Text(renewer));
+          namesystem.renewDelegationToken(token);
+          namesystem.cancelDelegationToken(token);
+          editLog.logSync();
+        } catch (IOException e) {
+          System.out.println("Transaction " + i + " encountered exception " +
+                             e);
+        }
+      }
+    }
+  }
+
+  /**
+   * Tests transaction logging in dfs.
+   */
+  public void testEditLog() throws IOException {
+
+    // start a cluster 
+    Configuration conf = new HdfsConfiguration();
+    MiniDFSCluster cluster = null;
+    FileSystem fileSys = null;
+
+    try {
+      cluster = new MiniDFSCluster(conf, NUM_DATA_NODES, true, null);
+      cluster.waitActive();
+      fileSys = cluster.getFileSystem();
+      final FSNamesystem namesystem = cluster.getNamesystem();
+      namesystem.getDelegationTokenSecretManager().startThreads();
+  
+      for (Iterator<URI> it = cluster.getNameDirs().iterator(); it.hasNext(); ) {
+        File dir = new File(it.next().getPath());
+        System.out.println(dir);
+      }
+      
+      FSImage fsimage = namesystem.getFSImage();
+      FSEditLog editLog = fsimage.getEditLog();
+  
+      // set small size of flush buffer
+      editLog.setBufferCapacity(2048);
+      editLog.close();
+      editLog.open();
+    
+      // Create threads and make them run transactions concurrently.
+      Thread threadId[] = new Thread[NUM_THREADS];
+      for (int i = 0; i < NUM_THREADS; i++) {
+        Transactions trans = new Transactions(namesystem, NUM_TRANSACTIONS);
+        threadId[i] = new Thread(trans, "TransactionThread-" + i);
+        threadId[i].start();
+      }
+  
+      // wait for all transactions to get over
+      for (int i = 0; i < NUM_THREADS; i++) {
+        try {
+          threadId[i].join();
+        } catch (InterruptedException e) {
+          i--;      // retry 
+        }
+      } 
+      
+      editLog.close();
+      editLog.open();
+  
+      // Verify that we can read in all the transactions that we have written.
+      // If there were any corruptions, it is likely that the reading in
+      // of these transactions will throw an exception.
+      //
+      namesystem.getDelegationTokenSecretManager().stopThreads();
+      for (Iterator<StorageDirectory> it = 
+              fsimage.dirIterator(NameNodeDirType.EDITS); it.hasNext();) {
+        File editFile = FSImage.getImageFile(it.next(), NameNodeFile.EDITS);
+        System.out.println("Verifying file: " + editFile);
+        int numEdits = namesystem.getEditLog().loadFSEdits(
+                                  new EditLogFileInputStream(editFile));
+        assertTrue("Verification for " + editFile + " failed. " +
+                   "Expected " + (NUM_THREADS * opsPerTrans * NUM_TRANSACTIONS + 2) + " transactions. "+
+                   "Found " + numEdits + " transactions.",
+                   numEdits == NUM_THREADS * opsPerTrans * NUM_TRANSACTIONS +2);
+  
+      }
+    } finally {
+      if(fileSys != null) fileSys.close();
+      if(cluster != null) cluster.shutdown();
+    }
+  }
+}