/*
 * 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.metrics;

import alluxio.cli.CommandUtils;
import alluxio.cli.fs.FileSystemShellUtils;
import alluxio.cli.fsadmin.command.AbstractFsAdminCommand;
import alluxio.cli.fsadmin.command.Context;
import alluxio.client.block.stream.BlockWorkerClient;
import alluxio.client.file.FileSystemContext;
import alluxio.conf.AlluxioConfiguration;
import alluxio.exception.status.InvalidArgumentException;
import alluxio.grpc.ClearMetricsRequest;
import alluxio.resource.CloseableResource;
import alluxio.util.ThreadFactoryUtils;
import alluxio.wire.WorkerInfo;
import alluxio.wire.WorkerNetAddress;

import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * Clear the leading master (and workers) metrics.
 */
public final class ClearCommand extends AbstractFsAdminCommand {
  private static final String MASTER_OPTION_NAME = "master";
  private static final String WORKERS_OPTION_NAME = "workers";
  private static final String PARALLELISM_OPTION_NAME = "parallelism";
  private static final int DEFAULT_PARALLELISM = 8;

  private static final Option MASTER_OPTION =
      Option.builder()
          .longOpt(MASTER_OPTION_NAME)
          .required(false)
          .hasArg(false)
          .desc("Clear the metrics of Alluxio leading master")
          .build();
  private static final Option WORKERS_OPTION =
      Option.builder()
          .longOpt(WORKERS_OPTION_NAME)
          .required(false)
          .hasArg(true)
          .desc("Clear metrics of specified workers. "
              + "Pass in the worker hostnames separated by comma")
          .build();
  private static final Option PARALLELISM_OPTION =
      Option.builder()
          .longOpt(PARALLELISM_OPTION_NAME)
          .required(false)
          .hasArg(true)
          .argName("# concurrent operations")
          .desc("Number of concurrent worker metrics clear operations, "
              + "default: " + DEFAULT_PARALLELISM)
          .build();

  private final AlluxioConfiguration mAlluxioConf;

  /**
   * @param context fsadmin command context
   * @param alluxioConf Alluxio configuration
   */
  public ClearCommand(Context context, AlluxioConfiguration alluxioConf) {
    super(context);
    mAlluxioConf = alluxioConf;
  }

  @Override
  public String getCommandName() {
    return "clear";
  }

  @Override
  public Options getOptions() {
    return new Options().addOption(MASTER_OPTION)
        .addOption(WORKERS_OPTION)
        .addOption(PARALLELISM_OPTION);
  }

  @Override
  public void validateArgs(CommandLine cl) throws InvalidArgumentException {
    CommandUtils.checkNumOfArgsNoMoreThan(this, cl, 3);
  }

  @Override
  public int run(CommandLine cl) throws IOException {
    Option[] options = cl.getOptions();
    // No options or only --parallelism: clear master + workers metrics
    // with --master: clear master metrics
    // with --workers <hostnames>: clear worker metrics
    boolean clearWorkers = options.length == 0 || cl.hasOption(WORKERS_OPTION_NAME)
        || (options.length == 1 && cl.hasOption(PARALLELISM_OPTION_NAME));
    boolean clearMaster = options.length == 0 || cl.hasOption(MASTER_OPTION_NAME)
        || (options.length == 1 && cl.hasOption(PARALLELISM_OPTION_NAME));

    // Clear worker metrics
    if (clearWorkers) {
      int globalParallelism = FileSystemShellUtils
          .getIntArg(cl, PARALLELISM_OPTION, DEFAULT_PARALLELISM);
      try (FileSystemContext context = FileSystemContext.sFileSystemContextFactory
          .create(mAlluxioConf)) {
        List<WorkerNetAddress> addressList = context.getCachedWorkers().stream()
            .map(WorkerInfo::getAddress).collect(Collectors.toList());

        if (cl.hasOption(WORKERS_OPTION_NAME)) {
          String workersValue = cl.getOptionValue(WORKERS_OPTION_NAME);
          Set<String> workersRequired = new HashSet<>(Arrays.asList(workersValue.split(",")));
          List<WorkerNetAddress> workersToClear = new ArrayList<>();
          for (WorkerNetAddress worker : addressList) {
            if (workersRequired.contains(worker.getHost())) {
              workersToClear.add(worker);
              workersRequired.remove(worker.getHost());
            }
          }
          if (workersRequired.size() != 0) {
            System.out.printf("Cannot find workers of hostnames %s%n",
                String.join(",", workersRequired));
            System.out.printf("Valid workers include %s%n", addressListToString(addressList));
            return -1;
          }
          if (!clearWorkers(workersToClear, context, globalParallelism)) {
            System.out.printf("Failed to clear metrics of workers %s%n",
                addressListToString(workersToClear));
            return -1;
          }
        } else {
          if (!clearWorkers(addressList, context, globalParallelism)) {
            System.out.printf("Failed to clear metrics of workers %s%n",
                addressListToString(addressList));
            return -1;
          }
        }
      }
    }

    // Clear master metrics
    if (clearMaster) {
      // Clear worker metrics before master metrics since worker metrics report
      // may be flaky during metrics clearance
      try {
        mMetricsClient.clearMetrics();
        System.out.printf("Successfully cleared metrics of Alluxio leading master.%n");
      } catch (Exception e) {
        System.out.println("Fatal error: " + e);
        return -1;
      }
    }
    return 0;
  }

  /**
   * Clears the metrics of a list of workers.
   *
   * @param workers the workers to clear metrics of
   * @param context FileSystemContext
   * @param globalParallelism the global parallelism value
   * @return true if clear succeed, false otherwise
   */
  private boolean clearWorkers(List<WorkerNetAddress> workers,
      FileSystemContext context, int globalParallelism) throws IOException {
    int workerNum = workers.size();
    if (workerNum == 0) {
      System.out.println("No worker metrics to clear.");
      return true;
    } else if (workerNum == 1) {
      clearWorkerMetrics(workers.get(0), context);
    } else {
      List<Future<Void>> futures = new ArrayList<>();
      int parallelism = Math.min(workerNum, globalParallelism);
      ExecutorService service = Executors.newFixedThreadPool(parallelism,
          ThreadFactoryUtils.build("metrics-clear-cli-%d", true));
      for (WorkerNetAddress worker : workers) {
        futures.add(service.submit(new ClearCallable(worker, context)));
      }
      try {
        for (Future<Void> future : futures) {
          future.get();
        }
      } catch (ExecutionException e) {
        System.out.println("Fatal error: " + e);
        return false;
      } catch (InterruptedException e) {
        System.out.println("Metrics clearance interrupted, exiting.");
        return false;
      } finally {
        service.shutdownNow();
      }
    }
    return true;
  }

  /**
   * Get string value of worker address list.
   *
   * @param addressList the addressList to transform
   * @return string value of worker address list
   */
  private String addressListToString(List<WorkerNetAddress> addressList) {
    return Arrays.toString(addressList.stream()
        .map(WorkerNetAddress::getHost).toArray(String[]::new));
  }

  /**
   * Thread that clears the metrics of a specific worker.
   */
  private class ClearCallable implements Callable<Void> {
    private final WorkerNetAddress mWorker;
    private final FileSystemContext mContext;

    ClearCallable(WorkerNetAddress worker, FileSystemContext context) {
      mWorker = worker;
      mContext = context;
    }

    @Override
    public Void call() throws Exception {
      clearWorkerMetrics(mWorker, mContext);
      return null;
    }
  }

  /**
   * Clears the worker metrics.
   *
   * @param worker the worker to clear metrics of
   * @param context the file system context
   */
  private void clearWorkerMetrics(WorkerNetAddress worker,
      FileSystemContext context) throws IOException {

    try (CloseableResource<BlockWorkerClient> blockWorkerClient =
             context.acquireBlockWorkerClient(worker)) {
      blockWorkerClient.get().clearMetrics(ClearMetricsRequest.newBuilder().build());
    }
    System.out.printf("Successfully cleared metrics of worker %s.%n", worker.getHost());
  }

  @Override
  public String getUsage() {
    return String.format("%s [--%s] [--%s <worker_hostnames>] [--%s <#>] %n"
            + "\t--%s: %s%n"
            + "\t--%s: %s%n"
            + "\t--%s: %s%n",
        getCommandName(), MASTER_OPTION_NAME, WORKERS_OPTION_NAME,
        PARALLELISM_OPTION_NAME, MASTER_OPTION_NAME, MASTER_OPTION.getDescription(),
        WORKERS_OPTION_NAME, WORKERS_OPTION.getDescription(),
        PARALLELISM_OPTION_NAME, PARALLELISM_OPTION.getDescription());
  }

  /**
   * @return command's description
   */
  @VisibleForTesting
  public static String description() {
    return "Clear the metrics of the whole cluster by default. "
        + "Users can pass in options to decide metrics of which nodes to be cleared. "
        + "This command is useful when getting metrics information in short-term testing. "
        + "This command should be used sparingly as it may affect the current metrics "
        + "recording and reporting which may lead to metrics incorrectness "
        + "and affect worker/client heartbeats with leading master.";
  }

  @Override
  public String getDescription() {
    return description();
  }
}
