alluxio-master-2.x/integration/fuse/src/main/java/alluxio/fuse/AlluxioFuse.java
/*
 * 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.fuse;

import alluxio.Constants;
import alluxio.ProjectConstants;
import alluxio.RuntimeConstants;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemContext;
import alluxio.client.file.options.FileSystemOptions;
import alluxio.client.file.options.UfsFileSystemOptions;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.Configuration;
import alluxio.conf.InstancedConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.conf.Source;
import alluxio.exception.runtime.FailedPreconditionRuntimeException;
import alluxio.exception.runtime.InvalidArgumentRuntimeException;
import alluxio.fuse.meta.UpdateChecker;
import alluxio.fuse.options.FuseOptions;
import alluxio.heartbeat.FixedIntervalSupplier;
import alluxio.heartbeat.HeartbeatContext;
import alluxio.heartbeat.HeartbeatThread;
import alluxio.jnifuse.LibFuse;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;
import alluxio.security.user.UserState;
import alluxio.util.CommonUtils;
import alluxio.util.JvmPauseMonitor;
import alluxio.util.network.NetworkAddressUtils;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import sun.misc.Signal;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Main entry point to Alluxio-FUSE.
 */
@ThreadSafe
public final class AlluxioFuse {
  private static final Logger LOG = LoggerFactory.getLogger(AlluxioFuse.class);
  private static final CommandLineParser PARSER = new DefaultParser();

  private static final String MOUNT_POINT_OPTION_NAME = "m";
  private static final String MOUNT_ALLUXIO_PATH_OPTION_NAME = "a";
  private static final String MOUNT_ROOT_UFS_OPTION_NAME = "u";
  private static final String MOUNT_OPTIONS_OPTION_NAME = "o";
  private static final String HELP_OPTION_NAME = "h";
  private static final String UPDATE_CHECK_OPTION_NAME = "c";

  private static final Option MOUNT_POINT_OPTION = Option.builder(MOUNT_POINT_OPTION_NAME)
      .hasArg()
      .required(false)
      .longOpt("mount-point")
      .desc("The absolute local filesystem path that standalone Fuse will mount Alluxio path to.")
      .build();
  private static final Option MOUNT_ALLUXIO_PATH_OPTION
      = Option.builder(MOUNT_ALLUXIO_PATH_OPTION_NAME)
      .hasArg()
      .required(false)
      .longOpt("alluxio-path")
      .desc("The Alluxio path to mount to the given Fuse mount point "
          + "(for example, mount alluxio path `/alluxio` to fuse mount point `/mnt/alluxio-fuse`; "
          + "local operations like `mkdir /mnt/alluxio-fuse/folder` will be translated to "
          + "`alluxio fs mkdir /alluxio/folder`)")
      .build();
  private static final Option MOUNT_ROOT_UFS_OPTION
      = Option.builder(MOUNT_ROOT_UFS_OPTION_NAME)
      .hasArg()
      .required(false)
      .longOpt("root-ufs")
      .desc("The storage address of the UFS to mount to the given Fuse mount point. "
          + "All operations against the FUSE mount point "
          + "will be redirected to this storage address. "
          + "(for example, mount storage address `s3://my_bucket/my_folder` "
          + "to local FUSE mount point `/mnt/alluxio-fuse`; "
          + "local operations like `mkdir /mnt/alluxio-fuse/folder` will be translated to "
          + "`mkdir s3://my_bucket/my_folder/folder`)")
      .build();
  private static final Option MOUNT_OPTIONS = Option.builder(MOUNT_OPTIONS_OPTION_NAME)
      .valueSeparator(',')
      .required(false)
      .hasArgs()
      .desc("Providing mount options separating by comma. "
          + "Mount options includes operating system mount options, "
          + "many FUSE specific mount options (e.g. direct_io,attr_timeout=10s.allow_other), "
          + "Alluxio property key=value pairs, and Alluxio FUSE special mount options "
          + "data_cache=<local_cache_directory>,data_cache_size=<size>,"
          + "metadata_cache_size=<size>,metadata_cache_expire=<timeout>")
      .build();
  private static final Option UPDATE_CHECK_OPTION = Option.builder(UPDATE_CHECK_OPTION_NAME)
      .required(false)
      .longOpt("update-check")
      .hasArg()
      .desc("Enables or disables the FUSE version update check. "
          + "Disabled by default when connecting to Alluxio system cache. "
          + "Enabled by default when connecting an under storage directly.")
      .build();
  private static final Option HELP_OPTION = Option.builder(HELP_OPTION_NAME)
      .required(false)
      .desc("Print this help message")
      .build();
  private static final Options OPTIONS = new Options()
      .addOption(MOUNT_POINT_OPTION)
      .addOption(MOUNT_ALLUXIO_PATH_OPTION)
      .addOption(MOUNT_ROOT_UFS_OPTION)
      .addOption(MOUNT_OPTIONS)
      .addOption(UPDATE_CHECK_OPTION)
      .addOption(HELP_OPTION);

  // prevent instantiation
  private AlluxioFuse() {}

  /**
   * Running this class will mount the file system according to the options passed to this function.
   * The user-space fuse application will stay on the foreground and keep the file system mounted.
   * The user can unmount the file system by gracefully killing (SIGINT) the process.
   *
   * @param args arguments to run the command line
   */
  public static void main(String[] args) throws ParseException {
    CommandLine cli = PARSER.parse(OPTIONS, args);

    if (cli.hasOption(HELP_OPTION_NAME)) {
      final HelpFormatter fmt = new HelpFormatter();
      fmt.printHelp(AlluxioFuse.class.getName(), OPTIONS);
      return;
    }

    LOG.info("Alluxio version: {}-{}", RuntimeConstants.VERSION, ProjectConstants.REVISION);
    setConfigurationFromInput(cli, Configuration.modifiableGlobal());
    AlluxioConfiguration conf = Configuration.global();
    FuseOptions fuseOptions = getFuseOptions(cli, conf);

    FileSystemContext fsContext = FileSystemContext.create(conf);
    if (!fuseOptions.getFileSystemOptions().getUfsFileSystemOptions().isPresent()) {
      conf = AlluxioFuseUtils.tryLoadingConfigFromMaster(fsContext);
    }

    CommonUtils.PROCESS_TYPE.set(CommonUtils.ProcessType.CLIENT);
    MetricsSystem.startSinks(conf.getString(PropertyKey.METRICS_CONF_FILE));
    if (conf.getBoolean(PropertyKey.FUSE_WEB_ENABLED)) {
      FuseWebServer webServer = new FuseWebServer(
          NetworkAddressUtils.ServiceType.FUSE_WEB.getServiceName(),
          NetworkAddressUtils.getBindAddress(
              NetworkAddressUtils.ServiceType.FUSE_WEB,
              Configuration.global()));
      webServer.start();
    }
    startJvmMonitorProcess();
    ExecutorService executor = null;
    if (fuseOptions.updateCheckEnabled()) {
      executor = Executors.newSingleThreadExecutor();
      executor.submit(new HeartbeatThread(HeartbeatContext.FUSE_UPDATE_CHECK,
          UpdateChecker.create(fuseOptions), () -> new FixedIntervalSupplier(Constants.DAY_MS),
          Configuration.global(), UserState.Factory.create(conf)));
    }
    try (FileSystem fs = FileSystem.Factory.create(fsContext, fuseOptions.getFileSystemOptions())) {
      launchFuse(fsContext, fs, fuseOptions, true);
    } catch (Throwable t) {
      if (executor != null) {
        executor.shutdown();
      }
      // TODO(lu) FUSE unmount gracefully
      LOG.error("Failed to launch FUSE", t);
      System.exit(-1);
    }
  }

  /**
   * Launches Fuse application.
   *
   * @param fsContext file system context for Fuse client to communicate to servers
   * @param fs file system for Fuse client to communicate to servers
   * @param fuseOptions Fuse options
   * @param blocking whether the Fuse application is blocking or not
   * @return the Fuse application handler for future Fuse umount operation
   */
  public static FuseUmountable launchFuse(FileSystemContext fsContext, FileSystem fs,
       FuseOptions fuseOptions, boolean blocking) {
    AlluxioConfiguration conf = fsContext.getClusterConf();
    validateFuseConfAndOptions(conf, fuseOptions);

    LibFuse.loadLibrary(AlluxioFuseUtils.getLibfuseVersion(conf));

    String mountPoint = conf.getString(PropertyKey.FUSE_MOUNT_POINT);
    Path mountPath = Paths.get(mountPoint);
    if (!Files.exists(mountPath)) {
      LOG.warn("Mount point on local filesystem does not exist, creating {}", mountPoint);
      try {
        Files.createDirectories(mountPath);
      } catch (IOException e) {
        throw new FailedPreconditionRuntimeException("Failed to create mount point");
      }
    }

    final boolean debugEnabled = conf.getBoolean(PropertyKey.FUSE_DEBUG_ENABLED);
    if (conf.getBoolean(PropertyKey.FUSE_JNIFUSE_ENABLED)) {
      final AlluxioJniFuseFileSystem fuseFs
          = new AlluxioJniFuseFileSystem(fsContext, fs, fuseOptions);

      FuseSignalHandler fuseSignalHandler = new FuseSignalHandler(fuseFs);
      Signal.handle(new Signal("TERM"), fuseSignalHandler);

      try {
        LOG.info("Mounting AlluxioJniFuseFileSystem: mount point=\"{}\", OPTIONS=\"{}\"",
            mountPoint, String.join(",", fuseOptions.getFuseMountOptions()));
        fuseFs.mount(blocking, debugEnabled, fuseOptions.getFuseMountOptions());
        return fuseFs;
      } catch (RuntimeException e) {
        fuseFs.umount(true);
        throw e;
      }
    } else {
      final AlluxioJnrFuseFileSystem fuseFs = new AlluxioJnrFuseFileSystem(fs, conf, fuseOptions);
      try {
        fuseFs.mount(mountPath, blocking, debugEnabled, fuseOptions.getFuseMountOptions()
            .stream().map(a -> "-o" + a).toArray(String[]::new));
        return fuseFs;
      } catch (Throwable t) {
        // only try to umount file system when exception occurred.
        // jnr-fuse registers JVM shutdown hook to ensure fs.umount()
        // will be executed when this process is exiting.
        fuseFs.umount();
        throw t;
      }
    }
  }

  /**
   * Updates Alluxio configuration according to command line input.
   *
   * @param cli the command line inputs
   * @param conf the modifiable configuration to update
   */
  private static void setConfigurationFromInput(CommandLine cli, InstancedConfiguration conf) {
    if (cli.hasOption(MOUNT_POINT_OPTION_NAME)) {
      conf.set(PropertyKey.FUSE_MOUNT_POINT,
          cli.getOptionValue(MOUNT_POINT_OPTION_NAME), Source.RUNTIME);
    }
    if (cli.hasOption(MOUNT_ALLUXIO_PATH_OPTION_NAME)) {
      conf.set(PropertyKey.FUSE_MOUNT_ALLUXIO_PATH,
          cli.getOptionValue(MOUNT_ALLUXIO_PATH_OPTION_NAME), Source.RUNTIME);
    }
    if (cli.hasOption(MOUNT_ROOT_UFS_OPTION_NAME)) {
      // Disable connections between FUSE and server
      conf.set(PropertyKey.USER_METRICS_COLLECTION_ENABLED, false, Source.RUNTIME);
      conf.set(PropertyKey.USER_UPDATE_FILE_ACCESSTIME_DISABLED, true, Source.RUNTIME);
    }
    if (cli.hasOption(MOUNT_OPTIONS_OPTION_NAME)) {
      List<String> fuseOptions = new ArrayList<>();
      String[] mountOptionsArray = cli.getOptionValues(MOUNT_OPTIONS_OPTION_NAME);
      for (String opt : mountOptionsArray) {
        String trimedOpt = opt.trim();
        if (trimedOpt.isEmpty()) {
          continue;
        }
        String[] optArray = trimedOpt.split("=");
        if (optArray.length == 1) {
          fuseOptions.add(trimedOpt);
          continue;
        }
        String key = optArray[0];
        String value = optArray[1];
        if (PropertyKey.isValid(key)) {
          PropertyKey propertyKey = PropertyKey.fromString(key);
          conf.set(propertyKey, propertyKey.parseValue(value), Source.RUNTIME);
          LOG.info("Set Alluxio property key({}={}) from command line input", key, value);
        } else if (key.equals("data_cache")) {
          conf.set(PropertyKey.USER_CLIENT_CACHE_ENABLED, true, Source.RUNTIME);
          conf.set(PropertyKey.USER_CLIENT_CACHE_DIRS,
              PropertyKey.USER_CLIENT_CACHE_DIRS.parseValue(value), Source.RUNTIME);
          LOG.info("Set data cache to {} from command line input", value);
        } else if (key.equals("data_cache_size")) {
          conf.set(PropertyKey.USER_CLIENT_CACHE_SIZE,
              PropertyKey.USER_CLIENT_CACHE_SIZE.parseValue(value), Source.RUNTIME);
          LOG.info("Set data cache size as {} from command line input", value);
        } else if (key.equals("metadata_cache_size")) {
          if (value.equals("0")) {
            continue;
          }
          conf.set(PropertyKey.USER_METADATA_CACHE_ENABLED, true, Source.RUNTIME);
          conf.set(PropertyKey.USER_METADATA_CACHE_MAX_SIZE,
              PropertyKey.USER_METADATA_CACHE_MAX_SIZE.parseValue(value), Source.RUNTIME);
          LOG.info("Set metadata cache size as {} from command line input", value);
        } else if (key.equals("metadata_cache_expire")) {
          conf.set(PropertyKey.USER_METADATA_CACHE_EXPIRATION_TIME,
              PropertyKey.USER_METADATA_CACHE_EXPIRATION_TIME.parseValue(value), Source.RUNTIME);
          LOG.info("Set metadata cache expiration time as {} from command line input", value);
        } else {
          fuseOptions.add(trimedOpt);
        }
      }
      if (!fuseOptions.isEmpty()) {
        conf.set(PropertyKey.FUSE_MOUNT_OPTIONS, fuseOptions, Source.RUNTIME);
        LOG.info("Set fuse mount point options as {} from command line input",
            String.join(",", fuseOptions));
      }
    }
  }

  private static FuseOptions getFuseOptions(CommandLine cli, AlluxioConfiguration conf) {
    boolean updateCheckEnabled = false;
    if (cli.hasOption(UPDATE_CHECK_OPTION_NAME)) {
      updateCheckEnabled = Boolean.parseBoolean(cli.getOptionValue(UPDATE_CHECK_OPTION_NAME));
    } else if (cli.hasOption(MOUNT_ROOT_UFS_OPTION_NAME)) {
      updateCheckEnabled = true;
    }
    return cli.hasOption(MOUNT_ROOT_UFS_OPTION_NAME)
        ? FuseOptions.create(conf, FileSystemOptions.create(conf,
        Optional.of(new UfsFileSystemOptions(cli.getOptionValue(MOUNT_ROOT_UFS_OPTION_NAME)))),
        updateCheckEnabled) : FuseOptions.create(conf, updateCheckEnabled);
  }

  private static void validateFuseConfAndOptions(AlluxioConfiguration conf, FuseOptions options) {
    String mountPoint = conf.getString(PropertyKey.FUSE_MOUNT_POINT);
    if (mountPoint.isEmpty()) {
      throw new InvalidArgumentRuntimeException(
          String.format("%s should be set and should not be empty",
              PropertyKey.FUSE_MOUNT_POINT.getName()));
    }
    if (!options.getFileSystemOptions().getUfsFileSystemOptions().isPresent()
        && conf.getString(PropertyKey.FUSE_MOUNT_ALLUXIO_PATH).isEmpty()) {
      throw new InvalidArgumentRuntimeException(
          String.format("%s should be set and should not be empty",
              PropertyKey.FUSE_MOUNT_ALLUXIO_PATH.getName()));
    }
    if (Files.isRegularFile(Paths.get(mountPoint))) {
      LOG.error("Mount point {} is not a directory but a file", mountPoint);
      throw new InvalidArgumentRuntimeException("Failed to launch fuse, mount point is a file");
    }
  }

  /**
   * Starts jvm monitor process, to monitor jvm.
   */
  private static void startJvmMonitorProcess() {
    if (Configuration.getBoolean(PropertyKey.STANDALONE_FUSE_JVM_MONITOR_ENABLED)) {
      JvmPauseMonitor jvmPauseMonitor = new JvmPauseMonitor(
          Configuration.getMs(PropertyKey.JVM_MONITOR_SLEEP_INTERVAL_MS),
          Configuration.getMs(PropertyKey.JVM_MONITOR_WARN_THRESHOLD_MS),
          Configuration.getMs(PropertyKey.JVM_MONITOR_INFO_THRESHOLD_MS));
      jvmPauseMonitor.start();
      MetricsSystem.registerGaugeIfAbsent(
          MetricsSystem.getMetricName(MetricKey.TOTAL_EXTRA_TIME.getName()),
          jvmPauseMonitor::getTotalExtraTime);
      MetricsSystem.registerGaugeIfAbsent(
          MetricsSystem.getMetricName(MetricKey.INFO_TIME_EXCEEDED.getName()),
          jvmPauseMonitor::getInfoTimeExceeded);
      MetricsSystem.registerGaugeIfAbsent(
          MetricsSystem.getMetricName(MetricKey.WARN_TIME_EXCEEDED.getName()),
          jvmPauseMonitor::getWarnTimeExceeded);
    }
  }
}
