alluxio-master-2.x/microbench/src/main/java/alluxio/fsmaster/BenchStandaloneGrpcServer.java
/*
 * 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.fsmaster;

import alluxio.Constants;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.conf.Source;
import alluxio.executor.ExecutorServiceBuilder;
import alluxio.grpc.ConfigProperty;
import alluxio.grpc.FileInfo;
import alluxio.grpc.FileSystemMasterClientServiceGrpc;
import alluxio.grpc.GetConfigHashPOptions;
import alluxio.grpc.GetConfigHashPResponse;
import alluxio.grpc.GetConfigurationPOptions;
import alluxio.grpc.GetConfigurationPResponse;
import alluxio.grpc.GetServiceVersionPRequest;
import alluxio.grpc.GetServiceVersionPResponse;
import alluxio.grpc.GetStatusPRequest;
import alluxio.grpc.GetStatusPResponse;
import alluxio.grpc.GrpcServer;
import alluxio.grpc.GrpcServerAddress;
import alluxio.grpc.GrpcServerBuilder;
import alluxio.grpc.GrpcService;
import alluxio.grpc.MetaMasterConfigurationServiceGrpc;
import alluxio.grpc.PAcl;
import alluxio.grpc.ServiceType;
import alluxio.grpc.ServiceVersionClientServiceGrpc;
import alluxio.grpc.TtlAction;
import alluxio.master.AlluxioExecutorService;
import alluxio.master.meta.PathProperties;
import alluxio.security.authentication.AuthType;
import alluxio.wire.ConfigHash;

import io.grpc.Server;
import io.grpc.ServerBuilder;
import io.grpc.ServerInterceptors;
import io.grpc.ServerServiceDefinition;
import io.grpc.stub.StreamObserver;
import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.junit.Assert;

import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.TimeUnit;

public class BenchStandaloneGrpcServer {
  ServerServiceDefinition mFsMasterClientService = ServerInterceptors.intercept(
      new FileSystemMasterClientServiceGrpc.FileSystemMasterClientServiceImplBase() {
        // dummy response based on a simple './bin/alluxio fs stat /' command
        private final GetStatusPResponse mResponse = GetStatusPResponse.newBuilder()
            .setFileInfo(
                FileInfo.newBuilder()
                    .setFileId(0)
                    .setName("")
                    .setPath("/")
                    .setUfsPath("/tmp/alluxio-tmp/alluxio-0/underFSStorage")
                    .setLength(0)
                    .setBlockSizeBytes(0)
                    .setCreationTimeMs(1658424362194L)
                    .setCompleted(true)
                    .setFolder(true)
                    .setPinned(false)
                    .setCacheable(false)
                    .setPersisted(true)
                    .setLastModificationTimeMs(1658424362194L)
                    .setTtl(-1)
                    .setOwner("arthurjenoudet")
                    .setGroup("staff")
                    .setMode(493)
                    .setPersistenceState("PERSISTED")
                    .setMountPoint(false)
                    .setTtlAction(TtlAction.DELETE)
                    .setMountId(1)
                    .setInAlluxioPercentage(0)
                    .setInMemoryPercentage(0)
                    .setUfsFingerprint("")
                    .setAcl(PAcl.newBuilder()
                        .setOwner("arthurjenoudet")
                        .setOwningGroup("staff")
                        .setMode(493)
                        .setIsDefault(false)
                        .build())
                    .setDefaultAcl(PAcl.newBuilder()
                        .setOwner("arthurjenoudet")
                        .setOwningGroup("staff")
                        .setMode(0)
                        .setIsDefault(true)
                        .setIsDefaultEmpty(true)
                        .build())
                    .setReplicationMax(0)
                    .setReplicationMin(0)
                    .setLastAccessTimeMs(1658424362194L)
                    .build()
            ).build();

        @Override
        public void getStatus(GetStatusPRequest request,
            StreamObserver<GetStatusPResponse> responseObserver) {
          responseObserver.onNext(mResponse);
          responseObserver.onCompleted();
        }
      }
  );
  ServerServiceDefinition mMetaMasterConfService = ServerInterceptors.intercept(
      new MetaMasterConfigurationServiceGrpc.MetaMasterConfigurationServiceImplBase() {
        private final GetConfigurationPResponse mGetConfigResponse =
            GetConfigurationPResponse.newBuilder()
                .addClusterConfigs(
                    ConfigProperty.newBuilder()
                        .setName(PropertyKey.VERSION.getName())
                        .setSource(Source.CLUSTER_DEFAULT.toString())
                        .setValue(Configuration.getString(PropertyKey.VERSION))
                        .build()
                ).build();

        @Override
        public void getConfiguration(GetConfigurationPOptions request,
            StreamObserver<GetConfigurationPResponse> responseObserver) {
          responseObserver.onNext(mGetConfigResponse);
          responseObserver.onCompleted();
        }

        private final GetConfigHashPResponse mGetConfigHashResponse =
            new ConfigHash(Configuration.hash(), new PathProperties().hash(), 0, 0).toProto();

        @Override
        public void getConfigHash(GetConfigHashPOptions request,
            StreamObserver<GetConfigHashPResponse> responseObserver) {
          responseObserver.onNext(mGetConfigHashResponse);
          responseObserver.onCompleted();
        }
      });
  ServerServiceDefinition mServiceVersionService = ServerInterceptors.intercept(
      new ServiceVersionClientServiceGrpc.ServiceVersionClientServiceImplBase() {
        private final GetServiceVersionPResponse mResponse = GetServiceVersionPResponse.newBuilder()
            .setVersion(Constants.FILE_SYSTEM_MASTER_CLIENT_SERVICE_VERSION)
            .build();

        @Override
        public void getServiceVersion(GetServiceVersionPRequest request,
            StreamObserver<GetServiceVersionPResponse> responseObserver) {
          responseObserver.onNext(mResponse);
          responseObserver.onCompleted();
        }
      });
  private GrpcServer mAlluxioServer;
  private Server mBasicServer;
  private FileSystemBase.ServerType mServerType;

  public void setServerType(FileSystemBase.ServerType serverType) {
    mServerType = serverType;
  }

  public void startServer(int port) throws IOException {
    Assert.assertNotNull(mServerType);
    if (mServerType == FileSystemBase.ServerType.ALLUXIO_GRPC_SERVER) {
      AlluxioExecutorService executor = ExecutorServiceBuilder.buildExecutorService(
          ExecutorServiceBuilder.RpcExecutorHost.MASTER);
      mAlluxioServer = GrpcServerBuilder
          .forAddress(GrpcServerAddress.create("localhost", new InetSocketAddress(port)),
              Configuration.global())
          .executor(executor)
          .flowControlWindow(
              (int) Configuration.getBytes(PropertyKey.MASTER_NETWORK_FLOWCONTROL_WINDOW))
          .keepAliveTime(
              Configuration.getMs(PropertyKey.MASTER_NETWORK_KEEPALIVE_TIME_MS),
              TimeUnit.MILLISECONDS)
          .keepAliveTimeout(
              Configuration.getMs(PropertyKey.MASTER_NETWORK_KEEPALIVE_TIMEOUT_MS),
              TimeUnit.MILLISECONDS)
          .permitKeepAlive(
              Configuration.getMs(PropertyKey.MASTER_NETWORK_PERMIT_KEEPALIVE_TIME_MS),
              TimeUnit.MILLISECONDS)
          .maxInboundMessageSize((int) Configuration.getBytes(
              PropertyKey.MASTER_NETWORK_MAX_INBOUND_MESSAGE_SIZE))
          .addService(ServiceType.FILE_SYSTEM_MASTER_CLIENT_SERVICE,
              new GrpcService(mFsMasterClientService))
          .addService(ServiceType.META_MASTER_CONFIG_SERVICE,
              new GrpcService(mMetaMasterConfService))
          .addService(new GrpcService(mServiceVersionService).disableAuthentication())
          .build()
          .start();
    } else if (mServerType == FileSystemBase.ServerType.BASIC_GRPC_SERVER) {
      mBasicServer = ServerBuilder
          .forPort(port)
          .addService(mServiceVersionService)
          .addService(mFsMasterClientService)
          .addService(mMetaMasterConfService)
          .build()
          .start();
    }
  }

  private int getPort() {
    Assert.assertNotNull(mServerType);
    if (mServerType == FileSystemBase.ServerType.ALLUXIO_GRPC_SERVER) {
      return mAlluxioServer.getBindPort();
    } else if (mServerType == FileSystemBase.ServerType.BASIC_GRPC_SERVER) {
      return mBasicServer.getPort();
    }
    return -1;
  }

  public void shutdown() {
    if (mServerType == FileSystemBase.ServerType.ALLUXIO_GRPC_SERVER) {
      mAlluxioServer.shutdown();
    } else if (mServerType == FileSystemBase.ServerType.BASIC_GRPC_SERVER) {
      mBasicServer.shutdown();
    }
  }

  /**
   * This main function can be used to start a standalone server and use the stressbench with
   * said standalone server.
   * @param args -s | --server will specify the {@link FileSystemBase.ServerType}
   *             -p | --port will specify the port the server should bind to (0 will bind to any
   *             available open port)
   *             -cs | --client-socket to specify which port this process will report the server
   *             port chosen for the benchmark
   */
  public static void main(String[] args) throws Exception {
    Options options = new Options();
    Option serverTypeOpt = new Option("s", "server", true,
        "server type (either " + FileSystemBase.ServerType.BASIC_GRPC_SERVER.name() + " or "
            + FileSystemBase.ServerType.ALLUXIO_GRPC_SERVER.name() + ").");
    serverTypeOpt.setRequired(true);
    Option portOpt = new Option("p", "port", true, "port of server.");
    Option clientSocketOpt = new Option("cs", "client-socket", true, "Internal benchmark use. If "
        + "the server port specified is 0, the server will pick any available open port to bind "
        + "to. It will then use the port specified in this option to tell anyone listening which "
        + "port the server bound to.");
    options.addOption(serverTypeOpt);
    options.addOption(portOpt);
    options.addOption(clientSocketOpt);

    CommandLineParser parser = new DefaultParser();
    HelpFormatter formatter = new HelpFormatter();
    CommandLine cmd = null;
    try {
      cmd = parser.parse(options, args);
    } catch (ParseException e) {
      System.out.println(e.getMessage());
      formatter.printHelp("standalone-server", options);
      System.exit(1);
    }

    Configuration.set(PropertyKey.SECURITY_AUTHENTICATION_TYPE, AuthType.NOSASL);
    FileSystemBase.ServerType serverType =
        FileSystemBase.ServerType.valueOf(cmd.getOptionValue(serverTypeOpt.getOpt()));
    int port = 0;
    if (cmd.hasOption(portOpt.getOpt())) {
      port = Integer.parseInt(cmd.getOptionValue("port"));
    }
    if (serverType == FileSystemBase.ServerType.STANDALONE) {
      System.out.println("Cannot use server type " + serverType);
      formatter.printHelp("standalone-server", options);
      System.exit(1);
    }
    BenchStandaloneGrpcServer standaloneServer = new BenchStandaloneGrpcServer();
    standaloneServer.setServerType(serverType);
    standaloneServer.startServer(port);
    System.out.println("Starting standalone server of type " + serverType + " at "
        + standaloneServer.getPort());
    if (cmd.hasOption(clientSocketOpt.getOpt())) {
      int clientSocketPort = Integer.parseInt(cmd.getOptionValue(clientSocketOpt.getOpt()));
      try (Socket socket = new Socket("localhost", clientSocketPort);
           DataOutputStream stream = new DataOutputStream(socket.getOutputStream())) {
        stream.writeInt(standaloneServer.getPort());
      }
    }
    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
      System.out.println("Closing standalone server");
      standaloneServer.shutdown();
    }));
    // kill the server using ctrl+c
    Thread.currentThread().join();
  }
}
