svn merge -c 1505160 merging to branch-2 to fix HDFS-5010.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1505161 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -3105,6 +3105,9 @@ Release 0.23.10 - UNRELEASED
 
   IMPROVEMENTS
 
+    HDFS-5010. Reduce the frequency of getCurrentUser() calls from namenode
+    (kihwal)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/BlockManager.java

@@ -860,8 +860,10 @@ public class BlockManager {
   public void setBlockToken(final LocatedBlock b,
       final BlockTokenSecretManager.AccessMode mode) throws IOException {
     if (isBlockTokenEnabled()) {
-      b.setBlockToken(blockTokenSecretManager.generateToken(b.getBlock(), 
-          EnumSet.of(mode)));
+      // Use cached UGI if serving RPC calls.
+      b.setBlockToken(blockTokenSecretManager.generateToken(
+          NameNode.getRemoteUser().getShortUserName(),
+          b.getBlock(), EnumSet.of(mode)));
     }    
   }
 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -168,6 +168,7 @@ import org.apache.hadoop.hdfs.server.common.Util;
 import org.apache.hadoop.hdfs.server.namenode.INode.BlocksMapUpdateInfo;
 import org.apache.hadoop.hdfs.server.namenode.JournalSet.JournalAndStream;
 import org.apache.hadoop.hdfs.server.namenode.LeaseManager.Lease;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory;
 import org.apache.hadoop.hdfs.server.namenode.startupprogress.Phase;
 import org.apache.hadoop.hdfs.server.namenode.startupprogress.StartupProgress;
@@ -2926,7 +2927,11 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
     
   private FSPermissionChecker getPermissionChecker()
       throws AccessControlException {
-    return new FSPermissionChecker(fsOwnerShortUserName, supergroup);
+    try {
+      return new FSPermissionChecker(fsOwnerShortUserName, supergroup, getRemoteUser());
+    } catch (IOException ioe) {
+      throw new AccessControlException(ioe);
+    }
   }
   /**
    * Remove a file/directory from the namespace.
@@ -3136,9 +3141,7 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
       return !INodeFile.valueOf(dir.getINode(src), src).isUnderConstruction();
     } catch (AccessControlException e) {
       if (isAuditEnabled() && isExternalInvocation()) {
-        logAuditEvent(false, UserGroupInformation.getCurrentUser(),
-                      getRemoteIp(),
-                      "isFileClosed", src, null, null);
+        logAuditEvent(false, "isFileClosed", src);
       }
       throw e;
     } finally {
@@ -5808,11 +5811,7 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
   // optimize ugi lookup for RPC operations to avoid a trip through
   // UGI.getCurrentUser which is synch'ed
   private static UserGroupInformation getRemoteUser() throws IOException {
-    UserGroupInformation ugi = null;
-    if (Server.isRpcInvocation()) {
-      ugi = Server.getRemoteUser();
-    }
-    return (ugi != null) ? ugi : UserGroupInformation.getCurrentUser();
+    return NameNode.getRemoteUser();
   }
   
   /**
@@ -6317,8 +6316,7 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
     readLock();
     try {
       checkOperation(OperationCategory.READ);
-      FSPermissionChecker checker = new FSPermissionChecker(
-          fsOwner.getShortUserName(), supergroup);
+      FSPermissionChecker checker = getPermissionChecker();
       final String user = checker.isSuperUser()? null : checker.getUser();
       status = snapshotManager.getSnapshottableDirListing(user);
     } finally {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java

@@ -56,14 +56,10 @@ class FSPermissionChecker {
   /** A set with group namess. Not synchronized since it is unmodifiable */
   private final Set<String> groups;
   private final boolean isSuper;
-  
-  FSPermissionChecker(String fsOwner, String supergroup
-      ) throws AccessControlException{
-    try {
-      ugi = UserGroupInformation.getCurrentUser();
-    } catch (IOException e) {
-      throw new AccessControlException(e); 
-    } 
+
+  FSPermissionChecker(String fsOwner, String supergroup,
+      UserGroupInformation callerUgi) {
+    ugi = callerUgi;
     HashSet<String> s = new HashSet<String>(Arrays.asList(ugi.getGroupNames()));
     groups = Collections.unmodifiableSet(s);
     user = ugi.getShortUserName();

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNode.java

@@ -433,6 +433,15 @@ public class NameNode {
     return nodeRegistration;
   }
 
+  /* optimize ugi lookup for RPC operations to avoid a trip through
+   * UGI.getCurrentUser which is synch'ed
+   */
+  public static UserGroupInformation getRemoteUser() throws IOException {
+    UserGroupInformation ugi = Server.getRemoteUser();
+    return (ugi != null) ? ugi : UserGroupInformation.getCurrentUser();
+  }
+
+
   /**
    * Login as the configured user for the NameNode.
    */

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java

@@ -90,6 +90,7 @@ import org.apache.hadoop.hdfs.security.token.block.ExportedBlockKeys;
 import org.apache.hadoop.hdfs.security.token.delegation.DelegationTokenIdentifier;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants.NamenodeRole;
 import org.apache.hadoop.hdfs.server.common.IncorrectVersionException;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory;
 import org.apache.hadoop.hdfs.server.namenode.metrics.NameNodeMetrics;
 import org.apache.hadoop.hdfs.server.namenode.web.resources.NamenodeWebHdfsMethods;
@@ -340,6 +341,11 @@ class NameNodeRpcServer implements NamenodeProtocols {
     return clientRpcAddress;
   }
 
+  private static UserGroupInformation getRemoteUser() throws IOException {
+    return NameNode.getRemoteUser();
+  }
+
+
   /////////////////////////////////////////////////////
   // NamenodeProtocol
   /////////////////////////////////////////////////////
@@ -448,7 +454,7 @@ class NameNodeRpcServer implements NamenodeProtocols {
           + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");
     }
     HdfsFileStatus fileStatus = namesystem.startFile(src, new PermissionStatus(
-        UserGroupInformation.getCurrentUser().getShortUserName(), null, masked),
+        getRemoteUser().getShortUserName(), null, masked),
         clientName, clientMachine, flag.get(), createParent, replication,
         blockSize);
     metrics.incrFilesCreated();
@@ -681,7 +687,7 @@ class NameNodeRpcServer implements NamenodeProtocols {
                             + MAX_PATH_LENGTH + " characters, " + MAX_PATH_DEPTH + " levels.");
     }
     return namesystem.mkdirs(src,
-        new PermissionStatus(UserGroupInformation.getCurrentUser().getShortUserName(),
+        new PermissionStatus(getRemoteUser().getShortUserName(),
             null, masked), createParent);
   }
 
@@ -873,7 +879,7 @@ class NameNodeRpcServer implements NamenodeProtocols {
     if ("".equals(target)) {
       throw new IOException("Invalid symlink target");
     }
-    final UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+    final UserGroupInformation ugi = getRemoteUser();
     namesystem.createSymlink(target, link,
       new PermissionStatus(ugi.getShortUserName(), null, dirPerms), createParent);
   }
@@ -1009,7 +1015,7 @@ class NameNodeRpcServer implements NamenodeProtocols {
   @Override // RefreshAuthorizationPolicyProtocol
   public void refreshUserToGroupsMappings() throws IOException {
     LOG.info("Refreshing all user-to-groups mappings. Requested by user: " + 
-             UserGroupInformation.getCurrentUser().getShortUserName());
+             getRemoteUser().getShortUserName());
     Groups.getUserToGroupsMappingService().refresh();
   }