HDFS-6304. Consolidate the logic of path resolution in FSDirectory. Contributed by Haohui Mai.

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

Author: Haohui Mai

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

@@ -332,6 +332,9 @@ Release 2.5.0 - UNRELEASED
     HDFS-6269. NameNode Audit Log should differentiate between webHDFS open and
     HDFS open. (Eric Payne via jeagles)
 
+    HDFS-6304. Consolidate the logic of path resolution in FSDirectory.
+    (wheat9)
+
   OPTIMIZATIONS
 
     HDFS-6214. Webhdfs has poor throughput for files >2GB (daryn)

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

@@ -494,7 +494,7 @@ private int loadDirectory(DataInput in, Counter counter) throws IOException {
      // Rename .snapshot paths if we're doing an upgrade
      parentPath = renameReservedPathsOnUpgrade(parentPath, getLayoutVersion());
      final INodeDirectory parent = INodeDirectory.valueOf(
-         namesystem.dir.rootDir.getNode(parentPath, true), parentPath);
+         namesystem.dir.getNode(parentPath, true), parentPath);
      return loadChildren(parent, in, counter);
    }
 

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

@@ -5507,7 +5507,7 @@ private void checkPermission(FSPermissionChecker pc,
       dir.waitForReady();
       readLock();
       try {
-        pc.checkPermission(path, dir.rootDir, doCheckOwner, ancestorAccess,
+        pc.checkPermission(path, dir, doCheckOwner, ancestorAccess,
             parentAccess, access, subAccess, resolveLink);
       } finally {
         readUnlock();

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

@@ -144,7 +144,7 @@ public void checkSuperuserPrivilege()
    * Guarded by {@link FSNamesystem#readLock()}
    * Caller of this method must hold that lock.
    */
-  void checkPermission(String path, INodeDirectory root, boolean doCheckOwner,
+  void checkPermission(String path, FSDirectory dir, boolean doCheckOwner,
       FsAction ancestorAccess, FsAction parentAccess, FsAction access,
       FsAction subAccess, boolean resolveLink)
       throws AccessControlException, UnresolvedLinkException {
@@ -159,7 +159,7 @@ void checkPermission(String path, INodeDirectory root, boolean doCheckOwner,
     }
     // check if (parentAccess != null) && file exists, then check sb
     // If resolveLink, the check is performed on the link target.
-    final INodesInPath inodesInPath = root.getINodesInPath(path, resolveLink);
+    final INodesInPath inodesInPath = dir.getINodesInPath(path, resolveLink);
     final int snapshotId = inodesInPath.getPathSnapshotId();
     final INode[] inodes = inodesInPath.getINodes();
     int ancestorIndex = inodes.length - 2;

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

@@ -403,53 +403,6 @@ private ReadOnlyList<INode> getCurrentChildrenList() {
         : ReadOnlyList.Util.asReadOnlyList(children);
   }
 
-  /** @return the {@link INodesInPath} containing only the last inode. */
-  INodesInPath getLastINodeInPath(String path, boolean resolveLink
-      ) throws UnresolvedLinkException {
-    return INodesInPath.resolve(this, getPathComponents(path), 1, resolveLink);
-  }
-
-  /** @return the {@link INodesInPath} containing all inodes in the path. */
-  INodesInPath getINodesInPath(String path, boolean resolveLink
-      ) throws UnresolvedLinkException {
-    final byte[][] components = getPathComponents(path);
-    return INodesInPath.resolve(this, components, components.length, resolveLink);
-  }
-
-  /** @return the last inode in the path. */
-  INode getNode(String path, boolean resolveLink) 
-    throws UnresolvedLinkException {
-    return getLastINodeInPath(path, resolveLink).getINode(0);
-  }
-
-  /**
-   * @return the INode of the last component in src, or null if the last
-   * component does not exist.
-   * @throws UnresolvedLinkException if symlink can't be resolved
-   * @throws SnapshotAccessControlException if path is in RO snapshot
-   */
-  INode getINode4Write(String src, boolean resolveLink)
-      throws UnresolvedLinkException, SnapshotAccessControlException {
-    return getINodesInPath4Write(src, resolveLink).getLastINode();
-  }
-
-  /**
-   * @return the INodesInPath of the components in src
-   * @throws UnresolvedLinkException if symlink can't be resolved
-   * @throws SnapshotAccessControlException if path is in RO snapshot
-   */
-  INodesInPath getINodesInPath4Write(String src, boolean resolveLink)
-      throws UnresolvedLinkException, SnapshotAccessControlException {
-    final byte[][] components = INode.getPathComponents(src);
-    INodesInPath inodesInPath = INodesInPath.resolve(this, components,
-        components.length, resolveLink);
-    if (inodesInPath.isSnapshot()) {
-      throw new SnapshotAccessControlException(
-          "Modification on a read-only snapshot is disallowed");
-    }
-    return inodesInPath;
-  }
-
   /**
    * Given a child's name, return the index of the next child
    *

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

@@ -1306,7 +1306,7 @@ private static void assertAclFeature(boolean expectAclFeature)
    */
   private static void assertAclFeature(Path pathToCheck,
       boolean expectAclFeature) throws IOException {
-    INode inode = cluster.getNamesystem().getFSDirectory().getRoot()
+    INode inode = cluster.getNamesystem().getFSDirectory()
       .getNode(pathToCheck.toUri().getPath(), false);
     assertNotNull(inode);
     AclFeature aclFeature = inode.getAclFeature();

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/FSAclBaseTest.java

@@ -26,6 +26,7 @@
 import java.io.IOException;
 import java.util.Arrays;
 
+import org.apache.hadoop.conf.Configuration;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -38,7 +39,10 @@
 import org.apache.hadoop.hdfs.server.namenode.snapshot.Snapshot;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
+import static org.mockito.Mockito.*;
 /**
  * Unit tests covering FSPermissionChecker.  All tests in this suite have been
  * cross-validated against Linux setfacl/getfacl to check for consistency of the
@@ -56,14 +60,24 @@ public class TestFSPermissionChecker {
   private static final UserGroupInformation CLARK =
     UserGroupInformation.createUserForTesting("clark", new String[] { "execs" });
 
+  private FSDirectory dir;
   private INodeDirectory inodeRoot;
 
   @Before
   public void setUp() {
-    PermissionStatus permStatus = PermissionStatus.createImmutable(SUPERUSER,
-      SUPERGROUP, FsPermission.createImmutable((short)0755));
-    inodeRoot = new INodeDirectory(INodeId.ROOT_INODE_ID,
-      INodeDirectory.ROOT_NAME, permStatus, 0L);
+    Configuration conf = new Configuration();
+    FSNamesystem fsn = mock(FSNamesystem.class);
+    doAnswer(new Answer() {
+      @Override
+      public Object answer(InvocationOnMock invocation) throws Throwable {
+        Object[] args = invocation.getArguments();
+        FsPermission perm = (FsPermission) args[0];
+        return new PermissionStatus(SUPERUSER, SUPERGROUP, perm);
+      }
+    }).when(fsn).createFsOwnerPermissions(any(FsPermission.class));
+    FSImage image = mock(FSImage.class);
+    dir = new FSDirectory(image, fsn, conf);
+    inodeRoot = dir.getRoot();
   }
 
   @Test
@@ -379,14 +393,14 @@ private void addAcl(INodeWithAdditionalFields inode, AclEntry... acl)
   private void assertPermissionGranted(UserGroupInformation user, String path,
       FsAction access) throws IOException {
     new FSPermissionChecker(SUPERUSER, SUPERGROUP, user).checkPermission(path,
-      inodeRoot, false, null, null, access, null, true);
+      dir, false, null, null, access, null, true);
   }
 
   private void assertPermissionDenied(UserGroupInformation user, String path,
       FsAction access) throws IOException {
     try {
       new FSPermissionChecker(SUPERUSER, SUPERGROUP, user).checkPermission(path,
-        inodeRoot, false, null, null, access, null, true);
+        dir, false, null, null, access, null, true);
       fail("expected AccessControlException for user + " + user + ", path = " +
         path + ", access = " + access);
     } catch (AccessControlException e) {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestFSPermissionChecker.java

@@ -699,7 +699,7 @@ public void testFsckError() throws Exception {
       DFSTestUtil.waitReplication(fs, filePath, (short)1);
 
       // intentionally corrupt NN data structure
-      INodeFile node = (INodeFile)cluster.getNamesystem().dir.rootDir.getNode(
+      INodeFile node = (INodeFile)cluster.getNamesystem().dir.getNode(
           fileName, true);
       final BlockInfo[] blocks = node.getBlocks(); 
       assertEquals(blocks.length, 1);