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.master;
  
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertTrue;
  
  import alluxio.Constants;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.status.UnavailableException;
  import alluxio.grpc.NodeState;
  import alluxio.master.journal.JournalUtils;
  import alluxio.master.journal.noop.NoopJournalSystem;
  import alluxio.master.journal.raft.RaftJournalSystem;
  import alluxio.master.journal.ufs.UfsJournalSingleMasterPrimarySelector;
  import alluxio.master.service.metrics.MetricsService;
  import alluxio.master.service.rpc.RpcServerService;
  import alluxio.master.service.web.WebServerService;
  import alluxio.util.CommonUtils;
  import alluxio.util.WaitForOptions;
  import alluxio.util.io.FileUtils;
  import alluxio.util.io.PathUtils;
  import alluxio.util.network.NetworkAddressUtils;
  import alluxio.util.network.NetworkAddressUtils.ServiceType;
  
  import com.google.common.base.Preconditions;
  import com.google.common.collect.ImmutableMap;
  import org.junit.Assert;
  import org.junit.Before;
  import org.junit.Ignore;
  import org.junit.Rule;
  import org.junit.Test;
  import org.junit.rules.TemporaryFolder;
  import org.junit.runner.RunWith;
  import org.junit.runners.Parameterized;
  import org.mockito.Mockito;
  
  import java.io.IOException;
  import java.net.BindException;
  import java.net.ConnectException;
  import java.net.InetSocketAddress;
  import java.net.ServerSocket;
  import java.net.Socket;
  import java.net.URL;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Map;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  /**
   * Tests for {@link AlluxioMasterProcess}.
   */
  @RunWith(Parameterized.class)
  public final class AlluxioMasterProcessTest {
    @Rule
    public PortReservationRule mRpcPortRule = new PortReservationRule();
    @Rule
    public PortReservationRule mWebPortRule = new PortReservationRule();
    @Rule
    public TemporaryFolder mFolder = new TemporaryFolder();
  
    @Parameterized.Parameters
    public static Collection<Object[]> data() {
      return Arrays.asList(new Object[][] {
          {new ImmutableMap.Builder<Object, Boolean>()
              .put(PropertyKey.STANDBY_MASTER_WEB_ENABLED, true)
              .put(PropertyKey.STANDBY_MASTER_METRICS_SINK_ENABLED, true)
              .build()},
          {new ImmutableMap.Builder<Object, Boolean>()
              .put(PropertyKey.STANDBY_MASTER_WEB_ENABLED, false)
              .put(PropertyKey.STANDBY_MASTER_METRICS_SINK_ENABLED, false)
              .build()},
          {new ImmutableMap.Builder<Object, Boolean>()
              .put(PropertyKey.STANDBY_MASTER_WEB_ENABLED, true)
              .put(PropertyKey.STANDBY_MASTER_METRICS_SINK_ENABLED, false)
              .build()},
          {new ImmutableMap.Builder<Object, Boolean>()
              .put(PropertyKey.STANDBY_MASTER_WEB_ENABLED, false)
              .put(PropertyKey.STANDBY_MASTER_METRICS_SINK_ENABLED, true)
              .build()},
      });
    }
  
    public ImmutableMap<PropertyKey, Object> mConfigMap;
  
    public AlluxioMasterProcessTest(ImmutableMap<PropertyKey, Object> propMap) {
      mConfigMap = propMap;
    }
  
    @Before
    public void before() throws Exception {
      Configuration.reloadProperties();
      Configuration.set(PropertyKey.MASTER_RPC_PORT, mRpcPortRule.getPort());
      Configuration.set(PropertyKey.MASTER_WEB_PORT, mWebPortRule.getPort());
      Configuration.set(PropertyKey.MASTER_METASTORE_DIR, mFolder.newFolder("metastore"));
      Configuration.set(PropertyKey.USER_METRICS_COLLECTION_ENABLED, false);
      Configuration.set(PropertyKey.MASTER_JOURNAL_FOLDER, mFolder.newFolder("journal"));
      for (Map.Entry<PropertyKey, Object> entry : mConfigMap.entrySet()) {
        Configuration.set(entry.getKey(), entry.getValue());
      }
    }
  
    @Test
    public void startStopPrimary() throws Exception {
      AlluxioMasterProcess master = new AlluxioMasterProcess(new NoopJournalSystem(),
          new UfsJournalSingleMasterPrimarySelector());
      master.registerService(
          RpcServerService.Factory.create(master.getRpcBindAddress(), master, master.getRegistry()));
      master.registerService(WebServerService.Factory.create(master.getWebBindAddress(), master));
      master.registerService(MetricsService.Factory.create());
      Thread t = new Thread(() -> {
        try {
          master.start();
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      });
      t.start();
      master.waitForReady(10_000);
      startStopTest(master);
      t.interrupt();
      t.join();
    }
  
    @Test
    public void startStopStandby() throws Exception {
      AlluxioMasterProcess master = new AlluxioMasterProcess(
          new NoopJournalSystem(), new AlwaysStandbyPrimarySelector());
      master.registerService(
          RpcServerService.Factory.create(master.getRpcBindAddress(), master, master.getRegistry()));
      master.registerService(WebServerService.Factory.create(master.getWebBindAddress(), master));
      master.registerService(MetricsService.Factory.create());
      Thread t = new Thread(() -> {
        try {
          master.start();
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      });
      t.start();
      startStopTest(master, false,
          Configuration.getBoolean(PropertyKey.STANDBY_MASTER_WEB_ENABLED),
          Configuration.getBoolean(PropertyKey.STANDBY_MASTER_METRICS_SINK_ENABLED));
    }
  
    @Test
    public void startMastersThrowsUnavailableException() throws InterruptedException, IOException {
      ControllablePrimarySelector primarySelector = new ControllablePrimarySelector();
      primarySelector.setState(NodeState.PRIMARY);
      Configuration.set(PropertyKey.MASTER_JOURNAL_EXIT_ON_DEMOTION, true);
      AlluxioMasterProcess master = new AlluxioMasterProcess(
          new NoopJournalSystem(), primarySelector);
      AlluxioMasterProcess spy = Mockito.spy(master);
      Mockito.doAnswer(invocation -> { throw new UnavailableException("unavailable"); })
          .when(spy).startMasterComponents(true);
  
      AtomicBoolean success = new AtomicBoolean(true);
      Thread t = new Thread(() -> {
        try {
          spy.start();
        } catch (UnavailableException ue) {
          success.set(false);
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      });
      t.start();
      final int WAIT_TIME_TO_THROW_EXC = 500; // in ms
      t.join(WAIT_TIME_TO_THROW_EXC);
      t.interrupt();
      Assert.assertTrue(success.get());
    }
  
    @Test
    @Ignore
    public void stopAfterStandbyTransition() throws Exception {
      ControllablePrimarySelector primarySelector = new ControllablePrimarySelector();
      primarySelector.setState(NodeState.PRIMARY);
      Configuration.set(PropertyKey.MASTER_JOURNAL_EXIT_ON_DEMOTION, true);
      AlluxioMasterProcess master = new AlluxioMasterProcess(
          new NoopJournalSystem(), primarySelector);
      master.registerService(
          RpcServerService.Factory.create(master.getRpcBindAddress(), master, master.getRegistry()));
      master.registerService(WebServerService.Factory.create(master.getWebBindAddress(), master));
      Thread t = new Thread(() -> {
        try {
          master.start();
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      });
      t.start();
      waitForSocketServing(ServiceType.MASTER_RPC);
      waitForSocketServing(ServiceType.MASTER_WEB);
      int rpcPort = master.getRpcAddress().getPort();
      int webPort = master.getWebAddress().getPort();
      assertTrue(isBound(rpcPort));
      assertTrue(isBound(webPort));
      primarySelector.setState(NodeState.STANDBY);
      t.join(10000);
      CommonUtils.waitFor("Master to be stopped", () -> !master.isRunning(),
          WaitForOptions.defaults().setTimeoutMs(3 * Constants.MINUTE_MS));
      CommonUtils.waitFor("Master to be stopped", () -> !isBound(rpcPort),
          WaitForOptions.defaults().setTimeoutMs(Constants.MINUTE_MS));
      CommonUtils.waitFor("Master to be stopped", () -> !isBound(webPort),
          WaitForOptions.defaults().setTimeoutMs(Constants.MINUTE_MS));
    }
  
    /**
     * This test ensures that given a root UFS which is <i>not</i> local, that we can still provide a
     * local path to an Alluxio backup and start the master.
     *
     * This test is ignored because it requires adding a new UFS dependency in the test scope, but
     * adding the alluxio-underfs-web dependency causes the module to hang in the  builds even after
     * the test completes. The hang seems to be due to a bug in maven 3.5.4.
     */
    @Test
    @Ignore
    public void restoreFromBackupLocal() throws Exception {
      URL backupResource = this.getClass().getResource("/alluxio-local-backup.gz");
      Preconditions.checkNotNull(backupResource);
      String backupPath = backupResource.toURI().toString();
      String journalPath = PathUtils.concatPath(mFolder.getRoot(), "journal");
      FileUtils.createDir(journalPath);
      String ufsPath = PathUtils.concatPath(mFolder.getRoot(), "ufs");
      FileUtils.createDir(ufsPath);
      ufsPath = "http://other_ufs/";
      Configuration.set(PropertyKey.MASTER_EMBEDDED_JOURNAL_MIN_ELECTION_TIMEOUT, "550");
      Configuration.set(PropertyKey.MASTER_EMBEDDED_JOURNAL_MAX_ELECTION_TIMEOUT, "1100");
      Configuration.set(PropertyKey.MASTER_JOURNAL_INIT_FROM_BACKUP, backupPath);
      Configuration.set(PropertyKey.MASTER_JOURNAL_FOLDER, journalPath);
      Configuration.set(PropertyKey.MASTER_MOUNT_TABLE_ROOT_UFS, ufsPath);
      AlluxioMasterProcess master = new AlluxioMasterProcess(
          new RaftJournalSystem(JournalUtils.getJournalLocation(), ServiceType.MASTER_RAFT),
          new UfsJournalSingleMasterPrimarySelector());
      master.registerService(
          RpcServerService.Factory.create(master.getRpcBindAddress(), master, master.getRegistry()));
      master.registerService(WebServerService.Factory.create(master.getWebBindAddress(), master));
      master.registerService(MetricsService.Factory.create());
      Thread t = new Thread(() -> {
        try {
          master.start();
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      });
      t.start();
      startStopTest(master);
    }
  
    @Test
    public void startStopStandbyStandbyServer() throws Exception {
      Configuration.set(PropertyKey.STANDBY_MASTER_GRPC_ENABLED, true);
      AlluxioMasterProcess master =
          new AlluxioMasterProcess(new NoopJournalSystem(), new AlwaysStandbyPrimarySelector());
      master.registerService(
          RpcServerService.Factory.create(
              master.getRpcBindAddress(), master, master.getRegistry()));
      master.registerService(WebServerService.Factory.create(master.getWebBindAddress(), master));
      master.registerService(MetricsService.Factory.create());
  
      Thread t = new Thread(() -> {
        try {
          master.start();
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      });
      t.start();
      startStopTest(master,
          true,
          Configuration.getBoolean(PropertyKey.STANDBY_MASTER_WEB_ENABLED),
          Configuration.getBoolean(PropertyKey.STANDBY_MASTER_METRICS_SINK_ENABLED));
    }
  
    private void startStopTest(AlluxioMasterProcess master) throws Exception {
      startStopTest(master, true, true, true);
    }
  
    private void startStopTest(AlluxioMasterProcess master, boolean expectGrpcServiceStarted,
        boolean expectWebServiceStarted, boolean expectMetricsSinkStarted) throws Exception {
      final int TIMEOUT_MS = 10_000;
      // rpc and web ports will be bound either by the serving server, or by the rejecting server
      waitForSocketServing(ServiceType.MASTER_RPC);
      waitForSocketServing(ServiceType.MASTER_WEB);
      assertTrue(isBound(master.getRpcAddress().getPort()));
      assertTrue(isBound(master.getWebAddress().getPort()));
      if (expectGrpcServiceStarted) {
        CommonUtils.waitFor("grpc server to serve",
            () -> master.mServices.stream().anyMatch(service -> service instanceof RpcServerService
            && ((RpcServerService) service).isServing()),
            WaitForOptions.defaults().setTimeoutMs(TIMEOUT_MS));
      }
      if (expectWebServiceStarted) {
        assertTrue(master.waitForWebServerReady(TIMEOUT_MS));
      }
      if (expectMetricsSinkStarted) {
        assertTrue(master.waitForMetricSinkServing(TIMEOUT_MS));
      }
      master.stop();
      assertFalse(isBound(master.getRpcAddress().getPort()));
      assertFalse(isBound(master.getWebAddress().getPort()));
    }
  
    private void waitForSocketServing(ServiceType service)
        throws TimeoutException, InterruptedException {
      InetSocketAddress addr =
          NetworkAddressUtils.getBindAddress(service, Configuration.global());
      CommonUtils.waitFor(service + " to be serving", () -> {
        try {
          Socket s = new Socket(addr.getAddress(), addr.getPort());
          s.close();
          return true;
        } catch (ConnectException e) {
          return false;
        } catch (IOException e) {
          throw new RuntimeException(e);
        }
      }, WaitForOptions.defaults().setTimeoutMs(5 * Constants.MINUTE_MS));
    }
  
    private boolean isBound(int port) {
      try {
        ServerSocket socket = new ServerSocket(port);
        socket.setReuseAddress(true);
        socket.close();
        return false;
      } catch (BindException e) {
        return true;
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
  }
  