From 7d7f7a9e9b45614f3a861fbc14821c7c238c5106 Mon Sep 17 00:00:00 2001 From: Xing Lin Date: Sun, 9 Oct 2022 08:33:48 -0700 Subject: [PATCH] HDFS-16024. RBF: Rename data to the Trash should be based on src location (#4962) (cherry picked from commit e18d8062126951b01d04d14374cc04053167735b) Reviewed-by: Dinesh Chitlangia Signed-off-by: Akira Ajisaka --- .../resolver/MountTableResolver.java | 79 ++++- .../federation/resolver/RemoteLocation.java | 14 + .../federation/router/TestRouterTrash.java | 297 ++++++++++++++++++ 3 files changed, 382 insertions(+), 8 deletions(-) create mode 100644 hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterTrash.java diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java index 797006ab1de..0b6e599194f 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/MountTableResolver.java @@ -45,11 +45,15 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.regex.Pattern; +import java.util.ArrayList; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.hdfs.server.federation.resolver.order.DestinationOrder; import org.apache.hadoop.hdfs.server.federation.router.Router; +import org.apache.hadoop.hdfs.server.federation.router.RouterRpcServer; import org.apache.hadoop.hdfs.server.federation.store.MountTableStore; import org.apache.hadoop.hdfs.server.federation.store.StateStoreCache; import org.apache.hadoop.hdfs.server.federation.store.StateStoreService; @@ -104,6 +108,8 @@ public class MountTableResolver private final Lock readLock = readWriteLock.readLock(); private final Lock writeLock = readWriteLock.writeLock(); + /** Trash Current matching pattern. */ + private static final String TRASH_PATTERN = "/(Current|[0-9]+)"; @VisibleForTesting public MountTableResolver(Configuration conf) { @@ -337,6 +343,52 @@ public class MountTableResolver this.init = true; } + /** + * Check if PATH is the trail associated with the Trash. + * + * @param path A path. + */ + @VisibleForTesting + public static boolean isTrashPath(String path) throws IOException { + Pattern pattern = Pattern.compile( + "^" + getTrashRoot() + TRASH_PATTERN + "/"); + return pattern.matcher(path).find(); + } + + @VisibleForTesting + public static String getTrashRoot() throws IOException { + // Gets the Trash directory for the current user. + return FileSystem.USER_HOME_PREFIX + "/" + + RouterRpcServer.getRemoteUser().getUserName() + "/" + + FileSystem.TRASH_PREFIX; + } + + /** + * Subtract a TrashCurrent to get a new path. + * + * @param path A path. + */ + @VisibleForTesting + public static String subtractTrashCurrentPath(String path) + throws IOException { + return path.replaceAll("^" + + getTrashRoot() + TRASH_PATTERN, ""); + } + + /** + * If path is a path related to the trash can, + * subtract TrashCurrent to return a new path. + * + * @param path A path. + */ + private static String processTrashPath(String path) throws IOException { + if (isTrashPath(path)) { + return subtractTrashCurrentPath(path); + } else { + return path; + } + } + /** * Replaces the current in-memory cached of the mount table with a new * version fetched from the data store. @@ -381,18 +433,26 @@ public class MountTableResolver public PathLocation getDestinationForPath(final String path) throws IOException { verifyMountTable(); + PathLocation res; readLock.lock(); try { if (this.locationCache == null) { - return lookupLocation(path); + res = lookupLocation(processTrashPath(path)); + } else { + Callable meh = (Callable) () -> + lookupLocation(processTrashPath(path)); + res = this.locationCache.get(processTrashPath(path), meh); } - Callable meh = new Callable() { - @Override - public PathLocation call() throws Exception { - return lookupLocation(path); + if (isTrashPath(path)) { + List remoteLocations = new ArrayList<>(); + for (RemoteLocation remoteLocation : res.getDestinations()) { + remoteLocations.add(new RemoteLocation(remoteLocation, path)); } - }; - return this.locationCache.get(path, meh); + return new PathLocation(path, remoteLocations, + res.getDestinationOrder()); + } else { + return res; + } } catch (ExecutionException e) { Throwable cause = e.getCause(); final IOException ioe; @@ -450,8 +510,11 @@ public class MountTableResolver @Override public List getMountPoints(final String str) throws IOException { verifyMountTable(); - final String path = RouterAdmin.normalizeFileSystemPath(str); + String path = RouterAdmin.normalizeFileSystemPath(str); + if (isTrashPath(path)) { + path = subtractTrashCurrentPath(path); + } Set children = new TreeSet<>(); readLock.lock(); try { diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/RemoteLocation.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/RemoteLocation.java index 77d050062e7..4cb6516e4b2 100644 --- a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/RemoteLocation.java +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/RemoteLocation.java @@ -61,6 +61,20 @@ public class RemoteLocation extends RemoteLocationContext { this.srcPath = sPath; } + /** + * Use the Ns and Nn of a remote location + * and another path to create a new remote location pointing. + * + * @param remoteLocation A remoteLocation. + * @param path Path in the destination namespace. + */ + public RemoteLocation(RemoteLocation remoteLocation, String path) { + this.nameserviceId = remoteLocation.nameserviceId; + this.namenodeId = remoteLocation.namenodeId; + this.dstPath = path; + this.srcPath = path; + } + @Override public String getNameserviceId() { String ret = this.nameserviceId; diff --git a/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterTrash.java b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterTrash.java new file mode 100644 index 00000000000..acd7b87a14b --- /dev/null +++ b/hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterTrash.java @@ -0,0 +1,297 @@ +/** + * 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.federation.router; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.fs.Trash; +import org.apache.hadoop.fs.permission.FsPermission; +import org.apache.hadoop.hdfs.DFSClient; +import org.apache.hadoop.hdfs.DFSTestUtil; +import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster; +import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder; +import org.apache.hadoop.hdfs.server.federation.StateStoreDFSCluster; +import org.apache.hadoop.hdfs.server.federation.resolver.MountTableManager; +import org.apache.hadoop.hdfs.server.federation.resolver.MountTableResolver; +import org.apache.hadoop.hdfs.server.federation.store.protocol.*; +import org.apache.hadoop.hdfs.server.federation.store.records.MountTable; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.hadoop.util.Time; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.net.URISyntaxException; +import java.util.Collections; + +import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_TRASH_INTERVAL_KEY; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +/** + * This is a test through the Router move data to the Trash. + */ +public class TestRouterTrash { + + public static final Logger LOG = + LoggerFactory.getLogger(TestRouterTrash.class); + + private static StateStoreDFSCluster cluster; + private static MiniRouterDFSCluster.RouterContext routerContext; + private static MountTableResolver mountTable; + private static FileSystem routerFs; + private static FileSystem nnFs; + private static final String TEST_USER = "test-trash"; + private static MiniRouterDFSCluster.NamenodeContext nnContext; + private static String ns0; + private static String ns1; + private static final String MOUNT_POINT = "/home/data"; + private static final String FILE = MOUNT_POINT + "/file1"; + private static final String TRASH_ROOT = "/user/" + TEST_USER + "/.Trash"; + private static final String CURRENT = "/Current"; + + @BeforeClass + public static void globalSetUp() throws Exception { + // Build and start a federated cluster + cluster = new StateStoreDFSCluster(false, 2); + Configuration conf = new RouterConfigBuilder() + .stateStore() + .admin() + .rpc() + .http() + .build(); + conf.set(FS_TRASH_INTERVAL_KEY, "100"); + cluster.addRouterOverrides(conf); + cluster.startCluster(); + cluster.startRouters(); + cluster.waitClusterUp(); + + ns0 = cluster.getNameservices().get(0); + ns1 = cluster.getNameservices().get(1); + + routerContext = cluster.getRandomRouter(); + routerFs = routerContext.getFileSystem(); + nnContext = cluster.getNamenode(ns0, null); + nnFs = nnContext.getFileSystem(); + Router router = routerContext.getRouter(); + mountTable = (MountTableResolver) router.getSubclusterResolver(); + } + + @AfterClass + public static void tearDown() { + if (cluster != null) { + cluster.stopRouter(routerContext); + cluster.shutdown(); + cluster = null; + } + } + + @After + public void clearMountTable() throws IOException { + RouterClient client = routerContext.getAdminClient(); + MountTableManager mountTableManager = client.getMountTableManager(); + GetMountTableEntriesRequest req1 = + GetMountTableEntriesRequest.newInstance("/"); + GetMountTableEntriesResponse response = + mountTableManager.getMountTableEntries(req1); + for (MountTable entry : response.getEntries()) { + RemoveMountTableEntryRequest req2 = + RemoveMountTableEntryRequest.newInstance(entry.getSourcePath()); + mountTableManager.removeMountTableEntry(req2); + } + } + + @After + public void clearFile() throws IOException { + FileStatus[] fileStatuses = nnFs.listStatus(new Path("/")); + for (FileStatus file : fileStatuses) { + nnFs.delete(file.getPath(), true); + } + } + + private boolean addMountTable(final MountTable entry) throws IOException { + RouterClient client = routerContext.getAdminClient(); + MountTableManager mountTableManager = client.getMountTableManager(); + AddMountTableEntryRequest addRequest = + AddMountTableEntryRequest.newInstance(entry); + AddMountTableEntryResponse addResponse = + mountTableManager.addMountTableEntry(addRequest); + // Reload the Router cache + mountTable.loadCache(true); + return addResponse.getStatus(); + } + + @Test + public void testMoveToTrashNoMountPoint() throws IOException, + URISyntaxException, InterruptedException { + MountTable addEntry = MountTable.newInstance(MOUNT_POINT, + Collections.singletonMap(ns0, MOUNT_POINT)); + assertTrue(addMountTable(addEntry)); + // current user client + DFSClient client = nnContext.getClient(); + client.setOwner("/", TEST_USER, TEST_USER); + UserGroupInformation ugi = UserGroupInformation. + createRemoteUser(TEST_USER); + // test user client + client = nnContext.getClient(ugi); + client.mkdirs(MOUNT_POINT, new FsPermission("777"), true); + assertTrue(client.exists(MOUNT_POINT)); + // create test file + client.create(FILE, true); + Path filePath = new Path(FILE); + + FileStatus[] fileStatuses = routerFs.listStatus(filePath); + assertEquals(1, fileStatuses.length); + assertEquals(TEST_USER, fileStatuses[0].getOwner()); + // move to Trash + Configuration routerConf = routerContext.getConf(); + FileSystem fs = + DFSTestUtil.getFileSystemAs(ugi, routerConf); + Trash trash = new Trash(fs, routerConf); + assertTrue(trash.moveToTrash(filePath)); + fileStatuses = nnFs.listStatus( + new Path(TRASH_ROOT + CURRENT + MOUNT_POINT)); + assertEquals(1, fileStatuses.length); + assertTrue(nnFs.exists(new Path(TRASH_ROOT + CURRENT + FILE))); + assertTrue(nnFs.exists(new Path("/user/" + + TEST_USER + "/.Trash/Current" + FILE))); + // When the target path in Trash already exists. + client.create(FILE, true); + filePath = new Path(FILE); + fileStatuses = routerFs.listStatus(filePath); + assertEquals(1, fileStatuses.length); + assertTrue(trash.moveToTrash(filePath)); + fileStatuses = routerFs.listStatus( + new Path(TRASH_ROOT + CURRENT + MOUNT_POINT)); + assertEquals(2, fileStatuses.length); + } + + @Test + public void testDeleteToTrashExistMountPoint() throws IOException, + URISyntaxException, InterruptedException { + MountTable addEntry = MountTable.newInstance(MOUNT_POINT, + Collections.singletonMap(ns0, MOUNT_POINT)); + assertTrue(addMountTable(addEntry)); + // add Trash mount point + addEntry = MountTable.newInstance(TRASH_ROOT, + Collections.singletonMap(ns1, TRASH_ROOT)); + assertTrue(addMountTable(addEntry)); + // current user client + DFSClient client = nnContext.getClient(); + client.setOwner("/", TEST_USER, TEST_USER); + UserGroupInformation ugi = UserGroupInformation. + createRemoteUser(TEST_USER); + // test user client + client = nnContext.getClient(ugi); + client.mkdirs(MOUNT_POINT, new FsPermission("777"), true); + assertTrue(client.exists(MOUNT_POINT)); + // create test file + client.create(FILE, true); + Path filePath = new Path(FILE); + + FileStatus[] fileStatuses = routerFs.listStatus(filePath); + assertEquals(1, fileStatuses.length); + assertEquals(TEST_USER, fileStatuses[0].getOwner()); + + // move to Trash + Configuration routerConf = routerContext.getConf(); + FileSystem fs = + DFSTestUtil.getFileSystemAs(ugi, routerConf); + Trash trash = new Trash(fs, routerConf); + assertTrue(trash.moveToTrash(filePath)); + fileStatuses = nnFs.listStatus( + new Path(TRASH_ROOT + CURRENT + MOUNT_POINT)); + assertEquals(1, fileStatuses.length); + assertTrue(nnFs.exists(new Path(TRASH_ROOT + CURRENT + FILE))); + // When the target path in Trash already exists. + client.create(FILE, true); + filePath = new Path(FILE); + + fileStatuses = nnFs.listStatus(filePath); + assertEquals(1, fileStatuses.length); + assertTrue(trash.moveToTrash(filePath)); + fileStatuses = nnFs.listStatus( + new Path(TRASH_ROOT + CURRENT + MOUNT_POINT)); + assertEquals(2, fileStatuses.length); + } + + @Test + public void testIsTrashPath() throws IOException { + UserGroupInformation ugi = UserGroupInformation.getCurrentUser(); + assertNotNull(ugi); + assertTrue(MountTableResolver.isTrashPath( + "/user/" + ugi.getUserName() + "/.Trash/Current" + MOUNT_POINT)); + assertTrue(MountTableResolver.isTrashPath( + "/user/" + ugi.getUserName() + + "/.Trash/" + Time.now() + MOUNT_POINT)); + assertFalse(MountTableResolver.isTrashPath(MOUNT_POINT)); + + // Contains TrashCurrent but does not begin with TrashCurrent. + assertFalse(MountTableResolver.isTrashPath("/home/user/" + + ugi.getUserName() + "/.Trash/Current" + MOUNT_POINT)); + assertFalse(MountTableResolver.isTrashPath("/home/user/" + + ugi.getUserName() + "/.Trash/" + Time.now() + MOUNT_POINT)); + + // Special cases. + assertFalse(MountTableResolver.isTrashPath("")); + assertFalse(MountTableResolver.isTrashPath( + "/home/user/empty.Trash/Current")); + assertFalse(MountTableResolver.isTrashPath( + "/home/user/.Trash")); + assertFalse(MountTableResolver.isTrashPath( + "/.Trash/Current")); + } + + @Test + public void testSubtractTrashCurrentPath() throws IOException { + UserGroupInformation ugi = UserGroupInformation.getCurrentUser(); + assertNotNull(ugi); + assertEquals(MOUNT_POINT, MountTableResolver.subtractTrashCurrentPath( + "/user/" + ugi.getUserName() + "/.Trash/Current" + MOUNT_POINT)); + assertEquals(MOUNT_POINT, MountTableResolver.subtractTrashCurrentPath( + "/user/" + ugi.getUserName() + + "/.Trash/" + Time.now() + MOUNT_POINT)); + + // Contains TrashCurrent but does not begin with TrashCurrent. + assertEquals("/home/user/" + ugi.getUserName() + + "/.Trash/Current" + MOUNT_POINT, MountTableResolver. + subtractTrashCurrentPath("/home/user/" + + ugi.getUserName() + "/.Trash/Current" + MOUNT_POINT)); + long time = Time.now(); + assertEquals("/home/user/" + ugi.getUserName() + + "/.Trash/" + time + MOUNT_POINT, MountTableResolver. + subtractTrashCurrentPath("/home/user/" + ugi.getUserName() + + "/.Trash/" + time + MOUNT_POINT)); + // Special cases. + assertEquals("", MountTableResolver.subtractTrashCurrentPath("")); + assertEquals("/home/user/empty.Trash/Current", MountTableResolver. + subtractTrashCurrentPath("/home/user/empty.Trash/Current")); + assertEquals("/home/user/.Trash", MountTableResolver. + subtractTrashCurrentPath("/home/user/.Trash")); + assertEquals("/.Trash/Current", MountTableResolver. + subtractTrashCurrentPath("/.Trash/Current")); + } +} \ No newline at end of file