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

import alluxio.conf.AlluxioConfiguration;
import alluxio.conf.PropertyKey;
import alluxio.exception.status.AlluxioStatusException;
import alluxio.grpc.GetServiceVersionPRequest;
import alluxio.grpc.GrpcChannel;
import alluxio.grpc.GrpcChannelBuilder;
import alluxio.grpc.GrpcServerAddress;
import alluxio.grpc.ServiceVersionClientServiceGrpc;
import alluxio.security.user.UserState;
import alluxio.util.OSUtils;
import alluxio.wire.WorkerNetAddress;

import com.google.common.base.Preconditions;
import io.netty.channel.unix.DomainSocketAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.Inet4Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.util.Collections;
import java.util.Enumeration;
import java.util.List;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Common network address related utilities shared by all components in Alluxio.
 */
@ThreadSafe
public final class NetworkAddressUtils {
  private static final Logger LOG = LoggerFactory.getLogger(NetworkAddressUtils.class);

  public static final String WILDCARD_ADDRESS = "0.0.0.0";

  /**
   * Checks if the underlying OS is Windows.
   */
  public static final boolean WINDOWS = OSUtils.isWindows();

  private static String sLocalHost;
  private static String sLocalHostMetricName;
  private static String sLocalIP;

  private NetworkAddressUtils() {}

  /**
   * An interface to get service attributes.
   */
  public interface ServiceAttributeProvider {
    /**
     * Gets service name.
     *
     * @return service name
     */
    String getServiceName();

    /**
     * Gets the key of connect hostname.
     *
     * @return key of connect hostname
     */
    PropertyKey getHostNameKey();

    /**
     * Gets the key of bind hostname.
     *
     * @return key of bind hostname
     */
    PropertyKey getBindHostKey();

    /**
     * Gets the key of service port.
     *
     * @return key of service port
     */
    PropertyKey getPortKey();

    /**
     * Gets the default port number on service.
     *
     * @return default port
     */
    int getDefaultPort();
  }

  /**
   * Different types of services that client uses to connect. These types also indicate the service
   * bind address.
   */
  public enum ServiceType implements ServiceAttributeProvider {
    /**
     * FUSE web service (Jetty).
     */
    FUSE_WEB("Alluxio FUSE Web service", PropertyKey.FUSE_WEB_HOSTNAME,
        PropertyKey.FUSE_WEB_BIND_HOST, PropertyKey.FUSE_WEB_PORT),
    /**
     * Job master Raft service (Netty). The bind and connect hosts are the same because the
     * underlying Raft implementation doesn't differentiate between bind and connect hosts.
     */
    JOB_MASTER_RAFT("Alluxio Job Master Raft service", PropertyKey.JOB_MASTER_HOSTNAME,
        PropertyKey.JOB_MASTER_HOSTNAME, PropertyKey.JOB_MASTER_EMBEDDED_JOURNAL_PORT),

    /**
     * Master Raft service (Netty). The bind and connect hosts are the same because the
     * underlying Raft implementation doesn't differentiate between bind and connect hosts.
     */
    MASTER_RAFT("Alluxio Master Raft service", PropertyKey.MASTER_HOSTNAME,
        PropertyKey.MASTER_HOSTNAME, PropertyKey.MASTER_EMBEDDED_JOURNAL_PORT),

    /**
     * Job master RPC service (gRPC).
     */
    JOB_MASTER_RPC("Alluxio Job Manager Master RPC service", PropertyKey.JOB_MASTER_HOSTNAME,
        PropertyKey.JOB_MASTER_BIND_HOST, PropertyKey.JOB_MASTER_RPC_PORT),

    /**
     * Job master web service (Jetty).
     */
    JOB_MASTER_WEB("Alluxio Job Manager Master Web service", PropertyKey.JOB_MASTER_WEB_HOSTNAME,
        PropertyKey.JOB_MASTER_WEB_BIND_HOST, PropertyKey.JOB_MASTER_WEB_PORT),

    /**
     * Job worker RPC service (gRPC).
     */
    JOB_WORKER_RPC("Alluxio Job Manager Worker RPC service", PropertyKey.JOB_WORKER_HOSTNAME,
        PropertyKey.JOB_WORKER_BIND_HOST, PropertyKey.JOB_WORKER_RPC_PORT),

    /**
     * Job master web service (Jetty).
     */
    JOB_WORKER_WEB("Alluxio Job Manager Worker Web service", PropertyKey.WORKER_WEB_HOSTNAME,
        PropertyKey.JOB_WORKER_WEB_BIND_HOST, PropertyKey.JOB_WORKER_WEB_PORT),

    /**
     * Master RPC service (gRPC).
     */
    MASTER_RPC("Alluxio Master RPC service", PropertyKey.MASTER_HOSTNAME,
        PropertyKey.MASTER_BIND_HOST, PropertyKey.MASTER_RPC_PORT),

    /**
     * Master web service (Jetty).
     */
    MASTER_WEB("Alluxio Master Web service", PropertyKey.MASTER_WEB_HOSTNAME,
        PropertyKey.MASTER_WEB_BIND_HOST, PropertyKey.MASTER_WEB_PORT),

    /**
     * Worker RPC service (gRPC).
     */
    WORKER_RPC("Alluxio Worker RPC service", PropertyKey.WORKER_HOSTNAME,
        PropertyKey.WORKER_BIND_HOST, PropertyKey.WORKER_RPC_PORT),

    /**
     * Worker data service (Netty).
     */
    WORKER_DATA("Alluxio Worker data service", PropertyKey.WORKER_DATA_HOSTNAME,
        PropertyKey.WORKER_DATA_BIND_HOST, PropertyKey.WORKER_DATA_PORT),

    /**
     * Worker s3 service (Netty).
     */
    WORKER_REST("Alluxio Worker S3 service", PropertyKey.WORKER_WEB_HOSTNAME,
        PropertyKey.WORKER_WEB_BIND_HOST, PropertyKey.WORKER_REST_PORT),

    /**
     * Worker web service (Jetty).
     */
    WORKER_WEB("Alluxio Worker Web service", PropertyKey.WORKER_WEB_HOSTNAME,
        PropertyKey.WORKER_WEB_BIND_HOST, PropertyKey.WORKER_WEB_PORT),

    /**
     * Proxy web service (Jetty).
     */
    PROXY_WEB("Alluxio Proxy Web service", PropertyKey.PROXY_WEB_HOSTNAME,
        PropertyKey.PROXY_WEB_BIND_HOST, PropertyKey.PROXY_WEB_PORT),
    ;

    // service name
    private final String mServiceName;

    // the key of connect hostname
    private final PropertyKey mHostNameKey;

    // the key of bind hostname
    private final PropertyKey mBindHostKey;

    // the key of service port
    private final PropertyKey mPortKey;

    ServiceType(String serviceName, PropertyKey hostNameKey, PropertyKey bindHostKey,
        PropertyKey portKey) {
      mServiceName = serviceName;
      mHostNameKey = hostNameKey;
      mBindHostKey = bindHostKey;
      mPortKey = portKey;
    }

    /**
     * Gets service name.
     *
     * @return service name
     */
    public String getServiceName() {
      return mServiceName;
    }

    /**
     * Gets the key of connect hostname.
     *
     * @return key of connect hostname
     */
    public PropertyKey getHostNameKey() {
      return mHostNameKey;
    }

    /**
     * Gets the key of bind hostname.
     *
     * @return key of bind hostname
     */
    public PropertyKey getBindHostKey() {
      return mBindHostKey;
    }

    /**
     * Gets the key of service port.
     *
     * @return key of service port
     */
    public PropertyKey getPortKey() {
      return mPortKey;
    }

    /**
     * Gets the default port number on service.
     *
     * @return default port
     */
    public int getDefaultPort() {
      return (int) mPortKey.getDefaultValue();
    }
  }

  /**
   * Checks if the given port is valid.
   *
   * @param port the port to check
   */
  public static void assertValidPort(final int port) {
    Preconditions.checkArgument(port < 65536, "Port must be less than 65536");
    Preconditions.checkArgument(port >= 0, "Port must be non-negative");
  }

  /**
   * Helper method to get the {@link InetSocketAddress} address for client to communicate with the
   * service.
   *
   * @param service the service name used to connect
   * @param conf the configuration containing the address to connect to
   * @return the service address that a client (typically outside the service machine) uses to
   *         communicate with service.
   */
  public static InetSocketAddress getConnectAddress(ServiceAttributeProvider service,
                                                    AlluxioConfiguration conf) {
    return InetSocketAddress.createUnresolved(getConnectHost(service, conf),
        getPort(service, conf));
  }

  /**
   * Provides an externally resolvable hostname for client to communicate with the service. If the
   * hostname is not explicitly specified, Alluxio will try to use the bind host. If the bind host
   * is wildcard, Alluxio will automatically determine an appropriate hostname from local machine.
   * The various possibilities shown in the following table:
   * <table>
   * <caption>Hostname Scenarios</caption>
   * <thead>
   * <tr>
   * <th>Specified Hostname</th>
   * <th>Specified Bind Host</th>
   * <th>Enable Network IP Address Used</th>
   * <th>Returned Connect Host</th>
   * </tr>
   * </thead>
   * <tbody>
   * <tr>
   * <td>hostname</td>
   * <td>hostname</td>
   * <td>true/false</td>
   * <td>hostname</td>
   * </tr>
   * <tr>
   * <td>not defined</td>
   * <td>hostname</td>
   * <td>true/false</td>
   * <td>hostname</td>
   * </tr>
   * <tr>
   * <td>hostname</td>
   * <td>0.0.0.0 or not defined</td>
   * <td>true/false</td>
   * <td>hostname</td>
   * </tr>
   * <tr>
   * <td>not defined</td>
   * <td>0.0.0.0 or not defined</td>
   * <td>false</td>
   * <td>local hostname</td>
   * </tr>
   * <tr>
   * <td>not defined</td>
   * <td>0.0.0.0 or not defined</td>
   * <td>true</td>
   * <td>local IP address</td>
   * </tr>
   * </tbody>
   * </table>
   *
   * @param service Service type used to connect
   * @param conf configuration
   * @return the externally resolvable hostname that the client can use to communicate with the
   *         service.
   */
  public static String getConnectHost(ServiceAttributeProvider service, AlluxioConfiguration conf) {
    if (conf.isSet(service.getHostNameKey())) {
      String connectHost = conf.getString(service.getHostNameKey());
      if (!connectHost.isEmpty() && !connectHost.equals(WILDCARD_ADDRESS)) {
        return connectHost;
      }
    }
    if (conf.isSet(service.getBindHostKey())) {
      String bindHost = conf.getString(service.getBindHostKey());
      if (!bindHost.isEmpty() && !bindHost.equals(WILDCARD_ADDRESS)) {
        return bindHost;
      }
    }
    if (conf.getBoolean(PropertyKey.NETWORK_IP_ADDRESS_USED)) {
      return getLocalIpAddress((int) conf.getMs(PropertyKey.NETWORK_HOST_RESOLUTION_TIMEOUT_MS));
    }
    return getLocalHostName((int) conf.getMs(PropertyKey.NETWORK_HOST_RESOLUTION_TIMEOUT_MS));
  }

  /**
   * Gets the port number on a given service type. If user defined port number is not explicitly
   * specified, Alluxio will use the default service port.
   *
   * @param service Service type used to connect
   * @param conf Alluxio configuration
   * @return the service port number
   */
  public static int getPort(ServiceAttributeProvider service, AlluxioConfiguration conf) {
    return conf.getInt(service.getPortKey());
  }

  /**
   * Helper method to get the bind hostname for a given service.
   *
   * @param service the service name
   * @param conf Alluxio configuration
   * @return the InetSocketAddress the service will bind to
   */
  public static InetSocketAddress getBindAddress(ServiceAttributeProvider service,
                                                 AlluxioConfiguration conf) {
    int port = getPort(service, conf);
    assertValidPort(port);
    return new InetSocketAddress(getBindHost(service, conf), port);
  }

  /**
   * Helper method to get the {@link InetSocketAddress} bind address on a given service.
   * <p>
   * Host bind information searching order:
   * <ol>
   * <li>System properties or environment variables via alluxio-env.sh
   * <li>Default properties via alluxio-default.properties file
   * <li>A externally resolvable local hostname for the host this JVM is running on
   * </ol>
   *
   * @param service the service name
   * @param conf Alluxio configuration
   * @return the bind hostname
   */
  public static String getBindHost(ServiceAttributeProvider service, AlluxioConfiguration conf) {
    if (conf.isSet(service.getBindHostKey())
            && !conf.getString(service.getBindHostKey()).isEmpty()) {
      return conf.getString(service.getBindHostKey());
    } else {
      return getLocalHostName((int) conf.getMs(PropertyKey.NETWORK_HOST_RESOLUTION_TIMEOUT_MS));
    }
  }

  /**
   * Gets the local hostname to be used by the client. If this isn't configured, a
   * non-loopback local hostname will be looked up.
   *
   * @param conf Alluxio configuration
   * @return the local hostname for the client
   */
  public static String getClientHostName(AlluxioConfiguration conf) {
    if (conf.isSet(PropertyKey.USER_HOSTNAME)) {
      return conf.getString(PropertyKey.USER_HOSTNAME);
    }
    return getLocalHostName((int) conf.getMs(PropertyKey.NETWORK_HOST_RESOLUTION_TIMEOUT_MS));
  }

  /**
   * Gets a local host name for the host this JVM is running on.
   *
   * @param timeoutMs Timeout in milliseconds to use for checking that a possible local host is
   *        reachable
   * @return the local host name, which is not based on a loopback ip address
   */
  public static synchronized String getLocalHostName(int timeoutMs) {
    if (sLocalHost != null) {
      return sLocalHost;
    }

    try {
      sLocalHost = InetAddress.getByName(getLocalIpAddress(timeoutMs)).getCanonicalHostName();
      return sLocalHost;
    } catch (UnknownHostException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * Gets a local hostname for the host this JVM is running on with '.' replaced with '_' for
   * metrics usage.
   *
   * @param timeoutMs Timeout in milliseconds to use for checking that a possible local host is
   * reachable
   * @return the metrics system friendly local host name
   */
  public static synchronized String getLocalHostMetricName(int timeoutMs) {
    if (sLocalHostMetricName != null) {
      return sLocalHostMetricName;
    }
    sLocalHostMetricName = getLocalHostName(timeoutMs).replace('.', '_');
    return sLocalHostMetricName;
  }

  /**
   * Gets a local IP address for the host this JVM is running on.
   *
   * @param timeoutMs Timeout in milliseconds to use for checking that a possible local IP is
   *        reachable
   * @return the local ip address, which is not a loopback address and is reachable
   */
  public static synchronized String getLocalIpAddress(int timeoutMs) {
    if (sLocalIP != null) {
      return sLocalIP;
    }

    try {
      InetAddress address = InetAddress.getLocalHost();
      LOG.debug("address: {} isLoopbackAddress: {}, with host {} {}", address,
          address.isLoopbackAddress(), address.getHostAddress(), address.getHostName());

      // Make sure that the address is actually reachable since in some network configurations
      // it is possible for the InetAddress.getLocalHost() call to return a non-reachable
      // address e.g. a broadcast address
      if (!isValidAddress(address, timeoutMs)) {
        Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();

        // Make getNetworkInterfaces have the same order of network interfaces as listed on
        // unix-like systems. This optimization can help avoid getting some special addresses, such
        // as loopback address"127.0.0.1", virtual bridge address "192.168.122.1" as far as
        // possible.
        if (!WINDOWS) {
          List<NetworkInterface> netIFs = Collections.list(networkInterfaces);
          Collections.reverse(netIFs);
          networkInterfaces = Collections.enumeration(netIFs);
        }

        while (networkInterfaces.hasMoreElements()) {
          NetworkInterface ni = networkInterfaces.nextElement();
          Enumeration<InetAddress> addresses = ni.getInetAddresses();
          while (addresses.hasMoreElements()) {
            address = addresses.nextElement();

            // Address must not be link local or loopback. And it must be reachable
            if (isValidAddress(address, timeoutMs)) {
              sLocalIP = address.getHostAddress();
              return sLocalIP;
            }
          }
        }

        LOG.warn("Your hostname, {} resolves to a loopback/non-reachable address: {}, "
                  +  "but we couldn't find any external IP address!",
                  InetAddress.getLocalHost().getHostName(), address.getHostAddress());
      }

      sLocalIP = address.getHostAddress();
      return sLocalIP;
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  /**
   * @param host the host to try to connect to
   * @param port the port to try to connect on
   * @return whether a socket connection can be made to the given host on the given port
   */
  public static boolean isServing(String host, int port) {
    if (port < 0) {
      return false;
    }
    try {
      Socket socket = new Socket(host, port);
      socket.close();
      return true;
    } catch (IOException e) {
      return false;
    }
  }

  /**
   * @param targetAddress the target address, hostname or IP
   * @param timeoutMs Timeout in milliseconds to use for checking that a possible local host is
   *        reachable
   * @return  true if the target address is the local address, false otherwise
   */
  public static boolean isLocalAddress(String targetAddress, int timeoutMs) {
    return getLocalHostName(timeoutMs).equals(targetAddress)
        || getLocalIpAddress(timeoutMs).equals(targetAddress);
  }

  /**
   * Tests if the address is externally resolvable. Address must not be wildcard, link local,
   * loopback address, non-IPv4, or other unreachable addresses.
   *
   * @param address The testing address
   * @param timeoutMs Timeout in milliseconds to use for checking that a possible local IP is
   *        reachable
   * @return a {@code boolean} indicating if the given address is externally resolvable address
   */
  private static boolean isValidAddress(InetAddress address, int timeoutMs) throws IOException {
    return !address.isAnyLocalAddress() && !address.isLinkLocalAddress()
        && !address.isLoopbackAddress() && address.isReachable(timeoutMs)
        && (address instanceof Inet4Address);
  }

  /**
   * Resolves a given hostname by a canonical hostname. When a hostname alias (e.g., those specified
   * in /etc/hosts) is given, the alias may not be resolvable on other hosts in a cluster unless the
   * same alias is defined there.
   *
   * @param hostname the input hostname, which could be an alias
   * @return the canonical form of the hostname, or null if it is null or empty
   * @throws UnknownHostException if the given hostname cannot be resolved
   */
  @Nullable
  public static String resolveHostName(String hostname) throws UnknownHostException {
    if (hostname == null || hostname.isEmpty()) {
      return null;
    }

    return InetAddress.getByName(hostname).getCanonicalHostName();
  }

  /**
   * Resolves a given hostname IP address.
   *
   * @param hostname the input hostname, which could be an alias
   * @return the hostname IP address
   * @throws UnknownHostException if the given hostname cannot be resolved
   */
  public static String resolveIpAddress(String hostname) throws UnknownHostException {
    Preconditions.checkNotNull(hostname, "hostname");
    Preconditions.checkArgument(!hostname.isEmpty(),
            "Cannot resolve IP address for empty hostname");
    return InetAddress.getByName(hostname).getHostAddress();
  }

  /**
   * Gets FQDN(Full Qualified Domain Name) from Java representations of network address, except
   * String representation which should be handled by {@link #resolveHostName(String)} which will
   * handle the situation where hostname is null.
   *
   * @param addr the input network address representation, can not be null
   * @return the resolved FQDN host name
   */
  public static String getFqdnHost(InetSocketAddress addr) {
    Preconditions.checkNotNull(addr.getAddress(), "the address of " + addr + " is invalid.");
    return addr.getAddress().getCanonicalHostName();
  }

  /**
   * Gets FQDN(Full Qualified Domain Name) from Alluxio representation of network address.
   *
   * @param addr the input network address representation
   * @return the resolved FQDN host name
   * @throws UnknownHostException if the host is not known
   */
  public static String getFqdnHost(WorkerNetAddress addr) throws UnknownHostException {
    return resolveHostName(addr.getHost());
  }

  /**
   * Parses {@link InetSocketAddress} from a String.
   *
   * @param address socket address to parse
   * @return InetSocketAddress of the String
   */
  @Nullable
  public static InetSocketAddress parseInetSocketAddress(String address) throws IOException {
    if (address == null) {
      return null;
    }
    String[] strArr = address.split(":");
    if (strArr.length != 2) {
      throw new IOException("Invalid InetSocketAddress " + address);
    }
    // a typical resolved InetSocketAddress has a string representation of form
    // <hostname>/<address>:port, e.g., "localhost/127.0.0.1:9000".
    // extract the <hostname> part by splitting at "/"
    String hostname = strArr[0].split("/")[0];
    return InetSocketAddress.createUnresolved(hostname, Integer.parseInt(strArr[1]));
  }

  /**
   * Extracts dataPort socket address from Alluxio representation of network address.
   *
   * @param netAddress the input network address representation
   * @param conf Alluxio configuration
   * @return the socket address
   */
  public static SocketAddress getDataPortSocketAddress(WorkerNetAddress netAddress,
      AlluxioConfiguration conf) {
    SocketAddress address;
    if (NettyUtils.isDomainSocketAccessible(netAddress, conf)) {
      address = new DomainSocketAddress(netAddress.getDomainSocketPath());
    } else {
      String host = netAddress.getHost();
      // issues#11172: If the worker is in a container, use the container hostname
      // to establish the connection.
      if (!netAddress.getContainerHost().equals("")) {
        LOG.debug("Worker is in a container. Use container host {} instead of physical host {}",
                netAddress.getContainerHost(), host);
        host = netAddress.getContainerHost();
      }

      int port = netAddress.getDataPort();
      address = new InetSocketAddress(host, port);
    }
    return address;
  }

  /**
   * Extracts rpcPort socket address from Alluxio representation of network address.
   *
   * @param netAddress the input network address representation
   * @param conf Alluxio configuration
   * @return the socket address
   */
  public static SocketAddress getRpcPortSocketAddress(WorkerNetAddress netAddress,
                                                       AlluxioConfiguration conf) {
    SocketAddress address;
    if (NettyUtils.isDomainSocketAccessible(netAddress, conf)) {
      address = new DomainSocketAddress(netAddress.getDomainSocketPath());
    } else {
      String host = netAddress.getHost();
      // issues#11172: If the worker is in a container, use the container hostname
      // to establish the connection.
      if (!netAddress.getContainerHost().equals("")) {
        LOG.debug("Worker is in a container. Use container host {} instead of physical host {}",
            netAddress.getContainerHost(), host);
        host = netAddress.getContainerHost();
      }

      int port = netAddress.getRpcPort();
      address = new InetSocketAddress(host, port);
    }
    return address;
  }

  /**
   * Test if the input address is serving an Alluxio service. This method make use of the
   * gRPC protocol for performing service communication.
   * This method throws UnauthenticatedException if the user is not authenticated,
   * StatusRuntimeException If the host not reachable or does not serve the given service.
   *
   * @param address the network address to ping
   * @param serviceType the Alluxio service type
   * @param conf Alluxio configuration
   * @param userState the UserState
   */
  public static void pingService(InetSocketAddress address, alluxio.grpc.ServiceType serviceType,
      AlluxioConfiguration conf, UserState userState)
      throws AlluxioStatusException {
    Preconditions.checkNotNull(address, "address");
    Preconditions.checkNotNull(serviceType, "serviceType");
    GrpcChannel channel = GrpcChannelBuilder.newBuilder(GrpcServerAddress.create(address), conf)
        .disableAuthentication().setSubject(userState.getSubject())
        .build();
    try {
      ServiceVersionClientServiceGrpc.ServiceVersionClientServiceBlockingStub versionClient =
          ServiceVersionClientServiceGrpc.newBlockingStub(channel)
              .withDeadlineAfter(conf.getMs(PropertyKey.USER_MASTER_POLLING_TIMEOUT),
                  TimeUnit.MILLISECONDS);
      versionClient.getServiceVersion(
          GetServiceVersionPRequest.newBuilder().setServiceType(serviceType).build());
    } catch (Throwable t) {
      throw AlluxioStatusException.fromThrowable(t);
    } finally {
      channel.shutdown();
    }
  }

  /**
   * @param clusterAddresses addresses of all nodes in the Raft cluster
   * @param conf Alluxio configuration
   * @return true if the cluster addresses contain the local IP, false otherwise
   */
  public static boolean containsLocalIp(List<InetSocketAddress> clusterAddresses,
      AlluxioConfiguration conf) {
    String localAddressIp = getLocalIpAddress((int) conf.getMs(PropertyKey
        .NETWORK_HOST_RESOLUTION_TIMEOUT_MS));
    for (InetSocketAddress addr : clusterAddresses) {
      String clusterNodeIp;
      try {
        clusterNodeIp = InetAddress.getByName(addr.getHostName()).getHostAddress();
        if (clusterNodeIp.equals(localAddressIp)) {
          return true;
        }
      } catch (UnknownHostException e) {
        LOG.error("Get raft cluster node ip by hostname({}) failed",
            addr.getHostName(), e);
      }
    }
    return false;
  }
}
