test/cpp/interop/stress_test.cc

   1 /*
   2  *
   3  * Copyright 2015 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *is % allowed in string
  17  */
  18
  19 #include <grpc/support/log.h>
  20 #include <grpc/support/time.h>
  21 #include <grpcpp/create_channel.h>
  22 #include <grpcpp/grpcpp.h>
  23
  24 #include <memory>
  25 #include <string>
  26 #include <thread>
  27 #include <utility>
  28 #include <vector>
  29
  30 #include "absl/flags/flag.h"
  31 #include "src/proto/grpc/testing/metrics.grpc.pb.h"
  32 #include "src/proto/grpc/testing/metrics.pb.h"
  33 #include "test/cpp/interop/interop_client.h"
  34 #include "test/cpp/interop/stress_interop_client.h"
  35 #include "test/cpp/util/create_test_channel.h"
  36 #include "test/cpp/util/metrics_server.h"
  37 #include "test/cpp/util/test_config.h"
  38
  39 extern void gpr_default_log(gpr_log_func_args* args);
  40
  41 ABSL_FLAG(int32_t, metrics_port, 8081, "The metrics server port.");
  42
  43 // TODO(Capstan): Consider using absl::Duration
  44 ABSL_FLAG(int32_t, sleep_duration_ms, 0,
  45           "The duration (in millisec) between two"
  46           " consecutive test calls (per server) issued by the server.");
  47
  48 // TODO(Capstan): Consider using absl::Duration
  49 ABSL_FLAG(int32_t, test_duration_secs, -1,
  50           "The length of time (in seconds) to run"
  51           " the test. Enter -1 if the test should run continuously until"
  52           " forcefully terminated.");
  53
  54 ABSL_FLAG(std::string, server_addresses, "localhost:8080",
  55           "The list of server addresses. The format is: \n"
  56           " \"<name_1>:<port_1>,<name_2>:<port_1>...<name_N>:<port_N>\"\n"
  57           " Note: <name> can be servername or IP address.");
  58
  59 ABSL_FLAG(int32_t, num_channels_per_server, 1,
  60           "Number of channels for each server");
  61
  62 ABSL_FLAG(int32_t, num_stubs_per_channel, 1,
  63           "Number of stubs per each channels to server. This number also "
  64           "indicates the max number of parallel RPC calls on each channel "
  65           "at any given time.");
  66
  67 // TODO(sreek): Add more test cases here in future
  68 ABSL_FLAG(std::string, test_cases, "",
  69           "List of test cases to call along with the"
  70           " relative weights in the following format:\n"
  71           " \"<testcase_1:w_1>,<testcase_2:w_2>...<testcase_n:w_n>\"\n"
  72           " The following testcases are currently supported:\n"
  73           "   empty_unary\n"
  74           "   large_unary\n"
  75           "   large_compressed_unary\n"
  76           "   client_streaming\n"
  77           "   server_streaming\n"
  78           "   server_compressed_streaming\n"
  79           "   slow_consumer\n"
  80           "   half_duplex\n"
  81           "   ping_pong\n"
  82           "   cancel_after_begin\n"
  83           "   cancel_after_first_response\n"
  84           "   timeout_on_sleeping_server\n"
  85           "   empty_stream\n"
  86           "   status_code_and_message\n"
  87           "   custom_metadata\n"
  88           " Example: \"empty_unary:20,large_unary:10,empty_stream:70\"\n"
  89           " The above will execute 'empty_unary', 20% of the time,"
  90           " 'large_unary', 10% of the time and 'empty_stream' the remaining"
  91           " 70% of the time");
  92
  93 ABSL_FLAG(int32_t, log_level, GPR_LOG_SEVERITY_INFO,
  94           "Severity level of messages that should be logged. Any messages "
  95           "greater than or equal to the level set here will be logged. "
  96           "The choices are: 0 (GPR_LOG_SEVERITY_DEBUG), 1 "
  97           "(GPR_LOG_SEVERITY_INFO) and 2 (GPR_LOG_SEVERITY_ERROR)");
  98
  99 ABSL_FLAG(bool, do_not_abort_on_transient_failures, true,
 100           "If set to 'true', abort() is not called in case of transient "
 101           "failures like temporary connection failures.");
 102
 103 // Options from client.cc (for compatibility with interop test).
 104 // TODO(sreek): Consolidate overlapping options
 105 ABSL_FLAG(bool, use_alts, false,
 106           "Whether to use alts. Enable alts will disable tls.");
 107 ABSL_FLAG(bool, use_tls, false, "Whether to use tls.");
 108 ABSL_FLAG(bool, use_test_ca, false, "False to use SSL roots for google");
 109 ABSL_FLAG(std::string, server_host_override, "",
 110           "Override the server host which is sent in HTTP header");
 111
 112 using grpc::testing::ALTS;
 113 using grpc::testing::INSECURE;
 114 using grpc::testing::kTestCaseList;
 115 using grpc::testing::MetricsServiceImpl;
 116 using grpc::testing::StressTestInteropClient;
 117 using grpc::testing::TestCaseType;
 118 using grpc::testing::TLS;
 119 using grpc::testing::transport_security;
 120 using grpc::testing::UNKNOWN_TEST;
 121 using grpc::testing::WeightedRandomTestSelector;
 122
 123 static int log_level = GPR_LOG_SEVERITY_DEBUG;
 124
 125 // A simple wrapper to grp_default_log() function. This only logs messages at or
 126 // above the current log level (set in 'log_level' variable)
 127 void TestLogFunction(gpr_log_func_args* args) {
 128   if (args->severity >= log_level) {
 129     gpr_default_log(args);
 130   }
 131 }
 132
 133 TestCaseType GetTestTypeFromName(const std::string& test_name) {
 134   TestCaseType test_case = UNKNOWN_TEST;
 135
 136   for (auto it = kTestCaseList.begin(); it != kTestCaseList.end(); it++) {
 137     if (test_name == it->second) {
 138       test_case = it->first;
 139       break;
 140     }
 141   }
 142
 143   return test_case;
 144 }
 145
 146 // Converts a string of comma delimited tokens to a vector of tokens
 147 bool ParseCommaDelimitedString(const std::string& comma_delimited_str,
 148                                std::vector<std::string>& tokens) {
 149   size_t bpos = 0;
 150   size_t epos = std::string::npos;
 151
 152   while ((epos = comma_delimited_str.find(',', bpos)) != std::string::npos) {
 153     tokens.emplace_back(comma_delimited_str.substr(bpos, epos - bpos));
 154     bpos = epos + 1;
 155   }
 156
 157   tokens.emplace_back(comma_delimited_str.substr(bpos));  // Last token
 158   return true;
 159 }
 160
 161 // Input: Test case string "<testcase_name:weight>,<testcase_name:weight>...."
 162 // Output:
 163 //   - Whether parsing was successful (return value)
 164 //   - Vector of (test_type_enum, weight) pairs returned via 'tests' parameter
 165 bool ParseTestCasesString(const std::string& test_cases,
 166                           std::vector<std::pair<TestCaseType, int>>& tests) {
 167   bool is_success = true;
 168
 169   std::vector<std::string> tokens;
 170   ParseCommaDelimitedString(test_cases, tokens);
 171
 172   for (auto it = tokens.begin(); it != tokens.end(); it++) {
 173     // Token is in the form <test_name>:<test_weight>
 174     size_t colon_pos = it->find(':');
 175     if (colon_pos == std::string::npos) {
 176       gpr_log(GPR_ERROR, "Error in parsing test case string: %s", it->c_str());
 177       is_success = false;
 178       break;
 179     }
 180
 181     std::string test_name = it->substr(0, colon_pos);
 182     int weight = std::stoi(it->substr(colon_pos + 1));
 183     TestCaseType test_case = GetTestTypeFromName(test_name);
 184     if (test_case == UNKNOWN_TEST) {
 185       gpr_log(GPR_ERROR, "Unknown test case: %s", test_name.c_str());
 186       is_success = false;
 187       break;
 188     }
 189
 190     tests.emplace_back(std::make_pair(test_case, weight));
 191   }
 192
 193   return is_success;
 194 }
 195
 196 // For debugging purposes
 197 void LogParameterInfo(const std::vector<std::string>& addresses,
 198                       const std::vector<std::pair<TestCaseType, int>>& tests) {
 199   gpr_log(GPR_INFO, "server_addresses: %s",
 200           absl::GetFlag(FLAGS_server_addresses).c_str());
 201   gpr_log(GPR_INFO, "test_cases : %s", absl::GetFlag(FLAGS_test_cases).c_str());
 202   gpr_log(GPR_INFO, "sleep_duration_ms: %d",
 203           absl::GetFlag(FLAGS_sleep_duration_ms));
 204   gpr_log(GPR_INFO, "test_duration_secs: %d",
 205           absl::GetFlag(FLAGS_test_duration_secs));
 206   gpr_log(GPR_INFO, "num_channels_per_server: %d",
 207           absl::GetFlag(FLAGS_num_channels_per_server));
 208   gpr_log(GPR_INFO, "num_stubs_per_channel: %d",
 209           absl::GetFlag(FLAGS_num_stubs_per_channel));
 210   gpr_log(GPR_INFO, "log_level: %d", absl::GetFlag(FLAGS_log_level));
 211   gpr_log(GPR_INFO, "do_not_abort_on_transient_failures: %s",
 212           absl::GetFlag(FLAGS_do_not_abort_on_transient_failures) ? "true"
 213                                                                   : "false");
 214
 215   int num = 0;
 216   for (auto it = addresses.begin(); it != addresses.end(); it++) {
 217     gpr_log(GPR_INFO, "%d:%s", ++num, it->c_str());
 218   }
 219
 220   num = 0;
 221   for (auto it = tests.begin(); it != tests.end(); it++) {
 222     TestCaseType test_case = it->first;
 223     int weight = it->second;
 224     gpr_log(GPR_INFO, "%d. TestCaseType: %d, Weight: %d", ++num, test_case,
 225             weight);
 226   }
 227 }
 228
 229 int main(int argc, char** argv) {
 230   grpc::testing::InitTest(&argc, &argv, true);
 231
 232   if (absl::GetFlag(FLAGS_log_level) > GPR_LOG_SEVERITY_ERROR ||
 233       absl::GetFlag(FLAGS_log_level) < GPR_LOG_SEVERITY_DEBUG) {
 234     gpr_log(GPR_ERROR, "log_level should be an integer between %d and %d",
 235             GPR_LOG_SEVERITY_DEBUG, GPR_LOG_SEVERITY_ERROR);
 236     return 1;
 237   }
 238
 239   // Change the default log function to TestLogFunction which respects the
 240   // log_level setting.
 241   log_level = absl::GetFlag(FLAGS_log_level);
 242   gpr_set_log_function(TestLogFunction);
 243
 244   srand(time(nullptr));
 245
 246   // Parse the server addresses
 247   std::vector<std::string> server_addresses;
 248   ParseCommaDelimitedString(absl::GetFlag(FLAGS_server_addresses),
 249                             server_addresses);
 250
 251   // Parse test cases and weights
 252   if (absl::GetFlag(FLAGS_test_cases).length() == 0) {
 253     gpr_log(GPR_ERROR, "No test cases supplied");
 254     return 1;
 255   }
 256
 257   std::vector<std::pair<TestCaseType, int>> tests;
 258   if (!ParseTestCasesString(absl::GetFlag(FLAGS_test_cases), tests)) {
 259     gpr_log(GPR_ERROR, "Error in parsing test cases string %s ",
 260             absl::GetFlag(FLAGS_test_cases).c_str());
 261     return 1;
 262   }
 263
 264   LogParameterInfo(server_addresses, tests);
 265
 266   WeightedRandomTestSelector test_selector(tests);
 267   MetricsServiceImpl metrics_service;
 268
 269   gpr_log(GPR_INFO, "Starting test(s)..");
 270
 271   std::vector<std::thread> test_threads;
 272   std::vector<std::unique_ptr<StressTestInteropClient>> clients;
 273
 274   // Create and start the test threads.
 275   // Note that:
 276   // - Each server can have multiple channels (as configured by
 277   // FLAGS_num_channels_per_server).
 278   //
 279   // - Each channel can have multiple stubs (as configured by
 280   // FLAGS_num_stubs_per_channel). This is to test calling multiple RPCs in
 281   // parallel on the same channel.
 282   int thread_idx = 0;
 283   int server_idx = -1;
 284   char buffer[256];
 285   transport_security security_type =
 286       absl::GetFlag(FLAGS_use_alts)
 287           ? ALTS
 288           : (absl::GetFlag(FLAGS_use_tls) ? TLS : INSECURE);
 289   for (auto it = server_addresses.begin(); it != server_addresses.end(); it++) {
 290     ++server_idx;
 291     // Create channel(s) for each server
 292     for (int channel_idx = 0;
 293          channel_idx < absl::GetFlag(FLAGS_num_channels_per_server);
 294          channel_idx++) {
 295       gpr_log(GPR_INFO, "Starting test with %s channel_idx=%d..", it->c_str(),
 296               channel_idx);
 297       grpc::testing::ChannelCreationFunc channel_creation_func =
 298           std::bind(static_cast<std::shared_ptr<grpc::Channel> (*)(
 299                         const std::string&, const std::string&,
 300                         grpc::testing::transport_security, bool)>(
 301                         grpc::CreateTestChannel),
 302                     *it, absl::GetFlag(FLAGS_server_host_override),
 303                     security_type, !absl::GetFlag(FLAGS_use_test_ca));
 304
 305       // Create stub(s) for each channel
 306       for (int stub_idx = 0;
 307            stub_idx < absl::GetFlag(FLAGS_num_stubs_per_channel); stub_idx++) {
 308         clients.emplace_back(new StressTestInteropClient(
 309             ++thread_idx, *it, channel_creation_func, test_selector,
 310             absl::GetFlag(FLAGS_test_duration_secs),
 311             absl::GetFlag(FLAGS_sleep_duration_ms),
 312             absl::GetFlag(FLAGS_do_not_abort_on_transient_failures)));
 313
 314         bool is_already_created = false;
 315         // QpsGauge name
 316         std::snprintf(buffer, sizeof(buffer),
 317                       "/stress_test/server_%d/channel_%d/stub_%d/qps",
 318                       server_idx, channel_idx, stub_idx);
 319
 320         test_threads.emplace_back(std::thread(
 321             &StressTestInteropClient::MainLoop, clients.back().get(),
 322             metrics_service.CreateQpsGauge(buffer, &is_already_created)));
 323
 324         // The QpsGauge should not have been already created
 325         GPR_ASSERT(!is_already_created);
 326       }
 327     }
 328   }
 329
 330   // Start metrics server before waiting for the stress test threads
 331   std::unique_ptr<grpc::Server> metrics_server;
 332   if (absl::GetFlag(FLAGS_metrics_port) > 0) {
 333     metrics_server =
 334         metrics_service.StartServer(absl::GetFlag(FLAGS_metrics_port));
 335   }
 336
 337   // Wait for the stress test threads to complete
 338   for (auto it = test_threads.begin(); it != test_threads.end(); it++) {
 339     it->join();
 340   }
 341
 342   return 0;
 343 }