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.client.cli.fsadmin.command;
  
  import alluxio.AlluxioURI;
  import alluxio.ConfigurationRule;
  import alluxio.SystemErrRule;
  import alluxio.SystemOutRule;
  import alluxio.cli.fsadmin.FileSystemAdminShell;
  import alluxio.cli.fsadmin.journal.QuorumCommand;
  import alluxio.cli.fsadmin.journal.QuorumElectCommand;
  import alluxio.cli.fsadmin.journal.QuorumInfoCommand;
  import alluxio.cli.fsadmin.journal.QuorumRemoveCommand;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.ExceptionMessage;
  import alluxio.grpc.JournalDomain;
  import alluxio.grpc.QuorumServerInfo;
  import alluxio.grpc.QuorumServerState;
  import alluxio.master.journal.JournalType;
  import alluxio.multi.process.MasterNetAddress;
  import alluxio.multi.process.MultiProcessCluster;
  import alluxio.multi.process.PortCoordination;
  import alluxio.testutils.BaseIntegrationTest;
  import alluxio.util.CommonUtils;
  import alluxio.util.WaitForOptions;
  
  import org.junit.After;
  import org.junit.Assert;
  import org.junit.Rule;
  import org.junit.Test;
  import org.junit.rules.ExpectedException;
  
  import java.io.ByteArrayOutputStream;
  import java.util.Arrays;
  import java.util.List;
  import java.util.stream.Collectors;
  
  /**
   * Integration tests for the embedded journal.
   */
  public final class QuorumCommandIntegrationTest extends BaseIntegrationTest {
    @Rule
    public ConfigurationRule mConf = new ConfigurationRule(
        PropertyKey.USER_METRICS_COLLECTION_ENABLED, false, Configuration.modifiableGlobal());
  
    public MultiProcessCluster mCluster;
  
    public ByteArrayOutputStream mOutput = new ByteArrayOutputStream();
    public ByteArrayOutputStream mErrOutput = new ByteArrayOutputStream();
  
    @Rule
    public ExpectedException mException = ExpectedException.none();
  
    @Rule
    public SystemOutRule mOutRule = new SystemOutRule(mOutput);
  
    @Rule
    public SystemErrRule mErrRule = new SystemErrRule(mErrOutput);
  
    @After
    public void after() throws Exception {
      if (mCluster != null) {
        mCluster.destroy();
      }
    }
  
    @Test
    public void quorumInfo() throws Exception {
      mCluster = MultiProcessCluster.newBuilder(PortCoordination.QUORUM_SHELL_INFO)
          .setClusterName("QuorumShellInfo").setNumMasters(3).setNumWorkers(0)
          .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.EMBEDDED)
          .addProperty(PropertyKey.MASTER_JOURNAL_FLUSH_TIMEOUT_MS, "5min")
          // To make the test run faster.
          .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MIN_ELECTION_TIMEOUT, "750ms")
          .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MAX_ELECTION_TIMEOUT, "1500ms").build();
      mCluster.start();
  
      String output;
  
      try (FileSystemAdminShell shell = new FileSystemAdminShell(Configuration.global())) {
        // Validate quorum state is dumped as expected.
        mOutput.reset();
        shell.run("journal", "quorum", "info", "-domain", "MASTER");
        output = mOutput.toString().trim();
        Assert.assertTrue(output.contains(
            String.format(QuorumInfoCommand.OUTPUT_HEADER_DOMAIN, JournalDomain.MASTER.name())));
        Assert.assertTrue(
            output.contains(String.format(QuorumInfoCommand.OUTPUT_HEADER_QUORUM_SIZE, 3)));
        List<String> journalAddresses =
            Configuration.getList(PropertyKey.MASTER_EMBEDDED_JOURNAL_ADDRESSES);
        for (String address : journalAddresses) {
          String format = String.format(QuorumInfoCommand.OUTPUT_SERVER_INFO,
                  QuorumServerState.AVAILABLE.name(), "0", address).trim();
          Assert.assertTrue(output.contains(format));
        }
  
        // Validate quorum state is updated as expected after a fail-over.
        mCluster.stopMaster(0);
        // Wait for up to 2 election timeouts until quorum notices the change.
        // Use shell "quorum info" for validation.
        CommonUtils.waitFor("Quorum noticing change.", () -> {
          mOutput.reset();
          shell.run("journal", "quorum", "info", "-domain", "MASTER");
          return mOutput.toString().trim().contains(QuorumServerState.UNAVAILABLE.name());
        }, WaitForOptions.defaults().setTimeoutMs(2 * (int) Configuration.getMs(
            PropertyKey.MASTER_EMBEDDED_JOURNAL_MAX_ELECTION_TIMEOUT)));
      }
      mCluster.notifySuccess();
    }
  
    @Test
    public void quorumRemove() throws Exception {
      mCluster = MultiProcessCluster.newBuilder(PortCoordination.QUORUM_SHELL_REMOVE)
          .setClusterName("QuorumShellRemove").setNumMasters(5).setNumWorkers(0)
          .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.EMBEDDED)
          .addProperty(PropertyKey.MASTER_JOURNAL_FLUSH_TIMEOUT_MS, "5min")
          // To make the test run faster.
          .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MIN_ELECTION_TIMEOUT, "750ms")
          .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MAX_ELECTION_TIMEOUT, "1500ms").build();
      mCluster.start();
  
      String output;
  
      try (FileSystemAdminShell shell = new FileSystemAdminShell(Configuration.global())) {
        AlluxioURI testDir = new AlluxioURI("/testDir");
        mCluster.getFileSystemClient().createDirectory(testDir);
        // Verify cluster is reachable.
        Assert.assertTrue(mCluster.getFileSystemClient().exists(testDir));
  
        // Validate quorum state is updated as expected after shutting down 2 masters.
        mCluster.stopMaster(0);
        mCluster.stopMaster(1);
  
        // Verify cluster is reachable.
        Assert.assertTrue(mCluster.getFileSystemClient().exists(testDir));
  
        // Wait for up to 2 election timeouts until quorum notices changes.
        CommonUtils.waitFor("Quorum noticing change.", () -> {
          try {
            return mCluster.getJournalMasterClientForMaster().getQuorumInfo().getServerInfoList()
                .stream().filter((info) -> info.getServerState() == QuorumServerState.UNAVAILABLE)
                .collect(Collectors.toList()).size() == 2;
          } catch (Exception e) {
            return false;
          }
        }, WaitForOptions.defaults().setTimeoutMs(6 * (int) Configuration.getMs(
            PropertyKey.MASTER_EMBEDDED_JOURNAL_MAX_ELECTION_TIMEOUT)));
  
        // Remove unavailable masters using shell.
        for (QuorumServerInfo serverInfo : mCluster.getJournalMasterClientForMaster().getQuorumInfo()
            .getServerInfoList()) {
          if (serverInfo.getServerState() == QuorumServerState.UNAVAILABLE) {
            mOutput.reset();
  
            String serverAddress = String.format("%s:%d", serverInfo.getServerAddress().getHost(),
                serverInfo.getServerAddress().getRpcPort());
            shell.run("journal", "quorum", "remove", "-domain", "MASTER", "-address", serverAddress);
  
            output = mOutput.toString().trim();
            Assert.assertEquals(String.format(QuorumRemoveCommand.OUTPUT_RESULT, serverAddress,
                JournalDomain.MASTER.name()), lastLine(output));
          }
        }
  
        // Verify quorum size is resized down to 3 after removing 2 masters.
        Assert.assertEquals(3,
            mCluster.getJournalMasterClientForMaster().getQuorumInfo().getServerInfoList().size());
  
        mCluster.stopMaster(2);
        // Verify cluster is reachable.
        Assert.assertTrue(mCluster.getFileSystemClient().exists(testDir));
      }
      mCluster.notifySuccess();
    }
  
    @Test
    public void elect() throws Exception {
      final int MASTER_INDEX_WAIT_TIME = 5_000;
      int numMasters = 3;
      mCluster = MultiProcessCluster.newBuilder(PortCoordination.QUORUM_SHELL_REMOVE)
              .setClusterName("QuorumShellElect").setNumMasters(numMasters).setNumWorkers(0)
              .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.EMBEDDED)
              .addProperty(PropertyKey.MASTER_JOURNAL_FLUSH_TIMEOUT_MS, "5min")
              // To make the test run faster.
              .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MIN_ELECTION_TIMEOUT, "750ms")
              .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MAX_ELECTION_TIMEOUT, "1500ms")
              .build();
      mCluster.start();
  
      try (FileSystemAdminShell shell = new FileSystemAdminShell(Configuration.global())) {
        int newLeaderIdx = (mCluster.getPrimaryMasterIndex(MASTER_INDEX_WAIT_TIME) + 1) % numMasters;
        // `getPrimaryMasterIndex` uses the same `mMasterAddresses` variable as getMasterAddresses
        // we can therefore access to the new leader's address this ways
        MasterNetAddress netAddress = mCluster.getMasterAddresses().get(newLeaderIdx);
        String newLeaderAddr = String.format("%s:%s", netAddress.getHostname(),
                netAddress.getEmbeddedJournalPort());
  
        mOutput.reset();
        shell.run("journal", "quorum", "elect", "-address" , newLeaderAddr);
        String output = mOutput.toString().trim();
        String expected = String.format("%s\n%s\n%s\n%s",
            String.format(QuorumElectCommand.TRANSFER_INIT, newLeaderAddr),
            String.format(QuorumElectCommand.TRANSFER_SUCCESS, newLeaderAddr),
            String.format(QuorumElectCommand.RESET_INIT, "successful"),
            QuorumElectCommand.RESET_SUCCESS);
        Assert.assertEquals(expected, output);
      }
      mCluster.notifySuccess();
    }
  
    @Test
    public void infoAfterElect() throws Exception {
      final int MASTER_INDEX_WAIT_TIME = 5_000;
      int numMasters = 3;
      mCluster = MultiProcessCluster.newBuilder(PortCoordination.QUORUM_SHELL_REMOVE)
          .setClusterName("QuorumShellElect").setNumMasters(numMasters).setNumWorkers(0)
          .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.EMBEDDED)
          .addProperty(PropertyKey.MASTER_JOURNAL_FLUSH_TIMEOUT_MS, "5min")
          // To make the test run faster.
          .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MIN_ELECTION_TIMEOUT, "750ms")
          .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MAX_ELECTION_TIMEOUT, "1500ms")
          .build();
      mCluster.start();
  
      try (FileSystemAdminShell shell = new FileSystemAdminShell(Configuration.global())) {
        int newLeaderIdx = (mCluster.getPrimaryMasterIndex(MASTER_INDEX_WAIT_TIME) + 1) % numMasters;
        // `getPrimaryMasterIndex` uses the same `mMasterAddresses` variable as getMasterAddresses
        // we can therefore access to the new leader's address this ways
        MasterNetAddress netAddress = mCluster.getMasterAddresses().get(newLeaderIdx);
        String newLeaderAddr = String.format("%s:%s", netAddress.getHostname(),
            netAddress.getEmbeddedJournalPort());
  
        int success = shell.run("journal", "quorum", "elect", "-address" , newLeaderAddr);
        Assert.assertEquals("elect command failed", 0, success);
        mOutput.reset();
        shell.run("journal", "quorum", "info", "-domain", "MASTER");
        String output = mOutput.toString().trim();
        for (MasterNetAddress masterAddr : mCluster.getMasterAddresses()) {
          String expected = String.format(QuorumInfoCommand.OUTPUT_SERVER_INFO, "AVAILABLE", "1",
              String.format("%s:%d", masterAddr.getHostname(), masterAddr.getEmbeddedJournalPort()));
          Assert.assertTrue(output.contains(expected.trim()));
        }
      }
      mCluster.notifySuccess();
    }
  
    @Test
    public void quorumCommand() throws Exception {
      mCluster = MultiProcessCluster.newBuilder(PortCoordination.QUORUM_SHELL)
          .setClusterName("QuorumShell").setNumMasters(3).setNumWorkers(0)
          .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.EMBEDDED)
          .addProperty(PropertyKey.MASTER_JOURNAL_FLUSH_TIMEOUT_MS, "5min")
          // To make the test run faster.
          .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MIN_ELECTION_TIMEOUT, "750ms")
          .addProperty(PropertyKey.MASTER_EMBEDDED_JOURNAL_MAX_ELECTION_TIMEOUT, "1500ms").build();
      mCluster.start();
      try (FileSystemAdminShell shell = new FileSystemAdminShell(Configuration.global())) {
        String output;
  
        // Validate quorum sub-commands are validated.
        mOutput.reset();
        shell.run("journal", "quorum", "nonexistentCommand");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumCommand.description(), lastLine(output));
  
        // Validate option counts are validated for "quorum info"
        mOutput.reset();
        shell.run("journal", "quorum", "info");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumInfoCommand.description(), lastLine(output));
  
        mOutput.reset();
        shell.run("journal", "quorum", "info", "-op1", "val1", "-op2", "val2");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumInfoCommand.description(), lastLine(output));
  
        // Validate option counts are validated for "quorum", "remove"
        mOutput.reset();
        shell.run("journal", "quorum", "remove");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumRemoveCommand.description(), lastLine(output));
        mOutput.reset();
        shell.run("journal", "quorum", "remove", "-op1", "val1");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumRemoveCommand.description(), lastLine(output));
        mOutput.reset();
        shell.run("journal", "quorum", "remove", "-op1", "val1", "-op2", "val2", "-op3", "val3");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumRemoveCommand.description(), lastLine(output));
  
        // Validate option counts are validated for "quorum", "elect"
        mOutput.reset();
        shell.run("journal", "quorum", "elect");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumElectCommand.description(), lastLine(output));
        mOutput.reset();
        shell.run("journal", "quorum", "elect", "-op1", "val1");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumElectCommand.description(), lastLine(output));
        mOutput.reset();
        shell.run("journal", "quorum", "elect", "-op1", "val1", "-op2", "val2", "-op3",
                "val3");
        output = mOutput.toString().trim();
        Assert.assertEquals(QuorumElectCommand.description(), lastLine(output));
  
        // Validate option validation works for "quorum info".
        mOutput.reset();
        shell.run("journal", "quorum", "info", "-domain", "UNKNOWN");
        output = mOutput.toString().trim();
        Assert.assertEquals(ExceptionMessage.INVALID_OPTION_VALUE.getMessage(
            QuorumInfoCommand.DOMAIN_OPTION_NAME, Arrays.toString(JournalDomain.values())), output);
  
        // Validate option validation works for "journal quorum remove"
        // Validate -domain is validated.
        mOutput.reset();
        shell.run("journal", "quorum", "remove", "-domain", "UNKNOWN", "-address", "host:0");
        output = mOutput.toString().trim();
        Assert.assertEquals(ExceptionMessage.INVALID_OPTION_VALUE.getMessage(
            QuorumInfoCommand.DOMAIN_OPTION_NAME, Arrays.toString(JournalDomain.values())), output);
  
        // Validate -address is validated.
        mOutput.reset();
        shell.run("journal", "quorum", "remove", "-domain", "JOB_MASTER", "-address",
            "hostname:invalid_port");
        output = mOutput.toString().trim();
        Assert.assertEquals(ExceptionMessage.INVALID_ADDRESS_VALUE.getMessage(), output);
  
        // Validate -address is validated for transferLeader.
        mOutput.reset();
        shell.run("journal", "quorum", "elect", "-address", "hostname:invalid_port");
        output = mOutput.toString().trim();
        Assert.assertEquals(ExceptionMessage.INVALID_ADDRESS_VALUE.getMessage(), output);
      }
      mCluster.notifySuccess();
    }
  
    private String lastLine(String output) {
      String[] lines = output.split("\n");
      if (lines.length > 0) {
        return lines[lines.length - 1];
      }
      return "";
    }
  }
  
  