Improve logging in interop soak test and add a flag to limit QPS (#29357)

* Improve logging in interop soak test and add a flag to limit QPS

doc/interop-test-descriptions.md

@@ -1045,6 +1045,9 @@ This test must be configurable via a few different command line flags:
   completed. This should default to
   `soak_per_iteration_max_acceptable_latency_ms` * `soak_iterations`.
 
+* `soak_min_time_ms_between_rpcs`: The minimum time in milliseconds between
+  consecutive RPCs. Useful for limiting QPS.
+
 The following is optional but encouraged to improve debuggability:
 
 * Implementations should log the number of milliseconds that each RPC takes.

test/cpp/interop/client.cc

@@ -110,6 +110,9 @@ ABSL_FLAG(int32_t, soak_overall_timeout_seconds, 0,
           "The overall number of seconds after which a soak test should "
           "stop and fail, if the desired number of iterations have not yet "
           "completed.");
+ABSL_FLAG(int32_t, soak_min_time_ms_between_rpcs, 0,
+          "The minimum time in milliseconds between consecutive RPCs in a "
+          "soak test (rpc_soak or channel_soak), useful for limiting QPS");
 ABSL_FLAG(int32_t, iteration_interval, 10,
           "The interval in seconds between rpcs. This is used by "
           "long_connection test");
@@ -300,12 +303,14 @@ int main(int argc, char** argv) {
       absl::GetFlag(FLAGS_soak_iterations),
       absl::GetFlag(FLAGS_soak_max_failures),
       absl::GetFlag(FLAGS_soak_per_iteration_max_acceptable_latency_ms),
+      absl::GetFlag(FLAGS_soak_min_time_ms_between_rpcs),
       absl::GetFlag(FLAGS_soak_overall_timeout_seconds));
   actions["rpc_soak"] = std::bind(
       &grpc::testing::InteropClient::DoRpcSoakTest, &client,
       absl::GetFlag(FLAGS_soak_iterations),
       absl::GetFlag(FLAGS_soak_max_failures),
       absl::GetFlag(FLAGS_soak_per_iteration_max_acceptable_latency_ms),
+      absl::GetFlag(FLAGS_soak_min_time_ms_between_rpcs),
       absl::GetFlag(FLAGS_soak_overall_timeout_seconds));
   actions["long_lived_channel"] =
       std::bind(&grpc::testing::InteropClient::DoLongLivedChannelTest, &client,

test/cpp/interop/interop_client.cc

@@ -1057,6 +1057,7 @@ void InteropClient::PerformSoakTest(
     const bool reset_channel_per_iteration, const int32_t soak_iterations,
     const int32_t max_failures,
     const int32_t max_acceptable_per_iteration_latency_ms,
+    const int32_t min_time_ms_between_rpcs,
     const int32_t overall_timeout_seconds) {
   std::vector<std::tuple<bool, int32_t, std::string>> results;
   grpc_histogram* latencies_ms_histogram = grpc_histogram_create(
@@ -1066,29 +1067,31 @@ void InteropClient::PerformSoakTest(
       gpr_now(GPR_CLOCK_MONOTONIC),
       gpr_time_from_seconds(overall_timeout_seconds, GPR_TIMESPAN));
   int32_t iterations_ran = 0;
+  int total_failures = 0;
   for (int i = 0;
        i < soak_iterations &&
        gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), overall_deadline) < 0;
        ++i) {
+    gpr_timespec earliest_next_start = gpr_time_add(
+        gpr_now(GPR_CLOCK_MONOTONIC),
+        gpr_time_from_millis(min_time_ms_between_rpcs, GPR_TIMESPAN));
     auto result = PerformOneSoakTestIteration(
         reset_channel_per_iteration, max_acceptable_per_iteration_latency_ms);
+    bool success = std::get<0>(result);
+    int32_t elapsed_ms = std::get<1>(result);
+    std::string debug_string = std::get<2>(result);
     results.push_back(result);
-    grpc_histogram_add(latencies_ms_histogram, std::get<1>(result));
-    iterations_ran++;
-  }
-  int total_failures = 0;
-  for (size_t i = 0; i < results.size(); i++) {
-    bool success = std::get<0>(results[i]);
-    int32_t elapsed_ms = std::get<1>(results[i]);
-    std::string debug_string = std::get<2>(results[i]);
     if (!success) {
-      gpr_log(GPR_DEBUG, "soak iteration: %ld elapsed_ms: %d failed: %s", i,
+      gpr_log(GPR_DEBUG, "soak iteration: %d elapsed_ms: %d failed: %s", i,
               elapsed_ms, debug_string.c_str());
       total_failures++;
     } else {
-      gpr_log(GPR_DEBUG, "soak iteration: %ld elapsed_ms: %d succeeded", i,
+      gpr_log(GPR_DEBUG, "soak iteration: %d elapsed_ms: %d succeeded", i,
               elapsed_ms);
     }
+    grpc_histogram_add(latencies_ms_histogram, std::get<1>(result));
+    iterations_ran++;
+    gpr_sleep_until(earliest_next_start);
   }
   double latency_ms_median =
       grpc_histogram_percentile(latencies_ms_histogram, 50);
@@ -1142,12 +1145,12 @@ void InteropClient::PerformSoakTest(
 bool InteropClient::DoRpcSoakTest(
     int32_t soak_iterations, int32_t max_failures,
     int64_t max_acceptable_per_iteration_latency_ms,
-    int32_t overall_timeout_seconds) {
+    int32_t soak_min_time_ms_between_rpcs, int32_t overall_timeout_seconds) {
   gpr_log(GPR_DEBUG, "Sending %d RPCs...", soak_iterations);
   GPR_ASSERT(soak_iterations > 0);
   PerformSoakTest(false /* reset channel per iteration */, soak_iterations,
                   max_failures, max_acceptable_per_iteration_latency_ms,
-                  overall_timeout_seconds);
+                  soak_min_time_ms_between_rpcs, overall_timeout_seconds);
   gpr_log(GPR_DEBUG, "rpc_soak test done.");
   return true;
 }
@@ -1155,13 +1158,13 @@ bool InteropClient::DoRpcSoakTest(
 bool InteropClient::DoChannelSoakTest(
     int32_t soak_iterations, int32_t max_failures,
     int64_t max_acceptable_per_iteration_latency_ms,
-    int32_t overall_timeout_seconds) {
+    int32_t soak_min_time_ms_between_rpcs, int32_t overall_timeout_seconds) {
   gpr_log(GPR_DEBUG, "Sending %d RPCs, tearing down the channel each time...",
           soak_iterations);
   GPR_ASSERT(soak_iterations > 0);
   PerformSoakTest(true /* reset channel per iteration */, soak_iterations,
                   max_failures, max_acceptable_per_iteration_latency_ms,
-                  overall_timeout_seconds);
+                  soak_min_time_ms_between_rpcs, overall_timeout_seconds);
   gpr_log(GPR_DEBUG, "channel_soak test done.");
   return true;
 }

test/cpp/interop/interop_client.h

@@ -80,9 +80,11 @@ class InteropClient {
   // languages
   bool DoChannelSoakTest(int32_t soak_iterations, int32_t max_failures,
                          int64_t max_acceptable_per_iteration_latency_ms,
+                         int32_t soak_min_time_ms_between_rpcs,
                          int32_t overall_timeout_seconds);
   bool DoRpcSoakTest(int32_t soak_iterations, int32_t max_failures,
                      int64_t max_acceptable_per_iteration_latency_ms,
+                     int32_t soak_min_time_ms_between_rpcs,
                      int32_t overall_timeout_seconds);
   bool DoLongLivedChannelTest(int32_t soak_iterations,
                               int32_t iteration_interval);
@@ -142,6 +144,7 @@ class InteropClient {
                        const int32_t soak_iterations,
                        const int32_t max_failures,
                        const int32_t max_acceptable_per_iteration_latency_ms,
+                       const int32_t min_time_ms_between_rpcs,
                        const int32_t overall_timeout_seconds);
 
   ServiceStub serviceStub_;