HADOOP-18144: getTrashRoot in ViewFileSystem should return a path in ViewFS. (#4123)

To get the new behavior, define fs.viewfs.trash.force-inside-mount-point to be true.

If the trash root for path p is in the same mount point as path p,
and one of:
* The mount point isn't at the top of the target fs.
* The resolved path of path is root (eg it is the fallback FS).
* The trash root isn't in user's target fs home directory.
get the corresponding viewFS path for the trash root and return it.
Otherwise, use <mnt>/.Trash/<user>.

Signed-off-by: Owen O'Malley <oomalley@linkedin.com>
(cherry picked from commit 8b8158f02df18424b2406fd66b34c1bdf3d7ab55)

Co-authored-by: Xing Lin <xinglin@linkedin.com>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/TrashPolicyDefault.java

@@ -191,8 +191,8 @@ public class TrashPolicyDefault extends TrashPolicy {
         cause = e;
       }
     }
-    throw (IOException)
-      new IOException("Failed to move to trash: " + path).initCause(cause);
+    throw new IOException("Failed to move " + path + " to trash " + trashPath,
+        cause);
   }
 
   @SuppressWarnings("deprecation")

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/Constants.java

@@ -125,4 +125,11 @@ public interface Constants {
       "fs.viewfs.ignore.port.in.mount.table.name";
 
   boolean CONFIG_VIEWFS_IGNORE_PORT_IN_MOUNT_TABLE_NAME_DEFAULT = false;
+
+  /**
+   * Force ViewFileSystem to return a trashRoot that is inside a mount point.
+   */
+  String CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT =
+      "fs.viewfs.trash.force-inside-mount-point";
+  boolean CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT_DEFAULT = false;
 }

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java

@@ -25,6 +25,8 @@ import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_IGNORE_PORT_IN
 import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS;
 import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_MOUNT_LINKS_AS_SYMLINKS_DEFAULT;
 import static org.apache.hadoop.fs.viewfs.Constants.PERMISSION_555;
+import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT;
+import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT_DEFAULT;
 
 import java.util.function.Function;
 import java.io.FileNotFoundException;
@@ -1138,16 +1140,79 @@ public class ViewFileSystem extends FileSystem {
    * Get the trash root directory for current user when the path
    * specified is deleted.
    *
+   * If FORCE_INSIDE_MOUNT_POINT flag is not set, return the default trash root
+   * from targetFS.
+   *
+   * When FORCE_INSIDE_MOUNT_POINT is set to true,
+   * <ol>
+   *   <li>
+   *     If the trash root for path p is in the same mount point as path p,
+   *       and one of:
+   *       <ol>
+   *         <li>The mount point isn't at the top of the target fs.</li>
+   *         <li>The resolved path of path is root (in fallback FS).</li>
+   *         <li>The trash isn't in user's target fs home directory
+   *            get the corresponding viewFS path for the trash root and return
+   *            it.
+   *         </li>
+   *       </ol>
+   *   </li>
+   *   <li>
+   *     else, return the trash root under the root of the mount point
+   *     (/{mntpoint}/.Trash/{user}).
+   *   </li>
+   * </ol>
+   *
+   * These conditions handle several different important cases:
+   * <ul>
+   *   <li>File systems may need to have more local trash roots, such as
+   *         encryption zones or snapshot roots.</li>
+   *   <li>The fallback mount should use the user's home directory.</li>
+   *   <li>Cloud storage systems should not use trash in an implicity defined
+   *        home directory, per a container, unless it is the fallback fs.</li>
+   * </ul>
+   *
    * @param path the trash root of the path to be determined.
    * @return the trash root path.
    */
   @Override
   public Path getTrashRoot(Path path) {
+
     try {
       InodeTree.ResolveResult<FileSystem> res =
           fsState.resolve(getUriPath(path), true);
-      return res.targetFileSystem.getTrashRoot(res.remainingPath);
-    } catch (Exception e) {
+      Path targetFSTrashRoot =
+          res.targetFileSystem.getTrashRoot(res.remainingPath);
+
+      // Allow clients to use old behavior of delegating to target fs.
+      if (!config.getBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT,
+          CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT_DEFAULT)) {
+        return targetFSTrashRoot;
+      }
+
+      // The trash root path from the target fs
+      String targetFSTrashRootPath = targetFSTrashRoot.toUri().getPath();
+      // The mount point path in the target fs
+      String mountTargetPath = res.targetFileSystem.getUri().getPath();
+      if (!mountTargetPath.endsWith("/")) {
+        mountTargetPath = mountTargetPath + "/";
+      }
+
+      Path targetFsUserHome = res.targetFileSystem.getHomeDirectory();
+      if (targetFSTrashRootPath.startsWith(mountTargetPath) &&
+          !(mountTargetPath.equals(ROOT_PATH.toString()) &&
+              !res.resolvedPath.equals(ROOT_PATH.toString()) &&
+              (targetFsUserHome != null && targetFSTrashRootPath.startsWith(
+                  targetFsUserHome.toUri().getPath())))) {
+        String relativeTrashRoot =
+            targetFSTrashRootPath.substring(mountTargetPath.length());
+        return makeQualified(new Path(res.resolvedPath, relativeTrashRoot));
+      } else {
+        // Return the trash root for the mount point.
+        return makeQualified(new Path(res.resolvedPath,
+            TRASH_PREFIX + "/" + ugi.getShortUserName()));
+      }
+    } catch (IOException | IllegalArgumentException e) {
       throw new NotInMountpointException(path, "getTrashRoot");
     }
   }
@@ -1155,16 +1220,78 @@ public class ViewFileSystem extends FileSystem {
   /**
    * Get all the trash roots for current user or all users.
    *
+   * When FORCE_INSIDE_MOUNT_POINT is set to true, we also return trash roots
+   * under the root of each mount point, with their viewFS paths.
+   *
    * @param allUsers return trash roots for all users if true.
    * @return all Trash root directories.
    */
   @Override
   public Collection<FileStatus> getTrashRoots(boolean allUsers) {
-    List<FileStatus> trashRoots = new ArrayList<>();
+    // A map from targetFSPath -> FileStatus.
+    // FileStatus can be from targetFS or viewFS.
+    HashMap<Path, FileStatus> trashRoots = new HashMap<>();
     for (FileSystem fs : getChildFileSystems()) {
-      trashRoots.addAll(fs.getTrashRoots(allUsers));
+      for (FileStatus trash : fs.getTrashRoots(allUsers)) {
+        trashRoots.put(trash.getPath(), trash);
+      }
     }
-    return trashRoots;
+
+    // Return trashRoots if FORCE_INSIDE_MOUNT_POINT is disabled.
+    if (!config.getBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT,
+        CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT_DEFAULT)) {
+      return trashRoots.values();
+    }
+
+    // Get trash roots in TRASH_PREFIX dir inside mount points and fallback FS.
+    List<InodeTree.MountPoint<FileSystem>> mountPoints =
+        fsState.getMountPoints();
+    // If we have a fallback FS, add a mount point for it as <"", fallback FS>.
+    // The source path of a mount point shall not end with '/', thus for
+    // fallback fs, we set its mount point src as "".
+    if (fsState.getRootFallbackLink() != null) {
+      mountPoints.add(new InodeTree.MountPoint<>("",
+          fsState.getRootFallbackLink()));
+    }
+
+    try {
+      for (InodeTree.MountPoint<FileSystem> mountPoint : mountPoints) {
+
+        Path trashRoot =
+            makeQualified(new Path(mountPoint.src + "/" + TRASH_PREFIX));
+
+        // Continue if trashRoot does not exist for this mount point
+        if (!exists(trashRoot)) {
+          continue;
+        }
+
+        FileSystem targetFS = mountPoint.target.getTargetFileSystem();
+        if (!allUsers) {
+          Path userTrashRoot = new Path(trashRoot, ugi.getShortUserName());
+          if (exists(userTrashRoot)) {
+            Path targetFSUserTrashRoot = targetFS.makeQualified(
+                new Path(targetFS.getUri().getPath(),
+                    TRASH_PREFIX + "/" + ugi.getShortUserName()));
+            trashRoots.put(targetFSUserTrashRoot, getFileStatus(userTrashRoot));
+          }
+        } else {
+          FileStatus[] mountPointTrashRoots = listStatus(trashRoot);
+          for (FileStatus trash : mountPointTrashRoots) {
+            // Remove the mountPoint and the leading '/' to get the
+            // relative targetFsTrash path
+            String targetFsTrash = trash.getPath().toUri().getPath()
+                .substring(mountPoint.src.length() + 1);
+            Path targetFsTrashPath = targetFS.makeQualified(
+                new Path(targetFS.getUri().getPath(), targetFsTrash));
+            trashRoots.put(targetFsTrashPath, trash);
+          }
+        }
+      }
+    } catch (IOException e) {
+      LOG.warn("Exception in get all trash roots for mount points", e);
+    }
+
+    return trashRoots.values();
   }
 
   @Override

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemLocalFileSystem.java

@@ -33,6 +33,8 @@ import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import static org.apache.hadoop.fs.FileSystem.TRASH_PREFIX;
+import org.apache.hadoop.security.UserGroupInformation;
 
 import org.junit.After;
 import org.junit.Before;
@@ -61,6 +63,13 @@ public class TestViewFileSystemLocalFileSystem extends ViewFileSystemBaseTest {
     
   }
 
+  @Override
+  Path getTrashRootInFallBackFS() throws IOException {
+    return new Path(
+        "/" + TRASH_PREFIX + "/" + UserGroupInformation.getCurrentUser()
+            .getShortUserName());
+  }
+
   @Test
   public void testNflyWriteSimple() throws IOException {
     LOG.info("Starting testNflyWriteSimple");

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemWithAuthorityLocalFileSystem.java

@@ -25,6 +25,9 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileSystemTestHelper;
 import org.apache.hadoop.fs.FsConstants;
 import org.apache.hadoop.fs.Path;
+import static org.apache.hadoop.fs.FileSystem.TRASH_PREFIX;
+import org.apache.hadoop.security.UserGroupInformation;
+import java.io.IOException;
 
 import org.junit.After;
 import org.junit.Assert;
@@ -63,6 +66,13 @@ public class TestViewFileSystemWithAuthorityLocalFileSystem extends ViewFileSyst
     super.tearDown();
   }
  
+  @Override
+  Path getTrashRootInFallBackFS() throws IOException {
+    return new Path(
+        "/" + TRASH_PREFIX + "/" + UserGroupInformation.getCurrentUser()
+            .getShortUserName());
+  }
+
   @Override
   @Test
   public void testBasicPaths() {

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java

@@ -68,6 +68,8 @@ import org.slf4j.LoggerFactory;
 import static org.apache.hadoop.fs.FileSystemTestHelper.*;
 import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_ENABLE_INNER_CACHE;
 import static org.apache.hadoop.fs.viewfs.Constants.PERMISSION_555;
+import static org.apache.hadoop.fs.viewfs.Constants.CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT;
+import static org.apache.hadoop.fs.FileSystem.TRASH_PREFIX;
 
 import org.junit.After;
 import org.junit.Assert;
@@ -1101,6 +1103,176 @@ abstract public class ViewFileSystemBaseTest {
     Assert.assertTrue("", fsView.getTrashRoots(true).size() > 0);
   }
 
+  // Default implementation of getTrashRoot for a fallback FS mounted at root:
+  // e.g., fallbackFS.uri.getPath = '/'
+  Path getTrashRootInFallBackFS() throws IOException {
+    return new Path(fsTarget.getHomeDirectory().toUri().getPath(),
+        TRASH_PREFIX);
+  }
+
+  /**
+   * Test TRASH_FORCE_INSIDE_MOUNT_POINT feature for getTrashRoot.
+   */
+  @Test
+  public void testTrashRootForceInsideMountPoint() throws IOException {
+    UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+    Configuration conf2 = new Configuration(conf);
+    conf2.setBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT, true);
+    ConfigUtil.addLinkFallback(conf2, targetTestRoot.toUri());
+    FileSystem fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+
+    // Case 1: path p in the /data mount point.
+    // Return a trash root within the /data mount point.
+    Path dataTestPath = new Path("/data/dir/file");
+    Path dataTrashRoot = fsView2.makeQualified(
+        new Path("/data/" + TRASH_PREFIX + "/" + ugi.getShortUserName()));
+    Assert.assertEquals(dataTrashRoot, fsView2.getTrashRoot(dataTestPath));
+
+    // Case 2: path p not found in mount table.
+    // Return a trash root in fallback FS.
+    Path nonExistentPath = new Path("/nonExistentDir/nonExistentFile");
+    Path expectedTrash =
+        fsView2.makeQualified(getTrashRootInFallBackFS());
+    Assert.assertEquals(expectedTrash, fsView2.getTrashRoot(nonExistentPath));
+
+    // Case 3: turn off the CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT flag.
+    // Return a trash root in user home dir.
+    conf2.setBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT, false);
+    fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+    Path targetFSUserHomeTrashRoot = fsTarget.makeQualified(
+        new Path(fsTarget.getHomeDirectory(), TRASH_PREFIX));
+    Assert.assertEquals(targetFSUserHomeTrashRoot,
+        fsView2.getTrashRoot(dataTestPath));
+
+    // Case 4: viewFS without fallback. Expect exception for a nonExistent path
+    conf2 = new Configuration(conf);
+    fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+    try {
+      fsView2.getTrashRoot(nonExistentPath);
+    } catch (NotInMountpointException ignored) {
+    }
+  }
+
+  /**
+   * A mocked FileSystem which returns a deep trash dir.
+   */
+  static class DeepTrashRootMockFS extends MockFileSystem {
+    public static final Path TRASH =
+        new Path("/vol/very/deep/deep/trash/dir/.Trash");
+
+    @Override
+    public Path getTrashRoot(Path path) {
+      return TRASH;
+    }
+  }
+
+  /**
+   * Test getTrashRoot that is very deep inside a mount point.
+   */
+  @Test
+  public void testTrashRootDeepTrashDir() throws IOException {
+
+    Configuration conf2 = ViewFileSystemTestSetup.createConfig();
+    conf2.setBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT, true);
+    conf2.setClass("fs.mocktrashfs.impl", DeepTrashRootMockFS.class,
+        FileSystem.class);
+    ConfigUtil.addLink(conf2, "/mnt/datavol1",
+        URI.create("mocktrashfs://localhost/vol"));
+    Path testPath = new Path("/mnt/datavol1/projs/proj");
+    FileSystem fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+    Path expectedTrash = fsView2.makeQualified(
+        new Path("/mnt/datavol1/very/deep/deep/trash/dir/.Trash"));
+    Assert.assertEquals(expectedTrash, fsView2.getTrashRoot(testPath));
+  }
+
+  /**
+   * Test getTrashRoots() for all users.
+   */
+  @Test
+  public void testTrashRootsAllUsers() throws IOException {
+    Configuration conf2 = new Configuration(conf);
+    conf2.setBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT, true);
+    FileSystem fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+
+    // Case 1: verify correct trash roots from fsView and fsView2
+    int beforeTrashRootNum = fsView.getTrashRoots(true).size();
+    int beforeTrashRootNum2 = fsView2.getTrashRoots(true).size();
+    Assert.assertEquals(beforeTrashRootNum, beforeTrashRootNum2);
+
+    fsView.mkdirs(new Path("/data/" + TRASH_PREFIX + "/user1"));
+    fsView.mkdirs(new Path("/data/" + TRASH_PREFIX + "/user2"));
+    fsView.mkdirs(new Path("/user/" + TRASH_PREFIX + "/user3"));
+    fsView.mkdirs(new Path("/user/" + TRASH_PREFIX + "/user4"));
+    fsView.mkdirs(new Path("/user2/" + TRASH_PREFIX + "/user5"));
+    int afterTrashRootsNum = fsView.getTrashRoots(true).size();
+    int afterTrashRootsNum2 = fsView2.getTrashRoots(true).size();
+    Assert.assertEquals(beforeTrashRootNum, afterTrashRootsNum);
+    Assert.assertEquals(beforeTrashRootNum2 + 5, afterTrashRootsNum2);
+
+    // Case 2: per-user mount point
+    fsTarget.mkdirs(new Path(targetTestRoot, "Users/userA/.Trash/userA"));
+    Configuration conf3 = new Configuration(conf2);
+    ConfigUtil.addLink(conf3, "/Users/userA",
+        new Path(targetTestRoot, "Users/userA").toUri());
+    FileSystem fsView3 = FileSystem.get(FsConstants.VIEWFS_URI, conf3);
+    int trashRootsNum3 = fsView3.getTrashRoots(true).size();
+    Assert.assertEquals(afterTrashRootsNum2 + 1, trashRootsNum3);
+
+    // Case 3: single /Users mount point for all users
+    fsTarget.mkdirs(new Path(targetTestRoot, "Users/.Trash/user1"));
+    fsTarget.mkdirs(new Path(targetTestRoot, "Users/.Trash/user2"));
+    Configuration conf4 = new Configuration(conf2);
+    ConfigUtil.addLink(conf4, "/Users",
+        new Path(targetTestRoot, "Users").toUri());
+    FileSystem fsView4 = FileSystem.get(FsConstants.VIEWFS_URI, conf4);
+    int trashRootsNum4 = fsView4.getTrashRoots(true).size();
+    Assert.assertEquals(afterTrashRootsNum2 + 2, trashRootsNum4);
+
+    // Case 4: test trash roots in fallback FS
+    fsTarget.mkdirs(new Path(targetTestRoot, ".Trash/user10"));
+    fsTarget.mkdirs(new Path(targetTestRoot, ".Trash/user11"));
+    fsTarget.mkdirs(new Path(targetTestRoot, ".Trash/user12"));
+    Configuration conf5 = new Configuration(conf2);
+    ConfigUtil.addLinkFallback(conf5, targetTestRoot.toUri());
+    FileSystem fsView5 = FileSystem.get(FsConstants.VIEWFS_URI, conf5);
+    int trashRootsNum5 = fsView5.getTrashRoots(true).size();
+    Assert.assertEquals(afterTrashRootsNum2 + 3, trashRootsNum5);
+  }
+
+  /**
+   * Test getTrashRoots() for current user.
+   */
+  @Test
+  public void testTrashRootsCurrentUser() throws IOException {
+    String currentUser =
+        UserGroupInformation.getCurrentUser().getShortUserName();
+    Configuration conf2 = new Configuration(conf);
+    conf2.setBoolean(CONFIG_VIEWFS_TRASH_FORCE_INSIDE_MOUNT_POINT, true);
+    FileSystem fsView2 = FileSystem.get(FsConstants.VIEWFS_URI, conf2);
+
+    int beforeTrashRootNum = fsView.getTrashRoots(false).size();
+    int beforeTrashRootNum2 = fsView2.getTrashRoots(false).size();
+    Assert.assertEquals(beforeTrashRootNum, beforeTrashRootNum2);
+
+    fsView.mkdirs(new Path("/data/" + TRASH_PREFIX + "/" + currentUser));
+    fsView.mkdirs(new Path("/data/" + TRASH_PREFIX + "/user2"));
+    fsView.mkdirs(new Path("/user/" + TRASH_PREFIX + "/" + currentUser));
+    fsView.mkdirs(new Path("/user/" + TRASH_PREFIX + "/user4"));
+    fsView.mkdirs(new Path("/user2/" + TRASH_PREFIX + "/user5"));
+    int afterTrashRootsNum = fsView.getTrashRoots(false).size();
+    int afterTrashRootsNum2 = fsView2.getTrashRoots(false).size();
+    Assert.assertEquals(beforeTrashRootNum, afterTrashRootsNum);
+    Assert.assertEquals(beforeTrashRootNum2 + 2, afterTrashRootsNum2);
+
+    // Test trash roots in fallback FS
+    Configuration conf3 = new Configuration(conf2);
+    fsTarget.mkdirs(new Path(targetTestRoot, TRASH_PREFIX + "/" + currentUser));
+    ConfigUtil.addLinkFallback(conf3, targetTestRoot.toUri());
+    FileSystem fsView3 = FileSystem.get(FsConstants.VIEWFS_URI, conf3);
+    int trashRootsNum3 = fsView3.getTrashRoots(false).size();
+    Assert.assertEquals(afterTrashRootsNum2 + 1, trashRootsNum3);
+  }
+
   @Test(expected = NotInMountpointException.class)
   public void testViewFileSystemUtil() throws Exception {
     Configuration newConf = new Configuration(conf);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemHdfs.java

@@ -55,6 +55,7 @@ import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.test.GenericTestUtils;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY;
+import static org.apache.hadoop.fs.FileSystem.TRASH_PREFIX;
 
 import org.junit.After;
 import org.junit.AfterClass;
@@ -181,6 +182,13 @@ public class TestViewFileSystemHdfs extends ViewFileSystemBaseTest {
     return 2;
   }
 
+  @Override
+  Path getTrashRootInFallBackFS() throws IOException {
+    return new Path(
+        "/" + TRASH_PREFIX + "/" + UserGroupInformation.getCurrentUser()
+            .getShortUserName());
+  }
+
   @Test
   public void testTrashRootsAfterEncryptionZoneDeletion() throws Exception {
     final Path zone = new Path("/EZ");