test/cpp/interop/xds_interop_client.cc

   1 /*
   2  *
   3  * Copyright 2020 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  *
  17  */
  18
  19 #include <atomic>
  20 #include <chrono>
  21 #include <condition_variable>
  22 #include <deque>
  23 #include <map>
  24 #include <mutex>
  25 #include <set>
  26 #include <sstream>
  27 #include <string>
  28 #include <thread>
  29 #include <vector>
  30
  31 #include "absl/algorithm/container.h"
  32 #include "absl/flags/flag.h"
  33 #include "absl/strings/str_split.h"
  34
  35 #include <grpcpp/ext/admin_services.h>
  36 #include <grpcpp/ext/proto_server_reflection_plugin.h>
  37 #include <grpcpp/grpcpp.h>
  38 #include <grpcpp/server.h>
  39 #include <grpcpp/server_builder.h>
  40 #include <grpcpp/server_context.h>
  41
  42 #include "src/core/lib/channel/status_util.h"
  43 #include "src/core/lib/gpr/env.h"
  44 #include "src/proto/grpc/testing/empty.pb.h"
  45 #include "src/proto/grpc/testing/messages.pb.h"
  46 #include "src/proto/grpc/testing/test.grpc.pb.h"
  47 #include "test/core/util/test_config.h"
  48 #include "test/cpp/util/test_config.h"
  49
  50 ABSL_FLAG(bool, fail_on_failed_rpc, false,
  51           "Fail client if any RPCs fail after first successful RPC.");
  52 ABSL_FLAG(int32_t, num_channels, 1, "Number of channels.");
  53 ABSL_FLAG(bool, print_response, false, "Write RPC response to stdout.");
  54 ABSL_FLAG(int32_t, qps, 1, "Qps per channel.");
  55 // TODO(Capstan): Consider using absl::Duration
  56 ABSL_FLAG(int32_t, rpc_timeout_sec, 30, "Per RPC timeout seconds.");
  57 ABSL_FLAG(std::string, server, "localhost:50051", "Address of server.");
  58 ABSL_FLAG(int32_t, stats_port, 50052,
  59           "Port to expose peer distribution stats service.");
  60 ABSL_FLAG(std::string, rpc, "UnaryCall",
  61           "a comma separated list of rpc methods.");
  62 ABSL_FLAG(std::string, metadata, "", "metadata to send with the RPC.");
  63 ABSL_FLAG(std::string, expect_status, "OK",
  64           "RPC status for the test RPC to be considered successful");
  65 ABSL_FLAG(
  66     bool, secure_mode, false,
  67     "If true, XdsCredentials are used, InsecureChannelCredentials otherwise");
  68
  69 using grpc::Channel;
  70 using grpc::ClientAsyncResponseReader;
  71 using grpc::ClientContext;
  72 using grpc::CompletionQueue;
  73 using grpc::Server;
  74 using grpc::ServerBuilder;
  75 using grpc::ServerContext;
  76 using grpc::Status;
  77 using grpc::testing::ClientConfigureRequest;
  78 using grpc::testing::ClientConfigureRequest_RpcType_Name;
  79 using grpc::testing::ClientConfigureResponse;
  80 using grpc::testing::Empty;
  81 using grpc::testing::LoadBalancerAccumulatedStatsRequest;
  82 using grpc::testing::LoadBalancerAccumulatedStatsResponse;
  83 using grpc::testing::LoadBalancerStatsRequest;
  84 using grpc::testing::LoadBalancerStatsResponse;
  85 using grpc::testing::LoadBalancerStatsService;
  86 using grpc::testing::SimpleRequest;
  87 using grpc::testing::SimpleResponse;
  88 using grpc::testing::TestService;
  89 using grpc::testing::XdsUpdateClientConfigureService;
  90
  91 class XdsStatsWatcher;
  92
  93 struct StatsWatchers {
  94   // Unique ID for each outgoing RPC
  95   int global_request_id = 0;
  96   // Unique ID for each outgoing RPC by RPC method type
  97   std::map<int, int> global_request_id_by_type;
  98   // Stores a set of watchers that should be notified upon outgoing RPC
  99   // completion
 100   std::set<XdsStatsWatcher*> watchers;
 101   // Global watcher for accumululated stats.
 102   XdsStatsWatcher* global_watcher;
 103   // Mutex for global_request_id and watchers
 104   std::mutex mu;
 105 };
 106 // Whether at least one RPC has succeeded, indicating xDS resolution completed.
 107 std::atomic<bool> one_rpc_succeeded(false);
 108 // RPC configuration detailing how RPC should be sent.
 109 struct RpcConfig {
 110   ClientConfigureRequest::RpcType type;
 111   std::vector<std::pair<std::string, std::string>> metadata;
 112   int timeout_sec = 0;
 113 };
 114 struct RpcConfigurationsQueue {
 115   // A queue of RPC configurations detailing how RPCs should be sent.
 116   std::deque<std::vector<RpcConfig>> rpc_configs_queue;
 117   // Mutex for rpc_configs_queue
 118   std::mutex mu_rpc_configs_queue;
 119 };
 120 struct AsyncClientCall {
 121   Empty empty_response;
 122   SimpleResponse simple_response;
 123   ClientContext context;
 124   Status status;
 125   int saved_request_id;
 126   ClientConfigureRequest::RpcType rpc_type;
 127   std::unique_ptr<ClientAsyncResponseReader<Empty>> empty_response_reader;
 128   std::unique_ptr<ClientAsyncResponseReader<SimpleResponse>>
 129       simple_response_reader;
 130 };
 131
 132 /** Records the remote peer distribution for a given range of RPCs. */
 133 class XdsStatsWatcher {
 134  public:
 135   XdsStatsWatcher(int start_id, int end_id)
 136       : start_id_(start_id), end_id_(end_id), rpcs_needed_(end_id - start_id) {}
 137
 138   // Upon the completion of an RPC, we will look at the request_id, the
 139   // rpc_type, and the peer the RPC was sent to in order to count
 140   // this RPC into the right stats bin.
 141   void RpcCompleted(AsyncClientCall* call, const std::string& peer) {
 142     // We count RPCs for global watcher or if the request_id falls into the
 143     // watcher's interested range of request ids.
 144     if ((start_id_ == 0 && end_id_ == 0) ||
 145         (start_id_ <= call->saved_request_id &&
 146          call->saved_request_id < end_id_)) {
 147       {
 148         std::lock_guard<std::mutex> lock(m_);
 149         if (peer.empty()) {
 150           no_remote_peer_++;
 151           ++no_remote_peer_by_type_[call->rpc_type];
 152         } else {
 153           // RPC is counted into both per-peer bin and per-method-per-peer bin.
 154           rpcs_by_peer_[peer]++;
 155           rpcs_by_type_[call->rpc_type][peer]++;
 156         }
 157         rpcs_needed_--;
 158         // Report accumulated stats.
 159         auto& stats_per_method = *accumulated_stats_.mutable_stats_per_method();
 160         auto& method_stat =
 161             stats_per_method[ClientConfigureRequest_RpcType_Name(
 162                 call->rpc_type)];
 163         auto& result = *method_stat.mutable_result();
 164         grpc_status_code code =
 165             static_cast<grpc_status_code>(call->status.error_code());
 166         auto& num_rpcs = result[code];
 167         ++num_rpcs;
 168         auto rpcs_started = method_stat.rpcs_started();
 169         method_stat.set_rpcs_started(++rpcs_started);
 170       }
 171       cv_.notify_one();
 172     }
 173   }
 174
 175   void WaitForRpcStatsResponse(LoadBalancerStatsResponse* response,
 176                                int timeout_sec) {
 177     std::unique_lock<std::mutex> lock(m_);
 178     cv_.wait_for(lock, std::chrono::seconds(timeout_sec),
 179                  [this] { return rpcs_needed_ == 0; });
 180     response->mutable_rpcs_by_peer()->insert(rpcs_by_peer_.begin(),
 181                                              rpcs_by_peer_.end());
 182     auto& response_rpcs_by_method = *response->mutable_rpcs_by_method();
 183     for (const auto& rpc_by_type : rpcs_by_type_) {
 184       std::string method_name;
 185       if (rpc_by_type.first == ClientConfigureRequest::EMPTY_CALL) {
 186         method_name = "EmptyCall";
 187       } else if (rpc_by_type.first == ClientConfigureRequest::UNARY_CALL) {
 188         method_name = "UnaryCall";
 189       } else {
 190         GPR_ASSERT(0);
 191       }
 192       // TODO(@donnadionne): When the test runner changes to accept EMPTY_CALL
 193       // and UNARY_CALL we will just use the name of the enum instead of the
 194       // method_name variable.
 195       auto& response_rpc_by_method = response_rpcs_by_method[method_name];
 196       auto& response_rpcs_by_peer =
 197           *response_rpc_by_method.mutable_rpcs_by_peer();
 198       for (const auto& rpc_by_peer : rpc_by_type.second) {
 199         auto& response_rpc_by_peer = response_rpcs_by_peer[rpc_by_peer.first];
 200         response_rpc_by_peer = rpc_by_peer.second;
 201       }
 202     }
 203     response->set_num_failures(no_remote_peer_ + rpcs_needed_);
 204   }
 205
 206   void GetCurrentRpcStats(LoadBalancerAccumulatedStatsResponse* response,
 207                           StatsWatchers* stats_watchers) {
 208     std::unique_lock<std::mutex> lock(m_);
 209     response->CopyFrom(accumulated_stats_);
 210     // TODO(@donnadionne): delete deprecated stats below when the test is no
 211     // longer using them.
 212     auto& response_rpcs_started_by_method =
 213         *response->mutable_num_rpcs_started_by_method();
 214     auto& response_rpcs_succeeded_by_method =
 215         *response->mutable_num_rpcs_succeeded_by_method();
 216     auto& response_rpcs_failed_by_method =
 217         *response->mutable_num_rpcs_failed_by_method();
 218     for (const auto& rpc_by_type : rpcs_by_type_) {
 219       auto total_succeeded = 0;
 220       for (const auto& rpc_by_peer : rpc_by_type.second) {
 221         total_succeeded += rpc_by_peer.second;
 222       }
 223       response_rpcs_succeeded_by_method[ClientConfigureRequest_RpcType_Name(
 224           rpc_by_type.first)] = total_succeeded;
 225       response_rpcs_started_by_method[ClientConfigureRequest_RpcType_Name(
 226           rpc_by_type.first)] =
 227           stats_watchers->global_request_id_by_type[rpc_by_type.first];
 228       response_rpcs_failed_by_method[ClientConfigureRequest_RpcType_Name(
 229           rpc_by_type.first)] = no_remote_peer_by_type_[rpc_by_type.first];
 230     }
 231   }
 232
 233  private:
 234   int start_id_;
 235   int end_id_;
 236   int rpcs_needed_;
 237   int no_remote_peer_ = 0;
 238   std::map<int, int> no_remote_peer_by_type_;
 239   // A map of stats keyed by peer name.
 240   std::map<std::string, int> rpcs_by_peer_;
 241   // A two-level map of stats keyed at top level by RPC method and second level
 242   // by peer name.
 243   std::map<int, std::map<std::string, int>> rpcs_by_type_;
 244   // Storing accumulated stats in the response proto format.
 245   LoadBalancerAccumulatedStatsResponse accumulated_stats_;
 246   std::mutex m_;
 247   std::condition_variable cv_;
 248 };
 249
 250 class TestClient {
 251  public:
 252   TestClient(const std::shared_ptr<Channel>& channel,
 253              StatsWatchers* stats_watchers)
 254       : stub_(TestService::NewStub(channel)), stats_watchers_(stats_watchers) {}
 255
 256   void AsyncUnaryCall(const RpcConfig& config) {
 257     SimpleResponse response;
 258     int saved_request_id;
 259     {
 260       std::lock_guard<std::mutex> lock(stats_watchers_->mu);
 261       saved_request_id = ++stats_watchers_->global_request_id;
 262       ++stats_watchers_
 263             ->global_request_id_by_type[ClientConfigureRequest::UNARY_CALL];
 264     }
 265     std::chrono::system_clock::time_point deadline =
 266         std::chrono::system_clock::now() +
 267         std::chrono::seconds(config.timeout_sec != 0
 268                                  ? config.timeout_sec
 269                                  : absl::GetFlag(FLAGS_rpc_timeout_sec));
 270     AsyncClientCall* call = new AsyncClientCall;
 271     for (const auto& data : config.metadata) {
 272       call->context.AddMetadata(data.first, data.second);
 273       // TODO(@donnadionne): move deadline to separate proto.
 274       if (data.first == "rpc-behavior" && data.second == "keep-open") {
 275         deadline =
 276             std::chrono::system_clock::now() + std::chrono::seconds(INT_MAX);
 277       }
 278     }
 279     call->context.set_deadline(deadline);
 280     call->saved_request_id = saved_request_id;
 281     call->rpc_type = ClientConfigureRequest::UNARY_CALL;
 282     call->simple_response_reader = stub_->PrepareAsyncUnaryCall(
 283         &call->context, SimpleRequest::default_instance(), &cq_);
 284     call->simple_response_reader->StartCall();
 285     call->simple_response_reader->Finish(&call->simple_response, &call->status,
 286                                          call);
 287   }
 288
 289   void AsyncEmptyCall(const RpcConfig& config) {
 290     Empty response;
 291     int saved_request_id;
 292     {
 293       std::lock_guard<std::mutex> lock(stats_watchers_->mu);
 294       saved_request_id = ++stats_watchers_->global_request_id;
 295       ++stats_watchers_
 296             ->global_request_id_by_type[ClientConfigureRequest::EMPTY_CALL];
 297     }
 298     std::chrono::system_clock::time_point deadline =
 299         std::chrono::system_clock::now() +
 300         std::chrono::seconds(config.timeout_sec != 0
 301                                  ? config.timeout_sec
 302                                  : absl::GetFlag(FLAGS_rpc_timeout_sec));
 303     AsyncClientCall* call = new AsyncClientCall;
 304     for (const auto& data : config.metadata) {
 305       call->context.AddMetadata(data.first, data.second);
 306       // TODO(@donnadionne): move deadline to separate proto.
 307       if (data.first == "rpc-behavior" && data.second == "keep-open") {
 308         deadline =
 309             std::chrono::system_clock::now() + std::chrono::seconds(INT_MAX);
 310       }
 311     }
 312     call->context.set_deadline(deadline);
 313     call->saved_request_id = saved_request_id;
 314     call->rpc_type = ClientConfigureRequest::EMPTY_CALL;
 315     call->empty_response_reader = stub_->PrepareAsyncEmptyCall(
 316         &call->context, Empty::default_instance(), &cq_);
 317     call->empty_response_reader->StartCall();
 318     call->empty_response_reader->Finish(&call->empty_response, &call->status,
 319                                         call);
 320   }
 321
 322   void AsyncCompleteRpc() {
 323     void* got_tag;
 324     bool ok = false;
 325     while (cq_.Next(&got_tag, &ok)) {
 326       AsyncClientCall* call = static_cast<AsyncClientCall*>(got_tag);
 327       GPR_ASSERT(ok);
 328       {
 329         std::lock_guard<std::mutex> lock(stats_watchers_->mu);
 330         auto server_initial_metadata = call->context.GetServerInitialMetadata();
 331         auto metadata_hostname =
 332             call->context.GetServerInitialMetadata().find("hostname");
 333         std::string hostname =
 334             metadata_hostname != call->context.GetServerInitialMetadata().end()
 335                 ? std::string(metadata_hostname->second.data(),
 336                               metadata_hostname->second.length())
 337                 : call->simple_response.hostname();
 338         for (auto watcher : stats_watchers_->watchers) {
 339           watcher->RpcCompleted(call, hostname);
 340         }
 341       }
 342
 343       if (!RpcStatusCheckSuccess(call)) {
 344         if (absl::GetFlag(FLAGS_print_response) ||
 345             absl::GetFlag(FLAGS_fail_on_failed_rpc)) {
 346           std::cout << "RPC failed: " << call->status.error_code() << ": "
 347                     << call->status.error_message() << std::endl;
 348         }
 349         if (absl::GetFlag(FLAGS_fail_on_failed_rpc) &&
 350             one_rpc_succeeded.load()) {
 351           abort();
 352         }
 353       } else {
 354         if (absl::GetFlag(FLAGS_print_response)) {
 355           auto metadata_hostname =
 356               call->context.GetServerInitialMetadata().find("hostname");
 357           std::string hostname =
 358               metadata_hostname !=
 359                       call->context.GetServerInitialMetadata().end()
 360                   ? std::string(metadata_hostname->second.data(),
 361                                 metadata_hostname->second.length())
 362                   : call->simple_response.hostname();
 363           std::cout << "Greeting: Hello world, this is " << hostname
 364                     << ", from " << call->context.peer() << std::endl;
 365         }
 366         one_rpc_succeeded = true;
 367       }
 368
 369       delete call;
 370     }
 371   }
 372
 373  private:
 374   static bool RpcStatusCheckSuccess(AsyncClientCall* call) {
 375     // Determine RPC success based on expected status.
 376     grpc_status_code code;
 377     GPR_ASSERT(grpc_status_code_from_string(
 378         absl::GetFlag(FLAGS_expect_status).c_str(), &code));
 379     return code == static_cast<grpc_status_code>(call->status.error_code());
 380   }
 381
 382   std::unique_ptr<TestService::Stub> stub_;
 383   StatsWatchers* stats_watchers_;
 384   CompletionQueue cq_;
 385 };
 386
 387 class LoadBalancerStatsServiceImpl : public LoadBalancerStatsService::Service {
 388  public:
 389   explicit LoadBalancerStatsServiceImpl(StatsWatchers* stats_watchers)
 390       : stats_watchers_(stats_watchers) {}
 391
 392   Status GetClientStats(ServerContext* /*context*/,
 393                         const LoadBalancerStatsRequest* request,
 394                         LoadBalancerStatsResponse* response) override {
 395     int start_id;
 396     int end_id;
 397     XdsStatsWatcher* watcher;
 398     {
 399       std::lock_guard<std::mutex> lock(stats_watchers_->mu);
 400       start_id = stats_watchers_->global_request_id + 1;
 401       end_id = start_id + request->num_rpcs();
 402       watcher = new XdsStatsWatcher(start_id, end_id);
 403       stats_watchers_->watchers.insert(watcher);
 404     }
 405     watcher->WaitForRpcStatsResponse(response, request->timeout_sec());
 406     {
 407       std::lock_guard<std::mutex> lock(stats_watchers_->mu);
 408       stats_watchers_->watchers.erase(watcher);
 409     }
 410     delete watcher;
 411     return Status::OK;
 412   }
 413
 414   Status GetClientAccumulatedStats(
 415       ServerContext* /*context*/,
 416       const LoadBalancerAccumulatedStatsRequest* /*request*/,
 417       LoadBalancerAccumulatedStatsResponse* response) override {
 418     std::lock_guard<std::mutex> lock(stats_watchers_->mu);
 419     stats_watchers_->global_watcher->GetCurrentRpcStats(response,
 420                                                         stats_watchers_);
 421     return Status::OK;
 422   }
 423
 424  private:
 425   StatsWatchers* stats_watchers_;
 426 };
 427
 428 class XdsUpdateClientConfigureServiceImpl
 429     : public XdsUpdateClientConfigureService::Service {
 430  public:
 431   explicit XdsUpdateClientConfigureServiceImpl(
 432       RpcConfigurationsQueue* rpc_configs_queue)
 433       : rpc_configs_queue_(rpc_configs_queue) {}
 434
 435   Status Configure(ServerContext* /*context*/,
 436                    const ClientConfigureRequest* request,
 437                    ClientConfigureResponse* /*response*/) override {
 438     std::map<int, std::vector<std::pair<std::string, std::string>>>
 439         metadata_map;
 440     for (const auto& data : request->metadata()) {
 441       metadata_map[data.type()].push_back({data.key(), data.value()});
 442     }
 443     std::vector<RpcConfig> configs;
 444     for (const auto& rpc : request->types()) {
 445       RpcConfig config;
 446       config.timeout_sec = request->timeout_sec();
 447       config.type = static_cast<ClientConfigureRequest::RpcType>(rpc);
 448       auto metadata_iter = metadata_map.find(rpc);
 449       if (metadata_iter != metadata_map.end()) {
 450         config.metadata = metadata_iter->second;
 451       }
 452       configs.push_back(std::move(config));
 453     }
 454     {
 455       std::lock_guard<std::mutex> lock(
 456           rpc_configs_queue_->mu_rpc_configs_queue);
 457       rpc_configs_queue_->rpc_configs_queue.emplace_back(std::move(configs));
 458     }
 459     return Status::OK;
 460   }
 461
 462  private:
 463   RpcConfigurationsQueue* rpc_configs_queue_;
 464 };
 465
 466 void RunTestLoop(std::chrono::duration<double> duration_per_query,
 467                  StatsWatchers* stats_watchers,
 468                  RpcConfigurationsQueue* rpc_configs_queue) {
 469   grpc::ChannelArguments channel_args;
 470   channel_args.SetInt(GRPC_ARG_ENABLE_RETRIES, 1);
 471   TestClient client(
 472       grpc::CreateCustomChannel(absl::GetFlag(FLAGS_server),
 473                                 absl::GetFlag(FLAGS_secure_mode)
 474                                     ? grpc::experimental::XdsCredentials(
 475                                           grpc::InsecureChannelCredentials())
 476                                     : grpc::InsecureChannelCredentials(),
 477                                 channel_args),
 478       stats_watchers);
 479   std::chrono::time_point<std::chrono::system_clock> start =
 480       std::chrono::system_clock::now();
 481   std::chrono::duration<double> elapsed;
 482
 483   std::thread thread = std::thread(&TestClient::AsyncCompleteRpc, &client);
 484
 485   std::vector<RpcConfig> configs;
 486   while (true) {
 487     {
 488       std::lock_guard<std::mutex> lockk(
 489           rpc_configs_queue->mu_rpc_configs_queue);
 490       if (!rpc_configs_queue->rpc_configs_queue.empty()) {
 491         configs = std::move(rpc_configs_queue->rpc_configs_queue.front());
 492         rpc_configs_queue->rpc_configs_queue.pop_front();
 493       }
 494     }
 495
 496     elapsed = std::chrono::system_clock::now() - start;
 497     if (elapsed > duration_per_query) {
 498       start = std::chrono::system_clock::now();
 499       for (const auto& config : configs) {
 500         if (config.type == ClientConfigureRequest::EMPTY_CALL) {
 501           client.AsyncEmptyCall(config);
 502         } else if (config.type == ClientConfigureRequest::UNARY_CALL) {
 503           client.AsyncUnaryCall(config);
 504         } else {
 505           GPR_ASSERT(0);
 506         }
 507       }
 508     }
 509   }
 510   GPR_UNREACHABLE_CODE(thread.join());
 511 }
 512
 513 void RunServer(const int port, StatsWatchers* stats_watchers,
 514                RpcConfigurationsQueue* rpc_configs_queue) {
 515   GPR_ASSERT(port != 0);
 516   std::ostringstream server_address;
 517   server_address << "0.0.0.0:" << port;
 518
 519   LoadBalancerStatsServiceImpl stats_service(stats_watchers);
 520   XdsUpdateClientConfigureServiceImpl client_config_service(rpc_configs_queue);
 521
 522   grpc::reflection::InitProtoReflectionServerBuilderPlugin();
 523   ServerBuilder builder;
 524   builder.RegisterService(&stats_service);
 525   builder.RegisterService(&client_config_service);
 526   grpc::AddAdminServices(&builder);
 527   builder.AddListeningPort(server_address.str(),
 528                            grpc::InsecureServerCredentials());
 529   std::unique_ptr<Server> server(builder.BuildAndStart());
 530   gpr_log(GPR_DEBUG, "Server listening on %s", server_address.str().c_str());
 531
 532   server->Wait();
 533 }
 534
 535 void BuildRpcConfigsFromFlags(RpcConfigurationsQueue* rpc_configs_queue) {
 536   // Store Metadata like
 537   // "EmptyCall:key1:value1,UnaryCall:key1:value1,UnaryCall:key2:value2" into a
 538   // map where the key is the RPC method and value is a vector of key:value
 539   // pairs. {EmptyCall, [{key1,value1}],
 540   //  UnaryCall, [{key1,value1}, {key2,value2}]}
 541   std::vector<std::string> rpc_metadata =
 542       absl::StrSplit(absl::GetFlag(FLAGS_metadata), ',', absl::SkipEmpty());
 543   std::map<int, std::vector<std::pair<std::string, std::string>>> metadata_map;
 544   for (auto& data : rpc_metadata) {
 545     std::vector<std::string> metadata =
 546         absl::StrSplit(data, ':', absl::SkipEmpty());
 547     GPR_ASSERT(metadata.size() == 3);
 548     if (metadata[0] == "EmptyCall") {
 549       metadata_map[ClientConfigureRequest::EMPTY_CALL].push_back(
 550           {metadata[1], metadata[2]});
 551     } else if (metadata[0] == "UnaryCall") {
 552       metadata_map[ClientConfigureRequest::UNARY_CALL].push_back(
 553           {metadata[1], metadata[2]});
 554     } else {
 555       GPR_ASSERT(0);
 556     }
 557   }
 558   std::vector<RpcConfig> configs;
 559   std::vector<std::string> rpc_methods =
 560       absl::StrSplit(absl::GetFlag(FLAGS_rpc), ',', absl::SkipEmpty());
 561   for (const std::string& rpc_method : rpc_methods) {
 562     RpcConfig config;
 563     if (rpc_method == "EmptyCall") {
 564       config.type = ClientConfigureRequest::EMPTY_CALL;
 565     } else if (rpc_method == "UnaryCall") {
 566       config.type = ClientConfigureRequest::UNARY_CALL;
 567     } else {
 568       GPR_ASSERT(0);
 569     }
 570     auto metadata_iter = metadata_map.find(config.type);
 571     if (metadata_iter != metadata_map.end()) {
 572       config.metadata = metadata_iter->second;
 573     }
 574     configs.push_back(std::move(config));
 575   }
 576   {
 577     std::lock_guard<std::mutex> lock(rpc_configs_queue->mu_rpc_configs_queue);
 578     rpc_configs_queue->rpc_configs_queue.emplace_back(std::move(configs));
 579   }
 580 }
 581
 582 int main(int argc, char** argv) {
 583   grpc::testing::TestEnvironment env(argc, argv);
 584   grpc::testing::InitTest(&argc, &argv, true);
 585   // Validate the expect_status flag.
 586   grpc_status_code code;
 587   GPR_ASSERT(grpc_status_code_from_string(
 588       absl::GetFlag(FLAGS_expect_status).c_str(), &code));
 589   StatsWatchers stats_watchers;
 590   RpcConfigurationsQueue rpc_config_queue;
 591
 592   {
 593     std::lock_guard<std::mutex> lock(stats_watchers.mu);
 594     stats_watchers.global_watcher = new XdsStatsWatcher(0, 0);
 595     stats_watchers.watchers.insert(stats_watchers.global_watcher);
 596   }
 597
 598   BuildRpcConfigsFromFlags(&rpc_config_queue);
 599
 600   std::chrono::duration<double> duration_per_query =
 601       std::chrono::nanoseconds(std::chrono::seconds(1)) /
 602       absl::GetFlag(FLAGS_qps);
 603
 604   std::vector<std::thread> test_threads;
 605   test_threads.reserve(absl::GetFlag(FLAGS_num_channels));
 606   for (int i = 0; i < absl::GetFlag(FLAGS_num_channels); i++) {
 607     test_threads.emplace_back(std::thread(&RunTestLoop, duration_per_query,
 608                                           &stats_watchers, &rpc_config_queue));
 609   }
 610
 611   RunServer(absl::GetFlag(FLAGS_stats_port), &stats_watchers,
 612             &rpc_config_queue);
 613
 614   for (auto it = test_threads.begin(); it != test_threads.end(); it++) {
 615     it->join();
 616   }
 617
 618   return 0;
 619 }