/*
 * 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 alluxio.util.network.NetworkAddressUtils.ServiceType;

import alluxio.Process;
import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.Configuration;
import alluxio.grpc.GrpcServerBuilder;
import alluxio.master.journal.JournalSystem;
import alluxio.master.service.SimpleService;
import alluxio.master.service.rpc.RpcServerService;
import alluxio.master.service.web.WebServerService;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;
import alluxio.util.CommonUtils;
import alluxio.util.ConfigurationUtils;
import alluxio.util.WaitForOptions;
import alluxio.util.network.NetworkAddressUtils;
import alluxio.web.WebServer;

import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeoutException;
import java.util.function.Supplier;

/**
 * Defines a set of methods which any {@link MasterProcess} should implement.
 *
 * This class serves as a common implementation for functions that both the
 * {@link AlluxioMasterProcess} and {@link AlluxioJobMasterProcess} use.
 * Each master should have an RPC server, web server, and journaling
 * system which can serve client requests.
 */
public abstract class MasterProcess implements Process {
  private static final Logger LOG = LoggerFactory.getLogger(MasterProcess.class);

  /** The journal system for writing journal entries and restoring master state. */
  protected final JournalSystem mJournalSystem;
  protected final PrimarySelector mLeaderSelector;
  /** The master registry. */
  protected final MasterRegistry mRegistry = new MasterRegistry();

  /** Rpc server bind address. **/
  final InetSocketAddress mRpcBindAddress;
  /** Web server bind address. **/
  final InetSocketAddress mWebBindAddress;
  /** The connection address for the rpc server. */
  final InetSocketAddress mRpcConnectAddress;
  /** The connection address for the web service. */
  final InetSocketAddress mWebConnectAddress;

  /** The start time for when the master started. */
  final long mStartTimeMs;
  // holds the simple services registered with the master, should only be accessed through
  // #registerService
  List<SimpleService> mServices = new ArrayList<>();

  /**
   * Prepares a {@link MasterProcess} journal, rpc and web server using the given sockets.
   *
   * @param journalSystem the journaling system
   * @param leaderSelector the leader selector
   * @param webService    the web service type
   * @param rpcService    the rpc service type
   */
  public MasterProcess(JournalSystem journalSystem, PrimarySelector leaderSelector,
      ServiceType webService, ServiceType rpcService) {
    mJournalSystem = Preconditions.checkNotNull(journalSystem, "journalSystem");
    mLeaderSelector = Preconditions.checkNotNull(leaderSelector, "leaderSelector");
    mRpcBindAddress = configureAddress(rpcService);
    mWebBindAddress = configureAddress(webService);
    mRpcConnectAddress = NetworkAddressUtils.getConnectAddress(rpcService, Configuration.global());
    mWebConnectAddress = NetworkAddressUtils.getConnectAddress(webService, Configuration.global());
    mStartTimeMs = System.currentTimeMillis();
    MetricsSystem.registerGaugeIfAbsent(MetricKey.MASTER_START_TIME.getName(), () -> mStartTimeMs);
  }

  private static InetSocketAddress configureAddress(ServiceType service) {
    AlluxioConfiguration conf = Configuration.global();
    int port = NetworkAddressUtils.getPort(service, conf);
    if (!ConfigurationUtils.isHaMode(conf) && port == 0) {
      throw new RuntimeException(
          String.format("%s port must be nonzero in single-master mode", service));
    }
    if (port == 0) {
      try (ServerSocket s = new ServerSocket(0)) {
        s.setReuseAddress(true);
        Configuration.set(service.getPortKey(), s.getLocalPort());
      } catch (IOException e) {
        throw new RuntimeException(e);
      }
    }
    return NetworkAddressUtils.getBindAddress(service, conf);
  }

  /**
   * Used in master factory after instantiating the master process and before starting the master
   * process. This allows you to define with services are necessary for the operation of the
   * master process.
   * @param service a service to register for the functioning of the master process
   */
  public void registerService(SimpleService service) {
    mServices.add(service);
  }

  /**
   * @return a fully configured rpc server except for the executor option (provided by the
   * {@link #createRpcExecutorService()}) and the rpc services (provided by {@link #mRegistry}
   */
  public abstract GrpcServerBuilder createBaseRpcServer();

  /**
   * This method is decoupled from {@link #createBaseRpcServer()} because the
   * {@link AlluxioExecutorService} needs to be managed (i.e. started and stopped) independently
   * of the rpc server that uses it
   * @return a custom executor service to be used in the rpc server
   */
  public Optional<AlluxioExecutorService> createRpcExecutorService() {
    return Optional.empty();
  }

  /**
   * @return the {@link SafeModeManager} if you have one
   */
  public Optional<SafeModeManager> getSafeModeManager() {
    return Optional.empty();
  }

  /**
   * @return this master's rpc bind address
   */
  public final InetSocketAddress getRpcBindAddress() {
    return mRpcBindAddress;
  }

  /**
   * @return this master's rpc connect address
   */
  public final InetSocketAddress getRpcAddress() {
    return mRpcConnectAddress;
  }

  /**
   * @param clazz the class of the master to retrieve
   * @return the desired master from the registry if it exists
   * @param <T> parameterized class of the desired master
   */
  public final <T extends Master> T getMaster(Class<T> clazz) {
    return mRegistry.get(clazz);
  }

  /**
   * Use only to instantiate {@link SimpleService}s.
   * @return the master registry
   */
  public final MasterRegistry getRegistry() {
    return mRegistry;
  }

  /**
   * @return the start time of the master in milliseconds
   */
  public long getStartTimeMs() {
    return mStartTimeMs;
  }

  /**
   * @return the uptime of the master in milliseconds
   */
  public long getUptimeMs() {
    return System.currentTimeMillis() - mStartTimeMs;
  }

  /**
   * @return a newly created web server for this master
   */
  public abstract WebServer createWebServer();

  /**
   * @return the master's web bind address
   */
  public final InetSocketAddress getWebBindAddress() {
    return mWebBindAddress;
  }

  /**
   * @return the master's web connect address
   */
  public final InetSocketAddress getWebAddress() {
    return mWebConnectAddress;
  }

  /**
   * @return true if the system is the leader (serving the rpc server), false otherwise
   */
  public boolean isGrpcServingAsLeader() {
    return mServices.stream().anyMatch(service -> service instanceof RpcServerService
        && ((RpcServerService) service).isServingLeader());
  }

  /**
   * @return true if the system is serving the web server, false otherwise
   */
  public boolean isWebServing() {
    return mServices.stream().anyMatch(service -> service instanceof WebServerService
        && ((WebServerService) service).isServing());
  }

  /**
   * @return true if the system is serving the metric sink, false otherwise
   */
  public boolean isMetricSinkServing() {
    return MetricsSystem.isStarted();
  }

  /**
   * Waits until the grpc server is ready to serve requests.
   *
   * @param timeoutMs how long to wait in milliseconds
   * @return whether the grpc server became ready before the specified timeout
   */
  public boolean waitForLeaderGrpcServerReady(int timeoutMs) {
    return pollFor(this + " to start", this::isGrpcServingAsLeader, timeoutMs);
  }

  /**
   * Waits until the web server is ready to serve requests.
   *
   * @param timeoutMs how long to wait in milliseconds
   * @return whether the web server became ready before the specified timeout
   */
  public boolean waitForWebServerReady(int timeoutMs) {
    return pollFor(this + " to start", this::isWebServing, timeoutMs);
  }

  /**
   * Wait until the metrics sinks have been started.
   *
   * @param timeoutMs how long to wait in milliseconds
   * @return whether the metrics sinks have begun serving before the specified timeout
   */
  public boolean waitForMetricSinkServing(int timeoutMs) {
    return pollFor("metrics sinks to start", this::isMetricSinkServing, timeoutMs);
  }

  private boolean pollFor(String message, Supplier<Boolean> waitFor, int timeoutMs) {
    try {
      CommonUtils.waitFor(message, waitFor, WaitForOptions.defaults().setTimeoutMs(timeoutMs));
      return true;
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      return false;
    } catch (TimeoutException e) {
      return false;
    }
  }

  @Override
  public boolean waitForReady(int timeoutMs) {
    return waitForLeaderGrpcServerReady(timeoutMs);
  }

  /**
   * @return the primary selector
   */
  public PrimarySelector getPrimarySelector() {
    return mLeaderSelector;
  }
}
