test/cpp/end2end/xds_end2end_test.cc

   1 /*
   2  *
   3  * Copyright 2017 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 <memory>
  20 #include <mutex>
  21 #include <set>
  22 #include <sstream>
  23 #include <thread>
  24
  25 #include <grpc/grpc.h>
  26 #include <grpc/support/alloc.h>
  27 #include <grpc/support/log.h>
  28 #include <grpc/support/string_util.h>
  29 #include <grpc/support/time.h>
  30 #include <grpcpp/channel.h>
  31 #include <grpcpp/client_context.h>
  32 #include <grpcpp/create_channel.h>
  33 #include <grpcpp/server.h>
  34 #include <grpcpp/server_builder.h>
  35
  36 #include "src/core/ext/filters/client_channel/backup_poller.h"
  37 #include "src/core/ext/filters/client_channel/parse_address.h"
  38 #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
  39 #include "src/core/ext/filters/client_channel/server_address.h"
  40 #include "src/core/lib/gpr/env.h"
  41 #include "src/core/lib/gprpp/ref_counted_ptr.h"
  42 #include "src/core/lib/iomgr/sockaddr.h"
  43 #include "src/core/lib/security/credentials/fake/fake_credentials.h"
  44 #include "src/cpp/client/secure_credentials.h"
  45 #include "src/cpp/server/secure_server_credentials.h"
  46
  47 #include "test/core/util/port.h"
  48 #include "test/core/util/test_config.h"
  49 #include "test/cpp/end2end/test_service_impl.h"
  50
  51 #include "src/proto/grpc/lb/v1/load_balancer.grpc.pb.h"
  52 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  53
  54 #include <gmock/gmock.h>
  55 #include <gtest/gtest.h>
  56
  57 // TODO(dgq): Other scenarios in need of testing:
  58 // - Send a serverlist with faulty ip:port addresses (port > 2^16, etc).
  59 // - Test reception of invalid serverlist
  60 // - Test against a non-LB server.
  61 // - Random LB server closing the stream unexpectedly.
  62 //
  63 // Findings from end to end testing to be covered here:
  64 // - Handling of LB servers restart, including reconnection after backing-off
  65 //   retries.
  66 // - Destruction of load balanced channel (and therefore of xds instance)
  67 //   while:
  68 //   1) the internal LB call is still active. This should work by virtue
  69 //   of the weak reference the LB call holds. The call should be terminated as
  70 //   part of the xds shutdown process.
  71 //   2) the retry timer is active. Again, the weak reference it holds should
  72 //   prevent a premature call to \a glb_destroy.
  73
  74 using std::chrono::system_clock;
  75
  76 using grpc::lb::v1::LoadBalanceRequest;
  77 using grpc::lb::v1::LoadBalanceResponse;
  78 using grpc::lb::v1::LoadBalancer;
  79
  80 namespace grpc {
  81 namespace testing {
  82 namespace {
  83
  84 template <typename ServiceType>
  85 class CountedService : public ServiceType {
  86  public:
  87   size_t request_count() {
  88     grpc::internal::MutexLock lock(&mu_);
  89     return request_count_;
  90   }
  91
  92   size_t response_count() {
  93     grpc::internal::MutexLock lock(&mu_);
  94     return response_count_;
  95   }
  96
  97   void IncreaseResponseCount() {
  98     grpc::internal::MutexLock lock(&mu_);
  99     ++response_count_;
 100   }
 101   void IncreaseRequestCount() {
 102     grpc::internal::MutexLock lock(&mu_);
 103     ++request_count_;
 104   }
 105
 106   void ResetCounters() {
 107     grpc::internal::MutexLock lock(&mu_);
 108     request_count_ = 0;
 109     response_count_ = 0;
 110   }
 111
 112  protected:
 113   grpc::internal::Mutex mu_;
 114
 115  private:
 116   size_t request_count_ = 0;
 117   size_t response_count_ = 0;
 118 };
 119
 120 using BackendService = CountedService<TestServiceImpl>;
 121 using BalancerService = CountedService<LoadBalancer::Service>;
 122
 123 const char g_kCallCredsMdKey[] = "Balancer should not ...";
 124 const char g_kCallCredsMdValue[] = "... receive me";
 125
 126 class BackendServiceImpl : public BackendService {
 127  public:
 128   BackendServiceImpl() {}
 129
 130   Status Echo(ServerContext* context, const EchoRequest* request,
 131               EchoResponse* response) override {
 132     // Backend should receive the call credentials metadata.
 133     auto call_credentials_entry =
 134         context->client_metadata().find(g_kCallCredsMdKey);
 135     EXPECT_NE(call_credentials_entry, context->client_metadata().end());
 136     if (call_credentials_entry != context->client_metadata().end()) {
 137       EXPECT_EQ(call_credentials_entry->second, g_kCallCredsMdValue);
 138     }
 139     IncreaseRequestCount();
 140     const auto status = TestServiceImpl::Echo(context, request, response);
 141     IncreaseResponseCount();
 142     AddClient(context->peer());
 143     return status;
 144   }
 145
 146   void Shutdown() {}
 147
 148   std::set<grpc::string> clients() {
 149     grpc::internal::MutexLock lock(&clients_mu_);
 150     return clients_;
 151   }
 152
 153  private:
 154   void AddClient(const grpc::string& client) {
 155     grpc::internal::MutexLock lock(&clients_mu_);
 156     clients_.insert(client);
 157   }
 158
 159   grpc::internal::Mutex mu_;
 160   grpc::internal::Mutex clients_mu_;
 161   std::set<grpc::string> clients_;
 162 };
 163
 164 grpc::string Ip4ToPackedString(const char* ip_str) {
 165   struct in_addr ip4;
 166   GPR_ASSERT(inet_pton(AF_INET, ip_str, &ip4) == 1);
 167   return grpc::string(reinterpret_cast<const char*>(&ip4), sizeof(ip4));
 168 }
 169
 170 struct ClientStats {
 171   size_t num_calls_started = 0;
 172   size_t num_calls_finished = 0;
 173   size_t num_calls_finished_with_client_failed_to_send = 0;
 174   size_t num_calls_finished_known_received = 0;
 175   std::map<grpc::string, size_t> drop_token_counts;
 176
 177   ClientStats& operator+=(const ClientStats& other) {
 178     num_calls_started += other.num_calls_started;
 179     num_calls_finished += other.num_calls_finished;
 180     num_calls_finished_with_client_failed_to_send +=
 181         other.num_calls_finished_with_client_failed_to_send;
 182     num_calls_finished_known_received +=
 183         other.num_calls_finished_known_received;
 184     for (const auto& p : other.drop_token_counts) {
 185       drop_token_counts[p.first] += p.second;
 186     }
 187     return *this;
 188   }
 189
 190   void Reset() {
 191     num_calls_started = 0;
 192     num_calls_finished = 0;
 193     num_calls_finished_with_client_failed_to_send = 0;
 194     num_calls_finished_known_received = 0;
 195     drop_token_counts.clear();
 196   }
 197 };
 198
 199 class BalancerServiceImpl : public BalancerService {
 200  public:
 201   using Stream = ServerReaderWriter<LoadBalanceResponse, LoadBalanceRequest>;
 202   using ResponseDelayPair = std::pair<LoadBalanceResponse, int>;
 203
 204   explicit BalancerServiceImpl(int client_load_reporting_interval_seconds)
 205       : client_load_reporting_interval_seconds_(
 206             client_load_reporting_interval_seconds) {}
 207
 208   Status BalanceLoad(ServerContext* context, Stream* stream) override {
 209     // TODO(juanlishen): Clean up the scoping.
 210     gpr_log(GPR_INFO, "LB[%p]: BalanceLoad", this);
 211     {
 212       grpc::internal::MutexLock lock(&mu_);
 213       if (serverlist_done_) goto done;
 214     }
 215     {
 216       // Balancer shouldn't receive the call credentials metadata.
 217       EXPECT_EQ(context->client_metadata().find(g_kCallCredsMdKey),
 218                 context->client_metadata().end());
 219       LoadBalanceRequest request;
 220       std::vector<ResponseDelayPair> responses_and_delays;
 221
 222       if (!stream->Read(&request)) {
 223         goto done;
 224       }
 225       IncreaseRequestCount();
 226       gpr_log(GPR_INFO, "LB[%p]: received initial message '%s'", this,
 227               request.DebugString().c_str());
 228
 229       {
 230         LoadBalanceResponse initial_response;
 231         initial_response.mutable_initial_response()
 232             ->mutable_client_stats_report_interval()
 233             ->set_seconds(client_load_reporting_interval_seconds_);
 234         stream->Write(initial_response);
 235       }
 236
 237       {
 238         grpc::internal::MutexLock lock(&mu_);
 239         responses_and_delays = responses_and_delays_;
 240       }
 241       for (const auto& response_and_delay : responses_and_delays) {
 242         SendResponse(stream, response_and_delay.first,
 243                      response_and_delay.second);
 244       }
 245       {
 246         grpc::internal::MutexLock lock(&mu_);
 247         serverlist_cond_.WaitUntil(&mu_, [this] { return serverlist_done_; });
 248       }
 249
 250       if (client_load_reporting_interval_seconds_ > 0) {
 251         request.Clear();
 252         if (stream->Read(&request)) {
 253           gpr_log(GPR_INFO, "LB[%p]: received client load report message '%s'",
 254                   this, request.DebugString().c_str());
 255           GPR_ASSERT(request.has_client_stats());
 256           // We need to acquire the lock here in order to prevent the notify_one
 257           // below from firing before its corresponding wait is executed.
 258           grpc::internal::MutexLock lock(&mu_);
 259           client_stats_.num_calls_started +=
 260               request.client_stats().num_calls_started();
 261           client_stats_.num_calls_finished +=
 262               request.client_stats().num_calls_finished();
 263           client_stats_.num_calls_finished_with_client_failed_to_send +=
 264               request.client_stats()
 265                   .num_calls_finished_with_client_failed_to_send();
 266           client_stats_.num_calls_finished_known_received +=
 267               request.client_stats().num_calls_finished_known_received();
 268           for (const auto& drop_token_count :
 269                request.client_stats().calls_finished_with_drop()) {
 270             client_stats_
 271                 .drop_token_counts[drop_token_count.load_balance_token()] +=
 272                 drop_token_count.num_calls();
 273           }
 274           load_report_ready_ = true;
 275           load_report_cond_.Signal();
 276         }
 277       }
 278     }
 279   done:
 280     gpr_log(GPR_INFO, "LB[%p]: done", this);
 281     return Status::OK;
 282   }
 283
 284   void add_response(const LoadBalanceResponse& response, int send_after_ms) {
 285     grpc::internal::MutexLock lock(&mu_);
 286     responses_and_delays_.push_back(std::make_pair(response, send_after_ms));
 287   }
 288
 289   void Shutdown() {
 290     grpc::internal::MutexLock lock(&mu_);
 291     NotifyDoneWithServerlistsLocked();
 292     responses_and_delays_.clear();
 293     client_stats_.Reset();
 294     gpr_log(GPR_INFO, "LB[%p]: shut down", this);
 295   }
 296
 297   static LoadBalanceResponse BuildResponseForBackends(
 298       const std::vector<int>& backend_ports,
 299       const std::map<grpc::string, size_t>& drop_token_counts) {
 300     LoadBalanceResponse response;
 301     for (const auto& drop_token_count : drop_token_counts) {
 302       for (size_t i = 0; i < drop_token_count.second; ++i) {
 303         auto* server = response.mutable_server_list()->add_servers();
 304         server->set_drop(true);
 305         server->set_load_balance_token(drop_token_count.first);
 306       }
 307     }
 308     for (const int& backend_port : backend_ports) {
 309       auto* server = response.mutable_server_list()->add_servers();
 310       server->set_ip_address(Ip4ToPackedString("127.0.0.1"));
 311       server->set_port(backend_port);
 312       static int token_count = 0;
 313       char* token;
 314       gpr_asprintf(&token, "token%03d", ++token_count);
 315       server->set_load_balance_token(token);
 316       gpr_free(token);
 317     }
 318     return response;
 319   }
 320
 321   const ClientStats& WaitForLoadReport() {
 322     grpc::internal::MutexLock lock(&mu_);
 323     load_report_cond_.WaitUntil(&mu_, [this] { return load_report_ready_; });
 324     load_report_ready_ = false;
 325     return client_stats_;
 326   }
 327
 328   void NotifyDoneWithServerlists() {
 329     grpc::internal::MutexLock lock(&mu_);
 330     NotifyDoneWithServerlistsLocked();
 331   }
 332
 333   void NotifyDoneWithServerlistsLocked() {
 334     if (!serverlist_done_) {
 335       serverlist_done_ = true;
 336       serverlist_cond_.Broadcast();
 337     }
 338   }
 339
 340  private:
 341   void SendResponse(Stream* stream, const LoadBalanceResponse& response,
 342                     int delay_ms) {
 343     gpr_log(GPR_INFO, "LB[%p]: sleeping for %d ms...", this, delay_ms);
 344     if (delay_ms > 0) {
 345       gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(delay_ms));
 346     }
 347     gpr_log(GPR_INFO, "LB[%p]: Woke up! Sending response '%s'", this,
 348             response.DebugString().c_str());
 349     IncreaseResponseCount();
 350     stream->Write(response);
 351   }
 352
 353   const int client_load_reporting_interval_seconds_;
 354   std::vector<ResponseDelayPair> responses_and_delays_;
 355   grpc::internal::Mutex mu_;
 356   grpc::internal::CondVar load_report_cond_;
 357   bool load_report_ready_ = false;
 358   grpc::internal::CondVar serverlist_cond_;
 359   bool serverlist_done_ = false;
 360   ClientStats client_stats_;
 361 };
 362
 363 class XdsEnd2endTest : public ::testing::Test {
 364  protected:
 365   XdsEnd2endTest(size_t num_backends, size_t num_balancers,
 366                  int client_load_reporting_interval_seconds)
 367       : server_host_("localhost"),
 368         num_backends_(num_backends),
 369         num_balancers_(num_balancers),
 370         client_load_reporting_interval_seconds_(
 371             client_load_reporting_interval_seconds) {}
 372
 373   void SetUp() override {
 374     response_generator_ =
 375         grpc_core::MakeRefCounted<grpc_core::FakeResolverResponseGenerator>();
 376     lb_channel_response_generator_ =
 377         grpc_core::MakeRefCounted<grpc_core::FakeResolverResponseGenerator>();
 378     // Start the backends.
 379     for (size_t i = 0; i < num_backends_; ++i) {
 380       backends_.emplace_back(new ServerThread<BackendServiceImpl>("backend"));
 381       backends_.back()->Start(server_host_);
 382     }
 383     // Start the load balancers.
 384     for (size_t i = 0; i < num_balancers_; ++i) {
 385       balancers_.emplace_back(new ServerThread<BalancerServiceImpl>(
 386           "balancer", client_load_reporting_interval_seconds_));
 387       balancers_.back()->Start(server_host_);
 388     }
 389     ResetStub();
 390   }
 391
 392   void TearDown() override {
 393     ShutdownAllBackends();
 394     for (auto& balancer : balancers_) balancer->Shutdown();
 395   }
 396
 397   void StartAllBackends() {
 398     for (auto& backend : backends_) backend->Start(server_host_);
 399   }
 400
 401   void StartBackend(size_t index) { backends_[index]->Start(server_host_); }
 402
 403   void ShutdownAllBackends() {
 404     for (auto& backend : backends_) backend->Shutdown();
 405   }
 406
 407   void ShutdownBackend(size_t index) { backends_[index]->Shutdown(); }
 408
 409   void ResetStub(int fallback_timeout = 0,
 410                  const grpc::string& expected_targets = "") {
 411     ChannelArguments args;
 412     // TODO(juanlishen): Add setter to ChannelArguments.
 413     if (fallback_timeout > 0) {
 414       args.SetInt(GRPC_ARG_XDS_FALLBACK_TIMEOUT_MS, fallback_timeout);
 415     }
 416     args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
 417                     response_generator_.get());
 418     if (!expected_targets.empty()) {
 419       args.SetString(GRPC_ARG_FAKE_SECURITY_EXPECTED_TARGETS, expected_targets);
 420     }
 421     std::ostringstream uri;
 422     uri << "fake:///" << kApplicationTargetName_;
 423     // TODO(dgq): templatize tests to run everything using both secure and
 424     // insecure channel credentials.
 425     grpc_channel_credentials* channel_creds =
 426         grpc_fake_transport_security_credentials_create();
 427     grpc_call_credentials* call_creds = grpc_md_only_test_credentials_create(
 428         g_kCallCredsMdKey, g_kCallCredsMdValue, false);
 429     std::shared_ptr<ChannelCredentials> creds(
 430         new SecureChannelCredentials(grpc_composite_channel_credentials_create(
 431             channel_creds, call_creds, nullptr)));
 432     call_creds->Unref();
 433     channel_creds->Unref();
 434     channel_ = ::grpc::CreateCustomChannel(uri.str(), creds, args);
 435     stub_ = grpc::testing::EchoTestService::NewStub(channel_);
 436   }
 437
 438   void ResetBackendCounters() {
 439     for (auto& backend : backends_) backend->service_.ResetCounters();
 440   }
 441
 442   ClientStats WaitForLoadReports() {
 443     ClientStats client_stats;
 444     for (auto& balancer : balancers_) {
 445       client_stats += balancer->service_.WaitForLoadReport();
 446     }
 447     return client_stats;
 448   }
 449
 450   bool SeenAllBackends(size_t start_index = 0, size_t stop_index = 0) {
 451     if (stop_index == 0) stop_index = backends_.size();
 452     for (size_t i = start_index; i < stop_index; ++i) {
 453       if (backends_[i]->service_.request_count() == 0) return false;
 454     }
 455     return true;
 456   }
 457
 458   void SendRpcAndCount(int* num_total, int* num_ok, int* num_failure,
 459                        int* num_drops) {
 460     const Status status = SendRpc();
 461     if (status.ok()) {
 462       ++*num_ok;
 463     } else {
 464       if (status.error_message() == "Call dropped by load balancing policy") {
 465         ++*num_drops;
 466       } else {
 467         ++*num_failure;
 468       }
 469     }
 470     ++*num_total;
 471   }
 472
 473   std::tuple<int, int, int> WaitForAllBackends(int num_requests_multiple_of = 1,
 474                                                size_t start_index = 0,
 475                                                size_t stop_index = 0) {
 476     int num_ok = 0;
 477     int num_failure = 0;
 478     int num_drops = 0;
 479     int num_total = 0;
 480     while (!SeenAllBackends(start_index, stop_index)) {
 481       SendRpcAndCount(&num_total, &num_ok, &num_failure, &num_drops);
 482     }
 483     while (num_total % num_requests_multiple_of != 0) {
 484       SendRpcAndCount(&num_total, &num_ok, &num_failure, &num_drops);
 485     }
 486     ResetBackendCounters();
 487     gpr_log(GPR_INFO,
 488             "Performed %d warm up requests (a multiple of %d) against the "
 489             "backends. %d succeeded, %d failed, %d dropped.",
 490             num_total, num_requests_multiple_of, num_ok, num_failure,
 491             num_drops);
 492     return std::make_tuple(num_ok, num_failure, num_drops);
 493   }
 494
 495   void WaitForBackend(size_t backend_idx) {
 496     do {
 497       (void)SendRpc();
 498     } while (backends_[backend_idx]->service_.request_count() == 0);
 499     ResetBackendCounters();
 500   }
 501
 502   grpc_core::ServerAddressList CreateLbAddressesFromPortList(
 503       const std::vector<int>& ports) {
 504     grpc_core::ServerAddressList addresses;
 505     for (int port : ports) {
 506       char* lb_uri_str;
 507       gpr_asprintf(&lb_uri_str, "ipv4:127.0.0.1:%d", port);
 508       grpc_uri* lb_uri = grpc_uri_parse(lb_uri_str, true);
 509       GPR_ASSERT(lb_uri != nullptr);
 510       grpc_resolved_address address;
 511       GPR_ASSERT(grpc_parse_uri(lb_uri, &address));
 512       std::vector<grpc_arg> args_to_add;
 513       grpc_channel_args* args = grpc_channel_args_copy_and_add(
 514           nullptr, args_to_add.data(), args_to_add.size());
 515       addresses.emplace_back(address.addr, address.len, args);
 516       grpc_uri_destroy(lb_uri);
 517       gpr_free(lb_uri_str);
 518     }
 519     return addresses;
 520   }
 521
 522   void SetNextResolution(const std::vector<int>& ports,
 523                          const char* service_config_json = nullptr,
 524                          grpc_core::FakeResolverResponseGenerator*
 525                              lb_channel_response_generator = nullptr) {
 526     grpc_core::ExecCtx exec_ctx;
 527     grpc_core::Resolver::Result result;
 528     result.addresses = CreateLbAddressesFromPortList(ports);
 529     if (service_config_json != nullptr) {
 530       grpc_error* error = GRPC_ERROR_NONE;
 531       result.service_config =
 532           grpc_core::ServiceConfig::Create(service_config_json, &error);
 533       GRPC_ERROR_UNREF(error);
 534     }
 535     grpc_arg arg = grpc_core::FakeResolverResponseGenerator::MakeChannelArg(
 536         lb_channel_response_generator == nullptr
 537             ? lb_channel_response_generator_.get()
 538             : lb_channel_response_generator);
 539     result.args = grpc_channel_args_copy_and_add(nullptr, &arg, 1);
 540     response_generator_->SetResponse(std::move(result));
 541   }
 542
 543   void SetNextResolutionForLbChannelAllBalancers(
 544       const char* service_config_json = nullptr,
 545       grpc_core::FakeResolverResponseGenerator* lb_channel_response_generator =
 546           nullptr) {
 547     std::vector<int> ports;
 548     for (size_t i = 0; i < balancers_.size(); ++i) {
 549       ports.emplace_back(balancers_[i]->port_);
 550     }
 551     SetNextResolutionForLbChannel(ports, service_config_json,
 552                                   lb_channel_response_generator);
 553   }
 554
 555   void SetNextResolutionForLbChannel(
 556       const std::vector<int>& ports, const char* service_config_json = nullptr,
 557       grpc_core::FakeResolverResponseGenerator* lb_channel_response_generator =
 558           nullptr) {
 559     grpc_core::ExecCtx exec_ctx;
 560     grpc_core::Resolver::Result result;
 561     result.addresses = CreateLbAddressesFromPortList(ports);
 562     if (service_config_json != nullptr) {
 563       grpc_error* error = GRPC_ERROR_NONE;
 564       result.service_config =
 565           grpc_core::ServiceConfig::Create(service_config_json, &error);
 566       GRPC_ERROR_UNREF(error);
 567     }
 568     if (lb_channel_response_generator == nullptr) {
 569       lb_channel_response_generator = lb_channel_response_generator_.get();
 570     }
 571     lb_channel_response_generator->SetResponse(std::move(result));
 572   }
 573
 574   void SetNextReresolutionResponse(const std::vector<int>& ports) {
 575     grpc_core::ExecCtx exec_ctx;
 576     grpc_core::Resolver::Result result;
 577     result.addresses = CreateLbAddressesFromPortList(ports);
 578     response_generator_->SetReresolutionResponse(std::move(result));
 579   }
 580
 581   const std::vector<int> GetBackendPorts(size_t start_index = 0,
 582                                          size_t stop_index = 0) const {
 583     if (stop_index == 0) stop_index = backends_.size();
 584     std::vector<int> backend_ports;
 585     for (size_t i = start_index; i < stop_index; ++i) {
 586       backend_ports.push_back(backends_[i]->port_);
 587     }
 588     return backend_ports;
 589   }
 590
 591   void ScheduleResponseForBalancer(size_t i,
 592                                    const LoadBalanceResponse& response,
 593                                    int delay_ms) {
 594     balancers_[i]->service_.add_response(response, delay_ms);
 595   }
 596
 597   Status SendRpc(EchoResponse* response = nullptr, int timeout_ms = 1000,
 598                  bool wait_for_ready = false) {
 599     const bool local_response = (response == nullptr);
 600     if (local_response) response = new EchoResponse;
 601     EchoRequest request;
 602     request.set_message(kRequestMessage_);
 603     ClientContext context;
 604     context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms));
 605     if (wait_for_ready) context.set_wait_for_ready(true);
 606     Status status = stub_->Echo(&context, request, response);
 607     if (local_response) delete response;
 608     return status;
 609   }
 610
 611   void CheckRpcSendOk(const size_t times = 1, const int timeout_ms = 1000,
 612                       bool wait_for_ready = false) {
 613     for (size_t i = 0; i < times; ++i) {
 614       EchoResponse response;
 615       const Status status = SendRpc(&response, timeout_ms, wait_for_ready);
 616       EXPECT_TRUE(status.ok()) << "code=" << status.error_code()
 617                                << " message=" << status.error_message();
 618       EXPECT_EQ(response.message(), kRequestMessage_);
 619     }
 620   }
 621
 622   void CheckRpcSendFailure() {
 623     const Status status = SendRpc();
 624     EXPECT_FALSE(status.ok());
 625   }
 626
 627   template <typename T>
 628   struct ServerThread {
 629     template <typename... Args>
 630     explicit ServerThread(const grpc::string& type, Args&&... args)
 631         : port_(grpc_pick_unused_port_or_die()),
 632           type_(type),
 633           service_(std::forward<Args>(args)...) {}
 634
 635     void Start(const grpc::string& server_host) {
 636       gpr_log(GPR_INFO, "starting %s server on port %d", type_.c_str(), port_);
 637       GPR_ASSERT(!running_);
 638       running_ = true;
 639       grpc::internal::Mutex mu;
 640       // We need to acquire the lock here in order to prevent the notify_one
 641       // by ServerThread::Serve from firing before the wait below is hit.
 642       grpc::internal::MutexLock lock(&mu);
 643       grpc::internal::CondVar cond;
 644       thread_.reset(new std::thread(
 645           std::bind(&ServerThread::Serve, this, server_host, &mu, &cond)));
 646       cond.Wait(&mu);
 647       gpr_log(GPR_INFO, "%s server startup complete", type_.c_str());
 648     }
 649
 650     void Serve(const grpc::string& server_host, grpc::internal::Mutex* mu,
 651                grpc::internal::CondVar* cond) {
 652       // We need to acquire the lock here in order to prevent the notify_one
 653       // below from firing before its corresponding wait is executed.
 654       grpc::internal::MutexLock lock(mu);
 655       std::ostringstream server_address;
 656       server_address << server_host << ":" << port_;
 657       ServerBuilder builder;
 658       std::shared_ptr<ServerCredentials> creds(new SecureServerCredentials(
 659           grpc_fake_transport_security_server_credentials_create()));
 660       builder.AddListeningPort(server_address.str(), creds);
 661       builder.RegisterService(&service_);
 662       server_ = builder.BuildAndStart();
 663       cond->Signal();
 664     }
 665
 666     void Shutdown() {
 667       if (!running_) return;
 668       gpr_log(GPR_INFO, "%s about to shutdown", type_.c_str());
 669       service_.Shutdown();
 670       server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0));
 671       thread_->join();
 672       gpr_log(GPR_INFO, "%s shutdown completed", type_.c_str());
 673       running_ = false;
 674     }
 675
 676     const int port_;
 677     grpc::string type_;
 678     T service_;
 679     std::unique_ptr<Server> server_;
 680     std::unique_ptr<std::thread> thread_;
 681     bool running_ = false;
 682   };
 683
 684   const grpc::string server_host_;
 685   const size_t num_backends_;
 686   const size_t num_balancers_;
 687   const int client_load_reporting_interval_seconds_;
 688   std::shared_ptr<Channel> channel_;
 689   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
 690   std::vector<std::unique_ptr<ServerThread<BackendServiceImpl>>> backends_;
 691   std::vector<std::unique_ptr<ServerThread<BalancerServiceImpl>>> balancers_;
 692   grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
 693       response_generator_;
 694   grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
 695       lb_channel_response_generator_;
 696   const grpc::string kRequestMessage_ = "Live long and prosper.";
 697   const grpc::string kApplicationTargetName_ = "application_target_name";
 698   const grpc::string kDefaultServiceConfig_ =
 699       "{\n"
 700       "  \"loadBalancingConfig\":[\n"
 701       "    { \"does_not_exist\":{} },\n"
 702       "    { \"xds_experimental\":{ \"balancerName\": \"fake:///lb\" } }\n"
 703       "  ]\n"
 704       "}";
 705 };
 706
 707 class SingleBalancerTest : public XdsEnd2endTest {
 708  public:
 709   SingleBalancerTest() : XdsEnd2endTest(4, 1, 0) {}
 710 };
 711
 712 TEST_F(SingleBalancerTest, Vanilla) {
 713   SetNextResolution({}, kDefaultServiceConfig_.c_str());
 714   SetNextResolutionForLbChannelAllBalancers();
 715   const size_t kNumRpcsPerAddress = 100;
 716   ScheduleResponseForBalancer(
 717       0, BalancerServiceImpl::BuildResponseForBackends(GetBackendPorts(), {}),
 718       0);
 719   // Make sure that trying to connect works without a call.
 720   channel_->GetState(true /* try_to_connect */);
 721   // We need to wait for all backends to come online.
 722   WaitForAllBackends();
 723   // Send kNumRpcsPerAddress RPCs per server.
 724   CheckRpcSendOk(kNumRpcsPerAddress * num_backends_);
 725   // Each backend should have gotten 100 requests.
 726   for (size_t i = 0; i < backends_.size(); ++i) {
 727     EXPECT_EQ(kNumRpcsPerAddress, backends_[i]->service_.request_count());
 728   }
 729   balancers_[0]->service_.NotifyDoneWithServerlists();
 730   // The balancer got a single request.
 731   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
 732   // and sent a single response.
 733   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
 734
 735   // Check LB policy name for the channel.
 736   EXPECT_EQ("xds_experimental", channel_->GetLoadBalancingPolicyName());
 737 }
 738
 739 TEST_F(SingleBalancerTest, SameBackendListedMultipleTimes) {
 740   SetNextResolution({}, kDefaultServiceConfig_.c_str());
 741   SetNextResolutionForLbChannelAllBalancers();
 742   // Same backend listed twice.
 743   std::vector<int> ports;
 744   ports.push_back(backends_[0]->port_);
 745   ports.push_back(backends_[0]->port_);
 746   const size_t kNumRpcsPerAddress = 10;
 747   ScheduleResponseForBalancer(
 748       0, BalancerServiceImpl::BuildResponseForBackends(ports, {}), 0);
 749   // We need to wait for the backend to come online.
 750   WaitForBackend(0);
 751   // Send kNumRpcsPerAddress RPCs per server.
 752   CheckRpcSendOk(kNumRpcsPerAddress * ports.size());
 753   // Backend should have gotten 20 requests.
 754   EXPECT_EQ(kNumRpcsPerAddress * 2, backends_[0]->service_.request_count());
 755   // And they should have come from a single client port, because of
 756   // subchannel sharing.
 757   EXPECT_EQ(1UL, backends_[0]->service_.clients().size());
 758   balancers_[0]->service_.NotifyDoneWithServerlists();
 759 }
 760
 761 TEST_F(SingleBalancerTest, SecureNaming) {
 762   // TODO(juanlishen): Use separate fake creds for the balancer channel.
 763   ResetStub(0, kApplicationTargetName_ + ";lb");
 764   SetNextResolution({}, kDefaultServiceConfig_.c_str());
 765   SetNextResolutionForLbChannel({balancers_[0]->port_});
 766   const size_t kNumRpcsPerAddress = 100;
 767   ScheduleResponseForBalancer(
 768       0, BalancerServiceImpl::BuildResponseForBackends(GetBackendPorts(), {}),
 769       0);
 770   // Make sure that trying to connect works without a call.
 771   channel_->GetState(true /* try_to_connect */);
 772   // We need to wait for all backends to come online.
 773   WaitForAllBackends();
 774   // Send kNumRpcsPerAddress RPCs per server.
 775   CheckRpcSendOk(kNumRpcsPerAddress * num_backends_);
 776
 777   // Each backend should have gotten 100 requests.
 778   for (size_t i = 0; i < backends_.size(); ++i) {
 779     EXPECT_EQ(kNumRpcsPerAddress, backends_[i]->service_.request_count());
 780   }
 781   // The balancer got a single request.
 782   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
 783   // and sent a single response.
 784   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
 785 }
 786
 787 TEST_F(SingleBalancerTest, SecureNamingDeathTest) {
 788   ::testing::FLAGS_gtest_death_test_style = "threadsafe";
 789   // Make sure that we blow up (via abort() from the security connector) when
 790   // the name from the balancer doesn't match expectations.
 791   ASSERT_DEATH(
 792       {
 793         ResetStub(0, kApplicationTargetName_ + ";lb");
 794         SetNextResolution({},
 795                           "{\n"
 796                           "  \"loadBalancingConfig\":[\n"
 797                           "    { \"does_not_exist\":{} },\n"
 798                           "    { \"xds_experimental\":{ \"balancerName\": "
 799                           "\"fake:///wrong_lb\" } }\n"
 800                           "  ]\n"
 801                           "}");
 802         SetNextResolutionForLbChannel({balancers_[0]->port_});
 803         channel_->WaitForConnected(grpc_timeout_seconds_to_deadline(1));
 804       },
 805       "");
 806 }
 807
 808 TEST_F(SingleBalancerTest, InitiallyEmptyServerlist) {
 809   SetNextResolution({}, kDefaultServiceConfig_.c_str());
 810   SetNextResolutionForLbChannelAllBalancers();
 811   const int kServerlistDelayMs = 500 * grpc_test_slowdown_factor();
 812   const int kCallDeadlineMs = kServerlistDelayMs * 2;
 813   // First response is an empty serverlist, sent right away.
 814   ScheduleResponseForBalancer(0, LoadBalanceResponse(), 0);
 815   // Send non-empty serverlist only after kServerlistDelayMs
 816   ScheduleResponseForBalancer(
 817       0, BalancerServiceImpl::BuildResponseForBackends(GetBackendPorts(), {}),
 818       kServerlistDelayMs);
 819   const auto t0 = system_clock::now();
 820   // Client will block: LB will initially send empty serverlist.
 821   CheckRpcSendOk(1, kCallDeadlineMs, true /* wait_for_ready */);
 822   const auto ellapsed_ms =
 823       std::chrono::duration_cast<std::chrono::milliseconds>(
 824           system_clock::now() - t0);
 825   // but eventually, the LB sends a serverlist update that allows the call to
 826   // proceed. The call delay must be larger than the delay in sending the
 827   // populated serverlist but under the call's deadline (which is enforced by
 828   // the call's deadline).
 829   EXPECT_GT(ellapsed_ms.count(), kServerlistDelayMs);
 830   balancers_[0]->service_.NotifyDoneWithServerlists();
 831   // The balancer got a single request.
 832   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
 833   // and sent two responses.
 834   EXPECT_EQ(2U, balancers_[0]->service_.response_count());
 835 }
 836
 837 TEST_F(SingleBalancerTest, AllServersUnreachableFailFast) {
 838   SetNextResolution({}, kDefaultServiceConfig_.c_str());
 839   SetNextResolutionForLbChannelAllBalancers();
 840   const size_t kNumUnreachableServers = 5;
 841   std::vector<int> ports;
 842   for (size_t i = 0; i < kNumUnreachableServers; ++i) {
 843     ports.push_back(grpc_pick_unused_port_or_die());
 844   }
 845   ScheduleResponseForBalancer(
 846       0, BalancerServiceImpl::BuildResponseForBackends(ports, {}), 0);
 847   const Status status = SendRpc();
 848   // The error shouldn't be DEADLINE_EXCEEDED.
 849   EXPECT_EQ(StatusCode::UNAVAILABLE, status.error_code());
 850   balancers_[0]->service_.NotifyDoneWithServerlists();
 851   // The balancer got a single request.
 852   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
 853   // and sent a single response.
 854   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
 855 }
 856
 857 TEST_F(SingleBalancerTest, Fallback) {
 858   const int kFallbackTimeoutMs = 200 * grpc_test_slowdown_factor();
 859   const int kServerlistDelayMs = 500 * grpc_test_slowdown_factor();
 860   const size_t kNumBackendsInResolution = backends_.size() / 2;
 861   ResetStub(kFallbackTimeoutMs);
 862   SetNextResolution(GetBackendPorts(0, kNumBackendsInResolution),
 863                     kDefaultServiceConfig_.c_str());
 864   SetNextResolutionForLbChannelAllBalancers();
 865   // Send non-empty serverlist only after kServerlistDelayMs.
 866   ScheduleResponseForBalancer(
 867       0,
 868       BalancerServiceImpl::BuildResponseForBackends(
 869           GetBackendPorts(kNumBackendsInResolution /* start_index */), {}),
 870       kServerlistDelayMs);
 871   // Wait until all the fallback backends are reachable.
 872   WaitForAllBackends(1 /* num_requests_multiple_of */, 0 /* start_index */,
 873                      kNumBackendsInResolution /* stop_index */);
 874   gpr_log(GPR_INFO, "========= BEFORE FIRST BATCH ==========");
 875   CheckRpcSendOk(kNumBackendsInResolution);
 876   gpr_log(GPR_INFO, "========= DONE WITH FIRST BATCH ==========");
 877   // Fallback is used: each backend returned by the resolver should have
 878   // gotten one request.
 879   for (size_t i = 0; i < kNumBackendsInResolution; ++i) {
 880     EXPECT_EQ(1U, backends_[i]->service_.request_count());
 881   }
 882   for (size_t i = kNumBackendsInResolution; i < backends_.size(); ++i) {
 883     EXPECT_EQ(0U, backends_[i]->service_.request_count());
 884   }
 885   // Wait until the serverlist reception has been processed and all backends
 886   // in the serverlist are reachable.
 887   WaitForAllBackends(1 /* num_requests_multiple_of */,
 888                      kNumBackendsInResolution /* start_index */);
 889   gpr_log(GPR_INFO, "========= BEFORE SECOND BATCH ==========");
 890   CheckRpcSendOk(backends_.size() - kNumBackendsInResolution);
 891   gpr_log(GPR_INFO, "========= DONE WITH SECOND BATCH ==========");
 892   // Serverlist is used: each backend returned by the balancer should
 893   // have gotten one request.
 894   for (size_t i = 0; i < kNumBackendsInResolution; ++i) {
 895     EXPECT_EQ(0U, backends_[i]->service_.request_count());
 896   }
 897   for (size_t i = kNumBackendsInResolution; i < backends_.size(); ++i) {
 898     EXPECT_EQ(1U, backends_[i]->service_.request_count());
 899   }
 900   // The balancer got a single request.
 901   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
 902   // and sent a single response.
 903   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
 904 }
 905
 906 TEST_F(SingleBalancerTest, FallbackUpdate) {
 907   const int kFallbackTimeoutMs = 200 * grpc_test_slowdown_factor();
 908   const int kServerlistDelayMs = 500 * grpc_test_slowdown_factor();
 909   const size_t kNumBackendsInResolution = backends_.size() / 3;
 910   const size_t kNumBackendsInResolutionUpdate = backends_.size() / 3;
 911   ResetStub(kFallbackTimeoutMs);
 912   SetNextResolution(GetBackendPorts(0, kNumBackendsInResolution),
 913                     kDefaultServiceConfig_.c_str());
 914   SetNextResolutionForLbChannelAllBalancers();
 915   // Send non-empty serverlist only after kServerlistDelayMs.
 916   ScheduleResponseForBalancer(
 917       0,
 918       BalancerServiceImpl::BuildResponseForBackends(
 919           GetBackendPorts(kNumBackendsInResolution +
 920                           kNumBackendsInResolutionUpdate /* start_index */),
 921           {}),
 922       kServerlistDelayMs);
 923   // Wait until all the fallback backends are reachable.
 924   WaitForAllBackends(1 /* num_requests_multiple_of */, 0 /* start_index */,
 925                      kNumBackendsInResolution /* stop_index */);
 926   gpr_log(GPR_INFO, "========= BEFORE FIRST BATCH ==========");
 927   CheckRpcSendOk(kNumBackendsInResolution);
 928   gpr_log(GPR_INFO, "========= DONE WITH FIRST BATCH ==========");
 929   // Fallback is used: each backend returned by the resolver should have
 930   // gotten one request.
 931   for (size_t i = 0; i < kNumBackendsInResolution; ++i) {
 932     EXPECT_EQ(1U, backends_[i]->service_.request_count());
 933   }
 934   for (size_t i = kNumBackendsInResolution; i < backends_.size(); ++i) {
 935     EXPECT_EQ(0U, backends_[i]->service_.request_count());
 936   }
 937   SetNextResolution(GetBackendPorts(kNumBackendsInResolution,
 938                                     kNumBackendsInResolution +
 939                                         kNumBackendsInResolutionUpdate),
 940                     kDefaultServiceConfig_.c_str());
 941   // Wait until the resolution update has been processed and all the new
 942   // fallback backends are reachable.
 943   WaitForAllBackends(1 /* num_requests_multiple_of */,
 944                      kNumBackendsInResolution /* start_index */,
 945                      kNumBackendsInResolution +
 946                          kNumBackendsInResolutionUpdate /* stop_index */);
 947   gpr_log(GPR_INFO, "========= BEFORE SECOND BATCH ==========");
 948   CheckRpcSendOk(kNumBackendsInResolutionUpdate);
 949   gpr_log(GPR_INFO, "========= DONE WITH SECOND BATCH ==========");
 950   // The resolution update is used: each backend in the resolution update should
 951   // have gotten one request.
 952   for (size_t i = 0; i < kNumBackendsInResolution; ++i) {
 953     EXPECT_EQ(0U, backends_[i]->service_.request_count());
 954   }
 955   for (size_t i = kNumBackendsInResolution;
 956        i < kNumBackendsInResolution + kNumBackendsInResolutionUpdate; ++i) {
 957     EXPECT_EQ(1U, backends_[i]->service_.request_count());
 958   }
 959   for (size_t i = kNumBackendsInResolution + kNumBackendsInResolutionUpdate;
 960        i < backends_.size(); ++i) {
 961     EXPECT_EQ(0U, backends_[i]->service_.request_count());
 962   }
 963   // Wait until the serverlist reception has been processed and all backends
 964   // in the serverlist are reachable.
 965   WaitForAllBackends(1 /* num_requests_multiple_of */,
 966                      kNumBackendsInResolution +
 967                          kNumBackendsInResolutionUpdate /* start_index */);
 968   gpr_log(GPR_INFO, "========= BEFORE THIRD BATCH ==========");
 969   CheckRpcSendOk(backends_.size() - kNumBackendsInResolution -
 970                  kNumBackendsInResolutionUpdate);
 971   gpr_log(GPR_INFO, "========= DONE WITH THIRD BATCH ==========");
 972   // Serverlist is used: each backend returned by the balancer should
 973   // have gotten one request.
 974   for (size_t i = 0;
 975        i < kNumBackendsInResolution + kNumBackendsInResolutionUpdate; ++i) {
 976     EXPECT_EQ(0U, backends_[i]->service_.request_count());
 977   }
 978   for (size_t i = kNumBackendsInResolution + kNumBackendsInResolutionUpdate;
 979        i < backends_.size(); ++i) {
 980     EXPECT_EQ(1U, backends_[i]->service_.request_count());
 981   }
 982   // The balancer got a single request.
 983   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
 984   // and sent a single response.
 985   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
 986 }
 987
 988 TEST_F(SingleBalancerTest, FallbackEarlyWhenBalancerChannelFails) {
 989   const int kFallbackTimeoutMs = 10000 * grpc_test_slowdown_factor();
 990   ResetStub(kFallbackTimeoutMs);
 991   // Return an unreachable balancer and one fallback backend.
 992   SetNextResolution({backends_[0]->port_}, kDefaultServiceConfig_.c_str());
 993   SetNextResolutionForLbChannel({grpc_pick_unused_port_or_die()});
 994   // Send RPC with deadline less than the fallback timeout and make sure it
 995   // succeeds.
 996   CheckRpcSendOk(/* times */ 1, /* timeout_ms */ 1000,
 997                  /* wait_for_ready */ false);
 998 }
 999
1000 TEST_F(SingleBalancerTest, FallbackEarlyWhenBalancerCallFails) {
1001   const int kFallbackTimeoutMs = 10000 * grpc_test_slowdown_factor();
1002   ResetStub(kFallbackTimeoutMs);
1003   // Return one balancer and one fallback backend.
1004   SetNextResolution({backends_[0]->port_}, kDefaultServiceConfig_.c_str());
1005   SetNextResolutionForLbChannelAllBalancers();
1006   // Balancer drops call without sending a serverlist.
1007   balancers_[0]->service_.NotifyDoneWithServerlists();
1008   // Send RPC with deadline less than the fallback timeout and make sure it
1009   // succeeds.
1010   CheckRpcSendOk(/* times */ 1, /* timeout_ms */ 1000,
1011                  /* wait_for_ready */ false);
1012 }
1013
1014 TEST_F(SingleBalancerTest, FallbackIfResponseReceivedButChildNotReady) {
1015   const int kFallbackTimeoutMs = 500 * grpc_test_slowdown_factor();
1016   ResetStub(kFallbackTimeoutMs);
1017   SetNextResolution({backends_[0]->port_}, kDefaultServiceConfig_.c_str());
1018   SetNextResolutionForLbChannelAllBalancers();
1019   // Send a serverlist that only contains an unreachable backend before fallback
1020   // timeout.
1021   ScheduleResponseForBalancer(0,
1022                               BalancerServiceImpl::BuildResponseForBackends(
1023                                   {grpc_pick_unused_port_or_die()}, {}),
1024                               0);
1025   // Because no child policy is ready before fallback timeout, we enter fallback
1026   // mode.
1027   WaitForBackend(0);
1028 }
1029
1030 TEST_F(SingleBalancerTest, FallbackModeIsExitedWhenBalancerSaysToDropAllCalls) {
1031   // Return an unreachable balancer and one fallback backend.
1032   SetNextResolution({backends_[0]->port_}, kDefaultServiceConfig_.c_str());
1033   SetNextResolutionForLbChannel({grpc_pick_unused_port_or_die()});
1034   // Enter fallback mode because the LB channel fails to connect.
1035   WaitForBackend(0);
1036   // Return a new balancer that sends an empty serverlist.
1037   ScheduleResponseForBalancer(
1038       0, BalancerServiceImpl::BuildResponseForBackends({}, {}), 0);
1039   SetNextResolutionForLbChannelAllBalancers();
1040   // Send RPCs until failure.
1041   gpr_timespec deadline = gpr_time_add(
1042       gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(5000, GPR_TIMESPAN));
1043   do {
1044     auto status = SendRpc();
1045     if (!status.ok()) break;
1046   } while (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0);
1047   CheckRpcSendFailure();
1048 }
1049
1050 TEST_F(SingleBalancerTest, FallbackModeIsExitedAfterChildRready) {
1051   // Return an unreachable balancer and one fallback backend.
1052   SetNextResolution({backends_[0]->port_}, kDefaultServiceConfig_.c_str());
1053   SetNextResolutionForLbChannel({grpc_pick_unused_port_or_die()});
1054   // Enter fallback mode because the LB channel fails to connect.
1055   WaitForBackend(0);
1056   // Return a new balancer that sends a dead backend.
1057   ShutdownBackend(1);
1058   ScheduleResponseForBalancer(
1059       0,
1060       BalancerServiceImpl::BuildResponseForBackends({backends_[1]->port_}, {}),
1061       0);
1062   SetNextResolutionForLbChannelAllBalancers();
1063   // The state (TRANSIENT_FAILURE) update from the child policy will be ignored
1064   // because we are still in fallback mode.
1065   gpr_timespec deadline = gpr_time_add(
1066       gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(5000, GPR_TIMESPAN));
1067   // Send 5 seconds worth of RPCs.
1068   do {
1069     CheckRpcSendOk();
1070   } while (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0);
1071   // After the backend is restarted, the child policy will eventually be READY,
1072   // and we will exit fallback mode.
1073   StartBackend(1);
1074   WaitForBackend(1);
1075   // We have exited fallback mode, so calls will go to the child policy
1076   // exclusively.
1077   CheckRpcSendOk(100);
1078   EXPECT_EQ(0U, backends_[0]->service_.request_count());
1079   EXPECT_EQ(100U, backends_[1]->service_.request_count());
1080 }
1081
1082 TEST_F(SingleBalancerTest, BackendsRestart) {
1083   SetNextResolution({}, kDefaultServiceConfig_.c_str());
1084   SetNextResolutionForLbChannelAllBalancers();
1085   ScheduleResponseForBalancer(
1086       0, BalancerServiceImpl::BuildResponseForBackends(GetBackendPorts(), {}),
1087       0);
1088   WaitForAllBackends();
1089   // Stop backends.  RPCs should fail.
1090   ShutdownAllBackends();
1091   CheckRpcSendFailure();
1092   // Restart all backends.  RPCs should start succeeding again.
1093   StartAllBackends();
1094   CheckRpcSendOk(1 /* times */, 2000 /* timeout_ms */,
1095                  true /* wait_for_ready */);
1096 }
1097
1098 class UpdatesTest : public XdsEnd2endTest {
1099  public:
1100   UpdatesTest() : XdsEnd2endTest(4, 3, 0) {}
1101 };
1102
1103 TEST_F(UpdatesTest, UpdateBalancersButKeepUsingOriginalBalancer) {
1104   SetNextResolution({}, kDefaultServiceConfig_.c_str());
1105   SetNextResolutionForLbChannelAllBalancers();
1106   const std::vector<int> first_backend{GetBackendPorts()[0]};
1107   const std::vector<int> second_backend{GetBackendPorts()[1]};
1108   ScheduleResponseForBalancer(
1109       0, BalancerServiceImpl::BuildResponseForBackends(first_backend, {}), 0);
1110   ScheduleResponseForBalancer(
1111       1, BalancerServiceImpl::BuildResponseForBackends(second_backend, {}), 0);
1112
1113   // Wait until the first backend is ready.
1114   WaitForBackend(0);
1115
1116   // Send 10 requests.
1117   gpr_log(GPR_INFO, "========= BEFORE FIRST BATCH ==========");
1118   CheckRpcSendOk(10);
1119   gpr_log(GPR_INFO, "========= DONE WITH FIRST BATCH ==========");
1120
1121   // All 10 requests should have gone to the first backend.
1122   EXPECT_EQ(10U, backends_[0]->service_.request_count());
1123
1124   // Balancer 0 got a single request.
1125   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
1126   // and sent a single response.
1127   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
1128   EXPECT_EQ(0U, balancers_[1]->service_.request_count());
1129   EXPECT_EQ(0U, balancers_[1]->service_.response_count());
1130   EXPECT_EQ(0U, balancers_[2]->service_.request_count());
1131   EXPECT_EQ(0U, balancers_[2]->service_.response_count());
1132
1133   gpr_log(GPR_INFO, "========= ABOUT TO UPDATE 1 ==========");
1134   SetNextResolutionForLbChannel({balancers_[1]->port_});
1135   gpr_log(GPR_INFO, "========= UPDATE 1 DONE ==========");
1136
1137   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1138   gpr_timespec deadline = gpr_time_add(
1139       gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(10000, GPR_TIMESPAN));
1140   // Send 10 seconds worth of RPCs
1141   do {
1142     CheckRpcSendOk();
1143   } while (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0);
1144   // The current LB call is still working, so xds continued using it to the
1145   // first balancer, which doesn't assign the second backend.
1146   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1147
1148   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
1149   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
1150   EXPECT_EQ(0U, balancers_[1]->service_.request_count());
1151   EXPECT_EQ(0U, balancers_[1]->service_.response_count());
1152   EXPECT_EQ(0U, balancers_[2]->service_.request_count());
1153   EXPECT_EQ(0U, balancers_[2]->service_.response_count());
1154 }
1155
1156 TEST_F(UpdatesTest, UpdateBalancerName) {
1157   SetNextResolution({}, kDefaultServiceConfig_.c_str());
1158   SetNextResolutionForLbChannelAllBalancers();
1159   const std::vector<int> first_backend{GetBackendPorts()[0]};
1160   const std::vector<int> second_backend{GetBackendPorts()[1]};
1161   ScheduleResponseForBalancer(
1162       0, BalancerServiceImpl::BuildResponseForBackends(first_backend, {}), 0);
1163   ScheduleResponseForBalancer(
1164       1, BalancerServiceImpl::BuildResponseForBackends(second_backend, {}), 0);
1165
1166   // Wait until the first backend is ready.
1167   WaitForBackend(0);
1168
1169   // Send 10 requests.
1170   gpr_log(GPR_INFO, "========= BEFORE FIRST BATCH ==========");
1171   CheckRpcSendOk(10);
1172   gpr_log(GPR_INFO, "========= DONE WITH FIRST BATCH ==========");
1173
1174   // All 10 requests should have gone to the first backend.
1175   EXPECT_EQ(10U, backends_[0]->service_.request_count());
1176
1177   // Balancer 0 got a single request.
1178   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
1179   // and sent a single response.
1180   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
1181   EXPECT_EQ(0U, balancers_[1]->service_.request_count());
1182   EXPECT_EQ(0U, balancers_[1]->service_.response_count());
1183   EXPECT_EQ(0U, balancers_[2]->service_.request_count());
1184   EXPECT_EQ(0U, balancers_[2]->service_.response_count());
1185
1186   std::vector<int> ports;
1187   ports.emplace_back(balancers_[1]->port_);
1188   auto new_lb_channel_response_generator =
1189       grpc_core::MakeRefCounted<grpc_core::FakeResolverResponseGenerator>();
1190   SetNextResolutionForLbChannel(ports, nullptr,
1191                                 new_lb_channel_response_generator.get());
1192   gpr_log(GPR_INFO, "========= ABOUT TO UPDATE BALANCER NAME ==========");
1193   SetNextResolution({},
1194                     "{\n"
1195                     "  \"loadBalancingConfig\":[\n"
1196                     "    { \"does_not_exist\":{} },\n"
1197                     "    { \"xds_experimental\":{ \"balancerName\": "
1198                     "\"fake:///updated_lb\" } }\n"
1199                     "  ]\n"
1200                     "}",
1201                     new_lb_channel_response_generator.get());
1202   gpr_log(GPR_INFO, "========= UPDATED BALANCER NAME ==========");
1203
1204   // Wait until update has been processed, as signaled by the second backend
1205   // receiving a request.
1206   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1207   WaitForBackend(1);
1208
1209   backends_[1]->service_.ResetCounters();
1210   gpr_log(GPR_INFO, "========= BEFORE SECOND BATCH ==========");
1211   CheckRpcSendOk(10);
1212   gpr_log(GPR_INFO, "========= DONE WITH SECOND BATCH ==========");
1213   // All 10 requests should have gone to the second backend.
1214   EXPECT_EQ(10U, backends_[1]->service_.request_count());
1215
1216   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
1217   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
1218   EXPECT_EQ(1U, balancers_[1]->service_.request_count());
1219   EXPECT_EQ(1U, balancers_[1]->service_.response_count());
1220   EXPECT_EQ(0U, balancers_[2]->service_.request_count());
1221   EXPECT_EQ(0U, balancers_[2]->service_.response_count());
1222 }
1223
1224 // Send an update with the same set of LBs as the one in SetUp() in order to
1225 // verify that the LB channel inside xds keeps the initial connection (which
1226 // by definition is also present in the update).
1227 TEST_F(UpdatesTest, UpdateBalancersRepeated) {
1228   SetNextResolution({}, kDefaultServiceConfig_.c_str());
1229   SetNextResolutionForLbChannelAllBalancers();
1230   const std::vector<int> first_backend{GetBackendPorts()[0]};
1231   const std::vector<int> second_backend{GetBackendPorts()[0]};
1232
1233   ScheduleResponseForBalancer(
1234       0, BalancerServiceImpl::BuildResponseForBackends(first_backend, {}), 0);
1235   ScheduleResponseForBalancer(
1236       1, BalancerServiceImpl::BuildResponseForBackends(second_backend, {}), 0);
1237
1238   // Wait until the first backend is ready.
1239   WaitForBackend(0);
1240
1241   // Send 10 requests.
1242   gpr_log(GPR_INFO, "========= BEFORE FIRST BATCH ==========");
1243   CheckRpcSendOk(10);
1244   gpr_log(GPR_INFO, "========= DONE WITH FIRST BATCH ==========");
1245
1246   // All 10 requests should have gone to the first backend.
1247   EXPECT_EQ(10U, backends_[0]->service_.request_count());
1248
1249   // Balancer 0 got a single request.
1250   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
1251   // and sent a single response.
1252   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
1253   EXPECT_EQ(0U, balancers_[1]->service_.request_count());
1254   EXPECT_EQ(0U, balancers_[1]->service_.response_count());
1255   EXPECT_EQ(0U, balancers_[2]->service_.request_count());
1256   EXPECT_EQ(0U, balancers_[2]->service_.response_count());
1257
1258   std::vector<int> ports;
1259   ports.emplace_back(balancers_[0]->port_);
1260   ports.emplace_back(balancers_[1]->port_);
1261   ports.emplace_back(balancers_[2]->port_);
1262   gpr_log(GPR_INFO, "========= ABOUT TO UPDATE 1 ==========");
1263   SetNextResolutionForLbChannel(ports);
1264   gpr_log(GPR_INFO, "========= UPDATE 1 DONE ==========");
1265
1266   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1267   gpr_timespec deadline = gpr_time_add(
1268       gpr_now(GPR_CLOCK_REALTIME), gpr_time_from_millis(10000, GPR_TIMESPAN));
1269   // Send 10 seconds worth of RPCs
1270   do {
1271     CheckRpcSendOk();
1272   } while (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0);
1273   // xds continued using the original LB call to the first balancer, which
1274   // doesn't assign the second backend.
1275   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1276
1277   ports.clear();
1278   ports.emplace_back(balancers_[0]->port_);
1279   ports.emplace_back(balancers_[1]->port_);
1280   gpr_log(GPR_INFO, "========= ABOUT TO UPDATE 2 ==========");
1281   SetNextResolutionForLbChannel(ports);
1282   gpr_log(GPR_INFO, "========= UPDATE 2 DONE ==========");
1283
1284   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1285   deadline = gpr_time_add(gpr_now(GPR_CLOCK_REALTIME),
1286                           gpr_time_from_millis(10000, GPR_TIMESPAN));
1287   // Send 10 seconds worth of RPCs
1288   do {
1289     CheckRpcSendOk();
1290   } while (gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), deadline) < 0);
1291   // xds continued using the original LB call to the first balancer, which
1292   // doesn't assign the second backend.
1293   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1294 }
1295
1296 TEST_F(UpdatesTest, UpdateBalancersDeadUpdate) {
1297   SetNextResolution({}, kDefaultServiceConfig_.c_str());
1298   SetNextResolutionForLbChannel({balancers_[0]->port_});
1299   const std::vector<int> first_backend{GetBackendPorts()[0]};
1300   const std::vector<int> second_backend{GetBackendPorts()[1]};
1301
1302   ScheduleResponseForBalancer(
1303       0, BalancerServiceImpl::BuildResponseForBackends(first_backend, {}), 0);
1304   ScheduleResponseForBalancer(
1305       1, BalancerServiceImpl::BuildResponseForBackends(second_backend, {}), 0);
1306
1307   // Start servers and send 10 RPCs per server.
1308   gpr_log(GPR_INFO, "========= BEFORE FIRST BATCH ==========");
1309   CheckRpcSendOk(10);
1310   gpr_log(GPR_INFO, "========= DONE WITH FIRST BATCH ==========");
1311   // All 10 requests should have gone to the first backend.
1312   EXPECT_EQ(10U, backends_[0]->service_.request_count());
1313
1314   // Kill balancer 0
1315   gpr_log(GPR_INFO, "********** ABOUT TO KILL BALANCER 0 *************");
1316   balancers_[0]->Shutdown();
1317   gpr_log(GPR_INFO, "********** KILLED BALANCER 0 *************");
1318
1319   // This is serviced by the existing child policy.
1320   gpr_log(GPR_INFO, "========= BEFORE SECOND BATCH ==========");
1321   CheckRpcSendOk(10);
1322   gpr_log(GPR_INFO, "========= DONE WITH SECOND BATCH ==========");
1323   // All 10 requests should again have gone to the first backend.
1324   EXPECT_EQ(20U, backends_[0]->service_.request_count());
1325   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1326
1327   // Balancer 0 got a single request.
1328   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
1329   // and sent a single response.
1330   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
1331   EXPECT_EQ(0U, balancers_[1]->service_.request_count());
1332   EXPECT_EQ(0U, balancers_[1]->service_.response_count());
1333   EXPECT_EQ(0U, balancers_[2]->service_.request_count());
1334   EXPECT_EQ(0U, balancers_[2]->service_.response_count());
1335
1336   gpr_log(GPR_INFO, "========= ABOUT TO UPDATE 1 ==========");
1337   SetNextResolutionForLbChannel({balancers_[1]->port_});
1338   gpr_log(GPR_INFO, "========= UPDATE 1 DONE ==========");
1339
1340   // Wait until update has been processed, as signaled by the second backend
1341   // receiving a request. In the meantime, the client continues to be serviced
1342   // (by the first backend) without interruption.
1343   EXPECT_EQ(0U, backends_[1]->service_.request_count());
1344   WaitForBackend(1);
1345
1346   // This is serviced by the updated RR policy
1347   backends_[1]->service_.ResetCounters();
1348   gpr_log(GPR_INFO, "========= BEFORE THIRD BATCH ==========");
1349   CheckRpcSendOk(10);
1350   gpr_log(GPR_INFO, "========= DONE WITH THIRD BATCH ==========");
1351   // All 10 requests should have gone to the second backend.
1352   EXPECT_EQ(10U, backends_[1]->service_.request_count());
1353
1354   EXPECT_EQ(1U, balancers_[0]->service_.request_count());
1355   EXPECT_EQ(1U, balancers_[0]->service_.response_count());
1356   // The second balancer, published as part of the first update, may end up
1357   // getting two requests (that is, 1 <= #req <= 2) if the LB call retry timer
1358   // firing races with the arrival of the update containing the second
1359   // balancer.
1360   EXPECT_GE(balancers_[1]->service_.request_count(), 1U);
1361   EXPECT_GE(balancers_[1]->service_.response_count(), 1U);
1362   EXPECT_LE(balancers_[1]->service_.request_count(), 2U);
1363   EXPECT_LE(balancers_[1]->service_.response_count(), 2U);
1364   EXPECT_EQ(0U, balancers_[2]->service_.request_count());
1365   EXPECT_EQ(0U, balancers_[2]->service_.response_count());
1366 }
1367
1368 // The re-resolution tests are deferred because they rely on the fallback mode,
1369 // which hasn't been supported.
1370
1371 // TODO(juanlishen): Add TEST_F(UpdatesTest, ReresolveDeadBackend).
1372
1373 // TODO(juanlishen): Add TEST_F(UpdatesWithClientLoadReportingTest,
1374 // ReresolveDeadBalancer)
1375
1376 // The drop tests are deferred because the drop handling hasn't been added yet.
1377
1378 // TODO(roth): Add TEST_F(SingleBalancerTest, Drop)
1379
1380 // TODO(roth): Add TEST_F(SingleBalancerTest, DropAllFirst)
1381
1382 // TODO(roth): Add TEST_F(SingleBalancerTest, DropAll)
1383
1384 class SingleBalancerWithClientLoadReportingTest : public XdsEnd2endTest {
1385  public:
1386   SingleBalancerWithClientLoadReportingTest() : XdsEnd2endTest(4, 1, 3) {}
1387 };
1388
1389 // The client load reporting tests are deferred because the client load
1390 // reporting hasn't been supported yet.
1391
1392 // TODO(vpowar): Add TEST_F(SingleBalancerWithClientLoadReportingTest, Vanilla)
1393
1394 // TODO(roth): Add TEST_F(SingleBalancerWithClientLoadReportingTest,
1395 // BalancerRestart)
1396
1397 // TODO(roth): Add TEST_F(SingleBalancerWithClientLoadReportingTest, Drop)
1398
1399 }  // namespace
1400 }  // namespace testing
1401 }  // namespace grpc
1402
1403 int main(int argc, char** argv) {
1404   // Make the backup poller poll very frequently in order to pick up
1405   // updates from all the subchannels's FDs.
1406   GPR_GLOBAL_CONFIG_SET(grpc_client_channel_backup_poll_interval_ms, 1);
1407   grpc_init();
1408   grpc::testing::TestEnvironment env(argc, argv);
1409   ::testing::InitGoogleTest(&argc, argv);
1410   const auto result = RUN_ALL_TESTS();
1411   grpc_shutdown();
1412   return result;
1413 }