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

import alluxio.AlluxioTestDirectory;
import alluxio.AlluxioURI;
import alluxio.ClientContext;
import alluxio.ConfigurationRule;
import alluxio.ConfigurationTestUtils;
import alluxio.Constants;
import alluxio.cli.Format;
import alluxio.client.block.RetryHandlingBlockMasterClient;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystem.Factory;
import alluxio.client.file.FileSystemContext;
import alluxio.client.file.RetryHandlingFileSystemMasterClient;
import alluxio.client.journal.JournalMasterClient;
import alluxio.client.journal.RetryHandlingJournalMasterClient;
import alluxio.client.meta.MetaMasterClient;
import alluxio.client.meta.RetryHandlingMetaMasterClient;
import alluxio.client.metrics.MetricsMasterClient;
import alluxio.client.metrics.RetryHandlingMetricsMasterClient;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.conf.Source;
import alluxio.exception.status.UnavailableException;
import alluxio.grpc.MasterInfo;
import alluxio.grpc.ServiceType;
import alluxio.master.LocalAlluxioCluster;
import alluxio.master.MasterClientContext;
import alluxio.master.MasterInquireClient;
import alluxio.master.PollingMasterInquireClient;
import alluxio.master.SingleMasterInquireClient;
import alluxio.master.ZkMasterInquireClient;
import alluxio.master.journal.JournalType;
import alluxio.membership.MembershipType;
import alluxio.multi.process.PortCoordination.ReservedPort;
import alluxio.security.user.ServerUserState;
import alluxio.util.CommonUtils;
import alluxio.util.WaitForOptions;
import alluxio.util.io.PathUtils;
import alluxio.util.network.NetworkAddressUtils;

import com.google.common.base.Preconditions;
import com.google.common.io.Closer;
import net.bytebuddy.utility.RandomString;
import org.apache.commons.io.Charsets;
import org.apache.commons.io.FileUtils;
import org.apache.curator.test.TestingServer;
import org.apache.log4j.Level;
import org.apache.log4j.LogManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.lang.ProcessBuilder.Redirect;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
import javax.security.auth.Subject;

/**
 * Class for starting, stopping, and interacting with an Alluxio cluster where each master and
 * worker runs in its own process.
 *
 * Compared to {@link LocalAlluxioCluster}, {@link MultiProcessCluster} is
 *   - Slower
 *   - Black box testing only (No access to master/worker internals)
 *   - Destructible (You can kill -9 masters/workers)
 *   - More realistic of real deployments
 *
 * Due to the slower speed, [@link LocalAlluxioCluster} should generally be preferred.
 * {@link MultiProcessCluster} is primarily for tests which want to stop or restart servers.
 *
 * The synchronization strategy for this class is to synchronize all public methods.
 */
@ThreadSafe
public final class MultiProcessCluster {
  public static final int PORTS_PER_MASTER = 3;
  public static final int PORTS_PER_WORKER = 5;
  private static final int MASTER_START_DELAY_MS = 5000; // in ms

  private static final Logger LOG = LoggerFactory.getLogger(MultiProcessCluster.class);
  private static final File ARTIFACTS_DIR = new File(Constants.TEST_ARTIFACTS_DIR);
  private static final File TESTS_LOG = new File(Constants.TESTS_LOG);
  private static final int WAIT_MASTER_SERVING_TIMEOUT_MS = 30000;

  private final Map<PropertyKey, Object> mProperties;
  private final Map<Integer, Map<PropertyKey, String>> mMasterProperties;
  private final Map<Integer, Map<PropertyKey, String>> mWorkerProperties;
  private int mNumMasters;
  private final int mNumWorkers;
  private final String mClusterName;
  /** Closer for closing all resources that must be closed when the cluster is destroyed. */
  private final Closer mCloser;
  private final List<Master> mMasters;
  private final List<Worker> mWorkers;
  private final List<ReservedPort> mPorts;
  private final boolean mNoFormat;

  private DeployMode mDeployMode;

  /** Base directory for storing configuration and logs. */
  private File mWorkDir;
  /** Addresses of all masters. Should have the same size as {@link #mMasters}. */
  private List<MasterNetAddress> mMasterAddresses;
  /** Unique IDs for all masters. Should have the same size as {@link #mMasters} */
  private List<String> mMasterIds = new ArrayList<>();
  private State mState;
  private TestingServer mCuratorServer;
  private FileSystemContext mFilesystemContext;
  /**
   * Tracks whether the test has succeeded. If mSuccess is never updated before {@link #destroy()},
   * the state of the cluster will be saved as a tarball in the artifacts directory.
   */
  private boolean mSuccess;

  private MultiProcessCluster(Map<PropertyKey, Object> properties,
      Map<Integer, Map<PropertyKey, String>> masterProperties,
      Map<Integer, Map<PropertyKey, String>> workerProperties, int numMasters, int numWorkers,
      String clusterName, boolean noFormat,
      List<PortCoordination.ReservedPort> ports) {
    LogManager.getLogger(MultiProcessCluster.class).setLevel(Level.DEBUG);

    Preconditions.checkState(
        ports.size() >= numMasters * PORTS_PER_MASTER + numWorkers * PORTS_PER_WORKER,
        "We require %s ports per master and %s ports per worker, but there are %s masters, "
            + "%s workers, and %s ports",
        PORTS_PER_MASTER, PORTS_PER_WORKER, numMasters, numWorkers, ports.size());
    mProperties = properties;
    mMasterProperties = masterProperties;
    mWorkerProperties = workerProperties;
    mNumMasters = numMasters;
    mNumWorkers = numWorkers;
    // Add a unique number so that different runs of the same test use different cluster names.
    mClusterName = clusterName + "-" + Math.abs(ThreadLocalRandom.current().nextInt());
    mNoFormat = noFormat;
    mMasters = new ArrayList<>();
    mWorkers = new ArrayList<>();
    mPorts = new ArrayList<>(ports);
    mCloser = Closer.create();
    mState = State.NOT_STARTED;
    mSuccess = false;

    JournalType journalType = (JournalType) mProperties.getOrDefault(
        PropertyKey.MASTER_JOURNAL_TYPE,
        Configuration.getEnum(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.class));
    mDeployMode = journalType == JournalType.EMBEDDED ? DeployMode.EMBEDDED
        : numMasters > 1 ? DeployMode.ZOOKEEPER_HA : DeployMode.UFS_NON_HA;
  }

  /**
   * Starts the cluster, launching all server processes.
   */
  public synchronized void start() throws Exception {
    Preconditions.checkState(mState != State.STARTED, "Cannot start while already started");
    Preconditions.checkState(mState != State.DESTROYED, "Cannot start a destroyed cluster");
    mWorkDir = AlluxioTestDirectory.createTemporaryDirectory(mClusterName);
    mState = State.STARTED;
    // Start servers
    LOG.info("Starting alluxio cluster {} with base directory {}", mClusterName,
        mWorkDir.getAbsolutePath());
    startNewMasters(mNumMasters, !mNoFormat);

    for (int i = 0; i < mNumWorkers; i++) {
      createWorker(i).start();
    }
    LOG.info("Starting alluxio cluster in directory {}", mWorkDir.getAbsolutePath());
    int primaryMasterIndex = getPrimaryMasterIndex(WAIT_MASTER_SERVING_TIMEOUT_MS);
    LOG.info("Alluxio primary master {} starts serving RPCs",
        mMasterAddresses.get(primaryMasterIndex));
  }

  /**
   * Start a number of new master nodes.
   *
   * @param count the count of the masters to start
   * @param format whether to format the master first
   * @throws Exception if any error occurs
   */
  public synchronized void startNewMasters(int count, boolean format) throws Exception {
    int startIndex = 0;
    if (mMasterAddresses != null) {
      startIndex = mMasterAddresses.size();
    } else {
      mMasterAddresses = new ArrayList<>();
    }
    List<MasterNetAddress> masterAddresses = generateMasterAddresses(count);
    for (MasterNetAddress newMasterAddress : masterAddresses) {
      String id = newMasterAddress.getEmbeddedJournalPort()
          + "-" + RandomString.make(RandomString.DEFAULT_LENGTH);
      mMasterIds.add(id);
    }
    mMasterAddresses.addAll(masterAddresses);
    mNumMasters = mMasterAddresses.size();

    LOG.info("Master addresses: {}", mMasterAddresses);
    switch (mDeployMode) {
      case UFS_NON_HA:
        MasterNetAddress masterAddress = mMasterAddresses.get(0);
        mProperties.put(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS);
        mProperties.put(PropertyKey.MASTER_HOSTNAME, masterAddress.getHostname());
        mProperties.put(PropertyKey.MASTER_RPC_PORT, masterAddress.getRpcPort());
        mProperties.put(PropertyKey.MASTER_WEB_PORT, masterAddress.getWebPort());
        break;
      case EMBEDDED:
        List<String> journalAddresses = new ArrayList<>();
        List<String> rpcAddresses = new ArrayList<>();
        for (MasterNetAddress address : mMasterAddresses) {
          journalAddresses
              .add(String.format("%s:%d", address.getHostname(), address.getEmbeddedJournalPort()));
          rpcAddresses.add(String.format("%s:%d", address.getHostname(), address.getRpcPort()));
        }
        mProperties.put(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.EMBEDDED);
        mProperties.put(PropertyKey.MASTER_EMBEDDED_JOURNAL_ADDRESSES,
            com.google.common.base.Joiner.on(",").join(journalAddresses));
        mProperties.put(PropertyKey.MASTER_RPC_ADDRESSES,
            com.google.common.base.Joiner.on(",").join(rpcAddresses));
        break;
      case ZOOKEEPER_HA:
        mCuratorServer = mCloser.register(
                new TestingServer(-1, AlluxioTestDirectory.createTemporaryDirectory("zk")));
        mProperties.put(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS);
        mProperties.put(PropertyKey.ZOOKEEPER_ENABLED, true);
        mProperties.put(PropertyKey.ZOOKEEPER_ADDRESS, mCuratorServer.getConnectString());
        break;
      default:
        throw new IllegalStateException("Unknown deploy mode: " + mDeployMode);
    }

    for (Entry<PropertyKey, Object> entry :
        ConfigurationTestUtils.testConfigurationDefaults(Configuration.global(),
        NetworkAddressUtils.getLocalHostName(
            (int) Configuration.getMs(PropertyKey.NETWORK_HOST_RESOLUTION_TIMEOUT_MS)),
            mWorkDir.getAbsolutePath()).entrySet()) {
      // Don't overwrite explicitly set properties.
      if (mProperties.containsKey(entry.getKey())) {
        continue;
      }
      // Keep the default RPC timeout.
      if (entry.getKey().equals(PropertyKey.USER_RPC_RETRY_MAX_DURATION)) {
        continue;
      }
      mProperties.put(entry.getKey(), entry.getValue());
    }
    if (!mProperties.containsKey(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS)) {
      mProperties.put(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS,
          PathUtils.concatPath(mWorkDir, "underFSStorage"));
      new File((String) mProperties.get(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS)).mkdirs();
    }

    mProperties.put(PropertyKey.DORA_CLIENT_UFS_ROOT,
        PathUtils.concatPath(mWorkDir, "underFSStorage"));
    mProperties.put(PropertyKey.MASTER_WORKER_REGISTER_LEASE_ENABLED, false);
    mProperties.put(PropertyKey.USER_NETTY_DATA_TRANSMISSION_ENABLED, true);

    if (format) {
      formatJournal();
    }
    writeConf();
    Configuration.merge(mProperties, Source.RUNTIME);

    for (int i = 0; i < count; i++) {
      createMaster(startIndex + i).start();
      wait(MASTER_START_DELAY_MS);
    }
    mFilesystemContext = null;
  }

  /**
   * Kills the primary master.
   *
   * If no master is currently primary, this method blocks until a primary has been elected, then
   * kills it.
   *
   * @param timeoutMs maximum amount of time to wait, in milliseconds
   * @return the ID of the killed master
   */
  public synchronized int waitForAndKillPrimaryMaster(int timeoutMs)
      throws TimeoutException, InterruptedException {
    int index = getPrimaryMasterIndex(timeoutMs);
    mMasters.get(index).close();
    return index;
  }

  /**
   * Gets the index of the primary master.
   *
   * @param timeoutMs maximum amount of time to wait, in milliseconds
   * @return the index of the primary master
   */
  public synchronized int getPrimaryMasterIndex(int timeoutMs)
      throws TimeoutException, InterruptedException {
    final FileSystem fs = getFileSystemClient();
    final MasterInquireClient inquireClient = getMasterInquireClient();
    CommonUtils.waitFor("a primary master to be serving", () -> {
      try {
        // Make sure the leader is serving.
        fs.getStatus(new AlluxioURI("/"));
        return true;
      } catch (Exception e) {
        LOG.error("Failed to get status of root directory:", e);
        return false;
      }
    }, WaitForOptions.defaults().setTimeoutMs(timeoutMs));
    int primaryRpcPort;
    try {
      primaryRpcPort = inquireClient.getPrimaryRpcAddress().getPort();
    } catch (UnavailableException e) {
      throw new RuntimeException(e);
    }
    // Returns the master whose RPC port matches the primary RPC port.
    for (int i = 0; i < mMasterAddresses.size(); i++) {
      if (mMasterAddresses.get(i).getRpcPort() == primaryRpcPort) {
        return i;
      }
    }
    throw new RuntimeException(
        String.format("No master found with RPC port %d. Master addresses: %s", primaryRpcPort,
            mMasterAddresses));
  }

  /**
   * Waits for all nodes to be registered.
   *
   * @param timeoutMs maximum amount of time to wait, in milliseconds
   */
  public synchronized void waitForAllNodesRegistered(int timeoutMs)
      throws TimeoutException, InterruptedException {
    MetaMasterClient metaMasterClient = getMetaMasterClient();
    int nodeCount = mNumMasters + mNumWorkers;
    CommonUtils.waitFor("all nodes registered", () -> {
      try {
        MasterInfo masterInfo = metaMasterClient.getMasterInfo(Collections.emptySet());
        int liveNodeNum = masterInfo.getMasterAddressesList().size()
            + masterInfo.getWorkerAddressesList().size();
        if (liveNodeNum == nodeCount) {
          return true;
        } else {
          LOG.info("Master addresses: {}. Worker addresses: {}",
              masterInfo.getMasterAddressesList(), masterInfo.getWorkerAddressesList());
          return false;
        }
      } catch (UnavailableException e) {
        return false;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, WaitForOptions.defaults().setInterval(200).setTimeoutMs(timeoutMs));
  }

  /**
   * @return the deploy mode
   */
  public synchronized DeployMode getDeployMode() {
    return mDeployMode;
  }

  /**
   * @return the {@link FileSystemContext} which can be used to access the cluster
   */
  public synchronized FileSystemContext getFilesystemContext() {
    if (mFilesystemContext == null) {
      mFilesystemContext = FileSystemContext.create(new Subject(), getMasterInquireClient(),
          Configuration.global());
      mCloser.register(mFilesystemContext);
    }
    return mFilesystemContext;
  }

  /**
   * @return a client for interacting with the cluster
   */
  public synchronized FileSystem getFileSystemClient() {
    Preconditions.checkState(mState == State.STARTED,
        "must be in the started state to create an fs client, but state was %s", mState);

    return Factory.create(getFilesystemContext());
  }

  /**
   * @return a meta master client
   */
  public synchronized MetaMasterClient getMetaMasterClient() {
    Preconditions.checkState(mState == State.STARTED,
        "must be in the started state to create a meta master client, but state was %s", mState);
    return new RetryHandlingMetaMasterClient(MasterClientContext
        .newBuilder(ClientContext.create(Configuration.global()))
        .setMasterInquireClient(getMasterInquireClient())
        .build());
  }

  /**
   * @return a metrics master client
   */
  public synchronized MetricsMasterClient getMetricsMasterClient() {
    Preconditions.checkState(mState == State.STARTED,
        "must be in the started state to create a metrics master client, but state was %s", mState);
    return new RetryHandlingMetricsMasterClient(MasterClientContext
        .newBuilder(ClientContext.create(Configuration.global()))
        .setMasterInquireClient(getMasterInquireClient())
        .build());
  }

  /**
   * @return a journal master client for alluxio master
   */
  public synchronized JournalMasterClient getJournalMasterClientForMaster() {
    Preconditions.checkState(mState == State.STARTED,
        "must be in the started state to create a journal master client, but state was %s", mState);

    return new RetryHandlingJournalMasterClient(
        MasterClientContext.newBuilder(ClientContext.create(Configuration.global()))
            .setMasterInquireClient(getMasterInquireClient()).build());
  }

  /**
   * @return clients for communicating with the cluster
   */
  public synchronized Clients getClients() {
    Preconditions.checkState(mState == State.STARTED,
        "must be in the started state to create a meta master client, but state was %s", mState);
    MasterClientContext config = MasterClientContext
        .newBuilder(ClientContext.create(Configuration.global()))
        .setMasterInquireClient(getMasterInquireClient()).build();
    return new Clients(getFileSystemClient(),
        new RetryHandlingFileSystemMasterClient(config),
        new RetryHandlingMetaMasterClient(config),
        new RetryHandlingBlockMasterClient(config));
  }

  /**
   * Informs the cluster that the test succeeded. If this method is never called, the cluster will
   * save a copy of its state during teardown.
   */
  public synchronized void notifySuccess() {
    mSuccess = true;
  }

  /**
   * Copies the work directory to the artifacts folder.
   */
  public synchronized void saveWorkdir() throws IOException {
    Preconditions.checkState(mState == State.STARTED,
        "cluster must be started before you can save its work directory");
    ARTIFACTS_DIR.mkdirs();

    File tarball = new File(mWorkDir.getParentFile(), mWorkDir.getName() + ".tar.gz");
    // Tar up the work directory.
    ProcessBuilder pb =
        new ProcessBuilder("tar", "-czf", tarball.getName(), mWorkDir.getName());
    pb.directory(mWorkDir.getParentFile());
    pb.redirectOutput(Redirect.appendTo(TESTS_LOG));
    pb.redirectError(Redirect.appendTo(TESTS_LOG));
    Process p = pb.start();
    try {
      p.waitFor();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new RuntimeException(e);
    }
    // Move tarball to artifacts directory.
    File finalTarball = new File(ARTIFACTS_DIR, tarball.getName());
    FileUtils.moveFile(tarball, finalTarball);
    LOG.info("Saved cluster {} to {}", mClusterName, finalTarball.getAbsolutePath());
  }

  /**
   * Destroys the cluster. It may not be re-started after being destroyed.
   */
  public synchronized void destroy() throws IOException {
    if (mState == State.DESTROYED) {
      return;
    }
    if (!mSuccess) {
      saveWorkdir();
    }
    mCloser.close();
    Configuration.reloadProperties();
    LOG.info("Destroyed cluster {}", mClusterName);
    mState = State.DESTROYED;
  }

  /**
   * Starts all masters.
   */
  public synchronized void startMasters() {
    mMasters.forEach(master -> master.start());
  }

  /**
   * Starts the specified master.
   *
   * @param i the index of the master to start
   */
  public synchronized void startMaster(int i) throws IOException {
    Preconditions.checkState(mState == State.STARTED,
        "Must be in a started state to start masters");
    mMasters.get(i).start();
  }

  /**
   * Starts the specified worker.
   *
   * @param i the index of the worker to start
   */
  public synchronized void startWorker(int i) throws IOException {
    Preconditions.checkState(mState == State.STARTED,
        "Must be in a started state to start workers");
    mWorkers.get(i).start();
  }

  /**
   * Stops all masters.
   */
  public synchronized void stopMasters() {
    mMasters.forEach(master -> master.close());
  }

  /**
   * @param i the index of the master to stop
   */
  public synchronized void stopMaster(int i) throws IOException {
    mMasters.get(i).close();
  }

  /**
   * Removes master i from the cluster.
   *
   * @param i index of the master to be removed
   * @throws IOException
   */
  public synchronized void stopAndRemoveMaster(int i) throws IOException {
    stopMaster(i);
    mMasterAddresses.remove(i);
    mMasterIds.remove(i);
    mMasters.remove(i);
    mMasterProperties.remove(i);
    mNumMasters--;
    switch (mDeployMode) {
      case EMBEDDED:
        String journalAddresses = mMasterAddresses.stream()
            .map(addr -> String.format("%s:%d", addr.getHostname(), addr.getEmbeddedJournalPort()))
            .collect(Collectors.joining(","));
        String rpcAddresses = mMasterAddresses.stream()
            .map(addr -> String.format("%s:%d", addr.getHostname(), addr.getRpcPort()))
            .collect(Collectors.joining(","));
        mProperties.put(PropertyKey.MASTER_EMBEDDED_JOURNAL_ADDRESSES, journalAddresses);
        mProperties.put(PropertyKey.MASTER_RPC_ADDRESSES, rpcAddresses);
        Configuration.set(PropertyKey.MASTER_EMBEDDED_JOURNAL_ADDRESSES, journalAddresses);
        Configuration.set(PropertyKey.MASTER_RPC_ADDRESSES, rpcAddresses);
        break;
      case ZOOKEEPER_HA: // zk will take care of fault tolerance
        break;
      default: // UFS_NON_HA: can't remove the only master in the cluster
        throw new IllegalStateException("Unimplemented deploy mode: " + mDeployMode);
    }
    // reset file system context after cluster size shrinks
    mFilesystemContext = null;
  }

  /**
   * Updates master configuration for all masters. This will take effect on a master the next time
   * the master is started.
   *
   * @param key the key to update
   * @param value the value to set, or null to unset the key
   */
  public synchronized void updateMasterConf(PropertyKey key, @Nullable String value) {
    mMasters.forEach(master -> master.updateConf(key, value));
  }

  /**
   * Updates internal master list with an external address.
   *
   * This API is to provide support when this cluster is joined by external masters.
   * Calling this API with an external master address will make this cluster aware of it
   * when creating clients/contexts etc.
   *
   * @param externalMasterAddress external master address
   */
  public synchronized void addExternalMasterAddress(MasterNetAddress externalMasterAddress) {
    mMasterAddresses.add(externalMasterAddress);
    // Reset current context to cause reinstantiation with new addresses.
    mFilesystemContext = null;
  }

  /**
   * Updates the cluster's deploy mode.
   *
   * @param mode the mode to set
   */
  public synchronized void updateDeployMode(DeployMode mode) {
    mDeployMode = mode;
    if (mDeployMode == DeployMode.EMBEDDED) {
      // Ensure that the journal properties are set correctly.
      for (int i = 0; i < mMasters.size(); i++) {
        Master master = mMasters.get(i);
        MasterNetAddress address = mMasterAddresses.get(i);
        master.updateConf(PropertyKey.MASTER_EMBEDDED_JOURNAL_PORT,
            address.getEmbeddedJournalPort());

        File journalDir = new File(mWorkDir, "journal" + i);
        journalDir.mkdirs();
        master.updateConf(PropertyKey.MASTER_JOURNAL_FOLDER, journalDir.getAbsolutePath());
      }
    }
  }

  /**
   * @param i the index of the worker to stop
   */
  public synchronized void stopWorker(int i) throws IOException {
    mWorkers.get(i).close();
  }

  /**
   * @return the journal directory
   */
  public synchronized String getJournalDir() {
    return (String) mProperties.get(PropertyKey.MASTER_JOURNAL_FOLDER);
  }

  /**
   * @param masterId index of the master
   * @return the journal directory for the specified master
   */
  public String getJournalDir(int masterId) {
    return (String) mMasters.get(masterId).getConf().get(PropertyKey.MASTER_JOURNAL_FOLDER);
  }

  /**
   * @return return the list of master addresses
   */
  public synchronized List<MasterNetAddress> getMasterAddresses() {
    return mMasterAddresses;
  }

  /**
   * Stops the Zookeeper cluster.
   */
  public synchronized void stopZk() throws IOException {
    mCuratorServer.stop();
  }

  /**
   * Restarts the Zookeeper cluster.
   */
  public synchronized void restartZk() throws Exception {
    Preconditions.checkNotNull(mCuratorServer, "mCuratorServer");
    mCuratorServer.restart();
  }

  /**
   * Creates the specified master without starting it.
   *
   * @param i the index of the master to create
   */
  private synchronized Master createMaster(int i) throws IOException {
    Preconditions.checkState(mState == State.STARTED,
        "Must be in a started state to create masters");
    MasterNetAddress address = mMasterAddresses.get(i);
    String extension = "-" + mMasterIds.get(i);
    File confDir = new File(mWorkDir, "conf-master" + extension);
    File metastoreDir = new File(mWorkDir, "metastore-master" + extension);
    File logsDir = new File(mWorkDir, "logs-master" + extension);
    logsDir.mkdirs();
    Map<PropertyKey, Object> conf = new HashMap<>();
    conf.put(PropertyKey.LOGGER_TYPE, "MASTER_LOGGER");
    conf.put(PropertyKey.CONF_DIR, confDir.getAbsolutePath());
    conf.put(PropertyKey.MASTER_METASTORE_DIR, metastoreDir.getAbsolutePath());
    conf.put(PropertyKey.LOGS_DIR, logsDir.getAbsolutePath());
    conf.put(PropertyKey.MASTER_HOSTNAME, address.getHostname());
    conf.put(PropertyKey.MASTER_RPC_PORT, address.getRpcPort());
    conf.put(PropertyKey.MASTER_WEB_PORT, address.getWebPort());
    conf.put(PropertyKey.MASTER_EMBEDDED_JOURNAL_PORT,
        address.getEmbeddedJournalPort());
    if (mDeployMode.equals(DeployMode.EMBEDDED)) {
      File journalDir = new File(mWorkDir, "journal" + extension);
      journalDir.mkdirs();
      conf.put(PropertyKey.MASTER_JOURNAL_FOLDER, journalDir.getAbsolutePath());
    }
    Master master = mCloser.register(new Master(logsDir, conf));
    mMasters.add(master);
    return master;
  }

  /**
   * Creates the specified worker without starting it.
   *
   * @param i the index of the worker to create
   */
  private synchronized Worker createWorker(int i) throws IOException {
    Preconditions.checkState(mState == State.STARTED,
        "Must be in a started state to create workers");
    File confDir = new File(mWorkDir, "conf-worker" + i);
    File logsDir = new File(mWorkDir, "logs-worker" + i);
    File ramdisk = new File(mWorkDir, "ramdisk" + i);
    File workerMetaStoreDir = new File(mWorkDir, "metastore-worker-" + i);

    logsDir.mkdirs();
    ramdisk.mkdirs();
    workerMetaStoreDir.mkdirs();
    int rpcPort = getNewPort();
    int dataPort = getNewPort();
    int webPort = getNewPort();
    int httpPort = getNewPort();
    int restPort = getNewPort();

    Map<PropertyKey, Object> conf = new HashMap<>();
    conf.put(PropertyKey.LOGGER_TYPE, "WORKER_LOGGER");
    conf.put(PropertyKey.CONF_DIR, confDir.getAbsolutePath());
    conf.put(PropertyKey.Template.WORKER_TIERED_STORE_LEVEL_DIRS_PATH.format(0),
        ramdisk.getAbsolutePath());
    conf.put(PropertyKey.LOGS_DIR, logsDir.getAbsolutePath());
    conf.put(PropertyKey.WORKER_RPC_PORT, rpcPort);
    conf.put(PropertyKey.WORKER_WEB_PORT, webPort);
    conf.put(PropertyKey.WORKER_DATA_PORT, dataPort);
    conf.put(PropertyKey.WORKER_HTTP_SERVER_PORT, httpPort);
    conf.put(PropertyKey.WORKER_REST_PORT, restPort);
    conf.put(PropertyKey.DORA_WORKER_METASTORE_ROCKSDB_DIR, workerMetaStoreDir.getAbsolutePath());

    conf.put(PropertyKey.DORA_CLIENT_UFS_ROOT, PathUtils.concatPath(mWorkDir, "underFSStorage"));
    conf.put(PropertyKey.MASTER_WORKER_REGISTER_LEASE_ENABLED, false);
    conf.put(PropertyKey.USER_NETTY_DATA_TRANSMISSION_ENABLED, true);

    Configuration.set(PropertyKey.WORKER_MEMBERSHIP_MANAGER_TYPE, MembershipType.MASTER);

    Worker worker = mCloser.register(new Worker(logsDir, conf));
    mWorkers.add(worker);
    LOG.info("Created worker with (rpc, data, web) ports ({}, {}, {})", rpcPort, dataPort,
        webPort);
    return worker;
  }

  /**
   * Formats the cluster journal.
   */
  public synchronized void formatJournal() throws IOException {
    if (mDeployMode == DeployMode.EMBEDDED) {
      for (Master master : mMasters) {
        File journalDir = new File((String) master.getConf()
                .get(PropertyKey.MASTER_JOURNAL_FOLDER));
        FileUtils.deleteDirectory(journalDir);
        journalDir.mkdirs();
      }
      return;
    }
    try (Closeable c = new ConfigurationRule(mProperties, Configuration.modifiableGlobal())
        .toResource()) {
      Format.format(Format.Mode.MASTER, Configuration.global());
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * @return a client for determining the serving master
   */
  public synchronized MasterInquireClient getMasterInquireClient() {
    switch (mDeployMode) {
      case UFS_NON_HA:
        Preconditions.checkState(mMasters.size() == 1,
            "Running with multiple masters requires Zookeeper or Embedded Journal");
        return new SingleMasterInquireClient(InetSocketAddress.createUnresolved(
            mMasterAddresses.get(0).getHostname(), mMasterAddresses.get(0).getRpcPort()));
      case EMBEDDED:
        if (mMasterAddresses.size() > 1) {
          List<InetSocketAddress> addresses = new ArrayList<>(mMasterAddresses.size());
          for (MasterNetAddress address : mMasterAddresses) {
            addresses.add(
                InetSocketAddress.createUnresolved(address.getHostname(), address.getRpcPort()));
          }
          return new PollingMasterInquireClient(addresses, Configuration.global(),
              ServerUserState.global(), ServiceType.META_MASTER_CLIENT_SERVICE);
        } else {
          return new SingleMasterInquireClient(InetSocketAddress.createUnresolved(
              mMasterAddresses.get(0).getHostname(), mMasterAddresses.get(0).getRpcPort()));
        }
      case ZOOKEEPER_HA:
        return ZkMasterInquireClient.getClient(mCuratorServer.getConnectString(),
            Configuration.getString(PropertyKey.ZOOKEEPER_ELECTION_PATH),
            Configuration.getString(PropertyKey.ZOOKEEPER_LEADER_PATH),
            Configuration.getInt(PropertyKey.ZOOKEEPER_LEADER_INQUIRY_RETRY_COUNT),
            Configuration.getBoolean(PropertyKey.ZOOKEEPER_AUTH_ENABLED));
      default:
        throw new IllegalStateException("Unknown deploy mode: " + mDeployMode);
    }
  }

  /**
   * Writes the contents of properties to the configuration file.
   */
  private void writeConf() throws IOException {
    for (int i = 0; i < mNumMasters; i++) {
      String extension = "-" + mMasterIds.get(i);
      File confDir = new File(mWorkDir, "conf-master" + extension);
      writeConfToFile(confDir, mMasterProperties.getOrDefault(i, new HashMap<>()));
    }
    for (int i = 0; i < mNumWorkers; i++) {
      File confDir = new File(mWorkDir, "conf-worker" + i);
      writeConfToFile(confDir, mWorkerProperties.getOrDefault(i, new HashMap<>()));
    }
  }

  private int getNewPort() throws IOException {
    Preconditions.checkState(!mPorts.isEmpty(), "Out of ports to reserve");
    return mPorts.remove(mPorts.size() - 1).getPort();
  }

  /**
   * Creates the conf directory and file.
   * Writes the properties to the generated file.
   *
   * @param dir the conf directory to create
   * @param properties the specific properties of the current node
   */
  private void writeConfToFile(File dir, Map<PropertyKey, String> properties) throws IOException {
    // Generates the full set of properties to write
    Map<PropertyKey, Object> map = new HashMap<>(mProperties);
    for (Map.Entry<PropertyKey, String> entry : properties.entrySet()) {
      map.put(entry.getKey(), entry.getValue());
    }

    StringBuilder sb = new StringBuilder();
    for (Entry<PropertyKey, Object> entry : map.entrySet()) {
      sb.append(String.format("%s=%s%n", entry.getKey(), entry.getValue()));
    }

    dir.mkdirs();
    try (FileOutputStream fos
        = new FileOutputStream(new File(dir, "alluxio-site.properties"))) {
      fos.write(sb.toString().getBytes(Charsets.UTF_8));
    }
  }

  private List<MasterNetAddress> generateMasterAddresses(int numMasters) throws IOException {
    List<MasterNetAddress> addrs = new ArrayList<>();
    for (int i = 0; i < numMasters; i++) {
      addrs.add(new MasterNetAddress(NetworkAddressUtils
          .getLocalHostName((int) Configuration
              .getMs(PropertyKey.NETWORK_HOST_RESOLUTION_TIMEOUT_MS)),
          getNewPort(), getNewPort(), getNewPort()));
    }
    return addrs;
  }

  private enum State {
    NOT_STARTED, STARTED, DESTROYED;
  }

  /**
   * Deploy mode for the cluster.
   */
  public enum DeployMode {
    EMBEDDED,
    UFS_NON_HA, ZOOKEEPER_HA
  }

  /**
   * Builder for {@link MultiProcessCluster}.
   */
  public static final class Builder {
    private final List<ReservedPort> mReservedPorts;

    private Map<PropertyKey, Object> mProperties = new HashMap<>();
    private Map<Integer, Map<PropertyKey, String>> mMasterProperties = new HashMap<>();
    private Map<Integer, Map<PropertyKey, String>> mWorkerProperties = new HashMap<>();
    private int mNumMasters = 1;
    private int mNumWorkers = 1;
    private String mClusterName = "AlluxioMiniCluster";
    private boolean mNoFormat = false;

    // Should only be instantiated by newBuilder().
    private Builder(List<ReservedPort> reservedPorts) {
      mReservedPorts = reservedPorts;
    }

    /**
     * @param key the property key to set
     * @param value the value to set
     * @return the builder
     */
    public Builder addProperty(PropertyKey key, Object value) {
      Preconditions.checkState(!key.equals(PropertyKey.ZOOKEEPER_ENABLED),
          "Enable Zookeeper via #setDeployMode instead of #addProperty");
      mProperties.put(key, value);
      return this;
    }

    /**
     * @param properties alluxio properties for launched masters and workers
     * @return the builder
     */
    public Builder addProperties(Map<PropertyKey, Object> properties) {
      for (Entry<PropertyKey, Object> entry : properties.entrySet()) {
        addProperty(entry.getKey(), entry.getValue());
      }
      return this;
    }

    /**
     * Sets master specific properties.
     * The keys of the properties are the indexes of masters
     * which are numbers between 0 to the number of masters (exclusive).
     *
     * @param properties the master properties to set
     * @return the builder
     */
    public Builder setMasterProperties(Map<Integer, Map<PropertyKey, String>> properties) {
      mMasterProperties = properties;
      return this;
    }

    /**
     * Sets worker specific properties.
     * The keys of the properties are the indexes of workers
     * which are numbers between 0 to the number of workers (exclusive).
     *
     * @param properties the worker properties to set
     * @return the builder
     */
    public Builder setWorkerProperties(Map<Integer, Map<PropertyKey, String>> properties) {
      mWorkerProperties = properties;
      return this;
    }

    /**
     * @param numMasters the number of masters for the cluster
     * @return the builder
     */
    public Builder setNumMasters(int numMasters) {
      mNumMasters = numMasters;
      return this;
    }

    /**
     * @param numWorkers the number of workers for the cluster
     * @return the builder
     */
    public Builder setNumWorkers(int numWorkers) {
      mNumWorkers = numWorkers;
      return this;
    }

    /**
     * @param clusterName a name for the cluster
     * @return the builder
     */
    public Builder setClusterName(String clusterName) {
      mClusterName = clusterName;
      return this;
    }

    /**
     * @param noFormat whether to skip formatting the journal
     * @return the builder
     */
    public Builder setNoFormat(boolean noFormat) {
      mNoFormat = noFormat;
      return this;
    }

    /**
     * @return a constructed {@link MultiProcessCluster}
     */
    public MultiProcessCluster build() {
      Preconditions.checkState(mMasterProperties.keySet()
              .stream().filter(key -> (key >= mNumMasters || key < 0)).count() == 0,
          "The master indexes in master properties should be bigger or equal to zero "
              + "and small than %s", mNumMasters);
      Preconditions.checkState(mWorkerProperties.keySet()
              .stream().filter(key ->  (key >= mNumWorkers || key < 0)).count() == 0,
          "The worker indexes in worker properties should be bigger or equal to zero "
              + "and small than %s", mNumWorkers);
      return new MultiProcessCluster(mProperties, mMasterProperties, mWorkerProperties,
          mNumMasters, mNumWorkers, mClusterName, mNoFormat, mReservedPorts);
    }
  }

  /**
   * @param reservedPorts ports reserved for usage by this cluster
   * @return a new builder for an {@link MultiProcessCluster}
   */
  public static Builder newBuilder(List<ReservedPort> reservedPorts) {
    return new Builder(reservedPorts);
  }
}
