Address comments and add unit tests

Author: zhuxiangyi

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/tools/federation/RouterAdmin.java

@@ -249,6 +249,10 @@ private boolean validateMin(String[] argv) {
       if (argv.length < 2) {
         return false;
       }
+    } else if ("-initViewFsToMountTable".equals(cmd)) {
+      if (argv.length < 2) {
+        return false;
+      }
     } else if ("-getDestination".equals(cmd)) {
       if (argv.length < 2) {
         return false;
@@ -392,8 +396,9 @@ public int run(String[] argv) throws Exception {
       } else if ("-refresh".equals(cmd)) {
         refresh(address);
       } else if ("-initViewFsToMountTable".equals(cmd)) {
-        if (initViewFsToMountTable(argv, i)) {
-          System.out.println("Successfully init ViewFs mapping to router " + argv[i]);
+        if (initViewFsToMountTable(argv[i])) {
+          System.out.println("Successfully init ViewFs mapping to router " +
+              argv[i]);
         } else {
           exitCode = -1;
         }
@@ -1052,53 +1057,63 @@ private boolean updateQuota(String mount, long nsQuota, long ssQuota)
 
   /**
    * initViewFsToMountTable.
-   *
-   * @param parameters The specified cluster to initialize.
-   * @param i Index in the parameters
+   * @param clusterName The specified cluster to initialize.
    * @return If the quota was updated.
    * @throws IOException Error adding the mount point.
    */
-  public boolean initViewFsToMountTable(String[] parameters, int i) throws IOException {
-    String clusterName = parameters[i++];
-    if (clusterName == null) {
-      System.out.println("Please enter the cluster name.");
-      return false;
-    }
+  public boolean initViewFsToMountTable(String clusterName)
+      throws IOException {
+    // fs.viewfs.mounttable.ClusterX.link./data
     final String mountTablePrefix =
         Constants.CONFIG_VIEWFS_PREFIX + "." + clusterName + "." +
             Constants.CONFIG_VIEWFS_LINK + "./";
     Map<String, String> viewFsMap = getConf().getValByRegex(mountTablePrefix);
-    if (viewFsMap.size() == 0) {
-      System.out.println("Please check the cluster name and veiwfs " +
-          "configuration.");
+    if (viewFsMap.isEmpty()) {
+      System.out.println("There is no ViewFs mapping to initialize.");
+      return true;
     }
-    for (String key : viewFsMap.keySet()) {
-      Path path = new Path(viewFsMap.get(key));
-      String owner = null;
-      String group = null;
-      FsPermission mode = null;
-      try {
-        FileSystem fs = path.getFileSystem(getConf());
-        if (fs.exists(path)) {
-          FileStatus fileStatus = fs.getFileStatus(path);
-          owner = fileStatus.getOwner();
-          group = fileStatus.getGroup();
-          mode = fileStatus.getPermission();
-        }
-      } catch (Exception e) {
-        LOG.warn("Exception encountered", e);
-      }
+    for (Entry<String, String> entry : viewFsMap.entrySet()) {
+      Path path = new Path(entry.getValue());
       DestinationOrder order = DestinationOrder.HASH;
-      String mount =
-          key.split(clusterName + "." + Constants.CONFIG_VIEWFS_LINK + ".")[1];
-      String dest = path.toUri().getPath();
+      String[] mount = entry.getKey().split(
+              clusterName + "." + Constants.CONFIG_VIEWFS_LINK + ".");
+      if (mount.length < 2) {
+        System.out.println("Added Mount Point failed " + entry.getKey());
+        continue;
+      }
       String[] nss = new String[]{path.toUri().getAuthority()};
-      addMount(mount, nss, dest, false, false, order,
-          new ACLEntity(owner, group, mode));
+      boolean added = addMount(
+          mount[1], nss, path.toUri().getPath(), false,
+          false, order, getACLEntityFormHdfsPath(path));
+      if (added) {
+        System.out.println("added mount point " + mount[1]);
+      }
     }
     return true;
   }
-  
+
+  /**
+   * Returns ACLEntity according to a HDFS pat.
+   * @param path A path of HDFS.
+   */
+  public ACLEntity getACLEntityFormHdfsPath(Path path){
+    String owner = null;
+    String group = null;
+    FsPermission mode = null;
+    try {
+      FileSystem fs = path.getFileSystem(getConf());
+      if (fs.exists(path)) {
+        FileStatus fileStatus = fs.getFileStatus(path);
+        owner = fileStatus.getOwner();
+        group = fileStatus.getGroup();
+        mode = fileStatus.getPermission();
+      }
+    } catch (IOException e) {
+      System.out.println("Exception encountered " + e);
+    }
+    return new ACLEntity(owner, group, mode);
+  }
+
   /**
    * Update storage type quota of specified mount table.
    *

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterAdminCLI.java

@@ -34,8 +34,11 @@
 import java.util.regex.Pattern;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.StorageType;
 import org.apache.hadoop.ha.HAServiceProtocol.HAServiceState;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.server.federation.MiniRouterDFSCluster.RouterContext;
 import org.apache.hadoop.hdfs.server.federation.RouterConfigBuilder;
@@ -78,6 +81,8 @@ public class TestRouterAdminCLI {
   private static RouterClient client;
   private static Router router;
 
+  private static DistributedFileSystem hdfs;
+
   private static final String TEST_USER = "test-user";
 
   private final ByteArrayOutputStream out = new ByteArrayOutputStream();
@@ -102,10 +107,12 @@ public static void globalSetUp() throws Exception {
 
     // Start routers
     cluster.startRouters();
+    cluster.startCluster();
 
     routerContext = cluster.getRandomRouter();
     router = routerContext.getRouter();
     stateStore = router.getStateStore();
+    hdfs = cluster.getCluster().getFileSystem();
 
     Configuration routerConf = new Configuration();
     InetSocketAddress routerSocket = router.getAdminServerAddress();
@@ -700,6 +707,46 @@ public void testAddMountTableIfParentExist() throws Exception {
     }
   }
 
+  @Test
+  public void testInitViewFsToMountTable() throws Exception {
+    // re-set system out for testing
+    System.setOut(new PrintStream(out));
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    String nnAddress = cluster.getRandomNamenode().getNamenode().getHostAndPort();
+
+    String src = "/data";
+    Path destPath = new Path("hdfs://" + nnAddress + "/data");
+    String user = "user1";
+    String group = "group1";
+    String clusterName = "ClusterX";
+
+    // 0.mkdir destPath
+    hdfs.mkdirs(destPath);
+    // 1.set owner
+    hdfs.setOwner(destPath, user, group);
+    // 2.set viewFs mapping
+    admin.getConf().set("fs.viewfs.mounttable.ClusterX.link." + src, destPath.toString());
+    // 3.run initialization
+    String[] argv = new String[]{"-initViewFsToMountTable", clusterName};
+    assertEquals(0, ToolRunner.run(admin, argv));
+    // 4.gets the mount point entries
+    stateStore.loadCache(MountTableStoreImpl.class, true);
+    GetMountTableEntriesRequest getRequest = GetMountTableEntriesRequest
+        .newInstance("/");
+    GetMountTableEntriesResponse getResponse = client.getMountTableManager()
+        .getMountTableEntries(getRequest);
+    List<MountTable> mountTables = getResponse.getEntries();
+    // 5.check
+    assertEquals(1, mountTables.size());
+    assertEquals(user, mountTables.get(0).getOwnerName());
+    assertEquals(group, mountTables.get(0).getGroupName());
+    assertEquals(destPath.toUri().getPath(), mountTables.get(0).
+        getDestinations().get(0).getDest());
+    assertEquals(nnAddress, mountTables.get(0).
+        getDestinations().get(0).getNameserviceId());
+    assertEquals(src, mountTables.get(0).getSourcePath());
+  }
+
   @Test
   public void testMountTablePermissions() throws Exception {
     // re-set system out for testing