HADOOP-17028. ViewFS should initialize mounted target filesystems lazily. Contributed by Abhishek Das (#2260)

Author: abhishekdas99

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.fs.viewfs;
 
+import java.util.function.Function;
 import org.apache.hadoop.thirdparty.com.google.common.base.Preconditions;
 import java.io.FileNotFoundException;
 import java.io.IOException;
@@ -257,7 +258,10 @@ enum LinkType {
    */
   static class INodeLink<T> extends INode<T> {
     final URI[] targetDirLinkList;
-    final T targetFileSystem;   // file system object created from the link.
+    private T targetFileSystem;   // file system object created from the link.
+    // Function to initialize file system. Only applicable for simple links
+    private Function<URI, T> fileSystemInitMethod;
+    private final Object lock = new Object();
 
     /**
      * Construct a mergeLink or nfly.
@@ -273,11 +277,13 @@ static class INodeLink<T> extends INode<T> {
      * Construct a simple link (i.e. not a mergeLink).
      */
     INodeLink(final String pathToNode, final UserGroupInformation aUgi,
-        final T targetFs, final URI aTargetDirLink) {
+        Function<URI, T> createFileSystemMethod,
+        final URI aTargetDirLink) {
       super(pathToNode, aUgi);
-      targetFileSystem = targetFs;
+      targetFileSystem = null;
       targetDirLinkList = new URI[1];
       targetDirLinkList[0] = aTargetDirLink;
+      this.fileSystemInitMethod = createFileSystemMethod;
     }
 
     /**
@@ -298,7 +304,30 @@ boolean isInternalDir() {
       return false;
     }
 
-    public T getTargetFileSystem() {
+    /**
+     * Get the instance of FileSystem to use, creating one if needed.
+     * @return An Initialized instance of T
+     * @throws IOException
+     */
+    public T getTargetFileSystem() throws IOException {
+      if (targetFileSystem != null) {
+        return targetFileSystem;
+      }
+      // For non NFLY and MERGE links, we initialize the FileSystem when the
+      // corresponding mount path is accessed.
+      if (targetDirLinkList.length == 1) {
+        synchronized (lock) {
+          if (targetFileSystem != null) {
+            return targetFileSystem;
+          }
+          targetFileSystem = fileSystemInitMethod.apply(targetDirLinkList[0]);
+          if (targetFileSystem == null) {
+            throw new IOException(
+                "Could not initialize target File System for URI : " +
+                    targetDirLinkList[0]);
+          }
+        }
+      }
       return targetFileSystem;
     }
   }
@@ -359,7 +388,7 @@ private void createLink(final String src, final String target,
     switch (linkType) {
     case SINGLE:
       newLink = new INodeLink<T>(fullPath, aUgi,
-          getTargetFileSystem(new URI(target)), new URI(target));
+          initAndGetTargetFs(), new URI(target));
       break;
     case SINGLE_FALLBACK:
     case MERGE_SLASH:
@@ -385,8 +414,7 @@ private void createLink(final String src, final String target,
    * 3 abstract methods.
    * @throws IOException
    */
-  protected abstract T getTargetFileSystem(URI uri)
-      throws UnsupportedFileSystemException, URISyntaxException, IOException;
+  protected abstract Function<URI, T> initAndGetTargetFs();
 
   protected abstract T getTargetFileSystem(INodeDir<T> dir)
       throws URISyntaxException, IOException;
@@ -589,7 +617,7 @@ protected InodeTree(final Configuration config, final String viewName,
     if (isMergeSlashConfigured) {
       Preconditions.checkNotNull(mergeSlashTarget);
       root = new INodeLink<T>(mountTableName, ugi,
-          getTargetFileSystem(new URI(mergeSlashTarget)),
+          initAndGetTargetFs(),
           new URI(mergeSlashTarget));
       mountPoints.add(new MountPoint<T>("/", (INodeLink<T>) root));
       rootFallbackLink = null;
@@ -608,8 +636,7 @@ protected InodeTree(final Configuration config, final String viewName,
                 + "not allowed.");
           }
           fallbackLink = new INodeLink<T>(mountTableName, ugi,
-              getTargetFileSystem(new URI(le.getTarget())),
-              new URI(le.getTarget()));
+              initAndGetTargetFs(), new URI(le.getTarget()));
           continue;
         case REGEX:
           addRegexMountEntry(le);
@@ -633,9 +660,8 @@ protected InodeTree(final Configuration config, final String viewName,
       FileSystem.LOG
           .info("Empty mount table detected for {} and considering itself "
               + "as a linkFallback.", theUri);
-      rootFallbackLink =
-          new INodeLink<T>(mountTableName, ugi, getTargetFileSystem(theUri),
-              theUri);
+      rootFallbackLink = new INodeLink<T>(mountTableName, ugi,
+          initAndGetTargetFs(), theUri);
       getRootDir().addFallbackLink(rootFallbackLink);
     }
   }
@@ -733,10 +759,10 @@ boolean isLastInternalDirLink() {
    * @param p - input path
    * @param resolveLastComponent
    * @return ResolveResult which allows further resolution of the remaining path
-   * @throws FileNotFoundException
+   * @throws IOException
    */
   ResolveResult<T> resolve(final String p, final boolean resolveLastComponent)
-      throws FileNotFoundException {
+      throws IOException {
     ResolveResult<T> resolveResult = null;
     String[] path = breakIntoPathComponents(p);
     if (path.length <= 1) { // special case for when path is "/"
@@ -880,19 +906,20 @@ protected ResolveResult<T> buildResolveResultForRegexMountPoint(
       ResultKind resultKind, String resolvedPathStr,
       String targetOfResolvedPathStr, Path remainingPath) {
     try {
-      T targetFs = getTargetFileSystem(
-          new URI(targetOfResolvedPathStr));
+      T targetFs = initAndGetTargetFs()
+          .apply(new URI(targetOfResolvedPathStr));
+      if (targetFs == null) {
+        LOGGER.error(String.format(
+            "Not able to initialize target file system."
+                + " ResultKind:%s, resolvedPathStr:%s,"
+                + " targetOfResolvedPathStr:%s, remainingPath:%s,"
+                + " will return null.",
+            resultKind, resolvedPathStr, targetOfResolvedPathStr,
+            remainingPath));
+        return null;
+      }
       return new ResolveResult<T>(resultKind, targetFs, resolvedPathStr,
           remainingPath, true);
-    } catch (IOException ex) {
-      LOGGER.error(String.format(
-          "Got Exception while build resolve result."
-              + " ResultKind:%s, resolvedPathStr:%s,"
-              + " targetOfResolvedPathStr:%s, remainingPath:%s,"
-              + " will return null.",
-          resultKind, resolvedPathStr, targetOfResolvedPathStr, remainingPath),
-          ex);
-      return null;
     } catch (URISyntaxException uex) {
       LOGGER.error(String.format(
           "Got Exception while build resolve result."

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

@@ -26,10 +26,12 @@
 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 java.util.function.Function;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -302,7 +304,7 @@ public void initialize(final URI theUri, final Configuration conf)
     enableInnerCache = config.getBoolean(CONFIG_VIEWFS_ENABLE_INNER_CACHE,
         CONFIG_VIEWFS_ENABLE_INNER_CACHE_DEFAULT);
     FsGetter fsGetter = fsGetter();
-    final InnerCache innerCache = new InnerCache(fsGetter);
+    cache = new InnerCache(fsGetter);
     // Now build  client side view (i.e. client side mount table) from config.
     final String authority = theUri.getAuthority();
     String tableName = authority;
@@ -318,15 +320,32 @@ public void initialize(final URI theUri, final Configuration conf)
       fsState = new InodeTree<FileSystem>(conf, tableName, myUri,
           initingUriAsFallbackOnNoMounts) {
         @Override
-        protected FileSystem getTargetFileSystem(final URI uri)
-          throws URISyntaxException, IOException {
-          FileSystem fs;
-          if (enableInnerCache) {
-            fs = innerCache.get(uri, config);
-          } else {
-            fs = fsGetter.get(uri, config);
-          }
-          return new ChRootedFileSystem(fs, uri);
+        protected Function<URI, FileSystem> initAndGetTargetFs() {
+          return new Function<URI, FileSystem>() {
+            @Override
+            public FileSystem apply(final URI uri) {
+              FileSystem fs;
+              try {
+                fs = ugi.doAs(new PrivilegedExceptionAction<FileSystem>() {
+                  @Override
+                  public FileSystem run() throws IOException {
+                    if (enableInnerCache) {
+                      synchronized (cache) {
+                        return cache.get(uri, config);
+                      }
+                    } else {
+                      return fsGetter().get(uri, config);
+                    }
+                  }
+                });
+                return new ChRootedFileSystem(fs, uri);
+              } catch (IOException | InterruptedException ex) {
+                LOG.error("Could not initialize the underlying FileSystem "
+                    + "object. Exception: " + ex.toString());
+              }
+              return null;
+            }
+          };
         }
 
         @Override
@@ -350,13 +369,6 @@ protected FileSystem getTargetFileSystem(final String settings,
     } catch (URISyntaxException e) {
       throw new IOException("URISyntax exception: " + theUri);
     }
-
-    if (enableInnerCache) {
-      // All fs instances are created and cached on startup. The cache is
-      // readonly after the initialize() so the concurrent access of the cache
-      // is safe.
-      cache = innerCache;
-    }
   }
 
   /**
@@ -388,7 +400,7 @@ public URI getUri() {
   @Override
   public Path resolvePath(final Path f) throws IOException {
     final InodeTree.ResolveResult<FileSystem> res;
-      res = fsState.resolve(getUriPath(f), true);
+    res = fsState.resolve(getUriPath(f), true);
     if (res.isInternalDir()) {
       return f;
     }
@@ -908,10 +920,35 @@ public void removeXAttr(Path path, String name) throws IOException {
   public void setVerifyChecksum(final boolean verifyChecksum) { 
     List<InodeTree.MountPoint<FileSystem>> mountPoints = 
         fsState.getMountPoints();
+    Map<String, FileSystem> fsMap = initializeMountedFileSystems(mountPoints);
     for (InodeTree.MountPoint<FileSystem> mount : mountPoints) {
-      mount.target.targetFileSystem.setVerifyChecksum(verifyChecksum);
+      fsMap.get(mount.src).setVerifyChecksum(verifyChecksum);
     }
   }
+
+  /**
+   * Initialize the target filesystem for all mount points.
+   * @param mountPoints The mount points
+   * @return Mapping of mount point and the initialized target filesystems
+   * @throws RuntimeException when the target file system cannot be initialized
+   */
+  private Map<String, FileSystem> initializeMountedFileSystems(
+      List<InodeTree.MountPoint<FileSystem>> mountPoints) {
+    FileSystem fs = null;
+    Map<String, FileSystem> fsMap = new HashMap<>(mountPoints.size());
+    for (InodeTree.MountPoint<FileSystem> mount : mountPoints) {
+      try {
+        fs = mount.target.getTargetFileSystem();
+        fsMap.put(mount.src, fs);
+      } catch (IOException ex) {
+        String errMsg = "Not able to initialize FileSystem for mount path " +
+            mount.src + " with exception " + ex;
+        LOG.error(errMsg);
+        throw new RuntimeException(errMsg, ex);
+      }
+    }
+    return fsMap;
+  }
 
   @Override
   public long getDefaultBlockSize() {
@@ -936,6 +973,9 @@ public long getDefaultBlockSize(Path f) {
       return res.targetFileSystem.getDefaultBlockSize(res.remainingPath);
     } catch (FileNotFoundException e) {
       throw new NotInMountpointException(f, "getDefaultBlockSize"); 
+    } catch (IOException e) {
+      throw new RuntimeException("Not able to initialize fs in "
+          + " getDefaultBlockSize for path " + f + " with exception", e);
     }
   }
 
@@ -947,6 +987,9 @@ public short getDefaultReplication(Path f) {
       return res.targetFileSystem.getDefaultReplication(res.remainingPath);
     } catch (FileNotFoundException e) {
       throw new NotInMountpointException(f, "getDefaultReplication"); 
+    } catch (IOException e) {
+      throw new RuntimeException("Not able to initialize fs in "
+          + " getDefaultReplication for path " + f + " with exception", e);
     }
   }
 
@@ -979,25 +1022,33 @@ public QuotaUsage getQuotaUsage(Path f) throws IOException {
   public void setWriteChecksum(final boolean writeChecksum) { 
     List<InodeTree.MountPoint<FileSystem>> mountPoints = 
         fsState.getMountPoints();
+    Map<String, FileSystem> fsMap = initializeMountedFileSystems(mountPoints);
     for (InodeTree.MountPoint<FileSystem> mount : mountPoints) {
-      mount.target.targetFileSystem.setWriteChecksum(writeChecksum);
+      fsMap.get(mount.src).setWriteChecksum(writeChecksum);
     }
   }
 
   @Override
   public FileSystem[] getChildFileSystems() {
     List<InodeTree.MountPoint<FileSystem>> mountPoints =
         fsState.getMountPoints();
+    Map<String, FileSystem> fsMap = initializeMountedFileSystems(mountPoints);
     Set<FileSystem> children = new HashSet<FileSystem>();
     for (InodeTree.MountPoint<FileSystem> mountPoint : mountPoints) {
-      FileSystem targetFs = mountPoint.target.targetFileSystem;
+      FileSystem targetFs = fsMap.get(mountPoint.src);
       children.addAll(Arrays.asList(targetFs.getChildFileSystems()));
     }
 
-    if (fsState.isRootInternalDir() && fsState.getRootFallbackLink() != null) {
-      children.addAll(Arrays.asList(
-          fsState.getRootFallbackLink().targetFileSystem
-              .getChildFileSystems()));
+    try {
+      if (fsState.isRootInternalDir() &&
+          fsState.getRootFallbackLink() != null) {
+        children.addAll(Arrays.asList(
+            fsState.getRootFallbackLink().getTargetFileSystem()
+                .getChildFileSystems()));
+      }
+    } catch (IOException ex) {
+      LOG.error("Could not add child filesystems for source path "
+          + fsState.getRootFallbackLink().fullPath + " with exception " + ex);
     }
     return children.toArray(new FileSystem[]{});
   }

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

@@ -348,7 +348,7 @@ public MountPathInfo<FileSystem> getMountPathInfo(Path path,
       FileSystem fs = res.isInternalDir() ?
           (fsState.getRootFallbackLink() != null ?
               ((ChRootedFileSystem) fsState
-                  .getRootFallbackLink().targetFileSystem).getMyFs() :
+                  .getRootFallbackLink().getTargetFileSystem()).getMyFs() :
               fsGetter().get(path.toUri(), conf)) :
           ((ChRootedFileSystem) res.targetFileSystem).getMyFs();
       return new MountPathInfo<FileSystem>(res.remainingPath, res.resolvedPath,
@@ -390,8 +390,13 @@ public FileSystem getFallbackFileSystem() {
     if (fsState.getRootFallbackLink() == null) {
       return null;
     }
-    return ((ChRootedFileSystem) fsState.getRootFallbackLink().targetFileSystem)
-        .getMyFs();
+    try {
+      return ((ChRootedFileSystem) fsState.getRootFallbackLink()
+          .getTargetFileSystem()).getMyFs();
+    } catch (IOException ex) {
+      LOG.error("Could not get fallback filesystem ");
+    }
+    return null;
   }
 
   @Override