HDFS-11145. Implement getTrashRoot() for ViewFileSystem. (Manoj Govindassamy via lei)

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

@@ -23,6 +23,7 @@ import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.EnumSet;
@@ -799,6 +800,39 @@ public class ViewFileSystem extends FileSystem {
     return allPolicies;
   }
 
+  /**
+   * Get the trash root directory for current user when the path
+   * specified is deleted.
+   *
+   * @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) {
+      throw new NotInMountpointException(path, "getTrashRoot");
+    }
+  }
+
+  /**
+   * Get all the trash roots for current user or all users.
+   *
+   * @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<>();
+    for (FileSystem fs : getChildFileSystems()) {
+      trashRoots.addAll(fs.getTrashRoots(allUsers));
+    }
+    return trashRoots;
+  }
+
   /**
    * An instance of this class represents an internal dir of the viewFs
    * that is internal dir of the mount table.

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

@@ -36,6 +36,7 @@ import org.apache.hadoop.fs.FsConstants;
 import org.apache.hadoop.fs.LocatedFileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.fs.Trash;
 import org.apache.hadoop.fs.UnsupportedFileSystemException;
 import org.apache.hadoop.fs.permission.AclEntry;
 import org.apache.hadoop.fs.permission.AclStatus;
@@ -952,4 +953,71 @@ abstract public class ViewFileSystemBaseTest {
     }
   }
 
+  @Test
+  public void testTrashRoot() throws IOException {
+
+    Path mountDataRootPath = new Path("/data");
+    Path fsTargetFilePath = new Path("debug.log");
+    Path mountDataFilePath = new Path(mountDataRootPath, fsTargetFilePath);
+    Path mountDataNonExistingFilePath = new Path(mountDataRootPath, "no.log");
+    fileSystemTestHelper.createFile(fsTarget, fsTargetFilePath);
+
+    // Get Trash roots for paths via ViewFileSystem handle
+    Path mountDataRootTrashPath = fsView.getTrashRoot(mountDataRootPath);
+    Path mountDataFileTrashPath = fsView.getTrashRoot(mountDataFilePath);
+
+    // Get Trash roots for the same set of paths via the mounted filesystem
+    Path fsTargetRootTrashRoot = fsTarget.getTrashRoot(mountDataRootPath);
+    Path fsTargetFileTrashPath = fsTarget.getTrashRoot(mountDataFilePath);
+
+    // Verify if Trash roots from ViewFileSystem matches that of the ones
+    // from the target mounted FileSystem.
+    assertEquals(mountDataRootTrashPath.toUri().getPath(),
+        fsTargetRootTrashRoot.toUri().getPath());
+    assertEquals(mountDataFileTrashPath.toUri().getPath(),
+        fsTargetFileTrashPath.toUri().getPath());
+    assertEquals(mountDataRootTrashPath.toUri().getPath(),
+        mountDataFileTrashPath.toUri().getPath());
+
+
+    // Verify trash root for an non-existing file but on a valid mountpoint.
+    Path trashRoot = fsView.getTrashRoot(mountDataNonExistingFilePath);
+    assertEquals(mountDataRootTrashPath.toUri().getPath(),
+        trashRoot.toUri().getPath());
+
+    // Verify trash root for invalid mounts.
+    Path invalidMountRootPath = new Path("/invalid_mount");
+    Path invalidMountFilePath = new Path(invalidMountRootPath, "debug.log");
+    try {
+      fsView.getTrashRoot(invalidMountRootPath);
+      fail("ViewFileSystem getTashRoot should fail for non-mountpoint paths.");
+    } catch (NotInMountpointException e) {
+      //expected exception
+    }
+    try {
+      fsView.getTrashRoot(invalidMountFilePath);
+      fail("ViewFileSystem getTashRoot should fail for non-mountpoint paths.");
+    } catch (NotInMountpointException e) {
+      //expected exception
+    }
+    try {
+      fsView.getTrashRoot(null);
+      fail("ViewFileSystem getTashRoot should fail for empty paths.");
+    } catch (NotInMountpointException e) {
+      //expected exception
+    }
+
+    // Move the file to trash
+    FileStatus fileStatus = fsTarget.getFileStatus(fsTargetFilePath);
+    Configuration newConf = new Configuration(conf);
+    newConf.setLong("fs.trash.interval", 1000);
+    Trash lTrash = new Trash(fsTarget, newConf);
+    boolean trashed = lTrash.moveToTrash(fsTargetFilePath);
+    Assert.assertTrue("File " + fileStatus + " move to " +
+        "trash failed.", trashed);
+
+    // Verify ViewFileSystem trash roots shows the ones from
+    // target mounted FileSystem.
+    Assert.assertTrue("", fsView.getTrashRoots(true).size() > 0);
+  }
 }

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

@@ -18,25 +18,37 @@
 package org.apache.hadoop.fs.viewfs;
 
 
+import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY;
+
+import java.io.File;
 import java.io.IOException;
 import java.net.URISyntaxException;
+import java.util.EnumSet;
 
 import javax.security.auth.login.LoginException;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.crypto.key.JavaKeyStoreProvider;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileSystemTestHelper;
 import org.apache.hadoop.fs.FsConstants;
+import org.apache.hadoop.fs.FsShell;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.DFSTestUtil;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
 import org.apache.hadoop.hdfs.MiniDFSNNTopology;
+import org.apache.hadoop.hdfs.client.CreateEncryptionZoneFlag;
+import org.apache.hadoop.hdfs.client.HdfsAdmin;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.junit.After;
 import org.junit.AfterClass;
+import org.junit.Assert;
 import org.junit.Before;
 import org.junit.BeforeClass;
+import org.junit.Test;
 
 
 public class TestViewFileSystemHdfs extends ViewFileSystemBaseTest {
@@ -58,6 +70,19 @@ public class TestViewFileSystemHdfs extends ViewFileSystemBaseTest {
   @BeforeClass
   public static void clusterSetupAtBegining() throws IOException,
       LoginException, URISyntaxException {
+
+    // Encryption Zone settings
+    FileSystemTestHelper fsHelper = new FileSystemTestHelper();
+    String testRoot = fsHelper.getTestRootDir();
+    CONF.set(CommonConfigurationKeysPublic.HADOOP_SECURITY_KEY_PROVIDER_PATH,
+        JavaKeyStoreProvider.SCHEME_NAME + "://file" +
+            new Path(new File(testRoot).getAbsoluteFile().toString(), "test" +
+                ".jks").toUri());
+    CONF.setBoolean(DFSConfigKeys
+        .DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_KEY, true);
+    CONF.setInt(DFSConfigKeys.DFS_NAMENODE_LIST_ENCRYPTION_ZONES_NUM_RESPONSES,
+        2);
+
     SupportsBlocks = true;
     CONF.setBoolean(
         DFSConfigKeys.DFS_NAMENODE_DELEGATION_TOKEN_ALWAYS_USE_KEY, true);
@@ -137,4 +162,36 @@ public class TestViewFileSystemHdfs extends ViewFileSystemBaseTest {
   int getExpectedDelegationTokenCountWithCredentials() {
     return 2;
   }
+
+  @Test
+  public void testTrashRootsAfterEncryptionZoneDeletion() throws Exception {
+    final Path zone = new Path("/EZ");
+    fsTarget.mkdirs(zone);
+    final Path zone1 = new Path("/EZ/zone1");
+    fsTarget.mkdirs(zone1);
+
+    DFSTestUtil.createKey("test_key", cluster, CONF);
+    HdfsAdmin hdfsAdmin = new HdfsAdmin(cluster.getURI(0), CONF);
+    final EnumSet<CreateEncryptionZoneFlag> provisionTrash =
+        EnumSet.of(CreateEncryptionZoneFlag.PROVISION_TRASH);
+    hdfsAdmin.createEncryptionZone(zone1, "test_key", provisionTrash);
+
+    final Path encFile = new Path(zone1, "encFile");
+    DFSTestUtil.createFile(fsTarget, encFile, 10240, (short) 1, 0xFEED);
+
+    Configuration clientConf = new Configuration(CONF);
+    clientConf.setLong(FS_TRASH_INTERVAL_KEY, 1);
+    clientConf.set("fs.default.name", fsTarget.getUri().toString());
+    FsShell shell = new FsShell(clientConf);
+
+    //Verify file deletion within EZ
+    DFSTestUtil.verifyDelete(shell, fsTarget, encFile, true);
+    Assert.assertTrue("ViewFileSystem trash roots should include EZ file trash",
+        (fsView.getTrashRoots(true).size() == 1));
+
+    //Verify deletion of EZ
+    DFSTestUtil.verifyDelete(shell, fsTarget, zone, true);
+    Assert.assertTrue("ViewFileSystem trash roots should include EZ zone trash",
+        (fsView.getTrashRoots(true).size() == 2));
+  }
 }