test/cpp/interop/interop_client.cc

   1 /*
   2  *
   3  * Copyright 2015-2016 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 "test/cpp/interop/interop_client.h"
  20
  21 #include <cinttypes>
  22 #include <fstream>
  23 #include <memory>
  24 #include <string>
  25 #include <type_traits>
  26 #include <utility>
  27
  28 #include "absl/strings/match.h"
  29 #include "absl/strings/str_format.h"
  30
  31 #include <grpc/grpc.h>
  32 #include <grpc/support/alloc.h>
  33 #include <grpc/support/log.h>
  34 #include <grpc/support/string_util.h>
  35 #include <grpc/support/time.h>
  36 #include <grpcpp/channel.h>
  37 #include <grpcpp/client_context.h>
  38 #include <grpcpp/security/credentials.h>
  39
  40 #include "src/proto/grpc/testing/empty.pb.h"
  41 #include "src/proto/grpc/testing/messages.pb.h"
  42 #include "src/proto/grpc/testing/test.grpc.pb.h"
  43 #include "test/core/util/histogram.h"
  44 #include "test/cpp/interop/client_helper.h"
  45
  46 namespace grpc {
  47 namespace testing {
  48
  49 namespace {
  50 // The same value is defined by the Java client.
  51 const std::vector<int> request_stream_sizes = {27182, 8, 1828, 45904};
  52 const std::vector<int> response_stream_sizes = {31415, 9, 2653, 58979};
  53 const int kNumResponseMessages = 2000;
  54 const int kResponseMessageSize = 1030;
  55 const int kReceiveDelayMilliSeconds = 20;
  56 const int kLargeRequestSize = 271828;
  57 const int kLargeResponseSize = 314159;
  58
  59 void NoopChecks(const InteropClientContextInspector& /*inspector*/,
  60                 const SimpleRequest* /*request*/,
  61                 const SimpleResponse* /*response*/) {}
  62
  63 void UnaryCompressionChecks(const InteropClientContextInspector& inspector,
  64                             const SimpleRequest* request,
  65                             const SimpleResponse* /*response*/) {
  66   const grpc_compression_algorithm received_compression =
  67       inspector.GetCallCompressionAlgorithm();
  68   if (request->response_compressed().value()) {
  69     if (received_compression == GRPC_COMPRESS_NONE) {
  70       // Requested some compression, got NONE. This is an error.
  71       gpr_log(GPR_ERROR,
  72               "Failure: Requested compression but got uncompressed response "
  73               "from server.");
  74       abort();
  75     }
  76     GPR_ASSERT(inspector.WasCompressed());
  77   } else {
  78     // Didn't request compression -> make sure the response is uncompressed
  79     GPR_ASSERT(!(inspector.WasCompressed()));
  80   }
  81 }
  82 }  // namespace
  83
  84 InteropClient::ServiceStub::ServiceStub(
  85     ChannelCreationFunc channel_creation_func, bool new_stub_every_call)
  86     : channel_creation_func_(std::move(channel_creation_func)),
  87       channel_(channel_creation_func_()),
  88       new_stub_every_call_(new_stub_every_call) {
  89   // If new_stub_every_call is false, then this is our chance to initialize
  90   // stub_. (see Get())
  91   if (!new_stub_every_call) {
  92     stub_ = TestService::NewStub(channel_);
  93   }
  94 }
  95
  96 TestService::Stub* InteropClient::ServiceStub::Get() {
  97   if (new_stub_every_call_) {
  98     stub_ = TestService::NewStub(channel_);
  99   }
 100
 101   return stub_.get();
 102 }
 103
 104 UnimplementedService::Stub*
 105 InteropClient::ServiceStub::GetUnimplementedServiceStub() {
 106   if (unimplemented_service_stub_ == nullptr) {
 107     unimplemented_service_stub_ = UnimplementedService::NewStub(channel_);
 108   }
 109   return unimplemented_service_stub_.get();
 110 }
 111
 112 void InteropClient::ServiceStub::ResetChannel() {
 113   channel_ = channel_creation_func_();
 114   if (!new_stub_every_call_) {
 115     stub_ = TestService::NewStub(channel_);
 116   }
 117 }
 118
 119 InteropClient::InteropClient(ChannelCreationFunc channel_creation_func,
 120                              bool new_stub_every_test_case,
 121                              bool do_not_abort_on_transient_failures)
 122     : serviceStub_(std::move(channel_creation_func), new_stub_every_test_case),
 123       do_not_abort_on_transient_failures_(do_not_abort_on_transient_failures) {}
 124
 125 bool InteropClient::AssertStatusOk(const Status& s,
 126                                    const std::string& optional_debug_string) {
 127   if (s.ok()) {
 128     return true;
 129   }
 130
 131   // Note: At this point, s.error_code is definitely not StatusCode::OK (we
 132   // already checked for s.ok() above). So, the following will call abort()
 133   // (unless s.error_code() corresponds to a transient failure and
 134   // 'do_not_abort_on_transient_failures' is true)
 135   return AssertStatusCode(s, StatusCode::OK, optional_debug_string);
 136 }
 137
 138 bool InteropClient::AssertStatusCode(const Status& s, StatusCode expected_code,
 139                                      const std::string& optional_debug_string) {
 140   if (s.error_code() == expected_code) {
 141     return true;
 142   }
 143
 144   gpr_log(GPR_ERROR,
 145           "Error status code: %d (expected: %d), message: %s,"
 146           " debug string: %s",
 147           s.error_code(), expected_code, s.error_message().c_str(),
 148           optional_debug_string.c_str());
 149
 150   // In case of transient transient/retryable failures (like a broken
 151   // connection) we may or may not abort (see TransientFailureOrAbort())
 152   if (s.error_code() == grpc::StatusCode::UNAVAILABLE) {
 153     return TransientFailureOrAbort();
 154   }
 155
 156   abort();
 157 }
 158
 159 bool InteropClient::DoEmpty() {
 160   gpr_log(GPR_DEBUG, "Sending an empty rpc...");
 161
 162   Empty request;
 163   Empty response;
 164   ClientContext context;
 165
 166   Status s = serviceStub_.Get()->EmptyCall(&context, request, &response);
 167
 168   if (!AssertStatusOk(s, context.debug_error_string())) {
 169     return false;
 170   }
 171
 172   gpr_log(GPR_DEBUG, "Empty rpc done.");
 173   return true;
 174 }
 175
 176 bool InteropClient::PerformLargeUnary(SimpleRequest* request,
 177                                       SimpleResponse* response) {
 178   return PerformLargeUnary(request, response, NoopChecks);
 179 }
 180
 181 bool InteropClient::PerformLargeUnary(SimpleRequest* request,
 182                                       SimpleResponse* response,
 183                                       const CheckerFn& custom_checks_fn) {
 184   ClientContext context;
 185   InteropClientContextInspector inspector(context);
 186   request->set_response_size(kLargeResponseSize);
 187   std::string payload(kLargeRequestSize, '\0');
 188   request->mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
 189   if (request->has_expect_compressed()) {
 190     if (request->expect_compressed().value()) {
 191       context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
 192     } else {
 193       context.set_compression_algorithm(GRPC_COMPRESS_NONE);
 194     }
 195   }
 196
 197   Status s = serviceStub_.Get()->UnaryCall(&context, *request, response);
 198   if (!AssertStatusOk(s, context.debug_error_string())) {
 199     return false;
 200   }
 201
 202   custom_checks_fn(inspector, request, response);
 203
 204   // Payload related checks.
 205   GPR_ASSERT(response->payload().body() ==
 206              std::string(kLargeResponseSize, '\0'));
 207   return true;
 208 }
 209
 210 bool InteropClient::DoComputeEngineCreds(
 211     const std::string& default_service_account,
 212     const std::string& oauth_scope) {
 213   gpr_log(GPR_DEBUG,
 214           "Sending a large unary rpc with compute engine credentials ...");
 215   SimpleRequest request;
 216   SimpleResponse response;
 217   request.set_fill_username(true);
 218   request.set_fill_oauth_scope(true);
 219
 220   if (!PerformLargeUnary(&request, &response)) {
 221     return false;
 222   }
 223
 224   gpr_log(GPR_DEBUG, "Got username %s", response.username().c_str());
 225   gpr_log(GPR_DEBUG, "Got oauth_scope %s", response.oauth_scope().c_str());
 226   GPR_ASSERT(!response.username().empty());
 227   GPR_ASSERT(response.username().c_str() == default_service_account);
 228   GPR_ASSERT(!response.oauth_scope().empty());
 229   const char* oauth_scope_str = response.oauth_scope().c_str();
 230   GPR_ASSERT(absl::StrContains(oauth_scope, oauth_scope_str));
 231   gpr_log(GPR_DEBUG, "Large unary with compute engine creds done.");
 232   return true;
 233 }
 234
 235 bool InteropClient::DoOauth2AuthToken(const std::string& username,
 236                                       const std::string& oauth_scope) {
 237   gpr_log(GPR_DEBUG,
 238           "Sending a unary rpc with raw oauth2 access token credentials ...");
 239   SimpleRequest request;
 240   SimpleResponse response;
 241   request.set_fill_username(true);
 242   request.set_fill_oauth_scope(true);
 243
 244   ClientContext context;
 245
 246   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
 247
 248   if (!AssertStatusOk(s, context.debug_error_string())) {
 249     return false;
 250   }
 251
 252   GPR_ASSERT(!response.username().empty());
 253   GPR_ASSERT(!response.oauth_scope().empty());
 254   GPR_ASSERT(username == response.username());
 255   const char* oauth_scope_str = response.oauth_scope().c_str();
 256   GPR_ASSERT(absl::StrContains(oauth_scope, oauth_scope_str));
 257   gpr_log(GPR_DEBUG, "Unary with oauth2 access token credentials done.");
 258   return true;
 259 }
 260
 261 bool InteropClient::DoPerRpcCreds(const std::string& json_key) {
 262   gpr_log(GPR_DEBUG, "Sending a unary rpc with per-rpc JWT access token ...");
 263   SimpleRequest request;
 264   SimpleResponse response;
 265   request.set_fill_username(true);
 266
 267   ClientContext context;
 268   std::chrono::seconds token_lifetime = std::chrono::hours(1);
 269   std::shared_ptr<CallCredentials> creds =
 270       ServiceAccountJWTAccessCredentials(json_key, token_lifetime.count());
 271
 272   context.set_credentials(creds);
 273
 274   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
 275
 276   if (!AssertStatusOk(s, context.debug_error_string())) {
 277     return false;
 278   }
 279
 280   GPR_ASSERT(!response.username().empty());
 281   GPR_ASSERT(json_key.find(response.username()) != std::string::npos);
 282   gpr_log(GPR_DEBUG, "Unary with per-rpc JWT access token done.");
 283   return true;
 284 }
 285
 286 bool InteropClient::DoJwtTokenCreds(const std::string& username) {
 287   gpr_log(GPR_DEBUG,
 288           "Sending a large unary rpc with JWT token credentials ...");
 289   SimpleRequest request;
 290   SimpleResponse response;
 291   request.set_fill_username(true);
 292
 293   if (!PerformLargeUnary(&request, &response)) {
 294     return false;
 295   }
 296
 297   GPR_ASSERT(!response.username().empty());
 298   GPR_ASSERT(username.find(response.username()) != std::string::npos);
 299   gpr_log(GPR_DEBUG, "Large unary with JWT token creds done.");
 300   return true;
 301 }
 302
 303 bool InteropClient::DoGoogleDefaultCredentials(
 304     const std::string& default_service_account) {
 305   gpr_log(GPR_DEBUG,
 306           "Sending a large unary rpc with GoogleDefaultCredentials...");
 307   SimpleRequest request;
 308   SimpleResponse response;
 309   request.set_fill_username(true);
 310
 311   if (!PerformLargeUnary(&request, &response)) {
 312     return false;
 313   }
 314
 315   gpr_log(GPR_DEBUG, "Got username %s", response.username().c_str());
 316   GPR_ASSERT(!response.username().empty());
 317   GPR_ASSERT(response.username().c_str() == default_service_account);
 318   gpr_log(GPR_DEBUG, "Large unary rpc with GoogleDefaultCredentials done.");
 319   return true;
 320 }
 321
 322 bool InteropClient::DoLargeUnary() {
 323   gpr_log(GPR_DEBUG, "Sending a large unary rpc...");
 324   SimpleRequest request;
 325   SimpleResponse response;
 326   if (!PerformLargeUnary(&request, &response)) {
 327     return false;
 328   }
 329   gpr_log(GPR_DEBUG, "Large unary done.");
 330   return true;
 331 }
 332
 333 bool InteropClient::DoClientCompressedUnary() {
 334   // Probing for compression-checks support.
 335   ClientContext probe_context;
 336   SimpleRequest probe_req;
 337   SimpleResponse probe_res;
 338
 339   probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE);
 340   probe_req.mutable_expect_compressed()->set_value(true);  // lies!
 341
 342   probe_req.set_response_size(kLargeResponseSize);
 343   probe_req.mutable_payload()->set_body(std::string(kLargeRequestSize, '\0'));
 344
 345   gpr_log(GPR_DEBUG, "Sending probe for compressed unary request.");
 346   const Status s =
 347       serviceStub_.Get()->UnaryCall(&probe_context, probe_req, &probe_res);
 348   if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) {
 349     // The server isn't able to evaluate incoming compression, making the rest
 350     // of this test moot.
 351     gpr_log(GPR_DEBUG, "Compressed unary request probe failed");
 352     return false;
 353   }
 354   gpr_log(GPR_DEBUG, "Compressed unary request probe succeeded. Proceeding.");
 355
 356   const std::vector<bool> compressions = {true, false};
 357   for (size_t i = 0; i < compressions.size(); i++) {
 358     std::string log_suffix =
 359         absl::StrFormat("(compression=%s)", compressions[i] ? "true" : "false");
 360
 361     gpr_log(GPR_DEBUG, "Sending compressed unary request %s.",
 362             log_suffix.c_str());
 363     SimpleRequest request;
 364     SimpleResponse response;
 365     request.mutable_expect_compressed()->set_value(compressions[i]);
 366     if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) {
 367       gpr_log(GPR_ERROR, "Compressed unary request failed %s",
 368               log_suffix.c_str());
 369       return false;
 370     }
 371
 372     gpr_log(GPR_DEBUG, "Compressed unary request failed %s",
 373             log_suffix.c_str());
 374   }
 375
 376   return true;
 377 }
 378
 379 bool InteropClient::DoServerCompressedUnary() {
 380   const std::vector<bool> compressions = {true, false};
 381   for (size_t i = 0; i < compressions.size(); i++) {
 382     std::string log_suffix =
 383         absl::StrFormat("(compression=%s)", compressions[i] ? "true" : "false");
 384
 385     gpr_log(GPR_DEBUG, "Sending unary request for compressed response %s.",
 386             log_suffix.c_str());
 387     SimpleRequest request;
 388     SimpleResponse response;
 389     request.mutable_response_compressed()->set_value(compressions[i]);
 390
 391     if (!PerformLargeUnary(&request, &response, UnaryCompressionChecks)) {
 392       gpr_log(GPR_ERROR, "Request for compressed unary failed %s",
 393               log_suffix.c_str());
 394       return false;
 395     }
 396
 397     gpr_log(GPR_DEBUG, "Request for compressed unary failed %s",
 398             log_suffix.c_str());
 399   }
 400
 401   return true;
 402 }
 403
 404 // Either abort() (unless do_not_abort_on_transient_failures_ is true) or return
 405 // false
 406 bool InteropClient::TransientFailureOrAbort() {
 407   if (do_not_abort_on_transient_failures_) {
 408     return false;
 409   }
 410
 411   abort();
 412 }
 413
 414 bool InteropClient::DoRequestStreaming() {
 415   gpr_log(GPR_DEBUG, "Sending request steaming rpc ...");
 416
 417   ClientContext context;
 418   StreamingInputCallRequest request;
 419   StreamingInputCallResponse response;
 420
 421   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
 422       serviceStub_.Get()->StreamingInputCall(&context, &response));
 423
 424   int aggregated_payload_size = 0;
 425   for (size_t i = 0; i < request_stream_sizes.size(); ++i) {
 426     Payload* payload = request.mutable_payload();
 427     payload->set_body(std::string(request_stream_sizes[i], '\0'));
 428     if (!stream->Write(request)) {
 429       gpr_log(GPR_ERROR, "DoRequestStreaming(): stream->Write() failed");
 430       return TransientFailureOrAbort();
 431     }
 432     aggregated_payload_size += request_stream_sizes[i];
 433   }
 434   GPR_ASSERT(stream->WritesDone());
 435
 436   Status s = stream->Finish();
 437   if (!AssertStatusOk(s, context.debug_error_string())) {
 438     return false;
 439   }
 440
 441   GPR_ASSERT(response.aggregated_payload_size() == aggregated_payload_size);
 442   return true;
 443 }
 444
 445 bool InteropClient::DoResponseStreaming() {
 446   gpr_log(GPR_DEBUG, "Receiving response streaming rpc ...");
 447
 448   ClientContext context;
 449   StreamingOutputCallRequest request;
 450   for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) {
 451     ResponseParameters* response_parameter = request.add_response_parameters();
 452     response_parameter->set_size(response_stream_sizes[i]);
 453   }
 454   StreamingOutputCallResponse response;
 455   std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
 456       serviceStub_.Get()->StreamingOutputCall(&context, request));
 457
 458   unsigned int i = 0;
 459   while (stream->Read(&response)) {
 460     GPR_ASSERT(response.payload().body() ==
 461                std::string(response_stream_sizes[i], '\0'));
 462     ++i;
 463   }
 464
 465   if (i < response_stream_sizes.size()) {
 466     // stream->Read() failed before reading all the expected messages. This is
 467     // most likely due to connection failure.
 468     gpr_log(GPR_ERROR,
 469             "DoResponseStreaming(): Read fewer streams (%d) than "
 470             "response_stream_sizes.size() (%" PRIuPTR ")",
 471             i, response_stream_sizes.size());
 472     return TransientFailureOrAbort();
 473   }
 474
 475   Status s = stream->Finish();
 476   if (!AssertStatusOk(s, context.debug_error_string())) {
 477     return false;
 478   }
 479
 480   gpr_log(GPR_DEBUG, "Response streaming done.");
 481   return true;
 482 }
 483
 484 bool InteropClient::DoClientCompressedStreaming() {
 485   // Probing for compression-checks support.
 486   ClientContext probe_context;
 487   StreamingInputCallRequest probe_req;
 488   StreamingInputCallResponse probe_res;
 489
 490   probe_context.set_compression_algorithm(GRPC_COMPRESS_NONE);
 491   probe_req.mutable_expect_compressed()->set_value(true);  // lies!
 492   probe_req.mutable_payload()->set_body(std::string(27182, '\0'));
 493
 494   gpr_log(GPR_DEBUG, "Sending probe for compressed streaming request.");
 495
 496   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> probe_stream(
 497       serviceStub_.Get()->StreamingInputCall(&probe_context, &probe_res));
 498
 499   if (!probe_stream->Write(probe_req)) {
 500     gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
 501     return TransientFailureOrAbort();
 502   }
 503   Status s = probe_stream->Finish();
 504   if (s.error_code() != grpc::StatusCode::INVALID_ARGUMENT) {
 505     // The server isn't able to evaluate incoming compression, making the rest
 506     // of this test moot.
 507     gpr_log(GPR_DEBUG, "Compressed streaming request probe failed");
 508     return false;
 509   }
 510   gpr_log(GPR_DEBUG,
 511           "Compressed streaming request probe succeeded. Proceeding.");
 512
 513   ClientContext context;
 514   StreamingInputCallRequest request;
 515   StreamingInputCallResponse response;
 516
 517   context.set_compression_algorithm(GRPC_COMPRESS_GZIP);
 518   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
 519       serviceStub_.Get()->StreamingInputCall(&context, &response));
 520
 521   request.mutable_payload()->set_body(std::string(27182, '\0'));
 522   request.mutable_expect_compressed()->set_value(true);
 523   gpr_log(GPR_DEBUG, "Sending streaming request with compression enabled");
 524   if (!stream->Write(request)) {
 525     gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
 526     return TransientFailureOrAbort();
 527   }
 528
 529   WriteOptions wopts;
 530   wopts.set_no_compression();
 531   request.mutable_payload()->set_body(std::string(45904, '\0'));
 532   request.mutable_expect_compressed()->set_value(false);
 533   gpr_log(GPR_DEBUG, "Sending streaming request with compression disabled");
 534   if (!stream->Write(request, wopts)) {
 535     gpr_log(GPR_ERROR, "%s(): stream->Write() failed", __func__);
 536     return TransientFailureOrAbort();
 537   }
 538   GPR_ASSERT(stream->WritesDone());
 539
 540   s = stream->Finish();
 541   return AssertStatusOk(s, context.debug_error_string());
 542 }
 543
 544 bool InteropClient::DoServerCompressedStreaming() {
 545   const std::vector<bool> compressions = {true, false};
 546   const std::vector<int> sizes = {31415, 92653};
 547
 548   ClientContext context;
 549   InteropClientContextInspector inspector(context);
 550   StreamingOutputCallRequest request;
 551
 552   GPR_ASSERT(compressions.size() == sizes.size());
 553   for (size_t i = 0; i < sizes.size(); i++) {
 554     std::string log_suffix =
 555         absl::StrFormat("(compression=%s; size=%d)",
 556                         compressions[i] ? "true" : "false", sizes[i]);
 557
 558     gpr_log(GPR_DEBUG, "Sending request streaming rpc %s.", log_suffix.c_str());
 559
 560     ResponseParameters* const response_parameter =
 561         request.add_response_parameters();
 562     response_parameter->mutable_compressed()->set_value(compressions[i]);
 563     response_parameter->set_size(sizes[i]);
 564   }
 565   std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
 566       serviceStub_.Get()->StreamingOutputCall(&context, request));
 567
 568   size_t k = 0;
 569   StreamingOutputCallResponse response;
 570   while (stream->Read(&response)) {
 571     // Payload size checks.
 572     GPR_ASSERT(response.payload().body() ==
 573                std::string(request.response_parameters(k).size(), '\0'));
 574
 575     // Compression checks.
 576     GPR_ASSERT(request.response_parameters(k).has_compressed());
 577     if (request.response_parameters(k).compressed().value()) {
 578       GPR_ASSERT(inspector.GetCallCompressionAlgorithm() > GRPC_COMPRESS_NONE);
 579       GPR_ASSERT(inspector.WasCompressed());
 580     } else {
 581       // requested *no* compression.
 582       GPR_ASSERT(!(inspector.WasCompressed()));
 583     }
 584     ++k;
 585   }
 586
 587   if (k < sizes.size()) {
 588     // stream->Read() failed before reading all the expected messages. This
 589     // is most likely due to a connection failure.
 590     gpr_log(GPR_ERROR,
 591             "%s(): Responses read (k=%" PRIuPTR
 592             ") is less than the expected number of  messages (%" PRIuPTR ").",
 593             __func__, k, sizes.size());
 594     return TransientFailureOrAbort();
 595   }
 596
 597   Status s = stream->Finish();
 598   return AssertStatusOk(s, context.debug_error_string());
 599 }
 600
 601 bool InteropClient::DoResponseStreamingWithSlowConsumer() {
 602   gpr_log(GPR_DEBUG, "Receiving response streaming rpc with slow consumer ...");
 603
 604   ClientContext context;
 605   StreamingOutputCallRequest request;
 606
 607   for (int i = 0; i < kNumResponseMessages; ++i) {
 608     ResponseParameters* response_parameter = request.add_response_parameters();
 609     response_parameter->set_size(kResponseMessageSize);
 610   }
 611   StreamingOutputCallResponse response;
 612   std::unique_ptr<ClientReader<StreamingOutputCallResponse>> stream(
 613       serviceStub_.Get()->StreamingOutputCall(&context, request));
 614
 615   int i = 0;
 616   while (stream->Read(&response)) {
 617     GPR_ASSERT(response.payload().body() ==
 618                std::string(kResponseMessageSize, '\0'));
 619     gpr_log(GPR_DEBUG, "received message %d", i);
 620     gpr_sleep_until(gpr_time_add(
 621         gpr_now(GPR_CLOCK_REALTIME),
 622         gpr_time_from_millis(kReceiveDelayMilliSeconds, GPR_TIMESPAN)));
 623     ++i;
 624   }
 625
 626   if (i < kNumResponseMessages) {
 627     gpr_log(GPR_ERROR,
 628             "DoResponseStreamingWithSlowConsumer(): Responses read (i=%d) is "
 629             "less than the expected messages (i.e kNumResponseMessages = %d)",
 630             i, kNumResponseMessages);
 631
 632     return TransientFailureOrAbort();
 633   }
 634
 635   Status s = stream->Finish();
 636   if (!AssertStatusOk(s, context.debug_error_string())) {
 637     return false;
 638   }
 639
 640   gpr_log(GPR_DEBUG, "Response streaming done.");
 641   return true;
 642 }
 643
 644 bool InteropClient::DoHalfDuplex() {
 645   gpr_log(GPR_DEBUG, "Sending half-duplex streaming rpc ...");
 646
 647   ClientContext context;
 648   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
 649                                      StreamingOutputCallResponse>>
 650       stream(serviceStub_.Get()->HalfDuplexCall(&context));
 651
 652   StreamingOutputCallRequest request;
 653   ResponseParameters* response_parameter = request.add_response_parameters();
 654   for (unsigned int i = 0; i < response_stream_sizes.size(); ++i) {
 655     response_parameter->set_size(response_stream_sizes[i]);
 656
 657     if (!stream->Write(request)) {
 658       gpr_log(GPR_ERROR, "DoHalfDuplex(): stream->Write() failed. i=%d", i);
 659       return TransientFailureOrAbort();
 660     }
 661   }
 662   stream->WritesDone();
 663
 664   unsigned int i = 0;
 665   StreamingOutputCallResponse response;
 666   while (stream->Read(&response)) {
 667     GPR_ASSERT(response.payload().body() ==
 668                std::string(response_stream_sizes[i], '\0'));
 669     ++i;
 670   }
 671
 672   if (i < response_stream_sizes.size()) {
 673     // stream->Read() failed before reading all the expected messages. This is
 674     // most likely due to a connection failure
 675     gpr_log(GPR_ERROR,
 676             "DoHalfDuplex(): Responses read (i=%d) are less than the expected "
 677             "number of messages response_stream_sizes.size() (%" PRIuPTR ")",
 678             i, response_stream_sizes.size());
 679     return TransientFailureOrAbort();
 680   }
 681
 682   Status s = stream->Finish();
 683   if (!AssertStatusOk(s, context.debug_error_string())) {
 684     return false;
 685   }
 686
 687   gpr_log(GPR_DEBUG, "Half-duplex streaming rpc done.");
 688   return true;
 689 }
 690
 691 bool InteropClient::DoPingPong() {
 692   gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
 693
 694   ClientContext context;
 695   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
 696                                      StreamingOutputCallResponse>>
 697       stream(serviceStub_.Get()->FullDuplexCall(&context));
 698
 699   StreamingOutputCallRequest request;
 700   ResponseParameters* response_parameter = request.add_response_parameters();
 701   Payload* payload = request.mutable_payload();
 702   StreamingOutputCallResponse response;
 703
 704   for (unsigned int i = 0; i < request_stream_sizes.size(); ++i) {
 705     response_parameter->set_size(response_stream_sizes[i]);
 706     payload->set_body(std::string(request_stream_sizes[i], '\0'));
 707
 708     if (!stream->Write(request)) {
 709       gpr_log(GPR_ERROR, "DoPingPong(): stream->Write() failed. i: %d", i);
 710       return TransientFailureOrAbort();
 711     }
 712
 713     if (!stream->Read(&response)) {
 714       gpr_log(GPR_ERROR, "DoPingPong(): stream->Read() failed. i:%d", i);
 715       return TransientFailureOrAbort();
 716     }
 717
 718     GPR_ASSERT(response.payload().body() ==
 719                std::string(response_stream_sizes[i], '\0'));
 720   }
 721
 722   stream->WritesDone();
 723
 724   GPR_ASSERT(!stream->Read(&response));
 725
 726   Status s = stream->Finish();
 727   if (!AssertStatusOk(s, context.debug_error_string())) {
 728     return false;
 729   }
 730
 731   gpr_log(GPR_DEBUG, "Ping pong streaming done.");
 732   return true;
 733 }
 734
 735 bool InteropClient::DoCancelAfterBegin() {
 736   gpr_log(GPR_DEBUG, "Sending request streaming rpc ...");
 737
 738   ClientContext context;
 739   StreamingInputCallRequest request;
 740   StreamingInputCallResponse response;
 741
 742   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
 743       serviceStub_.Get()->StreamingInputCall(&context, &response));
 744
 745   gpr_log(GPR_DEBUG, "Trying to cancel...");
 746   context.TryCancel();
 747   Status s = stream->Finish();
 748
 749   if (!AssertStatusCode(s, StatusCode::CANCELLED,
 750                         context.debug_error_string())) {
 751     return false;
 752   }
 753
 754   gpr_log(GPR_DEBUG, "Canceling streaming done.");
 755   return true;
 756 }
 757
 758 bool InteropClient::DoCancelAfterFirstResponse() {
 759   gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
 760
 761   ClientContext context;
 762   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
 763                                      StreamingOutputCallResponse>>
 764       stream(serviceStub_.Get()->FullDuplexCall(&context));
 765
 766   StreamingOutputCallRequest request;
 767   ResponseParameters* response_parameter = request.add_response_parameters();
 768   response_parameter->set_size(31415);
 769   request.mutable_payload()->set_body(std::string(27182, '\0'));
 770   StreamingOutputCallResponse response;
 771
 772   if (!stream->Write(request)) {
 773     gpr_log(GPR_ERROR, "DoCancelAfterFirstResponse(): stream->Write() failed");
 774     return TransientFailureOrAbort();
 775   }
 776
 777   if (!stream->Read(&response)) {
 778     gpr_log(GPR_ERROR, "DoCancelAfterFirstResponse(): stream->Read failed");
 779     return TransientFailureOrAbort();
 780   }
 781   GPR_ASSERT(response.payload().body() == std::string(31415, '\0'));
 782
 783   gpr_log(GPR_DEBUG, "Trying to cancel...");
 784   context.TryCancel();
 785
 786   Status s = stream->Finish();
 787   gpr_log(GPR_DEBUG, "Canceling pingpong streaming done.");
 788   return true;
 789 }
 790
 791 bool InteropClient::DoTimeoutOnSleepingServer() {
 792   gpr_log(GPR_DEBUG,
 793           "Sending Ping Pong streaming rpc with a short deadline...");
 794
 795   ClientContext context;
 796   std::chrono::system_clock::time_point deadline =
 797       std::chrono::system_clock::now() + std::chrono::milliseconds(1);
 798   context.set_deadline(deadline);
 799   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
 800                                      StreamingOutputCallResponse>>
 801       stream(serviceStub_.Get()->FullDuplexCall(&context));
 802
 803   StreamingOutputCallRequest request;
 804   request.mutable_payload()->set_body(std::string(27182, '\0'));
 805   stream->Write(request);
 806
 807   Status s = stream->Finish();
 808   if (!AssertStatusCode(s, StatusCode::DEADLINE_EXCEEDED,
 809                         context.debug_error_string())) {
 810     return false;
 811   }
 812
 813   gpr_log(GPR_DEBUG, "Pingpong streaming timeout done.");
 814   return true;
 815 }
 816
 817 bool InteropClient::DoEmptyStream() {
 818   gpr_log(GPR_DEBUG, "Starting empty_stream.");
 819
 820   ClientContext context;
 821   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
 822                                      StreamingOutputCallResponse>>
 823       stream(serviceStub_.Get()->FullDuplexCall(&context));
 824   stream->WritesDone();
 825   StreamingOutputCallResponse response;
 826   GPR_ASSERT(stream->Read(&response) == false);
 827
 828   Status s = stream->Finish();
 829   if (!AssertStatusOk(s, context.debug_error_string())) {
 830     return false;
 831   }
 832
 833   gpr_log(GPR_DEBUG, "empty_stream done.");
 834   return true;
 835 }
 836
 837 bool InteropClient::DoStatusWithMessage() {
 838   gpr_log(GPR_DEBUG,
 839           "Sending RPC with a request for status code 2 and message");
 840
 841   const grpc::StatusCode test_code = grpc::StatusCode::UNKNOWN;
 842   const std::string test_msg = "This is a test message";
 843
 844   // Test UnaryCall.
 845   ClientContext context;
 846   SimpleRequest request;
 847   SimpleResponse response;
 848   EchoStatus* requested_status = request.mutable_response_status();
 849   requested_status->set_code(test_code);
 850   requested_status->set_message(test_msg);
 851   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
 852   if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN,
 853                         context.debug_error_string())) {
 854     return false;
 855   }
 856   GPR_ASSERT(s.error_message() == test_msg);
 857
 858   // Test FullDuplexCall.
 859   ClientContext stream_context;
 860   std::shared_ptr<ClientReaderWriter<StreamingOutputCallRequest,
 861                                      StreamingOutputCallResponse>>
 862       stream(serviceStub_.Get()->FullDuplexCall(&stream_context));
 863   StreamingOutputCallRequest streaming_request;
 864   requested_status = streaming_request.mutable_response_status();
 865   requested_status->set_code(test_code);
 866   requested_status->set_message(test_msg);
 867   stream->Write(streaming_request);
 868   stream->WritesDone();
 869   StreamingOutputCallResponse streaming_response;
 870   while (stream->Read(&streaming_response)) {
 871   }
 872   s = stream->Finish();
 873   if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN,
 874                         context.debug_error_string())) {
 875     return false;
 876   }
 877   GPR_ASSERT(s.error_message() == test_msg);
 878
 879   gpr_log(GPR_DEBUG, "Done testing Status and Message");
 880   return true;
 881 }
 882
 883 bool InteropClient::DoSpecialStatusMessage() {
 884   gpr_log(
 885       GPR_DEBUG,
 886       "Sending RPC with a request for status code 2 and message - \\t\\ntest "
 887       "with whitespace\\r\\nand Unicode BMP ☺ and non-BMP 😈\\t\\n");
 888   const grpc::StatusCode test_code = grpc::StatusCode::UNKNOWN;
 889   const std::string test_msg =
 890       "\t\ntest with whitespace\r\nand Unicode BMP ☺ and non-BMP 😈\t\n";
 891   ClientContext context;
 892   SimpleRequest request;
 893   SimpleResponse response;
 894   EchoStatus* requested_status = request.mutable_response_status();
 895   requested_status->set_code(test_code);
 896   requested_status->set_message(test_msg);
 897   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
 898   if (!AssertStatusCode(s, grpc::StatusCode::UNKNOWN,
 899                         context.debug_error_string())) {
 900     return false;
 901   }
 902   GPR_ASSERT(s.error_message() == test_msg);
 903   gpr_log(GPR_DEBUG, "Done testing Special Status Message");
 904   return true;
 905 }
 906
 907 bool InteropClient::DoCacheableUnary() {
 908   gpr_log(GPR_DEBUG, "Sending RPC with cacheable response");
 909
 910   // Create request with current timestamp
 911   gpr_timespec ts = gpr_now(GPR_CLOCK_PRECISE);
 912   std::string timestamp =
 913       std::to_string(static_cast<long long unsigned>(ts.tv_nsec));
 914   SimpleRequest request;
 915   request.mutable_payload()->set_body(timestamp.c_str(), timestamp.size());
 916
 917   // Request 1
 918   ClientContext context1;
 919   SimpleResponse response1;
 920   context1.set_cacheable(true);
 921   // Add fake user IP since some proxy's (GFE) won't cache requests from
 922   // localhost.
 923   context1.AddMetadata("x-user-ip", "1.2.3.4");
 924   Status s1 =
 925       serviceStub_.Get()->CacheableUnaryCall(&context1, request, &response1);
 926   if (!AssertStatusOk(s1, context1.debug_error_string())) {
 927     return false;
 928   }
 929   gpr_log(GPR_DEBUG, "response 1 payload: %s",
 930           response1.payload().body().c_str());
 931
 932   // Request 2
 933   ClientContext context2;
 934   SimpleResponse response2;
 935   context2.set_cacheable(true);
 936   context2.AddMetadata("x-user-ip", "1.2.3.4");
 937   Status s2 =
 938       serviceStub_.Get()->CacheableUnaryCall(&context2, request, &response2);
 939   if (!AssertStatusOk(s2, context2.debug_error_string())) {
 940     return false;
 941   }
 942   gpr_log(GPR_DEBUG, "response 2 payload: %s",
 943           response2.payload().body().c_str());
 944
 945   // Check that the body is same for both requests. It will be the same if the
 946   // second response is a cached copy of the first response
 947   GPR_ASSERT(response2.payload().body() == response1.payload().body());
 948
 949   // Request 3
 950   // Modify the request body so it will not get a cache hit
 951   ts = gpr_now(GPR_CLOCK_PRECISE);
 952   timestamp = std::to_string(static_cast<long long unsigned>(ts.tv_nsec));
 953   SimpleRequest request1;
 954   request1.mutable_payload()->set_body(timestamp.c_str(), timestamp.size());
 955   ClientContext context3;
 956   SimpleResponse response3;
 957   context3.set_cacheable(true);
 958   context3.AddMetadata("x-user-ip", "1.2.3.4");
 959   Status s3 =
 960       serviceStub_.Get()->CacheableUnaryCall(&context3, request1, &response3);
 961   if (!AssertStatusOk(s3, context3.debug_error_string())) {
 962     return false;
 963   }
 964   gpr_log(GPR_DEBUG, "response 3 payload: %s",
 965           response3.payload().body().c_str());
 966
 967   // Check that the response is different from the previous response.
 968   GPR_ASSERT(response3.payload().body() != response1.payload().body());
 969   return true;
 970 }
 971
 972 bool InteropClient::DoPickFirstUnary() {
 973   const int rpcCount = 100;
 974   SimpleRequest request;
 975   SimpleResponse response;
 976   std::string server_id;
 977   request.set_fill_server_id(true);
 978   for (int i = 0; i < rpcCount; i++) {
 979     ClientContext context;
 980     Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
 981     if (!AssertStatusOk(s, context.debug_error_string())) {
 982       return false;
 983     }
 984     if (i == 0) {
 985       server_id = response.server_id();
 986       continue;
 987     }
 988     if (response.server_id() != server_id) {
 989       gpr_log(GPR_ERROR, "#%d rpc hits server_id %s, expect server_id %s", i,
 990               response.server_id().c_str(), server_id.c_str());
 991       return false;
 992     }
 993   }
 994   gpr_log(GPR_DEBUG, "pick first unary successfully finished");
 995   return true;
 996 }
 997
 998 bool InteropClient::DoCustomMetadata() {
 999   const std::string kEchoInitialMetadataKey("x-grpc-test-echo-initial");
1000   const std::string kInitialMetadataValue("test_initial_metadata_value");
1001   const std::string kEchoTrailingBinMetadataKey(
1002       "x-grpc-test-echo-trailing-bin");
1003   const std::string kTrailingBinValue("\x0a\x0b\x0a\x0b\x0a\x0b");
1004
1005   {
1006     gpr_log(GPR_DEBUG, "Sending RPC with custom metadata");
1007     ClientContext context;
1008     context.AddMetadata(kEchoInitialMetadataKey, kInitialMetadataValue);
1009     context.AddMetadata(kEchoTrailingBinMetadataKey, kTrailingBinValue);
1010     SimpleRequest request;
1011     SimpleResponse response;
1012     request.set_response_size(kLargeResponseSize);
1013     std::string payload(kLargeRequestSize, '\0');
1014     request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
1015
1016     Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
1017     if (!AssertStatusOk(s, context.debug_error_string())) {
1018       return false;
1019     }
1020
1021     const auto& server_initial_metadata = context.GetServerInitialMetadata();
1022     auto iter = server_initial_metadata.find(kEchoInitialMetadataKey);
1023     GPR_ASSERT(iter != server_initial_metadata.end());
1024     GPR_ASSERT(iter->second == kInitialMetadataValue);
1025     const auto& server_trailing_metadata = context.GetServerTrailingMetadata();
1026     iter = server_trailing_metadata.find(kEchoTrailingBinMetadataKey);
1027     GPR_ASSERT(iter != server_trailing_metadata.end());
1028     GPR_ASSERT(std::string(iter->second.begin(), iter->second.end()) ==
1029                kTrailingBinValue);
1030
1031     gpr_log(GPR_DEBUG, "Done testing RPC with custom metadata");
1032   }
1033
1034   {
1035     gpr_log(GPR_DEBUG, "Sending stream with custom metadata");
1036     ClientContext context;
1037     context.AddMetadata(kEchoInitialMetadataKey, kInitialMetadataValue);
1038     context.AddMetadata(kEchoTrailingBinMetadataKey, kTrailingBinValue);
1039     std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
1040                                        StreamingOutputCallResponse>>
1041         stream(serviceStub_.Get()->FullDuplexCall(&context));
1042
1043     StreamingOutputCallRequest request;
1044     ResponseParameters* response_parameter = request.add_response_parameters();
1045     response_parameter->set_size(kLargeResponseSize);
1046     std::string payload(kLargeRequestSize, '\0');
1047     request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
1048     StreamingOutputCallResponse response;
1049
1050     if (!stream->Write(request)) {
1051       gpr_log(GPR_ERROR, "DoCustomMetadata(): stream->Write() failed");
1052       return TransientFailureOrAbort();
1053     }
1054
1055     stream->WritesDone();
1056
1057     if (!stream->Read(&response)) {
1058       gpr_log(GPR_ERROR, "DoCustomMetadata(): stream->Read() failed");
1059       return TransientFailureOrAbort();
1060     }
1061
1062     GPR_ASSERT(response.payload().body() ==
1063                std::string(kLargeResponseSize, '\0'));
1064
1065     GPR_ASSERT(!stream->Read(&response));
1066
1067     Status s = stream->Finish();
1068     if (!AssertStatusOk(s, context.debug_error_string())) {
1069       return false;
1070     }
1071
1072     const auto& server_initial_metadata = context.GetServerInitialMetadata();
1073     auto iter = server_initial_metadata.find(kEchoInitialMetadataKey);
1074     GPR_ASSERT(iter != server_initial_metadata.end());
1075     GPR_ASSERT(iter->second == kInitialMetadataValue);
1076     const auto& server_trailing_metadata = context.GetServerTrailingMetadata();
1077     iter = server_trailing_metadata.find(kEchoTrailingBinMetadataKey);
1078     GPR_ASSERT(iter != server_trailing_metadata.end());
1079     GPR_ASSERT(std::string(iter->second.begin(), iter->second.end()) ==
1080                kTrailingBinValue);
1081
1082     gpr_log(GPR_DEBUG, "Done testing stream with custom metadata");
1083   }
1084
1085   return true;
1086 }
1087
1088 std::tuple<bool, int32_t, std::string>
1089 InteropClient::PerformOneSoakTestIteration(
1090     const bool reset_channel,
1091     const int32_t max_acceptable_per_iteration_latency_ms) {
1092   gpr_timespec start = gpr_now(GPR_CLOCK_MONOTONIC);
1093   SimpleRequest request;
1094   SimpleResponse response;
1095   // Don't set the deadline on the RPC, and instead just
1096   // record how long the RPC took and compare. This makes
1097   // debugging easier when looking at failure results.
1098   ClientContext context;
1099   InteropClientContextInspector inspector(context);
1100   request.set_response_size(kLargeResponseSize);
1101   std::string payload(kLargeRequestSize, '\0');
1102   request.mutable_payload()->set_body(payload.c_str(), kLargeRequestSize);
1103   if (reset_channel) {
1104     serviceStub_.ResetChannel();
1105   }
1106   Status s = serviceStub_.Get()->UnaryCall(&context, request, &response);
1107   gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
1108   int32_t elapsed_ms = gpr_time_to_millis(gpr_time_sub(now, start));
1109   if (!s.ok()) {
1110     return std::make_tuple(false, elapsed_ms, context.debug_error_string());
1111   } else if (elapsed_ms > max_acceptable_per_iteration_latency_ms) {
1112     std::string debug_string = absl::StrFormat(
1113         "%d ms exceeds max acceptable latency: %d ms, peer: %s", elapsed_ms,
1114         max_acceptable_per_iteration_latency_ms, context.peer());
1115     return std::make_tuple(false, elapsed_ms, std::move(debug_string));
1116   } else {
1117     return std::make_tuple(true, elapsed_ms, "");
1118   }
1119 }
1120
1121 void InteropClient::PerformSoakTest(
1122     const bool reset_channel_per_iteration, const int32_t soak_iterations,
1123     const int32_t max_failures,
1124     const int32_t max_acceptable_per_iteration_latency_ms,
1125     const int32_t overall_timeout_seconds) {
1126   std::vector<std::tuple<bool, int32_t, std::string>> results;
1127   grpc_histogram* latencies_ms_histogram = grpc_histogram_create(
1128       1 /* resolution */,
1129       500 * 1e3 /* largest bucket; 500 seconds is unlikely */);
1130   gpr_timespec overall_deadline = gpr_time_add(
1131       gpr_now(GPR_CLOCK_MONOTONIC),
1132       gpr_time_from_seconds(overall_timeout_seconds, GPR_TIMESPAN));
1133   int32_t iterations_ran = 0;
1134   for (int i = 0;
1135        i < soak_iterations &&
1136        gpr_time_cmp(gpr_now(GPR_CLOCK_MONOTONIC), overall_deadline) < 0;
1137        ++i) {
1138     auto result = PerformOneSoakTestIteration(
1139         reset_channel_per_iteration, max_acceptable_per_iteration_latency_ms);
1140     results.push_back(result);
1141     grpc_histogram_add(latencies_ms_histogram, std::get<1>(result));
1142     iterations_ran++;
1143   }
1144   int total_failures = 0;
1145   for (size_t i = 0; i < results.size(); i++) {
1146     bool success = std::get<0>(results[i]);
1147     int32_t elapsed_ms = std::get<1>(results[i]);
1148     std::string debug_string = std::get<2>(results[i]);
1149     if (!success) {
1150       gpr_log(GPR_DEBUG, "soak iteration: %ld elapsed_ms: %d failed: %s", i,
1151               elapsed_ms, debug_string.c_str());
1152       total_failures++;
1153     } else {
1154       gpr_log(GPR_DEBUG, "soak iteration: %ld elapsed_ms: %d succeeded", i,
1155               elapsed_ms);
1156     }
1157   }
1158   double latency_ms_median =
1159       grpc_histogram_percentile(latencies_ms_histogram, 50);
1160   double latency_ms_90th =
1161       grpc_histogram_percentile(latencies_ms_histogram, 90);
1162   double latency_ms_worst = grpc_histogram_maximum(latencies_ms_histogram);
1163   grpc_histogram_destroy(latencies_ms_histogram);
1164   if (iterations_ran < soak_iterations) {
1165     gpr_log(
1166         GPR_ERROR,
1167         "soak test consumed all %d seconds of time and quit early, only "
1168         "having ran %d out of desired %d iterations. "
1169         "total_failures: %d. "
1170         "max_failures_threshold: %d. "
1171         "median_soak_iteration_latency: %lf ms. "
1172         "90th_soak_iteration_latency: %lf ms. "
1173         "worst_soak_iteration_latency: %lf ms. "
1174         "Some or all of the iterations that did run were unexpectedly slow. "
1175         "See breakdown above for which iterations succeeded, failed, and "
1176         "why for more info.",
1177         overall_timeout_seconds, iterations_ran, soak_iterations,
1178         total_failures, max_failures, latency_ms_median, latency_ms_90th,
1179         latency_ms_worst);
1180     GPR_ASSERT(0);
1181   } else if (total_failures > max_failures) {
1182     gpr_log(GPR_ERROR,
1183             "soak test ran: %d iterations. total_failures: %d exceeds "
1184             "max_failures_threshold: %d. "
1185             "median_soak_iteration_latency: %lf ms. "
1186             "90th_soak_iteration_latency: %lf ms. "
1187             "worst_soak_iteration_latency: %lf ms. "
1188             "See breakdown above for which iterations succeeded, failed, and "
1189             "why for more info.",
1190             soak_iterations, total_failures, max_failures, latency_ms_median,
1191             latency_ms_90th, latency_ms_worst);
1192     GPR_ASSERT(0);
1193   } else {
1194     gpr_log(GPR_INFO,
1195             "soak test ran: %d iterations. total_failures: %d is within "
1196             "max_failures_threshold: %d. "
1197             "median_soak_iteration_latency: %lf ms. "
1198             "90th_soak_iteration_latency: %lf ms. "
1199             "worst_soak_iteration_latency: %lf ms. "
1200             "See breakdown above for which iterations succeeded, failed, and "
1201             "why for more info.",
1202             soak_iterations, total_failures, max_failures, latency_ms_median,
1203             latency_ms_90th, latency_ms_worst);
1204   }
1205 }
1206
1207 bool InteropClient::DoRpcSoakTest(
1208     int32_t soak_iterations, int32_t max_failures,
1209     int64_t max_acceptable_per_iteration_latency_ms,
1210     int32_t overall_timeout_seconds) {
1211   gpr_log(GPR_DEBUG, "Sending %d RPCs...", soak_iterations);
1212   GPR_ASSERT(soak_iterations > 0);
1213   PerformSoakTest(false /* reset channel per iteration */, soak_iterations,
1214                   max_failures, max_acceptable_per_iteration_latency_ms,
1215                   overall_timeout_seconds);
1216   gpr_log(GPR_DEBUG, "rpc_soak test done.");
1217   return true;
1218 }
1219
1220 bool InteropClient::DoChannelSoakTest(
1221     int32_t soak_iterations, int32_t max_failures,
1222     int64_t max_acceptable_per_iteration_latency_ms,
1223     int32_t overall_timeout_seconds) {
1224   gpr_log(GPR_DEBUG, "Sending %d RPCs, tearing down the channel each time...",
1225           soak_iterations);
1226   GPR_ASSERT(soak_iterations > 0);
1227   PerformSoakTest(true /* reset channel per iteration */, soak_iterations,
1228                   max_failures, max_acceptable_per_iteration_latency_ms,
1229                   overall_timeout_seconds);
1230   gpr_log(GPR_DEBUG, "channel_soak test done.");
1231   return true;
1232 }
1233
1234 bool InteropClient::DoLongLivedChannelTest(int32_t soak_iterations,
1235                                            int32_t iteration_interval) {
1236   gpr_log(GPR_DEBUG, "Sending %d RPCs...", soak_iterations);
1237   GPR_ASSERT(soak_iterations > 0);
1238   GPR_ASSERT(iteration_interval > 0);
1239   SimpleRequest request;
1240   SimpleResponse response;
1241   int num_failures = 0;
1242   for (int i = 0; i < soak_iterations; ++i) {
1243     gpr_log(GPR_DEBUG, "Sending RPC number %d...", i);
1244     if (!PerformLargeUnary(&request, &response)) {
1245       gpr_log(GPR_ERROR, "Iteration %d failed.", i);
1246       num_failures++;
1247     }
1248     gpr_sleep_until(
1249         gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
1250                      gpr_time_from_seconds(iteration_interval, GPR_TIMESPAN)));
1251   }
1252   if (num_failures == 0) {
1253     gpr_log(GPR_DEBUG, "long_lived_channel test done.");
1254     return true;
1255   } else {
1256     gpr_log(GPR_DEBUG, "long_lived_channel test failed with %d rpc failures.",
1257             num_failures);
1258     return false;
1259   }
1260 }
1261
1262 bool InteropClient::DoUnimplementedService() {
1263   gpr_log(GPR_DEBUG, "Sending a request for an unimplemented service...");
1264
1265   Empty request;
1266   Empty response;
1267   ClientContext context;
1268
1269   UnimplementedService::Stub* stub = serviceStub_.GetUnimplementedServiceStub();
1270
1271   Status s = stub->UnimplementedCall(&context, request, &response);
1272
1273   if (!AssertStatusCode(s, StatusCode::UNIMPLEMENTED,
1274                         context.debug_error_string())) {
1275     return false;
1276   }
1277
1278   gpr_log(GPR_DEBUG, "unimplemented service done.");
1279   return true;
1280 }
1281
1282 bool InteropClient::DoUnimplementedMethod() {
1283   gpr_log(GPR_DEBUG, "Sending a request for an unimplemented rpc...");
1284
1285   Empty request;
1286   Empty response;
1287   ClientContext context;
1288
1289   Status s =
1290       serviceStub_.Get()->UnimplementedCall(&context, request, &response);
1291
1292   if (!AssertStatusCode(s, StatusCode::UNIMPLEMENTED,
1293                         context.debug_error_string())) {
1294     return false;
1295   }
1296
1297   gpr_log(GPR_DEBUG, "unimplemented rpc done.");
1298   return true;
1299 }
1300
1301 }  // namespace testing
1302 }  // namespace grpc