testing: create test client for XDS federation integration tests (grp…

…c#9798)

This client can be used as a part of XDS federation integration tests. It can concurrently perform RPCs with different stubs using different underlying XDS servers.

For example, one might perform proxyless service mesh and DirectPath RPCs in the same process with flags like:

```
--server_uris=xds:///${PSM_TARGET},google-c2p:///${DIRECTPATH_TARGET} \
--credentials_types=INSECURE_CREDENTIALS,compute_engine_channel_creds \
--test_case=rpc_soak \
--soak_iterations=10  \
--soak_max_failures=0  \
--soak_per_iteration_max_acceptable_latency_ms=2500 \
--soak_overall_timeout_seconds=300
```

interop-testing/build.gradle

@@ -194,6 +194,13 @@ def xds_test_server = tasks.register("xds_test_server", CreateStartScripts) {
     classpath = startScriptsClasspath.get()
 }
 
+def xds_federation_test_client = tasks.register("xds_federation_test_client", CreateStartScripts) {
+    mainClass = "io.grpc.testing.integration.XdsFederationTestClient"
+    applicationName = "xds-federation-test-client"
+    outputDir = new File(project.buildDir, 'tmp/scripts/' + name)
+    classpath = startScriptsClasspath.get()
+}
+
 distributions.shadow.contents.into("bin") {
     from(test_client)
     from(test_server)
@@ -204,6 +211,7 @@ distributions.shadow.contents.into("bin") {
     from(grpclb_fallback_test_client)
     from(xds_test_client)
     from(xds_test_server)
+    from(xds_federation_test_client)
     fileMode = 0755
 }
 

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

@@ -2025,6 +2025,7 @@ private SoakIterationResult performOneSoakIteration(
     * and channel creation behavior.
    */
   public void performSoakTest(
+      String serverUri,
       boolean resetChannelPerIteration,
       int soakIterations,
       int maxFailures,
@@ -2057,21 +2058,22 @@ public void performSoakTest(
       SoakIterationResult result = performOneSoakIteration(soakStub);
       SocketAddress peer = clientCallCapture
           .get().getAttributes().get(Grpc.TRANSPORT_ATTR_REMOTE_ADDR);
-      System.err.print(
+      StringBuilder logStr = new StringBuilder(
           String.format(
               Locale.US,
-              "soak iteration: %d elapsed_ms: %d peer: %s",
-              i, result.getLatencyMs(), peer != null ? peer.toString() : "null"));
+              "soak iteration: %d elapsed_ms: %d peer: %s server_uri: %s",
+              i, result.getLatencyMs(), peer != null ? peer.toString() : "null", serverUri));
       if (!result.getStatus().equals(Status.OK)) {
         totalFailures++;
-        System.err.println(String.format(" failed: %s", result.getStatus()));
+        logStr.append(String.format(" failed: %s", result.getStatus()));
       } else if (result.getLatencyMs() > maxAcceptablePerIterationLatencyMs) {
         totalFailures++;
-        System.err.println(
+        logStr.append(
             " exceeds max acceptable latency: " + maxAcceptablePerIterationLatencyMs);
       } else {
-        System.err.println(" succeeded");
+        logStr.append(" succeeded");
       }
+      System.err.println(logStr.toString());
       iterationsDone++;
       latencies.recordValue(result.getLatencyMs());
       long remainingNs = earliestNextStartNs - System.nanoTime();
@@ -2084,29 +2086,22 @@ public void performSoakTest(
     System.err.println(
         String.format(
             Locale.US,
-            "soak test ran: %d / %d iterations\n"
-                + "total failures: %d\n"
-                + "max failures threshold: %d\n"
-                + "max acceptable per iteration latency ms: %d\n"
-                + " p50 soak iteration latency: %d ms\n"
-                + " p90 soak iteration latency: %d ms\n"
-                + "p100 soak iteration latency: %d ms\n"
-                + "See breakdown above for which iterations succeeded, failed, and "
-                + "why for more info.",
+            "(server_uri: %s) soak test ran: %d / %d iterations. total failures: %d. "
+                + "p50: %d ms, p90: %d ms, p100: %d ms",
+            serverUri,
             iterationsDone,
             soakIterations,
             totalFailures,
-            maxFailures,
-            maxAcceptablePerIterationLatencyMs,
             latencies.getValueAtPercentile(50),
             latencies.getValueAtPercentile(90),
             latencies.getValueAtPercentile(100)));
     // check if we timed out
     String timeoutErrorMessage =
         String.format(
             Locale.US,
-            "soak test consumed all %d seconds of time and quit early, only "
-                + "having ran %d out of desired %d iterations.",
+            "(server_uri: %s) soak test consumed all %d seconds of time and quit early, "
+                + "only having ran %d out of desired %d iterations.",
+            serverUri,
             overallTimeoutSeconds,
             iterationsDone,
             soakIterations);
@@ -2115,8 +2110,9 @@ public void performSoakTest(
     String tooManyFailuresErrorMessage =
         String.format(
             Locale.US,
-            "soak test total failures: %d exceeds max failures threshold: %d.",
-            totalFailures, maxFailures);
+            "(server_uri: %s) soak test total failures: %d exceeds max failures "
+                + "threshold: %d.",
+            serverUri, totalFailures, maxFailures);
     assertTrue(tooManyFailuresErrorMessage, totalFailures <= maxFailures);
   }
 

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

@@ -461,6 +461,7 @@ private void runTest(TestCases testCase) throws Exception {
 
       case RPC_SOAK: {
         tester.performSoakTest(
+            serverHost,
             false /* resetChannelPerIteration */,
             soakIterations,
             soakMaxFailures,
@@ -472,6 +473,7 @@ private void runTest(TestCases testCase) throws Exception {
 
       case CHANNEL_SOAK: {
         tester.performSoakTest(
+            serverHost,
             true /* resetChannelPerIteration */,
             soakIterations,
             soakMaxFailures,