<commit_before>/*
 * 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.Configuration;
import alluxio.ConfigurationRule;
import alluxio.ConfigurationTestUtils;
import alluxio.Constants;
import alluxio.PropertyKey;
import alluxio.cli.Format;
import alluxio.client.MetaMasterClient;
import alluxio.client.RetryHandlingMetaMasterClient;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystem.Factory;
import alluxio.client.file.FileSystemContext;
import alluxio.exception.status.UnavailableException;
import alluxio.master.LocalAlluxioCluster;
import alluxio.master.MasterClientConfig;
import alluxio.master.MasterInquireClient;
import alluxio.master.SingleMasterInquireClient;
import alluxio.master.ZkMasterInquireClient;
import alluxio.network.PortUtils;
import alluxio.util.CommonUtils;
import alluxio.util.WaitForOptions;
import alluxio.util.io.PathUtils;
import alluxio.util.network.NetworkAddressUtils;
import alluxio.wire.MasterInfo;
import alluxio.zookeeper.RestartableTestingServer;

import com.google.common.base.Preconditions;
import com.google.common.io.Closer;
import org.apache.commons.io.Charsets;
import org.apache.commons.io.FileUtils;
import org.junit.rules.TestRule;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
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.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 javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * 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 implements TestRule {
  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 final Map<PropertyKey, String> mProperties;
  private final Map<Integer, Map<PropertyKey, String>> mMasterProperties;
  private final Map<Integer, Map<PropertyKey, String>> mWorkerProperties;
  private final 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 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;
  private State mState;
  private RestartableTestingServer mCuratorServer;
  /**
   * 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, String> properties,
      Map<Integer, Map<PropertyKey, String>> masterProperties,
      Map<Integer, Map<PropertyKey, String>> workerProperties,
      int numMasters, int numWorkers, String clusterName, DeployMode mode) {
    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());
    mDeployMode = mode;
    mMasters = new ArrayList<>();
    mWorkers = new ArrayList<>();
    mCloser = Closer.create();
    mState = State.NOT_STARTED;
    mSuccess = false;
  }

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

    mMasterAddresses = generateMasterAddresses(mNumMasters);
    LOG.info("Master addresses: {}", mMasterAddresses);
    switch (mDeployMode) {
      case NON_HA:
        MasterNetAddress masterAddress = mMasterAddresses.get(0);
        mProperties.put(PropertyKey.MASTER_HOSTNAME, masterAddress.getHostname());
        mProperties.put(PropertyKey.MASTER_RPC_PORT, Integer.toString(masterAddress.getRpcPort()));
        mProperties.put(PropertyKey.MASTER_WEB_PORT, Integer.toString(masterAddress.getWebPort()));
        break;
      case ZOOKEEPER_HA:
        mCuratorServer = mCloser.register(
            new RestartableTestingServer(-1, AlluxioTestDirectory.createTemporaryDirectory("zk")));
        mProperties.put(PropertyKey.ZOOKEEPER_ENABLED, "true");
        mProperties.put(PropertyKey.ZOOKEEPER_ADDRESS, mCuratorServer.getConnectString());
        break;
      default:
        throw new IllegalStateException("Unknown deploy mode: " + mDeployMode.toString());
    }

    for (Entry<PropertyKey, String> entry : ConfigurationTestUtils.testConfigurationDefaults(
        NetworkAddressUtils.getLocalHostName(), 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());
    }
    mProperties.put(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS,
        PathUtils.concatPath(mWorkDir, "underFSStorage"));
    new File(mProperties.get(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS)).mkdirs();
    formatJournal();
    writeConf();

    // Start servers
    LOG.info("Starting alluxio cluster {} with base directory {}", mClusterName,
        mWorkDir.getAbsolutePath());
    for (int i = 0; i < mNumMasters; i++) {
      createMaster(i).start();
    }
    for (int i = 0; i < mNumWorkers; i++) {
      createWorker(i).start();
    }
  }

  /**
   * 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 (UnavailableException e) {
        return false;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, 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();
    CommonUtils.waitFor("all nodes registered", () -> {
      try {
        MasterInfo masterInfo = metaMasterClient.getMasterInfo(null);
        int liveNodeNum = masterInfo.getMasterAddresses().size()
            + masterInfo.getWorkerAddresses().size();
        return liveNodeNum == (mNumMasters + mNumWorkers);
      } catch (UnavailableException e) {
        return false;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, WaitForOptions.defaults().setTimeoutMs(timeoutMs));
  }

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

  /**
   * @return a meta master client
   */
  public synchronized MetaMasterClient getMetaMasterClient() {
    Preconditions.checkState(mState == State.STARTED,
        "must be in the started state to get a meta master client, but state was %s", mState);<commit_msg>Add logging to multiprocess cluster start polling<commit_after>/*
 * 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.Configuration;
import alluxio.ConfigurationRule;
import alluxio.ConfigurationTestUtils;
import alluxio.Constants;
import alluxio.PropertyKey;
import alluxio.cli.Format;
import alluxio.client.MetaMasterClient;
import alluxio.client.RetryHandlingMetaMasterClient;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystem.Factory;
import alluxio.client.file.FileSystemContext;
import alluxio.exception.status.UnavailableException;
import alluxio.master.LocalAlluxioCluster;
import alluxio.master.MasterClientConfig;
import alluxio.master.MasterInquireClient;
import alluxio.master.SingleMasterInquireClient;
import alluxio.master.ZkMasterInquireClient;
import alluxio.network.PortUtils;
import alluxio.util.CommonUtils;
import alluxio.util.WaitForOptions;
import alluxio.util.io.PathUtils;
import alluxio.util.network.NetworkAddressUtils;
import alluxio.wire.MasterInfo;
import alluxio.zookeeper.RestartableTestingServer;

import com.google.common.base.Preconditions;
import com.google.common.io.Closer;
import org.apache.commons.io.Charsets;
import org.apache.commons.io.FileUtils;
import org.junit.rules.TestRule;
import org.junit.runner.Description;
import org.junit.runners.model.Statement;
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.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 javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * 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 implements TestRule {
  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 final Map<PropertyKey, String> mProperties;
  private final Map<Integer, Map<PropertyKey, String>> mMasterProperties;
  private final Map<Integer, Map<PropertyKey, String>> mWorkerProperties;
  private final 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 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;
  private State mState;
  private RestartableTestingServer mCuratorServer;
  /**
   * 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, String> properties,
      Map<Integer, Map<PropertyKey, String>> masterProperties,
      Map<Integer, Map<PropertyKey, String>> workerProperties,
      int numMasters, int numWorkers, String clusterName, DeployMode mode) {
    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());
    mDeployMode = mode;
    mMasters = new ArrayList<>();
    mWorkers = new ArrayList<>();
    mCloser = Closer.create();
    mState = State.NOT_STARTED;
    mSuccess = false;
  }

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

    mMasterAddresses = generateMasterAddresses(mNumMasters);
    LOG.info("Master addresses: {}", mMasterAddresses);
    switch (mDeployMode) {
      case NON_HA:
        MasterNetAddress masterAddress = mMasterAddresses.get(0);
        mProperties.put(PropertyKey.MASTER_HOSTNAME, masterAddress.getHostname());
        mProperties.put(PropertyKey.MASTER_RPC_PORT, Integer.toString(masterAddress.getRpcPort()));
        mProperties.put(PropertyKey.MASTER_WEB_PORT, Integer.toString(masterAddress.getWebPort()));
        break;
      case ZOOKEEPER_HA:
        mCuratorServer = mCloser.register(
            new RestartableTestingServer(-1, AlluxioTestDirectory.createTemporaryDirectory("zk")));
        mProperties.put(PropertyKey.ZOOKEEPER_ENABLED, "true");
        mProperties.put(PropertyKey.ZOOKEEPER_ADDRESS, mCuratorServer.getConnectString());
        break;
      default:
        throw new IllegalStateException("Unknown deploy mode: " + mDeployMode.toString());
    }

    for (Entry<PropertyKey, String> entry : ConfigurationTestUtils.testConfigurationDefaults(
        NetworkAddressUtils.getLocalHostName(), 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());
    }
    mProperties.put(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS,
        PathUtils.concatPath(mWorkDir, "underFSStorage"));
    new File(mProperties.get(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS)).mkdirs();
    formatJournal();
    writeConf();

    // Start servers
    LOG.info("Starting alluxio cluster {} with base directory {}", mClusterName,
        mWorkDir.getAbsolutePath());
    for (int i = 0; i < mNumMasters; i++) {
      createMaster(i).start();
    }
    for (int i = 0; i < mNumWorkers; i++) {
      createWorker(i).start();
    }
  }

  /**
   * 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 (UnavailableException e) {
        return false;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, 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();
    CommonUtils.waitFor("all nodes registered", () -> {
      try {
        MasterInfo masterInfo = metaMasterClient.getMasterInfo(null);
        int liveNodeNum = masterInfo.getMasterAddresses().size()
            + masterInfo.getWorkerAddresses().size();
        if (liveNodeNum == (mNumMasters + mNumWorkers)) {
          return true;
        } else {
          LOG.info("Master addresses: {}. Worker addresses: {}", masterInfo.getMasterAddresses(),
              masterInfo.getWorkerAddresses());
          return false;
        }
      } catch (UnavailableException e) {
        return false;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, WaitForOptions.defaults().setInterval(200).setTimeoutMs(timeoutMs));
  }

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

  /**
   * @return a meta master client
   */
  public synchronized MetaMasterClient getMetaMasterClient() {
    Preconditions.checkState(mState == State.STARTED,
        "must be in the started state to get a meta master client, but state was %s", mState);
{'fraction_non_alphanumeric': 0.09187388987566607, 'fraction_numerical': 0.0011989342806394316, 'mean_word_length': 3.9594802906848714, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3404536', 'n_tokens_mistral': 6303, 'n_tokens_neox': 6222, 'n_words': 1796}