content: /*
   * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
   * (the "License"). You may not use this work except in compliance with the License, which is
   * available at www.apache.org/licenses/LICENSE-2.0
   *
   * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
   * either express or implied, as more fully set forth in the License.
   *
   * See the NOTICE file distributed with this work for information regarding copyright ownership.
   */
  
  package alluxio.hadoop;
  
  import static com.google.common.hash.Hashing.md5;
  import static java.nio.charset.StandardCharsets.UTF_8;
  
  import alluxio.AlluxioURI;
  import alluxio.client.file.CacheContext;
  import alluxio.client.file.FileInStream;
  import alluxio.client.file.URIStatus;
  import alluxio.client.file.cache.CacheManager;
  import alluxio.client.file.cache.LocalCacheFileInStream;
  import alluxio.client.file.cache.filter.CacheFilter;
  import alluxio.conf.AlluxioConfiguration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.AlluxioException;
  import alluxio.metrics.MetricsConfig;
  import alluxio.metrics.MetricsSystem;
  import alluxio.wire.FileInfo;
  
  import com.google.common.base.Preconditions;
  import org.apache.hadoop.fs.BlockLocation;
  import org.apache.hadoop.fs.FSDataInputStream;
  import org.apache.hadoop.fs.FSDataOutputStream;
  import org.apache.hadoop.fs.FileStatus;
  import org.apache.hadoop.fs.Path;
  import org.apache.hadoop.fs.permission.FsPermission;
  import org.apache.hadoop.util.Progressable;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.net.URI;
  import java.util.Map;
  import java.util.Optional;
  import java.util.Properties;
  
  /**
   * An Alluxio client compatible with Apache Hadoop {@link org.apache.hadoop.fs.FileSystem}
   * interface, using Alluxio local cache. This client will first consult the local cache before
   * requesting the remote Hadoop FileSystem in case of cache misses.
   */
  public class LocalCacheFileSystem extends org.apache.hadoop.fs.FileSystem {
    private static final Logger LOG = LoggerFactory.getLogger(LocalCacheFileSystem.class);
  
    /** The external Hadoop filesystem to query on cache miss. */
    private final org.apache.hadoop.fs.FileSystem mExternalFileSystem;
    private final HadoopFileOpener mHadoopFileOpener;
    private final LocalCacheFileInStream.FileInStreamOpener mAlluxioFileOpener;
    private CacheManager mCacheManager;
    private CacheFilter mCacheFilter;
    private org.apache.hadoop.conf.Configuration mHadoopConf;
    private AlluxioConfiguration mAlluxioConf;
  
    /**
     * @param fileSystem File System instance
     */
    public LocalCacheFileSystem(org.apache.hadoop.fs.FileSystem fileSystem) {
      this(fileSystem, uriStatus -> fileSystem.open(new Path(uriStatus.getPath())));
    }
  
    /**
     * @param fileSystem File System instance
     * @param fileOpener File opener instance
     */
    public LocalCacheFileSystem(org.apache.hadoop.fs.FileSystem fileSystem,
        HadoopFileOpener fileOpener) {
      mExternalFileSystem = Preconditions.checkNotNull(fileSystem, "filesystem");
      mHadoopFileOpener = Preconditions.checkNotNull(fileOpener, "fileOpener");
      mAlluxioFileOpener = status -> new AlluxioHdfsInputStream(mHadoopFileOpener.open(status));
    }
  
    @Override
    public synchronized void initialize(URI uri, org.apache.hadoop.conf.Configuration conf)
        throws IOException {
      super.initialize(uri, conf);
      mHadoopConf = conf;
      // Set statistics
      setConf(conf);
      mAlluxioConf = HadoopUtils.toAlluxioConf(mHadoopConf);
      // Handle metrics
      Properties metricsProperties = new Properties();
      for (Map.Entry<String, String> entry : conf) {
        metricsProperties.setProperty(entry.getKey(), entry.getValue());
      }
      MetricsSystem.startSinksFromConfig(new MetricsConfig(metricsProperties));
      mCacheManager = CacheManager.Factory.get(mAlluxioConf);
      mCacheFilter = CacheFilter.create(mAlluxioConf);
      LocalCacheFileInStream.registerMetrics();
    }
  
    @Override
    public void close() throws IOException {
      // super.close should be called first before releasing the resources in this instance, as the
      // super class may invoke other methods in this class. For example,
      // org.apache.hadoop.fs.FileSystem.close may check the existence of certain temp files before
      // closing
      super.close();
    }
  
    /**
     * @return scheme
     */
    // @Override This doesn't exist in Hadoop 1.x, so cannot put {@literal @Override}.
    public String getScheme() {
      return mExternalFileSystem.getScheme();
    }
  
    @Override
    public FSDataInputStream open(Path path, int bufferSize) throws IOException {
      if (mCacheManager == null) {
        return mExternalFileSystem.open(path, bufferSize);
      }
      FileStatus externalFileStatus = mExternalFileSystem.getFileStatus(path);
      // Note that, we don't set have fileId here because fileId is Alluxio specific
      FileInfo info = new FileInfo()
          .setLength(externalFileStatus.getLen())
          .setPath(externalFileStatus.getPath().toString())
          .setFolder(externalFileStatus.isDirectory())
          .setBlockSizeBytes(externalFileStatus.getBlockSize())
          .setLastModificationTimeMs(externalFileStatus.getModificationTime())
          .setLastAccessTimeMs(externalFileStatus.getAccessTime())
          .setOwner(externalFileStatus.getOwner())
          .setGroup(externalFileStatus.getGroup());
      // FilePath is a unique identifier for a file, however it can be a long string
      // hence using md5 hash of the file path as the identifier in the cache.
      String cacheIdentifier;
      if (mAlluxioConf.getBoolean(PropertyKey.USER_CLIENT_CACHE_IDENTIFIER_INCLUDE_MTIME)) {
        // include mtime to avoid consistency issues if the file may update
        cacheIdentifier = md5().hashString(externalFileStatus.getPath().toString()
            + externalFileStatus.getModificationTime(), UTF_8).toString();
      } else {
        cacheIdentifier = md5().hashString(externalFileStatus.getPath().toString(), UTF_8).toString();
      }
      CacheContext context = CacheContext.defaults().setCacheIdentifier(cacheIdentifier);
      URIStatus status = new URIStatus(info, context);
      return open(status, bufferSize);
    }
  
    /**
     * A wrapper method to default not enforce an open call.
     *
     * @param status the status of the file to open
     * @param bufferSize stream buffer size in bytes, currently unused
     * @return an {@link FSDataInputStream} at the indicated path of a file
     */
    public FSDataInputStream open(URIStatus status, int bufferSize) throws IOException {
      return open(status, bufferSize, false);
    }
  
    /**
     * Attempts to open the specified file for reading.
     *
     * @param status the status of the file to open
     * @param bufferSize stream buffer size in bytes, currently unused
     * @param enforceOpen flag to enforce calling open to external storage
     * @return an {@link FSDataInputStream} at the indicated path of a file
     */
    public FSDataInputStream open(URIStatus status, int bufferSize, boolean enforceOpen)
            throws IOException {
      if (mCacheManager == null || !mCacheFilter.needsCache(status)) {
        return mExternalFileSystem.open(HadoopUtils.toPath(new AlluxioURI(status.getPath())),
            bufferSize);
      }
      Optional<FileInStream> externalFileInStream;
      if (enforceOpen) {
        try {
          // making the open call right now, instead of later when called back
          externalFileInStream = Optional.of(mAlluxioFileOpener.open(status));
        } catch (AlluxioException e) {
          throw new IOException(e);
        }
      } else {
        externalFileInStream = Optional.empty();
      }
  
      return new FSDataInputStream(new HdfsFileInputStream(
          new LocalCacheFileInStream(status, mAlluxioFileOpener, mCacheManager, mAlluxioConf,
              externalFileInStream),
          statistics));
    }
  
    @Override
    public URI getUri() {
      return mExternalFileSystem.getUri();
    }
  
    @Override
    public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite,
        int bufferSize, short replication, long blockSize, Progressable progress) throws IOException {
      return mExternalFileSystem.create(f, permission, overwrite, bufferSize, replication, blockSize,
          progress);
    }
  
    @Override
    public FSDataOutputStream append(Path f, int bufferSize, Progressable progress)
        throws IOException {
      return mExternalFileSystem.append(f, bufferSize, progress);
    }
  
    @Override
    public boolean rename(Path src, Path dst) throws IOException {
      return mExternalFileSystem.rename(src, dst);
    }
  
    @Override
    public boolean delete(Path f, boolean recursive) throws IOException {
      return mExternalFileSystem.delete(f, recursive);
    }
  
    @Override
    public FileStatus[] listStatus(Path f) throws FileNotFoundException, IOException {
      return mExternalFileSystem.listStatus(f);
    }
  
    @Override
    public void setWorkingDirectory(Path new_dir) {
      mExternalFileSystem.setWorkingDirectory(new_dir);
    }
  
    @Override
    public Path getWorkingDirectory() {
      return mExternalFileSystem.getWorkingDirectory();
    }
  
    @Override
    public boolean mkdirs(Path f, FsPermission permission) throws IOException {
      return mExternalFileSystem.mkdirs(f, permission);
    }
  
    @Override
    public FileStatus getFileStatus(Path f) throws IOException {
      return mExternalFileSystem.getFileStatus(f);
    }
  
    @Override
    public BlockLocation[] getFileBlockLocations(FileStatus file, long start,
        long len) throws IOException {
      // Applications use the block information here to schedule/distribute the tasks.
      // Return the UFS locations directly instead of the local cache location,
      // so the application can schedule the tasks accordingly
      return mExternalFileSystem.getFileBlockLocations(file, start, len);
    }
  
    @Override
    public BlockLocation[] getFileBlockLocations(Path p, long start, long len)
        throws IOException {
      // Applications use the block information here to schedule/distribute the tasks.
      // Return the UFS locations directly instead of the local cache location,
      // so the application can schedule the tasks accordingly
      return mExternalFileSystem.getFileBlockLocations(p, start, len);
    }
  }
  