benchmarks: Modernize client to use target and credentials

This allows using LoadClient with xDS.

The changes to SocketAddressValidator are a bit hacky, but we really
don't care about cleanliness there. Eventually on client-side, we should
be deleting the unix special case entirely, as we'll have a unix name
resolver.

Fixes #8877

Author: ejona86

benchmarks/build.gradle

@@ -27,6 +27,7 @@ dependencies {
             project(':grpc-stub'),
             project(':grpc-protobuf'),
             project(':grpc-testing'),
+            project(path: ':grpc-xds', configuration: 'shadow'),
             libraries.hdrhistogram,
             libraries.netty_tcnative,
             libraries.netty_epoll,

benchmarks/src/main/java/io/grpc/benchmarks/SocketAddressValidator.java

@@ -33,6 +33,11 @@ public interface SocketAddressValidator {
     public boolean isValidSocketAddress(SocketAddress address) {
       return address instanceof InetSocketAddress;
     }
+
+    @Override
+    public boolean isValidSocketAddress(String address) {
+      return !address.startsWith("unix://");
+    }
   };
 
   /**
@@ -43,10 +48,20 @@ public boolean isValidSocketAddress(SocketAddress address) {
     public boolean isValidSocketAddress(SocketAddress address) {
       return "DomainSocketAddress".equals(address.getClass().getSimpleName());
     }
+
+    @Override
+    public boolean isValidSocketAddress(String address) {
+      return address.startsWith("unix://");
+    }
   };
 
   /**
    * Returns {@code true} if the given address is valid.
    */
   boolean isValidSocketAddress(SocketAddress address);
+
+  /**
+   * Returns {@code true} if the given address is valid.
+   */
+  boolean isValidSocketAddress(String address);
 }

benchmarks/src/main/java/io/grpc/benchmarks/Transport.java

@@ -16,8 +16,6 @@
 
 package io.grpc.benchmarks;
 
-import java.net.SocketAddress;
-
 /**
  * All of the supported transports.
  */
@@ -49,7 +47,7 @@ public enum Transport {
    *
    * @throws IllegalArgumentException if the given address is invalid for this transport.
    */
-  public void validateSocketAddress(SocketAddress address) {
+  public void validateSocketAddress(String address) {
     if (!socketAddressValidator.isValidSocketAddress(address)) {
       throw new IllegalArgumentException(
           "Invalid address " + address + " for transport " + this);

benchmarks/src/main/java/io/grpc/benchmarks/Utils.java

@@ -18,20 +18,22 @@
 
 import static java.util.concurrent.ForkJoinPool.defaultForkJoinWorkerThreadFactory;
 
+import com.google.common.base.Preconditions;
 import com.google.common.util.concurrent.UncaughtExceptionHandlers;
 import com.google.protobuf.ByteString;
+import io.grpc.ChannelCredentials;
+import io.grpc.InsecureChannelCredentials;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Status;
+import io.grpc.TlsChannelCredentials;
 import io.grpc.benchmarks.proto.Messages;
 import io.grpc.benchmarks.proto.Messages.Payload;
 import io.grpc.benchmarks.proto.Messages.SimpleRequest;
 import io.grpc.benchmarks.proto.Messages.SimpleResponse;
-import io.grpc.internal.testing.TestUtils;
-import io.grpc.netty.GrpcSslContexts;
 import io.grpc.netty.NettyChannelBuilder;
 import io.grpc.okhttp.OkHttpChannelBuilder;
-import io.grpc.okhttp.internal.Platform;
+import io.grpc.testing.TlsTesting;
 import io.netty.channel.epoll.EpollDomainSocketChannel;
 import io.netty.channel.epoll.EpollEventLoopGroup;
 import io.netty.channel.epoll.EpollSocketChannel;
@@ -79,15 +81,10 @@ public static boolean parseBoolean(String value) {
   /**
    * Parse a {@link SocketAddress} from the given string.
    */
-  public static SocketAddress parseSocketAddress(String value) {
+  public static SocketAddress parseServerSocketAddress(String value) {
     if (value.startsWith(UNIX_DOMAIN_SOCKET_PREFIX)) {
-      // Unix Domain Socket address.
-      // Create the underlying file for the Unix Domain Socket.
-      String filePath = value.substring(UNIX_DOMAIN_SOCKET_PREFIX.length());
-      File file = new File(filePath);
-      if (!file.isAbsolute()) {
-        throw new IllegalArgumentException("File path must be absolute: " + filePath);
-      }
+      DomainSocketAddress domainAddress = parseUnixSocketAddress(value);
+      File file = new File(domainAddress.path());
       try {
         if (file.createNewFile()) {
           // If this application created the file, delete it when the application exits.
@@ -96,8 +93,7 @@ public static SocketAddress parseSocketAddress(String value) {
       } catch (IOException ex) {
         throw new RuntimeException(ex);
       }
-      // Create the SocketAddress referencing the file.
-      return new DomainSocketAddress(file);
+      return domainAddress;
     } else {
       // Standard TCP/IP address.
       String[] parts = value.split(":", 2);
@@ -111,37 +107,24 @@ public static SocketAddress parseSocketAddress(String value) {
     }
   }
 
-  private static OkHttpChannelBuilder newOkHttpClientChannel(
-      SocketAddress address, boolean tls, boolean testca) {
-    InetSocketAddress addr = (InetSocketAddress) address;
-    OkHttpChannelBuilder builder =
-        OkHttpChannelBuilder.forAddress(addr.getHostName(), addr.getPort());
-    if (!tls) {
-      builder.usePlaintext();
-    } else if (testca) {
-      try {
-        builder.sslSocketFactory(TestUtils.newSslSocketFactoryForCa(
-            Platform.get().getProvider(),
-            TestUtils.loadCert("ca.pem")));
-      } catch (Exception e) {
-        throw new RuntimeException(e);
-      }
+  private static DomainSocketAddress parseUnixSocketAddress(String value) {
+    Preconditions.checkArgument(
+        value.startsWith(UNIX_DOMAIN_SOCKET_PREFIX),
+        "Must start with %s: %s", UNIX_DOMAIN_SOCKET_PREFIX, value);
+    // Unix Domain Socket address.
+    // Create the underlying file for the Unix Domain Socket.
+    String filePath = value.substring(UNIX_DOMAIN_SOCKET_PREFIX.length());
+    File file = new File(filePath);
+    if (!file.isAbsolute()) {
+      throw new IllegalArgumentException("File path must be absolute: " + filePath);
     }
-    return builder;
+    // Create the SocketAddress referencing the file.
+    return new DomainSocketAddress(file);
   }
 
-  private static NettyChannelBuilder newNettyClientChannel(Transport transport,
-      SocketAddress address, boolean tls, boolean testca, int flowControlWindow)
-      throws IOException {
-    NettyChannelBuilder builder =
-        NettyChannelBuilder.forAddress(address).flowControlWindow(flowControlWindow);
-    if (!tls) {
-      builder.usePlaintext();
-    } else if (testca) {
-      File cert = TestUtils.loadCert("ca.pem");
-      builder.sslContext(GrpcSslContexts.forClient().trustManager(cert).build());
-    }
-
+  private static NettyChannelBuilder configureNetty(
+      NettyChannelBuilder builder, Transport transport, int flowControlWindow) {
+    builder.flowControlWindow(flowControlWindow);
     DefaultThreadFactory tf = new DefaultThreadFactory("client-elg-", true /*daemon */);
     switch (transport) {
       case NETTY_NIO:
@@ -194,17 +177,38 @@ public ForkJoinWorkerThread newThread(ForkJoinPool pool) {
   /**
    * Create a {@link ManagedChannel} for the given parameters.
    */
-  public static ManagedChannel newClientChannel(Transport transport, SocketAddress address,
+  public static ManagedChannel newClientChannel(Transport transport, String target,
         boolean tls, boolean testca, @Nullable String authorityOverride,
         int flowControlWindow, boolean directExecutor) {
+    ChannelCredentials credentials;
+    if (tls) {
+      if (testca) {
+        try {
+          credentials = TlsChannelCredentials.newBuilder()
+              .trustManager(TlsTesting.loadCert("ca.pem"))
+              .build();
+        } catch (IOException ex) {
+          throw new RuntimeException(ex);
+        }
+      } else {
+        credentials = TlsChannelCredentials.create();
+      }
+    } else {
+      credentials = InsecureChannelCredentials.create();
+    }
     ManagedChannelBuilder<?> builder;
     if (transport == Transport.OK_HTTP) {
-      builder = newOkHttpClientChannel(address, tls, testca);
+      builder = OkHttpChannelBuilder.forTarget(target, credentials)
+          .flowControlWindow(flowControlWindow);
     } else {
-      try {
-        builder = newNettyClientChannel(transport, address, tls, testca, flowControlWindow);
-      } catch (Exception e) {
-        throw new RuntimeException(e);
+      if (target.startsWith(UNIX_DOMAIN_SOCKET_PREFIX)) {
+        builder = configureNetty(
+            NettyChannelBuilder.forAddress(parseUnixSocketAddress(target), credentials),
+            transport, flowControlWindow);
+      } else {
+        builder = configureNetty(
+            NettyChannelBuilder.forTarget(target, credentials),
+            transport, flowControlWindow);
       }
     }
     if (authorityOverride != null) {

benchmarks/src/main/java/io/grpc/benchmarks/driver/LoadClient.java

@@ -82,7 +82,7 @@ class LoadClient {
       channels[i] =
           Utils.newClientChannel(
               Epoll.isAvailable() ?  Transport.NETTY_EPOLL : Transport.NETTY_NIO,
-              Utils.parseSocketAddress(config.getServerTargets(i % config.getServerTargetsCount())),
+              config.getServerTargets(i % config.getServerTargetsCount()),
               config.hasSecurityParams(),
               config.hasSecurityParams() && config.getSecurityParams().getUseTestCa(),
               config.hasSecurityParams()

benchmarks/src/main/java/io/grpc/benchmarks/qps/ClientConfiguration.java

@@ -28,8 +28,6 @@
 import io.grpc.benchmarks.proto.Messages.PayloadType;
 import io.grpc.internal.testing.TestUtils;
 import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.SocketAddress;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashSet;
@@ -47,7 +45,7 @@ public class ClientConfiguration implements Configuration {
   String authorityOverride = TestUtils.TEST_SERVER_HOST;
   boolean useDefaultCiphers;
   boolean directExecutor;
-  SocketAddress address;
+  String target;
   int channels = 4;
   int outstandingRpcsPerChannel = 10;
   int serverPayload;
@@ -66,7 +64,7 @@ private ClientConfiguration() {
   }
 
   public ManagedChannel newChannel() throws IOException {
-    return Utils.newClientChannel(transport, address, tls, testca, authorityOverride,
+    return Utils.newClientChannel(transport, target, tls, testca, authorityOverride,
         flowControlWindow, directExecutor);
   }
 
@@ -106,18 +104,10 @@ protected ClientConfiguration build0(ClientConfiguration config) {
           throw new IllegalArgumentException(
               "Transport " + config.transport.name().toLowerCase() + " does not support TLS.");
         }
-
-        if (config.transport != Transport.OK_HTTP
-            && config.testca && config.address instanceof InetSocketAddress) {
-          // Override the socket address with the host from the testca.
-          InetSocketAddress address = (InetSocketAddress) config.address;
-          config.address = TestUtils.testServerAddress(address.getHostName(),
-                  address.getPort());
-        }
       }
 
       // Verify that the address type is correct for the transport type.
-      config.transport.validateSocketAddress(config.address);
+      config.transport.validateSocketAddress(config.target);
 
       return config;
     }
@@ -136,7 +126,7 @@ enum ClientParam implements AbstractConfigurationBuilder.Param {
         + "(unix:///path/to/file), depending on the transport selected.", null, true) {
       @Override
       protected void setClientValue(ClientConfiguration config, String value) {
-        config.address = Utils.parseSocketAddress(value);
+        config.target = value;
       }
     },
     CHANNELS("INT", "Number of Channels.", "" + DEFAULT.channels) {

benchmarks/src/main/java/io/grpc/benchmarks/qps/ServerConfiguration.java

@@ -142,7 +142,7 @@ enum ServerParam implements AbstractConfigurationBuilder.Param {
         + "(unix:///path/to/file), depending on the transport selected.", null, true) {
       @Override
       protected void setServerValue(ServerConfiguration config, String value) {
-        SocketAddress address = Utils.parseSocketAddress(value);
+        SocketAddress address = Utils.parseServerSocketAddress(value);
         if (address instanceof InetSocketAddress) {
           InetSocketAddress addr = (InetSocketAddress) address;
           int port = addr.getPort() == 0 ? Utils.pickUnusedPort() : addr.getPort();