Add stress test client. Fixes #1584

As of the discussion in #1584, the client does not
support TLS and interop tests that require auth are
yet to be implemented.
It has the same functionality as the C++ stress test client.

interop-testing/build.gradle

@@ -69,10 +69,19 @@ task reconnect_test_client(type: CreateStartScripts) {
     classpath = jar.outputs.files + configurations.runtime + configurations.tcnative
 }
 
+task stresstest_client(type: CreateStartScripts) {
+    mainClassName = "io.grpc.testing.integration.StressTestClient"
+    applicationName = "stresstest-client"
+    outputDir = new File(project.buildDir, 'tmp')
+    classpath = jar.outputs.files + configurations.runtime + configurations.tcnative
+    defaultJvmOpts = ["-verbose:gc"]
+}
+
 applicationDistribution.into("bin") {
     from(test_client)
     from(test_server)
     from(reconnect_test_client)
+    from(stresstest_client)
     fileMode = 0755
 }
 

interop-testing/src/generated/main/grpc/io/grpc/testing/integration/MetricsServiceGrpc.java

@@ -0,0 +1,254 @@
+package io.grpc.testing.integration;
+
+import static io.grpc.stub.ClientCalls.asyncUnaryCall;
+import static io.grpc.stub.ClientCalls.asyncServerStreamingCall;
+import static io.grpc.stub.ClientCalls.asyncClientStreamingCall;
+import static io.grpc.stub.ClientCalls.asyncBidiStreamingCall;
+import static io.grpc.stub.ClientCalls.blockingUnaryCall;
+import static io.grpc.stub.ClientCalls.blockingServerStreamingCall;
+import static io.grpc.stub.ClientCalls.futureUnaryCall;
+import static io.grpc.MethodDescriptor.generateFullMethodName;
+import static io.grpc.stub.ServerCalls.asyncUnaryCall;
+import static io.grpc.stub.ServerCalls.asyncServerStreamingCall;
+import static io.grpc.stub.ServerCalls.asyncClientStreamingCall;
+import static io.grpc.stub.ServerCalls.asyncBidiStreamingCall;
+import static io.grpc.stub.ServerCalls.asyncUnimplementedUnaryCall;
+import static io.grpc.stub.ServerCalls.asyncUnimplementedStreamingCall;
+
+@javax.annotation.Generated(
+    value = "by gRPC proto compiler (version 0.14.0-SNAPSHOT)",
+    comments = "Source: io/grpc/testing/integration/metrics.proto")
+public class MetricsServiceGrpc {
+
+  private MetricsServiceGrpc() {}
+
+  public static final String SERVICE_NAME = "grpc.testing.MetricsService";
+
+  // Static method descriptors that strictly reflect the proto.
+  @io.grpc.ExperimentalApi
+  public static final io.grpc.MethodDescriptor<io.grpc.testing.integration.Metrics.EmptyMessage,
+      io.grpc.testing.integration.Metrics.GaugeResponse> METHOD_GET_ALL_GAUGES =
+      io.grpc.MethodDescriptor.create(
+          io.grpc.MethodDescriptor.MethodType.SERVER_STREAMING,
+          generateFullMethodName(
+              "grpc.testing.MetricsService", "GetAllGauges"),
+          io.grpc.protobuf.ProtoUtils.marshaller(io.grpc.testing.integration.Metrics.EmptyMessage.getDefaultInstance()),
+          io.grpc.protobuf.ProtoUtils.marshaller(io.grpc.testing.integration.Metrics.GaugeResponse.getDefaultInstance()));
+  @io.grpc.ExperimentalApi
+  public static final io.grpc.MethodDescriptor<io.grpc.testing.integration.Metrics.GaugeRequest,
+      io.grpc.testing.integration.Metrics.GaugeResponse> METHOD_GET_GAUGE =
+      io.grpc.MethodDescriptor.create(
+          io.grpc.MethodDescriptor.MethodType.UNARY,
+          generateFullMethodName(
+              "grpc.testing.MetricsService", "GetGauge"),
+          io.grpc.protobuf.ProtoUtils.marshaller(io.grpc.testing.integration.Metrics.GaugeRequest.getDefaultInstance()),
+          io.grpc.protobuf.ProtoUtils.marshaller(io.grpc.testing.integration.Metrics.GaugeResponse.getDefaultInstance()));
+
+  public static MetricsServiceStub newStub(io.grpc.Channel channel) {
+    return new MetricsServiceStub(channel);
+  }
+
+  public static MetricsServiceBlockingStub newBlockingStub(
+      io.grpc.Channel channel) {
+    return new MetricsServiceBlockingStub(channel);
+  }
+
+  public static MetricsServiceFutureStub newFutureStub(
+      io.grpc.Channel channel) {
+    return new MetricsServiceFutureStub(channel);
+  }
+
+  public static interface MetricsService {
+
+    public void getAllGauges(io.grpc.testing.integration.Metrics.EmptyMessage request,
+        io.grpc.stub.StreamObserver<io.grpc.testing.integration.Metrics.GaugeResponse> responseObserver);
+
+    public void getGauge(io.grpc.testing.integration.Metrics.GaugeRequest request,
+        io.grpc.stub.StreamObserver<io.grpc.testing.integration.Metrics.GaugeResponse> responseObserver);
+  }
+
+  public static abstract class AbstractMetricsService implements MetricsService, io.grpc.BindableService {
+
+    @java.lang.Override
+    public void getAllGauges(io.grpc.testing.integration.Metrics.EmptyMessage request,
+        io.grpc.stub.StreamObserver<io.grpc.testing.integration.Metrics.GaugeResponse> responseObserver) {
+      asyncUnimplementedUnaryCall(METHOD_GET_ALL_GAUGES, responseObserver);
+    }
+
+    @java.lang.Override
+    public void getGauge(io.grpc.testing.integration.Metrics.GaugeRequest request,
+        io.grpc.stub.StreamObserver<io.grpc.testing.integration.Metrics.GaugeResponse> responseObserver) {
+      asyncUnimplementedUnaryCall(METHOD_GET_GAUGE, responseObserver);
+    }
+
+    @java.lang.Override public io.grpc.ServerServiceDefinition bindService() {
+      return MetricsServiceGrpc.bindService(this);
+    }
+  }
+
+  public static interface MetricsServiceBlockingClient {
+
+    public java.util.Iterator<io.grpc.testing.integration.Metrics.GaugeResponse> getAllGauges(
+        io.grpc.testing.integration.Metrics.EmptyMessage request);
+
+    public io.grpc.testing.integration.Metrics.GaugeResponse getGauge(io.grpc.testing.integration.Metrics.GaugeRequest request);
+  }
+
+  public static interface MetricsServiceFutureClient {
+
+    public com.google.common.util.concurrent.ListenableFuture<io.grpc.testing.integration.Metrics.GaugeResponse> getGauge(
+        io.grpc.testing.integration.Metrics.GaugeRequest request);
+  }
+
+  public static class MetricsServiceStub extends io.grpc.stub.AbstractStub<MetricsServiceStub>
+      implements MetricsService {
+    private MetricsServiceStub(io.grpc.Channel channel) {
+      super(channel);
+    }
+
+    private MetricsServiceStub(io.grpc.Channel channel,
+        io.grpc.CallOptions callOptions) {
+      super(channel, callOptions);
+    }
+
+    @java.lang.Override
+    protected MetricsServiceStub build(io.grpc.Channel channel,
+        io.grpc.CallOptions callOptions) {
+      return new MetricsServiceStub(channel, callOptions);
+    }
+
+    @java.lang.Override
+    public void getAllGauges(io.grpc.testing.integration.Metrics.EmptyMessage request,
+        io.grpc.stub.StreamObserver<io.grpc.testing.integration.Metrics.GaugeResponse> responseObserver) {
+      asyncServerStreamingCall(
+          getChannel().newCall(METHOD_GET_ALL_GAUGES, getCallOptions()), request, responseObserver);
+    }
+
+    @java.lang.Override
+    public void getGauge(io.grpc.testing.integration.Metrics.GaugeRequest request,
+        io.grpc.stub.StreamObserver<io.grpc.testing.integration.Metrics.GaugeResponse> responseObserver) {
+      asyncUnaryCall(
+          getChannel().newCall(METHOD_GET_GAUGE, getCallOptions()), request, responseObserver);
+    }
+  }
+
+  public static class MetricsServiceBlockingStub extends io.grpc.stub.AbstractStub<MetricsServiceBlockingStub>
+      implements MetricsServiceBlockingClient {
+    private MetricsServiceBlockingStub(io.grpc.Channel channel) {
+      super(channel);
+    }
+
+    private MetricsServiceBlockingStub(io.grpc.Channel channel,
+        io.grpc.CallOptions callOptions) {
+      super(channel, callOptions);
+    }
+
+    @java.lang.Override
+    protected MetricsServiceBlockingStub build(io.grpc.Channel channel,
+        io.grpc.CallOptions callOptions) {
+      return new MetricsServiceBlockingStub(channel, callOptions);
+    }
+
+    @java.lang.Override
+    public java.util.Iterator<io.grpc.testing.integration.Metrics.GaugeResponse> getAllGauges(
+        io.grpc.testing.integration.Metrics.EmptyMessage request) {
+      return blockingServerStreamingCall(
+          getChannel(), METHOD_GET_ALL_GAUGES, getCallOptions(), request);
+    }
+
+    @java.lang.Override
+    public io.grpc.testing.integration.Metrics.GaugeResponse getGauge(io.grpc.testing.integration.Metrics.GaugeRequest request) {
+      return blockingUnaryCall(
+          getChannel(), METHOD_GET_GAUGE, getCallOptions(), request);
+    }
+  }
+
+  public static class MetricsServiceFutureStub extends io.grpc.stub.AbstractStub<MetricsServiceFutureStub>
+      implements MetricsServiceFutureClient {
+    private MetricsServiceFutureStub(io.grpc.Channel channel) {
+      super(channel);
+    }
+
+    private MetricsServiceFutureStub(io.grpc.Channel channel,
+        io.grpc.CallOptions callOptions) {
+      super(channel, callOptions);
+    }
+
+    @java.lang.Override
+    protected MetricsServiceFutureStub build(io.grpc.Channel channel,
+        io.grpc.CallOptions callOptions) {
+      return new MetricsServiceFutureStub(channel, callOptions);
+    }
+
+    @java.lang.Override
+    public com.google.common.util.concurrent.ListenableFuture<io.grpc.testing.integration.Metrics.GaugeResponse> getGauge(
+        io.grpc.testing.integration.Metrics.GaugeRequest request) {
+      return futureUnaryCall(
+          getChannel().newCall(METHOD_GET_GAUGE, getCallOptions()), request);
+    }
+  }
+
+  private static final int METHODID_GET_ALL_GAUGES = 0;
+  private static final int METHODID_GET_GAUGE = 1;
+
+  private static class MethodHandlers<Req, Resp> implements
+      io.grpc.stub.ServerCalls.UnaryMethod<Req, Resp>,
+      io.grpc.stub.ServerCalls.ServerStreamingMethod<Req, Resp>,
+      io.grpc.stub.ServerCalls.ClientStreamingMethod<Req, Resp>,
+      io.grpc.stub.ServerCalls.BidiStreamingMethod<Req, Resp> {
+    private final MetricsService serviceImpl;
+    private final int methodId;
+
+    public MethodHandlers(MetricsService serviceImpl, int methodId) {
+      this.serviceImpl = serviceImpl;
+      this.methodId = methodId;
+    }
+
+    @java.lang.Override
+    @java.lang.SuppressWarnings("unchecked")
+    public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserver) {
+      switch (methodId) {
+        case METHODID_GET_ALL_GAUGES:
+          serviceImpl.getAllGauges((io.grpc.testing.integration.Metrics.EmptyMessage) request,
+              (io.grpc.stub.StreamObserver<io.grpc.testing.integration.Metrics.GaugeResponse>) responseObserver);
+          break;
+        case METHODID_GET_GAUGE:
+          serviceImpl.getGauge((io.grpc.testing.integration.Metrics.GaugeRequest) request,
+              (io.grpc.stub.StreamObserver<io.grpc.testing.integration.Metrics.GaugeResponse>) responseObserver);
+          break;
+        default:
+          throw new AssertionError();
+      }
+    }
+
+    @java.lang.Override
+    @java.lang.SuppressWarnings("unchecked")
+    public io.grpc.stub.StreamObserver<Req> invoke(
+        io.grpc.stub.StreamObserver<Resp> responseObserver) {
+      switch (methodId) {
+        default:
+          throw new AssertionError();
+      }
+    }
+  }
+
+  public static io.grpc.ServerServiceDefinition bindService(
+      final MetricsService serviceImpl) {
+    return io.grpc.ServerServiceDefinition.builder(SERVICE_NAME)
+        .addMethod(
+          METHOD_GET_ALL_GAUGES,
+          asyncServerStreamingCall(
+            new MethodHandlers<
+              io.grpc.testing.integration.Metrics.EmptyMessage,
+              io.grpc.testing.integration.Metrics.GaugeResponse>(
+                serviceImpl, METHODID_GET_ALL_GAUGES)))
+        .addMethod(
+          METHOD_GET_GAUGE,
+          asyncUnaryCall(
+            new MethodHandlers<
+              io.grpc.testing.integration.Metrics.GaugeRequest,
+              io.grpc.testing.integration.Metrics.GaugeResponse>(
+                serviceImpl, METHODID_GET_GAUGE)))
+        .build();
+  }
+}

interop-testing/src/main/java/io/grpc/testing/integration/StressTestClient.java

@@ -0,0 +1,588 @@
+/*
+ * Copyright 2016, Google Inc. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *    * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *    * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *
+ *    * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.grpc.testing.integration;
+
+import static java.util.Arrays.asList;
+import static java.util.Collections.shuffle;
+import static java.util.Collections.singletonList;
+import static java.util.concurrent.Executors.newFixedThreadPool;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import com.google.api.client.repackaged.com.google.common.base.Joiner;
+import com.google.api.client.repackaged.com.google.common.base.Objects;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Splitter;
+import com.google.common.collect.Iterators;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+import io.grpc.Status;
+import io.grpc.StatusException;
+import io.grpc.stub.StreamObserver;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * A stress test client following the
+ * <a href="https://github.com/grpc/grpc/blob/master/tools/run_tests/stress_test/STRESS_CLIENT_SPEC.md">
+ * specifications</a> of the gRPC stress testing framework.
+ */
+public class StressTestClient {
+
+  private static final Logger log = Logger.getLogger(StressTestClient.class.getName());
+
+  /**
+   * The main application allowing this client to be launched from the command line.
+   */
+  public static void main(String... args) throws Exception {
+    final StressTestClient client = new StressTestClient();
+    client.parseArgs(args);
+
+    // Attempt an orderly shutdown, if the JVM is shutdown via a signal.
+    Runtime.getRuntime().addShutdownHook(new Thread() {
+      @Override
+      public void run() {
+        client.shutdown();
+      }
+    });
+
+    try {
+      client.startMetricsService();
+      client.runStressTest();
+      client.blockUntilStressTestComplete();
+    } catch (Exception e) {
+      log.log(Level.WARNING, "The stress test client encountered an error!", e);
+    } finally {
+      client.shutdown();
+    }
+  }
+
+  private static final int WORKER_GRACE_PERIOD_SECS = 30;
+
+  private List<InetSocketAddress> addresses =
+      singletonList(new InetSocketAddress("localhost", 8080));
+  private List<TestCaseWeightPair> testCaseWeightPairs = new ArrayList<TestCaseWeightPair>();
+  private int durationSecs = -1;
+  private int channelsPerServer = 1;
+  private int stubsPerChannel = 1;
+  private int metricsPort = 8081;
+
+  private Server metricsServer;
+  private final Map<String, Metrics.GaugeResponse> gauges =
+      new ConcurrentHashMap<String, Metrics.GaugeResponse>();
+
+  private volatile boolean shutdown;
+
+  /**
+   * List of futures that {@link #blockUntilStressTestComplete()} waits for.
+   */
+  private final List<ListenableFuture<?>> workerFutures =
+      new ArrayList<ListenableFuture<?>>();
+  private final List<ManagedChannel> channels = new ArrayList<ManagedChannel>();
+  private ListeningExecutorService threadpool;
+
+  @VisibleForTesting
+  void parseArgs(String[] args) {
+    boolean usage = false;
+    for (String arg : args) {
+      if (!arg.startsWith("--")) {
+        System.err.println("All arguments must start with '--': " + arg);
+        usage = true;
+        break;
+      }
+      String[] parts = arg.substring(2).split("=", 2);
+      String key = parts[0];
+      if ("help".equals(key)) {
+        usage = true;
+        break;
+      }
+      if (parts.length != 2) {
+        System.err.println("All arguments must be of the form --arg=value");
+        usage = true;
+        break;
+      }
+      String value = parts[1];
+      if ("server_addresses".equals(key)) {
+        addresses = parseServerAddresses(value);
+        usage = addresses.isEmpty();
+      } else if ("test_cases".equals(key)) {
+        testCaseWeightPairs = parseTestCases(value);
+      } else if ("test_duration_secs".equals(key)) {
+        durationSecs = Integer.valueOf(value);
+      } else if ("num_channels_per_server".equals(key)) {
+        channelsPerServer = Integer.valueOf(value);
+      } else if ("num_stubs_per_channel".equals(key)) {
+        stubsPerChannel = Integer.valueOf(value);
+      } else if ("metrics_port".equals(key)) {
+        metricsPort = Integer.valueOf(value);
+      } else {
+        System.err.println("Unknown argument: " + key);
+        usage = true;
+        break;
+      }
+    }
+    if (usage) {
+      StressTestClient c = new StressTestClient();
+      System.err.println(
+          "Usage: [ARGS...]"
+              + "\n"
+              + "\n  --server_addresses=<name_1>:<port_1>,<name_2>:<port_2>...<name_N>:<port_N>"
+              + "\n    Default: " + serverAddressesToString(c.addresses)
+              + "\n  --test_cases=<testcase_1:w_1>,<testcase_2:w_2>...<testcase_n:w_n>"
+              + "\n    List of <testcase,weight> tuples. Weight is the relative frequency at which"
+              + " testcase is run."
+              + "\n    Valid Testcases:"
+              + validTestCasesHelpText()
+              + "\n  --test_duration_secs=SECONDS   '-1' for no limit. Default: " + c.durationSecs
+              + "\n  --num_channels_per_server=INT  Number of connections to each server address."
+              + " Default: " + c.channelsPerServer
+              + "\n  --num_stubs_per_channel=INT    Default: " + c.stubsPerChannel
+              + "\n  --metrics_port=PORT            Listening port of the metrics server."
+              + " Default: " + c.metricsPort
+      );
+      System.exit(1);
+    }
+  }
+
+  @VisibleForTesting
+  void startMetricsService() throws IOException {
+    Preconditions.checkState(!shutdown, "client was shutdown.");
+
+    metricsServer = ServerBuilder.forPort(metricsPort)
+        .addService(MetricsServiceGrpc.bindService(new MetricsServiceImpl()))
+        .build()
+        .start();
+  }
+
+  @VisibleForTesting
+  void runStressTest() throws Exception {
+    Preconditions.checkState(!shutdown, "client was shutdown.");
+    if (testCaseWeightPairs.isEmpty()) {
+      return;
+    }
+
+    int numChannels = addresses.size() * channelsPerServer;
+    int numThreads = numChannels * stubsPerChannel;
+    threadpool = MoreExecutors.listeningDecorator(newFixedThreadPool(numThreads));
+    int server_idx = -1;
+    for (InetSocketAddress address : addresses) {
+      server_idx++;
+      for (int i = 0; i < channelsPerServer; i++) {
+        ManagedChannel channel = createChannel(address);
+        channels.add(channel);
+        for (int j = 0; j < stubsPerChannel; j++) {
+          String gaugeName =
+              String.format("/stress_test/server_%d/channel_%d/stub_%d/qps", server_idx, i, j);
+          Worker worker =
+              new Worker(channel, testCaseWeightPairs, durationSecs, gaugeName);
+
+          workerFutures.add(threadpool.submit(worker));
+        }
+      }
+    }
+  }
+
+  @VisibleForTesting
+  void blockUntilStressTestComplete() throws Exception {
+    Preconditions.checkState(!shutdown, "client was shutdown.");
+
+    ListenableFuture<?> f = Futures.allAsList(workerFutures);
+    if (durationSecs == -1) {
+      // '-1' indicates that the stress test runs until terminated by the user.
+      f.get();
+    } else {
+      f.get(durationSecs + WORKER_GRACE_PERIOD_SECS, SECONDS);
+    }
+  }
+
+  @VisibleForTesting
+  void shutdown() {
+    if (shutdown) {
+      return;
+    }
+    shutdown = true;
+
+    for (ManagedChannel ch : channels) {
+      try {
+        ch.shutdownNow();
+        ch.awaitTermination(1, SECONDS);
+      } catch (Throwable t) {
+        log.log(Level.WARNING, "Error shutting down channel!", t);
+      }
+    }
+
+    try {
+      metricsServer.shutdownNow();
+    } catch (Throwable t) {
+      log.log(Level.WARNING, "Error shutting down metrics service!", t);
+    }
+
+    try {
+      if (threadpool != null) {
+        threadpool.shutdownNow();
+      }
+    } catch (Throwable t) {
+      log.log(Level.WARNING, "Error shutting down threadpool.", t);
+    }
+  }
+
+  private static List<InetSocketAddress> parseServerAddresses(String addressesStr) {
+    List<InetSocketAddress> addresses = new ArrayList<InetSocketAddress>();
+
+    for (List<String> namePort : parseCommaSeparatedTuples(addressesStr)) {
+      String name = namePort.get(0);
+      int port = Integer.valueOf(namePort.get(1));
+      addresses.add(new InetSocketAddress(name, port));
+    }
+
+    return addresses;
+  }
+
+  private static List<TestCaseWeightPair> parseTestCases(String testCasesStr) {
+    List<TestCaseWeightPair> testCaseWeightPairs = new ArrayList<TestCaseWeightPair>();
+
+    for (List<String> nameWeight : parseCommaSeparatedTuples(testCasesStr)) {
+      TestCases testCase = TestCases.fromString(nameWeight.get(0));
+      int weight = Integer.valueOf(nameWeight.get(1));
+      testCaseWeightPairs.add(new TestCaseWeightPair(testCase, weight));
+    }
+
+    return testCaseWeightPairs;
+  }
+
+  private static List<List<String>> parseCommaSeparatedTuples(String str) {
+    List<List<String>> tuples = new ArrayList<List<String>>();
+    for (String tupleStr : Splitter.on(',').split(str)) {
+      int splitIdx = tupleStr.lastIndexOf(':');
+      if (splitIdx == -1) {
+        throw new IllegalArgumentException("Illegal tuple format: '" + tupleStr + "'");
+      }
+      String part0 = tupleStr.substring(0, splitIdx);
+      String part1 = tupleStr.substring(splitIdx + 1);
+      tuples.add(asList(part0, part1));
+    }
+    return tuples;
+  }
+
+  private static ManagedChannel createChannel(InetSocketAddress address) {
+    return ManagedChannelBuilder.forAddress(address.getHostName(), address.getPort())
+        .usePlaintext(true)
+        .build();
+  }
+
+  private static String serverAddressesToString(List<InetSocketAddress> addresses) {
+    List<String> tmp = new ArrayList<String>();
+    for (InetSocketAddress address : addresses) {
+      URI uri;
+      try {
+        uri = new URI(null, null, address.getHostName(), address.getPort(), null, null, null);
+      } catch (URISyntaxException e) {
+        throw new RuntimeException(e);
+      }
+      tmp.add(uri.getAuthority());
+    }
+    return Joiner.on(',').join(tmp);
+  }
+
+  private static String validTestCasesHelpText() {
+    StringBuilder builder = new StringBuilder();
+    for (TestCases testCase : TestCases.values()) {
+      String strTestcase = testCase.name().toLowerCase();
+      builder.append("\n      ")
+          .append(strTestcase)
+          .append(": ")
+          .append(testCase.description());
+    }
+    return builder.toString();
+  }
+
+  /**
+   * A stress test worker. Every stub has its own stress test worker.
+   */
+  private class Worker implements Runnable {
+
+    // Interval at which the QPS stats of metrics service are updated.
+    private static final long METRICS_COLLECTION_INTERVAL_SECS = 5;
+
+    private final ManagedChannel channel;
+    private final List<TestCaseWeightPair> testCaseWeightPairs;
+    private final Integer durationSec;
+    private final String gaugeName;
+
+    Worker(ManagedChannel channel, List<TestCaseWeightPair> testCaseWeightPairs,
+        int durationSec, String gaugeName) {
+      Preconditions.checkArgument(durationSec >= -1, "durationSec must be gte -1.");
+      this.channel = Preconditions.checkNotNull(channel);
+      this.testCaseWeightPairs = Preconditions.checkNotNull(testCaseWeightPairs);
+      this.durationSec = durationSec == -1 ? null : durationSec;
+      this.gaugeName = Preconditions.checkNotNull(gaugeName);
+    }
+
+    @Override
+    public void run() {
+      // Simplify debugging if the worker crashes / never terminates.
+      Thread.currentThread().setName(gaugeName);
+
+      Tester tester = new Tester();
+      tester.setUp();
+      WeightedTestCaseSelector testCaseSelector = new WeightedTestCaseSelector(testCaseWeightPairs);
+      Long endTime = durationSec == null ? null : System.nanoTime() + SECONDS.toNanos(durationSecs);
+      long lastMetricsCollectionTime = initLastMetricsCollectionTime();
+      // Number of interop testcases run since the last time metrics have been updated.
+      long testCasesSinceLastMetricsCollection = 0;
+
+      while (!Thread.currentThread().isInterrupted() && !shutdown
+          && (endTime == null || endTime - System.nanoTime() > 0)) {
+        try {
+          runTestCase(tester, testCaseSelector.nextTestCase());
+        } catch (Exception e) {
+          throw new RuntimeException(e);
+        }
+
+        testCasesSinceLastMetricsCollection++;
+
+        double durationSecs = computeDurationSecs(lastMetricsCollectionTime);
+        if (durationSecs >= METRICS_COLLECTION_INTERVAL_SECS) {
+          long qps = (long) Math.ceil(testCasesSinceLastMetricsCollection / durationSecs);
+
+          Metrics.GaugeResponse gauge = Metrics.GaugeResponse
+              .newBuilder()
+              .setName(gaugeName)
+              .setLongValue(qps)
+              .build();
+
+          gauges.put(gaugeName, gauge);
+
+          lastMetricsCollectionTime = System.nanoTime();
+          testCasesSinceLastMetricsCollection = 0;
+        }
+      }
+    }
+
+    private long initLastMetricsCollectionTime() {
+      return System.nanoTime() - SECONDS.toNanos(METRICS_COLLECTION_INTERVAL_SECS);
+    }
+
+    private double computeDurationSecs(long lastMetricsCollectionTime) {
+      return (System.nanoTime() - lastMetricsCollectionTime) / 1000000000.0;
+    }
+
+    private void runTestCase(Tester tester, TestCases testCase) throws Exception {
+      // TODO(buchgr): Implement tests requiring auth, once C++ supports it.
+      switch (testCase) {
+        case EMPTY_UNARY:
+          tester.emptyUnary();
+          break;
+
+        case LARGE_UNARY:
+          tester.largeUnary();
+          break;
+
+        case CLIENT_STREAMING:
+          tester.clientStreaming();
+          break;
+
+        case SERVER_STREAMING:
+          tester.serverStreaming();
+          break;
+
+        case PING_PONG:
+          tester.pingPong();
+          break;
+
+        case EMPTY_STREAM:
+          tester.emptyStream();
+          break;
+
+        case UNIMPLEMENTED_METHOD: {
+          tester.unimplementedMethod();
+          break;
+        }
+
+        case CANCEL_AFTER_BEGIN: {
+          tester.cancelAfterBegin();
+          break;
+        }
+
+        case CANCEL_AFTER_FIRST_RESPONSE: {
+          tester.cancelAfterFirstResponse();
+          break;
+        }
+
+        case TIMEOUT_ON_SLEEPING_SERVER: {
+          tester.timeoutOnSleepingServer();
+          break;
+        }
+
+        default:
+          throw new IllegalArgumentException("Unknown test case: " + testCase);
+      }
+    }
+
+    class Tester extends AbstractInteropTest {
+      @Override
+      protected ManagedChannel createChannel() {
+        return Worker.this.channel;
+      }
+    }
+
+    class WeightedTestCaseSelector {
+      /**
+       * Randomly shuffled and cyclic sequence that contains each testcase proportionally
+       * to its weight.
+       */
+      final Iterator<TestCases> testCases;
+
+      WeightedTestCaseSelector(List<TestCaseWeightPair> testCaseWeightPairs) {
+        Preconditions.checkNotNull(testCaseWeightPairs);
+        Preconditions.checkArgument(testCaseWeightPairs.size() > 0);
+
+        List<TestCases> testCases = new ArrayList<TestCases>();
+        for (TestCaseWeightPair testCaseWeightPair : testCaseWeightPairs) {
+          for (int i = 0; i < testCaseWeightPair.weight; i++) {
+            testCases.add(testCaseWeightPair.testCase);
+          }
+        }
+
+        shuffle(testCases);
+
+        this.testCases = Iterators.cycle(testCases);
+      }
+
+      TestCases nextTestCase() {
+        return testCases.next();
+      }
+    }
+  }
+
+  /**
+   * Service that exports the QPS metrics of the stress test.
+   */
+  private class MetricsServiceImpl implements MetricsServiceGrpc.MetricsService {
+
+    @Override
+    public void getAllGauges(Metrics.EmptyMessage request,
+        StreamObserver<Metrics.GaugeResponse> responseObserver) {
+      for (Metrics.GaugeResponse gauge : gauges.values()) {
+        responseObserver.onNext(gauge);
+      }
+      responseObserver.onCompleted();
+    }
+
+    @Override
+    public void getGauge(Metrics.GaugeRequest request,
+        StreamObserver<Metrics.GaugeResponse> responseObserver) {
+      String gaugeName = request.getName();
+      Metrics.GaugeResponse gauge = gauges.get(gaugeName);
+      if (gauge != null) {
+        responseObserver.onNext(gauge);
+        responseObserver.onCompleted();
+      } else {
+        responseObserver.onError(new StatusException(Status.NOT_FOUND));
+      }
+    }
+  }
+
+  @VisibleForTesting
+  static class TestCaseWeightPair {
+    final TestCases testCase;
+    final int weight;
+
+    TestCaseWeightPair(TestCases testCase, int weight) {
+      Preconditions.checkArgument(weight >= 0, "weight must be positive.");
+      this.testCase = Preconditions.checkNotNull(testCase);
+      this.weight = weight;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+      if (!(other instanceof TestCaseWeightPair)) {
+        return false;
+      }
+      TestCaseWeightPair that = (TestCaseWeightPair) other;
+      return testCase.equals(that.testCase) && weight == that.weight;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hashCode(testCase, weight);
+    }
+  }
+
+  @VisibleForTesting
+  List<InetSocketAddress> addresses() {
+    return Collections.unmodifiableList(addresses);
+  }
+
+  @VisibleForTesting
+  List<TestCaseWeightPair> testCaseWeightPairs() {
+    return testCaseWeightPairs;
+  }
+
+  @VisibleForTesting
+  int durationSecs() {
+    return durationSecs;
+  }
+
+  @VisibleForTesting
+  int channelsPerServer() {
+    return channelsPerServer;
+  }
+
+  @VisibleForTesting
+  int stubsPerChannel() {
+    return stubsPerChannel;
+  }
+
+  @VisibleForTesting
+  int metricsPort() {
+    return metricsPort;
+  }
+}

interop-testing/src/main/java/io/grpc/testing/integration/TestCases.java

@@ -0,0 +1,77 @@
+/*
+ * Copyright 2016, Google Inc. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *    * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *    * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *
+ *    * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.grpc.testing.integration;
+
+import com.google.common.base.Preconditions;
+
+/**
+ * Enum of interop test cases.
+ */
+public enum TestCases {
+  EMPTY_UNARY("empty (zero bytes) request and response"),
+  LARGE_UNARY("single request and (large) response"),
+  CLIENT_STREAMING("request streaming with single response"),
+  SERVER_STREAMING("single request with response streaming"),
+  PING_PONG("full-duplex ping-pong streaming"),
+  EMPTY_STREAM("A stream that has zero-messages in both directions"),
+  COMPUTE_ENGINE_CREDS("large_unary with service_account auth"),
+  SERVICE_ACCOUNT_CREDS("large_unary with compute engine auth"),
+  JWT_TOKEN_CREDS("JWT-based auth"),
+  OAUTH2_AUTH_TOKEN("raw oauth2 access token auth"),
+  PER_RPC_CREDS("per rpc raw oauth2 access token auth"),
+  UNIMPLEMENTED_METHOD("call an unimplemented RPC method"),
+  CANCEL_AFTER_BEGIN("cancel stream after starting it"),
+  CANCEL_AFTER_FIRST_RESPONSE("cancel on first response"),
+  TIMEOUT_ON_SLEEPING_SERVER("timeout before receiving a response");
+
+  private final String description;
+
+  TestCases(String description) {
+    this.description = description;
+  }
+
+  /**
+   * Returns a description of the test case.
+   */
+  public String description() {
+    return description;
+  }
+
+  /**
+   * Returns the {@link TestCases} matching the string {@code s}. The
+   * matching is done case insensitive.
+   */
+  public static TestCases fromString(String s) {
+    Preconditions.checkNotNull(s);
+    return TestCases.valueOf(s.toUpperCase());
+  }
+}

interop-testing/src/main/java/io/grpc/testing/integration/TestServiceClient.java

@@ -56,6 +56,9 @@ import javax.net.ssl.SSLSocketFactory;
  * series of tests.
  */
 public class TestServiceClient {
+
+  private static final Charset UTF_8 = Charset.forName("UTF-8");
+
   /**
    * The main application allowing this client to be launched from the command line.
    */
@@ -160,21 +163,7 @@ public class TestServiceClient {
           + "\n  --server_port=PORT          Port to connect to. Default " + c.serverPort
           + "\n  --test_case=TESTCASE        Test case to run. Default " + c.testCase
           + "\n    Valid options:"
-          + "\n      empty_unary: empty (zero bytes) request and response"
+          + validTestCasesHelpText()
-          + "\n      large_unary: single request and (large) response"
-          + "\n      client_streaming: request streaming with single response"
-          + "\n      server_streaming: single request with response streaming"
-          + "\n      ping_pong: full-duplex ping-pong streaming"
-          + "\n      empty_stream: A stream that has zero-messages in both directions"
-          + "\n      service_account_creds: large_unary with service_account auth"
-          + "\n      compute_engine_creds: large_unary with compute engine auth"
-          + "\n      jwt_token_creds: JWT-based auth"
-          + "\n      oauth2_auth_token: raw oauth2 access token auth"
-          + "\n      per_rpc_creds: per rpc raw oauth2 access token auth"
-          + "\n      unimplemented_method: call an unimplemented RPC method"
-          + "\n      cancel_after_begin: cancel stream after starting it"
-          + "\n      cancel_after_first_response: cancel on first response"
-          + "\n      timeout_on_sleeping_server: timeout before receiving a response"
           + "\n  --use_tls=true|false        Whether to use TLS. Default " + c.useTls
           + "\n  --use_test_ca=true|false    Whether to trust our fake CA. Default " + c.useTestCa
           + "\n  --use_okhttp=true|false     Whether to use OkHttp instead of Netty. Default "
@@ -206,7 +195,7 @@ public class TestServiceClient {
   private void run() {
     System.out.println("Running test " + testCase);
     try {
-      runTest(testCase);
+      runTest(TestCases.fromString(testCase));
     } catch (RuntimeException ex) {
       throw ex;
     } catch (Exception ex) {
@@ -215,46 +204,85 @@ public class TestServiceClient {
     System.out.println("Test completed.");
   }
 
-  private void runTest(String testCase) throws Exception {
+  private void runTest(TestCases testCase) throws Exception {
-    if ("empty_unary".equals(testCase)) {
+    switch (testCase) {
-      tester.emptyUnary();
+      case EMPTY_UNARY:
-    } else if ("large_unary".equals(testCase)) {
+        tester.emptyUnary();
-      tester.largeUnary();
+        break;
-    } else if ("client_streaming".equals(testCase)) {
+
-      tester.clientStreaming();
+      case LARGE_UNARY:
-    } else if ("server_streaming".equals(testCase)) {
+        tester.largeUnary();
-      tester.serverStreaming();
+        break;
-    } else if ("ping_pong".equals(testCase)) {
+
-      tester.pingPong();
+      case CLIENT_STREAMING:
-    } else if ("empty_stream".equals(testCase)) {
+        tester.clientStreaming();
-      tester.emptyStream();
+        break;
-    } else if ("compute_engine_creds".equals(testCase)) {
+
-      tester.computeEngineCreds(defaultServiceAccount, oauthScope);
+      case SERVER_STREAMING:
-    } else if ("service_account_creds".equals(testCase)) {
+        tester.serverStreaming();
-      String jsonKey = Files.toString(new File(serviceAccountKeyFile), Charset.forName("UTF-8"));
+        break;
-      FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
+
-      tester.serviceAccountCreds(jsonKey, credentialsStream, oauthScope);
+      case PING_PONG:
-    } else if ("jwt_token_creds".equals(testCase)) {
+        tester.pingPong();
-      FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
+        break;
-      tester.jwtTokenCreds(credentialsStream);
+
-    } else if ("oauth2_auth_token".equals(testCase)) {
+      case EMPTY_STREAM:
-      String jsonKey = Files.toString(new File(serviceAccountKeyFile), Charset.forName("UTF-8"));
+        tester.emptyStream();
-      FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
+        break;
-      tester.oauth2AuthToken(jsonKey, credentialsStream, oauthScope);
+
-    } else if ("per_rpc_creds".equals(testCase)) {
+      case COMPUTE_ENGINE_CREDS:
-      String jsonKey = Files.toString(new File(serviceAccountKeyFile), Charset.forName("UTF-8"));
+        tester.computeEngineCreds(defaultServiceAccount, oauthScope);
-      FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
+        break;
-      tester.perRpcCreds(jsonKey, credentialsStream, oauthScope);
+
-    } else if ("unimplemented_method".equals(testCase)) {
+      case SERVICE_ACCOUNT_CREDS: {
-      tester.unimplementedMethod();
+        String jsonKey = Files.toString(new File(serviceAccountKeyFile), UTF_8);
-    } else if ("cancel_after_begin".equals(testCase)) {
+        FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
-      tester.cancelAfterBegin();
+        tester.serviceAccountCreds(jsonKey, credentialsStream, oauthScope);
-    } else if ("cancel_after_first_response".equals(testCase)) {
+        break;
-      tester.cancelAfterFirstResponse();
+      }
-    } else  if ("timeout_on_sleeping_server".equals(testCase)) {
+
-      tester.timeoutOnSleepingServer();
+      case JWT_TOKEN_CREDS: {
-    } else {
+        FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
-      throw new IllegalArgumentException("Unknown test case: " + testCase);
+        tester.jwtTokenCreds(credentialsStream);
+        break;
+      }
+
+      case OAUTH2_AUTH_TOKEN: {
+        String jsonKey = Files.toString(new File(serviceAccountKeyFile), UTF_8);
+        FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
+        tester.oauth2AuthToken(jsonKey, credentialsStream, oauthScope);
+        break;
+      }
+
+      case PER_RPC_CREDS: {
+        String jsonKey = Files.toString(new File(serviceAccountKeyFile), UTF_8);
+        FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
+        tester.perRpcCreds(jsonKey, credentialsStream, oauthScope);
+        break;
+      }
+
+      case UNIMPLEMENTED_METHOD: {
+        tester.unimplementedMethod();
+        break;
+      }
+
+      case CANCEL_AFTER_BEGIN: {
+        tester.cancelAfterBegin();
+        break;
+      }
+
+      case CANCEL_AFTER_FIRST_RESPONSE: {
+        tester.cancelAfterFirstResponse();
+        break;
+      }
+
+      case TIMEOUT_ON_SLEEPING_SERVER: {
+        tester.timeoutOnSleepingServer();
+        break;
+      }
+
+      default:
+        throw new IllegalArgumentException("Unknown test case: " + testCase);
     }
   }
 
@@ -307,4 +335,16 @@ public class TestServiceClient {
       }
     }
   }
+
+  private static String validTestCasesHelpText() {
+    StringBuilder builder = new StringBuilder();
+    for (TestCases testCase : TestCases.values()) {
+      String strTestcase = testCase.name().toLowerCase();
+      builder.append("\n      ")
+          .append(strTestcase)
+          .append(": ")
+          .append(testCase.description());
+    }
+    return builder.toString();
+  }
 }

interop-testing/src/main/java/io/grpc/testing/integration/TestServiceServer.java

@@ -31,6 +31,7 @@
 
 package io.grpc.testing.integration;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.util.concurrent.MoreExecutors;
 
 import io.grpc.Server;
@@ -83,7 +84,8 @@ public class TestServiceServer {
   private ScheduledExecutorService executor;
   private Server server;
 
-  private void parseArgs(String[] args) {
+  @VisibleForTesting
+  void parseArgs(String[] args) {
     boolean usage = false;
     for (String arg : args) {
       if (!arg.startsWith("--")) {
@@ -131,7 +133,8 @@ public class TestServiceServer {
     }
   }
 
-  private void start() throws Exception {
+  @VisibleForTesting
+  void start() throws Exception {
     executor = Executors.newSingleThreadScheduledExecutor();
     SslContext sslContext = null;
     if (useTls) {
@@ -146,7 +149,8 @@ public class TestServiceServer {
         .build().start();
   }
 
-  private void stop() throws Exception {
+  @VisibleForTesting
+  void stop() throws Exception {
     server.shutdownNow();
     if (!server.awaitTermination(5, TimeUnit.SECONDS)) {
       System.err.println("Timed out waiting for server shutdown");

interop-testing/src/main/proto/io/grpc/testing/integration/metrics.proto

@@ -0,0 +1,66 @@
+// Copyright 2015-2016, Google Inc.
+// All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+//     * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+//     * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+//     * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+// Contains the definitions for a metrics service and the type of metrics
+// exposed by the service.
+//
+// Currently, 'Gauge' (i.e a metric that represents the measured value of
+// something at an instant of time) is the only metric type supported by the
+// service.
+syntax = "proto3";
+
+package grpc.testing;
+
+option java_package = "io.grpc.testing.integration";
+
+// Reponse message containing the gauge name and value
+message GaugeResponse {
+  string name = 1;
+  oneof value {
+    int64 long_value = 2;
+    double double_value = 3;
+    string string_value = 4;
+  }
+}
+
+// Request message containing the gauge name
+message GaugeRequest {
+  string name = 1;
+}
+
+message EmptyMessage {}
+
+service MetricsService {
+  // Returns the values of all the gauges that are currently being maintained by
+  // the service
+  rpc GetAllGauges(EmptyMessage) returns (stream GaugeResponse);
+
+  // Returns the value of one gauge
+  rpc GetGauge(GaugeRequest) returns (GaugeResponse);
+}

interop-testing/src/test/java/io/grpc/testing/integration/StressTestClientTest.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright 2016, Google Inc. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *    * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *    * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *
+ *    * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.grpc.testing.integration;
+
+import static com.google.common.collect.Sets.newHashSet;
+import static java.util.Collections.singletonList;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
+import io.grpc.testing.TestUtils;
+import io.grpc.testing.integration.Metrics.EmptyMessage;
+import io.grpc.testing.integration.Metrics.GaugeResponse;
+import io.grpc.testing.integration.StressTestClient.TestCaseWeightPair;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+
+/** Unit tests for {@link StressTestClient}. */
+@RunWith(JUnit4.class)
+public class StressTestClientTest {
+
+  @Test
+  public void ipv6AddressesShouldBeSupported() {
+    StressTestClient client = new StressTestClient();
+    client.parseArgs(new String[] {"--server_addresses=[0:0:0:0:0:0:0:1]:8080,"
+        + "[1:2:3:4:f:e:a:b]:8083"});
+
+    assertEquals(2, client.addresses().size());
+    assertEquals(new InetSocketAddress("0:0:0:0:0:0:0:1", 8080), client.addresses().get(0));
+    assertEquals(new InetSocketAddress("1:2:3:4:f:e:a:b", 8083), client.addresses().get(1));
+  }
+
+  @Test
+  public void defaults() {
+    StressTestClient client = new StressTestClient();
+    assertEquals(singletonList(new InetSocketAddress("localhost", 8080)), client.addresses());
+    assertTrue(client.testCaseWeightPairs().isEmpty());
+    assertEquals(-1, client.durationSecs());
+    assertEquals(1, client.channelsPerServer());
+    assertEquals(1, client.stubsPerChannel());
+    assertEquals(8081, client.metricsPort());
+  }
+
+  @Test
+  public void allCommandlineSwitchesAreSupported() {
+    StressTestClient client = new StressTestClient();
+    client.parseArgs(new String[] {
+        "--server_addresses=localhost:8080,localhost:8081,localhost:8082",
+        "--test_cases=empty_unary:20,large_unary:50,server_streaming:30",
+        "--test_duration_secs=20",
+        "--num_channels_per_server=10",
+        "--num_stubs_per_channel=5",
+        "--metrics_port=9090"
+    });
+
+    List<InetSocketAddress> addresses = Arrays.asList(new InetSocketAddress("localhost", 8080),
+        new InetSocketAddress("localhost", 8081), new InetSocketAddress("localhost", 8082));
+    assertEquals(addresses, client.addresses());
+
+    List<TestCaseWeightPair> testCases = Arrays.asList(
+        new TestCaseWeightPair(TestCases.EMPTY_UNARY, 20),
+        new TestCaseWeightPair(TestCases.LARGE_UNARY, 50),
+        new TestCaseWeightPair(TestCases.SERVER_STREAMING, 30));
+    assertEquals(testCases, client.testCaseWeightPairs());
+
+    assertEquals(20, client.durationSecs());
+    assertEquals(10, client.channelsPerServer());
+    assertEquals(5, client.stubsPerChannel());
+    assertEquals(9090, client.metricsPort());
+  }
+
+  @Test(timeout = 5000)
+  public void gaugesShouldBeExported() throws Exception {
+    int serverPort = TestUtils.pickUnusedPort();
+    int metricsPort = TestUtils.pickUnusedPort();
+
+    TestServiceServer server = new TestServiceServer();
+    server.parseArgs(new String[]{"--port=" + serverPort, "--use_tls=false"});
+    server.start();
+
+    StressTestClient client = new StressTestClient();
+    client.parseArgs(new String[] {"--test_cases=empty_unary:1",
+        "--server_addresses=localhost:" + serverPort, "--metrics_port=" + metricsPort,
+        "--num_stubs_per_channel=2"});
+    client.startMetricsService();
+    client.runStressTest();
+
+    // Connect to the metrics service
+    ManagedChannel ch = ManagedChannelBuilder.forAddress("localhost", metricsPort)
+        .usePlaintext(true)
+        .build();
+
+    MetricsServiceGrpc.MetricsServiceBlockingStub stub = MetricsServiceGrpc.newBlockingStub(ch);
+
+    // Wait until gauges have been exported
+    Iterator<GaugeResponse> responseIt = stub.getAllGauges(EmptyMessage.getDefaultInstance());
+    while (!responseIt.hasNext()) {
+      LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));
+      responseIt = stub.getAllGauges(EmptyMessage.getDefaultInstance());
+    }
+
+    Set<String> gaugeNames = newHashSet("/stress_test/server_0/channel_0/stub_0/qps",
+        "/stress_test/server_0/channel_0/stub_1/qps");
+
+    while (responseIt.hasNext()) {
+      GaugeResponse response = responseIt.next();
+      String gaugeName = response.getName();
+
+      assertTrue("gaugeName: " + gaugeName, gaugeNames.contains(gaugeName));
+      assertTrue("qps: " + response.getLongValue(), response.getLongValue() > 0);
+      gaugeNames.remove(response.getName());
+
+      GaugeResponse response1 =
+          stub.getGauge(Metrics.GaugeRequest.newBuilder().setName(gaugeName).build());
+      assertEquals(gaugeName, response1.getName());
+      assertTrue("qps: " + response1.getLongValue(), response1.getLongValue() > 0);
+    }
+
+    assertTrue("gauges: " + gaugeNames, gaugeNames.isEmpty());
+
+    client.shutdown();
+    server.stop();
+  }
+
+}

interop-testing/src/test/java/io/grpc/testing/integration/TestCasesTest.java

@@ -0,0 +1,72 @@
+/*
+ * Copyright 2016, Google Inc. All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *    * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *    * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *
+ *    * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.grpc.testing.integration;
+
+import static io.grpc.testing.integration.TestCases.fromString;
+import static org.junit.Assert.assertEquals;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Unit tests for {@link TestCases}.
+ */
+@RunWith(JUnit4.class)
+public class TestCasesTest {
+
+  @Test(expected = IllegalArgumentException.class)
+  public void unknownStringThrowsException() {
+    fromString("does_not_exist_1234");
+  }
+
+  @Test
+  public void testCaseNamesShouldMapToEnums() {
+    // names of testcases as defined in the interop spec
+    String[] testCases = {"empty_unary", "large_unary", "client_streaming", "server_streaming",
+      "ping_pong", "empty_stream", "compute_engine_creds", "service_account_creds",
+      "jwt_token_creds", "oauth2_auth_token", "per_rpc_creds", "unimplemented_method",
+      "cancel_after_begin", "cancel_after_first_response", "timeout_on_sleeping_server"};
+
+    assertEquals(testCases.length, TestCases.values().length);
+
+    Set<TestCases> testCaseSet = new HashSet<TestCases>(testCases.length);
+    for (String testCase : testCases) {
+      testCaseSet.add(TestCases.fromString(testCase));
+    }
+
+    assertEquals(TestCases.values().length, testCaseSet.size());
+  }
+}