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.cli.fsadmin.command;
  
  import alluxio.annotation.PublicApi;
  import alluxio.cli.CommandUtils;
  import alluxio.conf.AlluxioConfiguration;
  import alluxio.exception.status.AlluxioStatusException;
  import alluxio.exception.status.InvalidArgumentException;
  import alluxio.grpc.BackupPOptions;
  import alluxio.grpc.BackupPRequest;
  import alluxio.grpc.BackupState;
  import alluxio.wire.BackupStatus;
  
  import org.apache.commons.cli.CommandLine;
  import org.apache.commons.cli.Option;
  import org.apache.commons.cli.Options;
  import org.apache.commons.lang3.StringUtils;
  
  import java.io.IOException;
  import java.util.UUID;
  
  /**
   * Command for backing up Alluxio master metadata.
   */
  @PublicApi
  public class BackupCommand extends AbstractFsAdminCommand {
  
    private static final Option LOCAL_OPTION =
        Option.builder()
            .longOpt("local")
            .required(false)
            .hasArg(false)
            .desc("whether to write the backup to the master's local filesystem"
                + " instead of the root UFS")
            .build();
    private static final Option ALLOW_LEADER_OPTION =
        Option.builder()
            .longOpt("allow-leader")
            .required(false)
            .hasArg(false)
            .desc("whether to allow leader to take the backup when"
                + " backup delegation is enabled and HA cluster has no standby master.")
            .build();
    private static final Option BYPASS_DELEGATION_OPTION =
        Option.builder()
            .longOpt("bypass-delegation")
            .required(false)
            .hasArg(false)
            .desc("when specified, the leading master will by-pass backup delegation,"
                + " if it was enabled by configuration.")
            .build();
  
    /**
     * @param context fsadmin command context
     * @param alluxioConf Alluxio configuration
     */
    public BackupCommand(Context context, AlluxioConfiguration alluxioConf) {
      super(context);
    }
  
    @Override
    public String getCommandName() {
      return "backup";
    }
  
    @Override
    public Options getOptions() {
      return new Options()
          .addOption(LOCAL_OPTION)
          .addOption(ALLOW_LEADER_OPTION)
          .addOption(BYPASS_DELEGATION_OPTION);
    }
  
    @Override
    public int run(CommandLine cl) throws IOException {
      String[] args = cl.getArgs();
      BackupPRequest.Builder opts = BackupPRequest.newBuilder();
      if (args.length >= 1) {
        opts.setTargetDirectory(args[0]);
      }
      opts.setOptions(
          BackupPOptions.newBuilder()
              .setRunAsync(true)
              .setLocalFileSystem(cl.hasOption(LOCAL_OPTION.getLongOpt()))
              .setAllowLeader(cl.hasOption(ALLOW_LEADER_OPTION.getLongOpt()))
              .setBypassDelegation(cl.hasOption(BYPASS_DELEGATION_OPTION.getLongOpt())));
      // Take backup in async mode.
      BackupStatus status = mMetaClient.backup(opts.build());
      UUID backupId = status.getBackupId();
      do {
        clearProgressLine();
        // Backup could be after a fail-over.
        if (status.getState() == BackupState.None) {
          mPrintStream.printf("Backup lost. Please check Alluxio logs.%n");
          return -1;
        }
        if (status.getState() == BackupState.Completed) {
          break;
        } else if (status.getState() == BackupState.Failed) {
          throw AlluxioStatusException.fromAlluxioException(status.getError());
        } else {
          // Generate progress line that will be replaced until backup is finished.
          String progressMessage = String.format(" Backup state: %s", status.getState());
          // Start showing entry count once backup started running.
          if (status.getState() == BackupState.Running) {
            progressMessage += String.format(" | Entries processed: %d", status.getEntryCount());
          }
          mPrintStream.write(progressMessage.getBytes());
          mPrintStream.write("\r".getBytes());
          mPrintStream.flush();
        }
        // Sleep half a sec before querying status again.
        try {
          Thread.sleep(500);
          status = mMetaClient.getBackupStatus(backupId);
        } catch (InterruptedException ie) {
          throw new RuntimeException("Interrupted while waiting for backup completion.");
        } finally {
          // In case exception is thrown.
          clearProgressLine();
        }
      } while (true);
      clearProgressLine();
      // Print final state.
      mPrintStream.printf("Backup Host        : %s%n", status.getHostname());
      mPrintStream.printf("Backup URI         : %s%n", status.getBackupUri());
      mPrintStream.printf("Backup Entry Count : %d%n", status.getEntryCount());
      return 0;
    }
  
    private void clearProgressLine() throws IOException {
      final int progressLineLength = 75;
      // Clear progress line.
      mPrintStream.write(StringUtils.repeat(" ", progressLineLength).getBytes());
      mPrintStream.write("\r".getBytes());
    }
  
    @Override
    public String getUsage() {
      return "backup [directory] [--local] [--allow-leader] [--bypass-delegation]";
    }
  
    @Override
    public String getDescription() {
      return "backup backs up all Alluxio metadata to the backup directory configured on master. The"
          + " directory to back up to can be overridden by specifying a directory here. The directory"
          + " path is relative to the root UFS. To write the backup to the local disk of the primary"
          + " master, use --local and specify a filesystem path. Backing up metadata"
          + " will be delegated to standby masters in HA cluster. Use --allow-leader for"
          + " leader to take the backup when there are no standby masters.(This will pause"
          + " metadata changes during the backup). Use --bypass-delegation to take the backup on"
          + " the leader even if there are standby masters.";
    }
  
    @Override
    public void validateArgs(CommandLine cl) throws InvalidArgumentException {
      CommandUtils.checkNumOfArgsNoMoreThan(this, cl, 2);
    }
  }
  