test/cpp/end2end/client_lb_end2end_test.cc

   1 /*
   2  *
   3  * Copyright 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 <algorithm>
  20 #include <memory>
  21 #include <mutex>
  22 #include <random>
  23 #include <set>
  24 #include <thread>
  25
  26 #include <grpc/grpc.h>
  27 #include <grpc/support/alloc.h>
  28 #include <grpc/support/atm.h>
  29 #include <grpc/support/log.h>
  30 #include <grpc/support/string_util.h>
  31 #include <grpc/support/time.h>
  32 #include <grpcpp/channel.h>
  33 #include <grpcpp/client_context.h>
  34 #include <grpcpp/create_channel.h>
  35 #include <grpcpp/health_check_service_interface.h>
  36 #include <grpcpp/impl/codegen/sync.h>
  37 #include <grpcpp/server.h>
  38 #include <grpcpp/server_builder.h>
  39
  40 #include "src/core/ext/filters/client_channel/backup_poller.h"
  41 #include "src/core/ext/filters/client_channel/global_subchannel_pool.h"
  42 #include "src/core/ext/filters/client_channel/parse_address.h"
  43 #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
  44 #include "src/core/ext/filters/client_channel/server_address.h"
  45 #include "src/core/ext/filters/client_channel/service_config.h"
  46 #include "src/core/lib/backoff/backoff.h"
  47 #include "src/core/lib/channel/channel_args.h"
  48 #include "src/core/lib/gpr/env.h"
  49 #include "src/core/lib/gprpp/debug_location.h"
  50 #include "src/core/lib/gprpp/ref_counted_ptr.h"
  51 #include "src/core/lib/iomgr/tcp_client.h"
  52 #include "src/core/lib/security/credentials/fake/fake_credentials.h"
  53 #include "src/cpp/client/secure_credentials.h"
  54 #include "src/cpp/server/secure_server_credentials.h"
  55
  56 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  57 #include "src/proto/grpc/testing/xds/orca_load_report_for_test.pb.h"
  58 #include "test/core/util/port.h"
  59 #include "test/core/util/test_config.h"
  60 #include "test/core/util/test_lb_policies.h"
  61 #include "test/cpp/end2end/test_service_impl.h"
  62
  63 #include <gmock/gmock.h>
  64 #include <gtest/gtest.h>
  65
  66 using grpc::testing::EchoRequest;
  67 using grpc::testing::EchoResponse;
  68 using std::chrono::system_clock;
  69
  70 // defined in tcp_client.cc
  71 extern grpc_tcp_client_vtable* grpc_tcp_client_impl;
  72
  73 static grpc_tcp_client_vtable* default_client_impl;
  74
  75 namespace grpc {
  76 namespace testing {
  77 namespace {
  78
  79 gpr_atm g_connection_delay_ms;
  80
  81 void tcp_client_connect_with_delay(grpc_closure* closure, grpc_endpoint** ep,
  82                                    grpc_pollset_set* interested_parties,
  83                                    const grpc_channel_args* channel_args,
  84                                    const grpc_resolved_address* addr,
  85                                    grpc_millis deadline) {
  86   const int delay_ms = gpr_atm_acq_load(&g_connection_delay_ms);
  87   if (delay_ms > 0) {
  88     gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(delay_ms));
  89   }
  90   default_client_impl->connect(closure, ep, interested_parties, channel_args,
  91                                addr, deadline + delay_ms);
  92 }
  93
  94 grpc_tcp_client_vtable delayed_connect = {tcp_client_connect_with_delay};
  95
  96 // Subclass of TestServiceImpl that increments a request counter for
  97 // every call to the Echo RPC.
  98 class MyTestServiceImpl : public TestServiceImpl {
  99  public:
 100   Status Echo(ServerContext* context, const EchoRequest* request,
 101               EchoResponse* response) override {
 102     const udpa::data::orca::v1::OrcaLoadReport* load_report = nullptr;
 103     {
 104       grpc::internal::MutexLock lock(&mu_);
 105       ++request_count_;
 106       load_report = load_report_;
 107     }
 108     AddClient(context->peer());
 109     if (load_report != nullptr) {
 110       // TODO(roth): Once we provide a more standard server-side API for
 111       // populating this data, use that API here.
 112       context->AddTrailingMetadata("x-endpoint-load-metrics-bin",
 113                                    load_report->SerializeAsString());
 114     }
 115     return TestServiceImpl::Echo(context, request, response);
 116   }
 117
 118   int request_count() {
 119     grpc::internal::MutexLock lock(&mu_);
 120     return request_count_;
 121   }
 122
 123   void ResetCounters() {
 124     grpc::internal::MutexLock lock(&mu_);
 125     request_count_ = 0;
 126   }
 127
 128   std::set<grpc::string> clients() {
 129     grpc::internal::MutexLock lock(&clients_mu_);
 130     return clients_;
 131   }
 132
 133   void set_load_report(udpa::data::orca::v1::OrcaLoadReport* load_report) {
 134     grpc::internal::MutexLock lock(&mu_);
 135     load_report_ = load_report;
 136   }
 137
 138  private:
 139   void AddClient(const grpc::string& client) {
 140     grpc::internal::MutexLock lock(&clients_mu_);
 141     clients_.insert(client);
 142   }
 143
 144   grpc::internal::Mutex mu_;
 145   int request_count_ = 0;
 146   const udpa::data::orca::v1::OrcaLoadReport* load_report_ = nullptr;
 147   grpc::internal::Mutex clients_mu_;
 148   std::set<grpc::string> clients_;
 149 };
 150
 151 class FakeResolverResponseGeneratorWrapper {
 152  public:
 153   FakeResolverResponseGeneratorWrapper()
 154       : response_generator_(grpc_core::MakeRefCounted<
 155                             grpc_core::FakeResolverResponseGenerator>()) {}
 156
 157   FakeResolverResponseGeneratorWrapper(
 158       FakeResolverResponseGeneratorWrapper&& other) {
 159     response_generator_ = std::move(other.response_generator_);
 160   }
 161
 162   void SetNextResolution(const std::vector<int>& ports,
 163                          const char* service_config_json = nullptr) {
 164     grpc_core::ExecCtx exec_ctx;
 165     response_generator_->SetResponse(
 166         BuildFakeResults(ports, service_config_json));
 167   }
 168
 169   void SetNextResolutionUponError(const std::vector<int>& ports) {
 170     grpc_core::ExecCtx exec_ctx;
 171     response_generator_->SetReresolutionResponse(BuildFakeResults(ports));
 172   }
 173
 174   void SetFailureOnReresolution() {
 175     grpc_core::ExecCtx exec_ctx;
 176     response_generator_->SetFailureOnReresolution();
 177   }
 178
 179   grpc_core::FakeResolverResponseGenerator* Get() const {
 180     return response_generator_.get();
 181   }
 182
 183  private:
 184   static grpc_core::Resolver::Result BuildFakeResults(
 185       const std::vector<int>& ports,
 186       const char* service_config_json = nullptr) {
 187     grpc_core::Resolver::Result result;
 188     for (const int& port : ports) {
 189       char* lb_uri_str;
 190       gpr_asprintf(&lb_uri_str, "ipv4:127.0.0.1:%d", port);
 191       grpc_uri* lb_uri = grpc_uri_parse(lb_uri_str, true);
 192       GPR_ASSERT(lb_uri != nullptr);
 193       grpc_resolved_address address;
 194       GPR_ASSERT(grpc_parse_uri(lb_uri, &address));
 195       result.addresses.emplace_back(address.addr, address.len,
 196                                     nullptr /* args */);
 197       grpc_uri_destroy(lb_uri);
 198       gpr_free(lb_uri_str);
 199     }
 200     if (service_config_json != nullptr) {
 201       result.service_config = grpc_core::ServiceConfig::Create(
 202           service_config_json, &result.service_config_error);
 203       GPR_ASSERT(result.service_config != nullptr);
 204     }
 205     return result;
 206   }
 207
 208   grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
 209       response_generator_;
 210 };
 211
 212 class ClientLbEnd2endTest : public ::testing::Test {
 213  protected:
 214   ClientLbEnd2endTest()
 215       : server_host_("localhost"),
 216         kRequestMessage_("Live long and prosper."),
 217         creds_(new SecureChannelCredentials(
 218             grpc_fake_transport_security_credentials_create())) {}
 219
 220   static void SetUpTestCase() {
 221     // Make the backup poller poll very frequently in order to pick up
 222     // updates from all the subchannels's FDs.
 223     GPR_GLOBAL_CONFIG_SET(grpc_client_channel_backup_poll_interval_ms, 1);
 224 #if TARGET_OS_IPHONE
 225     // Workaround Apple CFStream bug
 226     gpr_setenv("grpc_cfstream", "0");
 227 #endif
 228   }
 229
 230   void SetUp() override { grpc_init(); }
 231
 232   void TearDown() override {
 233     for (size_t i = 0; i < servers_.size(); ++i) {
 234       servers_[i]->Shutdown();
 235     }
 236     // Explicitly destroy all the members so that we can make sure grpc_shutdown
 237     // has finished by the end of this function, and thus all the registered
 238     // LB policy factories are removed.
 239     servers_.clear();
 240     creds_.reset();
 241     grpc_shutdown_blocking();
 242   }
 243
 244   void CreateServers(size_t num_servers,
 245                      std::vector<int> ports = std::vector<int>()) {
 246     servers_.clear();
 247     for (size_t i = 0; i < num_servers; ++i) {
 248       int port = 0;
 249       if (ports.size() == num_servers) port = ports[i];
 250       servers_.emplace_back(new ServerData(port));
 251     }
 252   }
 253
 254   void StartServer(size_t index) { servers_[index]->Start(server_host_); }
 255
 256   void StartServers(size_t num_servers,
 257                     std::vector<int> ports = std::vector<int>()) {
 258     CreateServers(num_servers, std::move(ports));
 259     for (size_t i = 0; i < num_servers; ++i) {
 260       StartServer(i);
 261     }
 262   }
 263
 264   std::vector<int> GetServersPorts(size_t start_index = 0) {
 265     std::vector<int> ports;
 266     for (size_t i = start_index; i < servers_.size(); ++i) {
 267       ports.push_back(servers_[i]->port_);
 268     }
 269     return ports;
 270   }
 271
 272   FakeResolverResponseGeneratorWrapper BuildResolverResponseGenerator() {
 273     return FakeResolverResponseGeneratorWrapper();
 274   }
 275
 276   std::unique_ptr<grpc::testing::EchoTestService::Stub> BuildStub(
 277       const std::shared_ptr<Channel>& channel) {
 278     return grpc::testing::EchoTestService::NewStub(channel);
 279   }
 280
 281   std::shared_ptr<Channel> BuildChannel(
 282       const grpc::string& lb_policy_name,
 283       const FakeResolverResponseGeneratorWrapper& response_generator,
 284       ChannelArguments args = ChannelArguments()) {
 285     if (lb_policy_name.size() > 0) {
 286       args.SetLoadBalancingPolicyName(lb_policy_name);
 287     }  // else, default to pick first
 288     args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
 289                     response_generator.Get());
 290     return ::grpc::CreateCustomChannel("fake:///", creds_, args);
 291   }
 292
 293   bool SendRpc(
 294       const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
 295       EchoResponse* response = nullptr, int timeout_ms = 1000,
 296       Status* result = nullptr, bool wait_for_ready = false) {
 297     const bool local_response = (response == nullptr);
 298     if (local_response) response = new EchoResponse;
 299     EchoRequest request;
 300     request.set_message(kRequestMessage_);
 301     ClientContext context;
 302     context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms));
 303     if (wait_for_ready) context.set_wait_for_ready(true);
 304     Status status = stub->Echo(&context, request, response);
 305     if (result != nullptr) *result = status;
 306     if (local_response) delete response;
 307     return status.ok();
 308   }
 309
 310   void CheckRpcSendOk(
 311       const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
 312       const grpc_core::DebugLocation& location, bool wait_for_ready = false) {
 313     EchoResponse response;
 314     Status status;
 315     const bool success =
 316         SendRpc(stub, &response, 2000, &status, wait_for_ready);
 317     ASSERT_TRUE(success) << "From " << location.file() << ":" << location.line()
 318                          << "\n"
 319                          << "Error: " << status.error_message() << " "
 320                          << status.error_details();
 321     ASSERT_EQ(response.message(), kRequestMessage_)
 322         << "From " << location.file() << ":" << location.line();
 323     if (!success) abort();
 324   }
 325
 326   void CheckRpcSendFailure(
 327       const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub) {
 328     const bool success = SendRpc(stub);
 329     EXPECT_FALSE(success);
 330   }
 331
 332   struct ServerData {
 333     int port_;
 334     std::unique_ptr<Server> server_;
 335     MyTestServiceImpl service_;
 336     std::unique_ptr<std::thread> thread_;
 337     bool server_ready_ = false;
 338     bool started_ = false;
 339
 340     explicit ServerData(int port = 0) {
 341       port_ = port > 0 ? port : grpc_pick_unused_port_or_die();
 342     }
 343
 344     void Start(const grpc::string& server_host) {
 345       gpr_log(GPR_INFO, "starting server on port %d", port_);
 346       started_ = true;
 347       grpc::internal::Mutex mu;
 348       grpc::internal::MutexLock lock(&mu);
 349       grpc::internal::CondVar cond;
 350       thread_.reset(new std::thread(
 351           std::bind(&ServerData::Serve, this, server_host, &mu, &cond)));
 352       cond.WaitUntil(&mu, [this] { return server_ready_; });
 353       server_ready_ = false;
 354       gpr_log(GPR_INFO, "server startup complete");
 355     }
 356
 357     void Serve(const grpc::string& server_host, grpc::internal::Mutex* mu,
 358                grpc::internal::CondVar* cond) {
 359       std::ostringstream server_address;
 360       server_address << server_host << ":" << port_;
 361       ServerBuilder builder;
 362       std::shared_ptr<ServerCredentials> creds(new SecureServerCredentials(
 363           grpc_fake_transport_security_server_credentials_create()));
 364       builder.AddListeningPort(server_address.str(), std::move(creds));
 365       builder.RegisterService(&service_);
 366       server_ = builder.BuildAndStart();
 367       grpc::internal::MutexLock lock(mu);
 368       server_ready_ = true;
 369       cond->Signal();
 370     }
 371
 372     void Shutdown() {
 373       if (!started_) return;
 374       server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0));
 375       thread_->join();
 376       started_ = false;
 377     }
 378
 379     void SetServingStatus(const grpc::string& service, bool serving) {
 380       server_->GetHealthCheckService()->SetServingStatus(service, serving);
 381     }
 382   };
 383
 384   void ResetCounters() {
 385     for (const auto& server : servers_) server->service_.ResetCounters();
 386   }
 387
 388   void WaitForServer(
 389       const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
 390       size_t server_idx, const grpc_core::DebugLocation& location,
 391       bool ignore_failure = false) {
 392     do {
 393       if (ignore_failure) {
 394         SendRpc(stub);
 395       } else {
 396         CheckRpcSendOk(stub, location, true);
 397       }
 398     } while (servers_[server_idx]->service_.request_count() == 0);
 399     ResetCounters();
 400   }
 401
 402   bool WaitForChannelState(
 403       Channel* channel, std::function<bool(grpc_connectivity_state)> predicate,
 404       bool try_to_connect = false, int timeout_seconds = 5) {
 405     const gpr_timespec deadline =
 406         grpc_timeout_seconds_to_deadline(timeout_seconds);
 407     while (true) {
 408       grpc_connectivity_state state = channel->GetState(try_to_connect);
 409       if (predicate(state)) break;
 410       if (!channel->WaitForStateChange(state, deadline)) return false;
 411     }
 412     return true;
 413   }
 414
 415   bool WaitForChannelNotReady(Channel* channel, int timeout_seconds = 5) {
 416     auto predicate = [](grpc_connectivity_state state) {
 417       return state != GRPC_CHANNEL_READY;
 418     };
 419     return WaitForChannelState(channel, predicate, false, timeout_seconds);
 420   }
 421
 422   bool WaitForChannelReady(Channel* channel, int timeout_seconds = 5) {
 423     auto predicate = [](grpc_connectivity_state state) {
 424       return state == GRPC_CHANNEL_READY;
 425     };
 426     return WaitForChannelState(channel, predicate, true, timeout_seconds);
 427   }
 428
 429   bool SeenAllServers() {
 430     for (const auto& server : servers_) {
 431       if (server->service_.request_count() == 0) return false;
 432     }
 433     return true;
 434   }
 435
 436   // Updates \a connection_order by appending to it the index of the newly
 437   // connected server. Must be called after every single RPC.
 438   void UpdateConnectionOrder(
 439       const std::vector<std::unique_ptr<ServerData>>& servers,
 440       std::vector<int>* connection_order) {
 441     for (size_t i = 0; i < servers.size(); ++i) {
 442       if (servers[i]->service_.request_count() == 1) {
 443         // Was the server index known? If not, update connection_order.
 444         const auto it =
 445             std::find(connection_order->begin(), connection_order->end(), i);
 446         if (it == connection_order->end()) {
 447           connection_order->push_back(i);
 448           return;
 449         }
 450       }
 451     }
 452   }
 453
 454   const grpc::string server_host_;
 455   std::vector<std::unique_ptr<ServerData>> servers_;
 456   const grpc::string kRequestMessage_;
 457   std::shared_ptr<ChannelCredentials> creds_;
 458 };
 459
 460 TEST_F(ClientLbEnd2endTest, ChannelStateConnectingWhenResolving) {
 461   const int kNumServers = 3;
 462   StartServers(kNumServers);
 463   auto response_generator = BuildResolverResponseGenerator();
 464   auto channel = BuildChannel("", response_generator);
 465   auto stub = BuildStub(channel);
 466   // Initial state should be IDLE.
 467   EXPECT_EQ(channel->GetState(false /* try_to_connect */), GRPC_CHANNEL_IDLE);
 468   // Tell the channel to try to connect.
 469   // Note that this call also returns IDLE, since the state change has
 470   // not yet occurred; it just gets triggered by this call.
 471   EXPECT_EQ(channel->GetState(true /* try_to_connect */), GRPC_CHANNEL_IDLE);
 472   // Now that the channel is trying to connect, we should be in state
 473   // CONNECTING.
 474   EXPECT_EQ(channel->GetState(false /* try_to_connect */),
 475             GRPC_CHANNEL_CONNECTING);
 476   // Return a resolver result, which allows the connection attempt to proceed.
 477   response_generator.SetNextResolution(GetServersPorts());
 478   // We should eventually transition into state READY.
 479   EXPECT_TRUE(WaitForChannelReady(channel.get()));
 480 }
 481
 482 TEST_F(ClientLbEnd2endTest, PickFirst) {
 483   // Start servers and send one RPC per server.
 484   const int kNumServers = 3;
 485   StartServers(kNumServers);
 486   auto response_generator = BuildResolverResponseGenerator();
 487   auto channel = BuildChannel(
 488       "", response_generator);  // test that pick first is the default.
 489   auto stub = BuildStub(channel);
 490   response_generator.SetNextResolution(GetServersPorts());
 491   for (size_t i = 0; i < servers_.size(); ++i) {
 492     CheckRpcSendOk(stub, DEBUG_LOCATION);
 493   }
 494   // All requests should have gone to a single server.
 495   bool found = false;
 496   for (size_t i = 0; i < servers_.size(); ++i) {
 497     const int request_count = servers_[i]->service_.request_count();
 498     if (request_count == kNumServers) {
 499       found = true;
 500     } else {
 501       EXPECT_EQ(0, request_count);
 502     }
 503   }
 504   EXPECT_TRUE(found);
 505   // Check LB policy name for the channel.
 506   EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
 507 }
 508
 509 TEST_F(ClientLbEnd2endTest, PickFirstProcessPending) {
 510   StartServers(1);  // Single server
 511   auto response_generator = BuildResolverResponseGenerator();
 512   auto channel = BuildChannel(
 513       "", response_generator);  // test that pick first is the default.
 514   auto stub = BuildStub(channel);
 515   response_generator.SetNextResolution({servers_[0]->port_});
 516   WaitForServer(stub, 0, DEBUG_LOCATION);
 517   // Create a new channel and its corresponding PF LB policy, which will pick
 518   // the subchannels in READY state from the previous RPC against the same
 519   // target (even if it happened over a different channel, because subchannels
 520   // are globally reused). Progress should happen without any transition from
 521   // this READY state.
 522   auto second_response_generator = BuildResolverResponseGenerator();
 523   auto second_channel = BuildChannel("", second_response_generator);
 524   auto second_stub = BuildStub(second_channel);
 525   second_response_generator.SetNextResolution({servers_[0]->port_});
 526   CheckRpcSendOk(second_stub, DEBUG_LOCATION);
 527 }
 528
 529 TEST_F(ClientLbEnd2endTest, PickFirstSelectsReadyAtStartup) {
 530   ChannelArguments args;
 531   constexpr int kInitialBackOffMs = 5000;
 532   args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
 533   // Create 2 servers, but start only the second one.
 534   std::vector<int> ports = {grpc_pick_unused_port_or_die(),
 535                             grpc_pick_unused_port_or_die()};
 536   CreateServers(2, ports);
 537   StartServer(1);
 538   auto response_generator1 = BuildResolverResponseGenerator();
 539   auto channel1 = BuildChannel("pick_first", response_generator1, args);
 540   auto stub1 = BuildStub(channel1);
 541   response_generator1.SetNextResolution(ports);
 542   // Wait for second server to be ready.
 543   WaitForServer(stub1, 1, DEBUG_LOCATION);
 544   // Create a second channel with the same addresses.  Its PF instance
 545   // should immediately pick the second subchannel, since it's already
 546   // in READY state.
 547   auto response_generator2 = BuildResolverResponseGenerator();
 548   auto channel2 = BuildChannel("pick_first", response_generator2, args);
 549   response_generator2.SetNextResolution(ports);
 550   // Check that the channel reports READY without waiting for the
 551   // initial backoff.
 552   EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1 /* timeout_seconds */));
 553 }
 554
 555 TEST_F(ClientLbEnd2endTest, PickFirstBackOffInitialReconnect) {
 556   ChannelArguments args;
 557   constexpr int kInitialBackOffMs = 100;
 558   args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
 559   const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
 560   const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
 561   auto response_generator = BuildResolverResponseGenerator();
 562   auto channel = BuildChannel("pick_first", response_generator, args);
 563   auto stub = BuildStub(channel);
 564   response_generator.SetNextResolution(ports);
 565   // The channel won't become connected (there's no server).
 566   ASSERT_FALSE(channel->WaitForConnected(
 567       grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 2)));
 568   // Bring up a server on the chosen port.
 569   StartServers(1, ports);
 570   // Now it will.
 571   ASSERT_TRUE(channel->WaitForConnected(
 572       grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 2)));
 573   const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
 574   const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
 575   gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
 576   // We should have waited at least kInitialBackOffMs. We substract one to
 577   // account for test and precision accuracy drift.
 578   EXPECT_GE(waited_ms, kInitialBackOffMs - 1);
 579   // But not much more.
 580   EXPECT_GT(
 581       gpr_time_cmp(
 582           grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 1.10), t1),
 583       0);
 584 }
 585
 586 TEST_F(ClientLbEnd2endTest, PickFirstBackOffMinReconnect) {
 587   ChannelArguments args;
 588   constexpr int kMinReconnectBackOffMs = 1000;
 589   args.SetInt(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, kMinReconnectBackOffMs);
 590   const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
 591   auto response_generator = BuildResolverResponseGenerator();
 592   auto channel = BuildChannel("pick_first", response_generator, args);
 593   auto stub = BuildStub(channel);
 594   response_generator.SetNextResolution(ports);
 595   // Make connection delay a 10% longer than it's willing to in order to make
 596   // sure we are hitting the codepath that waits for the min reconnect backoff.
 597   gpr_atm_rel_store(&g_connection_delay_ms, kMinReconnectBackOffMs * 1.10);
 598   default_client_impl = grpc_tcp_client_impl;
 599   grpc_set_tcp_client_impl(&delayed_connect);
 600   const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
 601   channel->WaitForConnected(
 602       grpc_timeout_milliseconds_to_deadline(kMinReconnectBackOffMs * 2));
 603   const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
 604   const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
 605   gpr_log(GPR_DEBUG, "Waited %" PRId64 " ms", waited_ms);
 606   // We should have waited at least kMinReconnectBackOffMs. We substract one to
 607   // account for test and precision accuracy drift.
 608   EXPECT_GE(waited_ms, kMinReconnectBackOffMs - 1);
 609   gpr_atm_rel_store(&g_connection_delay_ms, 0);
 610 }
 611
 612 TEST_F(ClientLbEnd2endTest, PickFirstResetConnectionBackoff) {
 613   ChannelArguments args;
 614   constexpr int kInitialBackOffMs = 1000;
 615   args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
 616   const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
 617   auto response_generator = BuildResolverResponseGenerator();
 618   auto channel = BuildChannel("pick_first", response_generator, args);
 619   auto stub = BuildStub(channel);
 620   response_generator.SetNextResolution(ports);
 621   // The channel won't become connected (there's no server).
 622   EXPECT_FALSE(
 623       channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
 624   // Bring up a server on the chosen port.
 625   StartServers(1, ports);
 626   const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
 627   // Wait for connect, but not long enough.  This proves that we're
 628   // being throttled by initial backoff.
 629   EXPECT_FALSE(
 630       channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
 631   // Reset connection backoff.
 632   experimental::ChannelResetConnectionBackoff(channel.get());
 633   // Wait for connect.  Should happen as soon as the client connects to
 634   // the newly started server, which should be before the initial
 635   // backoff timeout elapses.
 636   EXPECT_TRUE(
 637       channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(20)));
 638   const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
 639   const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
 640   gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
 641   // We should have waited less than kInitialBackOffMs.
 642   EXPECT_LT(waited_ms, kInitialBackOffMs);
 643 }
 644
 645 TEST_F(ClientLbEnd2endTest,
 646        PickFirstResetConnectionBackoffNextAttemptStartsImmediately) {
 647   ChannelArguments args;
 648   constexpr int kInitialBackOffMs = 1000;
 649   args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
 650   const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
 651   auto response_generator = BuildResolverResponseGenerator();
 652   auto channel = BuildChannel("pick_first", response_generator, args);
 653   auto stub = BuildStub(channel);
 654   response_generator.SetNextResolution(ports);
 655   // Wait for connect, which should fail ~immediately, because the server
 656   // is not up.
 657   gpr_log(GPR_INFO, "=== INITIAL CONNECTION ATTEMPT");
 658   EXPECT_FALSE(
 659       channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
 660   // Reset connection backoff.
 661   // Note that the time at which the third attempt will be started is
 662   // actually computed at this point, so we record the start time here.
 663   gpr_log(GPR_INFO, "=== RESETTING BACKOFF");
 664   const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
 665   experimental::ChannelResetConnectionBackoff(channel.get());
 666   // Trigger a second connection attempt.  This should also fail
 667   // ~immediately, but the retry should be scheduled for
 668   // kInitialBackOffMs instead of applying the multiplier.
 669   gpr_log(GPR_INFO, "=== POLLING FOR SECOND CONNECTION ATTEMPT");
 670   EXPECT_FALSE(
 671       channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
 672   // Bring up a server on the chosen port.
 673   gpr_log(GPR_INFO, "=== STARTING BACKEND");
 674   StartServers(1, ports);
 675   // Wait for connect.  Should happen within kInitialBackOffMs.
 676   // Give an extra 100ms to account for the time spent in the second and
 677   // third connection attempts themselves (since what we really want to
 678   // measure is the time between the two).  As long as this is less than
 679   // the 1.6x increase we would see if the backoff state was not reset
 680   // properly, the test is still proving that the backoff was reset.
 681   constexpr int kWaitMs = kInitialBackOffMs + 100;
 682   gpr_log(GPR_INFO, "=== POLLING FOR THIRD CONNECTION ATTEMPT");
 683   EXPECT_TRUE(channel->WaitForConnected(
 684       grpc_timeout_milliseconds_to_deadline(kWaitMs)));
 685   const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
 686   const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
 687   gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
 688   EXPECT_LT(waited_ms, kWaitMs);
 689 }
 690
 691 TEST_F(ClientLbEnd2endTest, PickFirstUpdates) {
 692   // Start servers and send one RPC per server.
 693   const int kNumServers = 3;
 694   StartServers(kNumServers);
 695   auto response_generator = BuildResolverResponseGenerator();
 696   auto channel = BuildChannel("pick_first", response_generator);
 697   auto stub = BuildStub(channel);
 698
 699   std::vector<int> ports;
 700
 701   // Perform one RPC against the first server.
 702   ports.emplace_back(servers_[0]->port_);
 703   response_generator.SetNextResolution(ports);
 704   gpr_log(GPR_INFO, "****** SET [0] *******");
 705   CheckRpcSendOk(stub, DEBUG_LOCATION);
 706   EXPECT_EQ(servers_[0]->service_.request_count(), 1);
 707
 708   // An empty update will result in the channel going into TRANSIENT_FAILURE.
 709   ports.clear();
 710   response_generator.SetNextResolution(ports);
 711   gpr_log(GPR_INFO, "****** SET none *******");
 712   grpc_connectivity_state channel_state;
 713   do {
 714     channel_state = channel->GetState(true /* try to connect */);
 715   } while (channel_state == GRPC_CHANNEL_READY);
 716   ASSERT_NE(channel_state, GRPC_CHANNEL_READY);
 717   servers_[0]->service_.ResetCounters();
 718
 719   // Next update introduces servers_[1], making the channel recover.
 720   ports.clear();
 721   ports.emplace_back(servers_[1]->port_);
 722   response_generator.SetNextResolution(ports);
 723   gpr_log(GPR_INFO, "****** SET [1] *******");
 724   WaitForServer(stub, 1, DEBUG_LOCATION);
 725   EXPECT_EQ(servers_[0]->service_.request_count(), 0);
 726
 727   // And again for servers_[2]
 728   ports.clear();
 729   ports.emplace_back(servers_[2]->port_);
 730   response_generator.SetNextResolution(ports);
 731   gpr_log(GPR_INFO, "****** SET [2] *******");
 732   WaitForServer(stub, 2, DEBUG_LOCATION);
 733   EXPECT_EQ(servers_[0]->service_.request_count(), 0);
 734   EXPECT_EQ(servers_[1]->service_.request_count(), 0);
 735
 736   // Check LB policy name for the channel.
 737   EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
 738 }
 739
 740 TEST_F(ClientLbEnd2endTest, PickFirstUpdateSuperset) {
 741   // Start servers and send one RPC per server.
 742   const int kNumServers = 3;
 743   StartServers(kNumServers);
 744   auto response_generator = BuildResolverResponseGenerator();
 745   auto channel = BuildChannel("pick_first", response_generator);
 746   auto stub = BuildStub(channel);
 747
 748   std::vector<int> ports;
 749
 750   // Perform one RPC against the first server.
 751   ports.emplace_back(servers_[0]->port_);
 752   response_generator.SetNextResolution(ports);
 753   gpr_log(GPR_INFO, "****** SET [0] *******");
 754   CheckRpcSendOk(stub, DEBUG_LOCATION);
 755   EXPECT_EQ(servers_[0]->service_.request_count(), 1);
 756   servers_[0]->service_.ResetCounters();
 757
 758   // Send and superset update
 759   ports.clear();
 760   ports.emplace_back(servers_[1]->port_);
 761   ports.emplace_back(servers_[0]->port_);
 762   response_generator.SetNextResolution(ports);
 763   gpr_log(GPR_INFO, "****** SET superset *******");
 764   CheckRpcSendOk(stub, DEBUG_LOCATION);
 765   // We stick to the previously connected server.
 766   WaitForServer(stub, 0, DEBUG_LOCATION);
 767   EXPECT_EQ(0, servers_[1]->service_.request_count());
 768
 769   // Check LB policy name for the channel.
 770   EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
 771 }
 772
 773 TEST_F(ClientLbEnd2endTest, PickFirstGlobalSubchannelPool) {
 774   // Start one server.
 775   const int kNumServers = 1;
 776   StartServers(kNumServers);
 777   std::vector<int> ports = GetServersPorts();
 778   // Create two channels that (by default) use the global subchannel pool.
 779   auto response_generator1 = BuildResolverResponseGenerator();
 780   auto channel1 = BuildChannel("pick_first", response_generator1);
 781   auto stub1 = BuildStub(channel1);
 782   response_generator1.SetNextResolution(ports);
 783   auto response_generator2 = BuildResolverResponseGenerator();
 784   auto channel2 = BuildChannel("pick_first", response_generator2);
 785   auto stub2 = BuildStub(channel2);
 786   response_generator2.SetNextResolution(ports);
 787   WaitForServer(stub1, 0, DEBUG_LOCATION);
 788   // Send one RPC on each channel.
 789   CheckRpcSendOk(stub1, DEBUG_LOCATION);
 790   CheckRpcSendOk(stub2, DEBUG_LOCATION);
 791   // The server receives two requests.
 792   EXPECT_EQ(2, servers_[0]->service_.request_count());
 793   // The two requests are from the same client port, because the two channels
 794   // share subchannels via the global subchannel pool.
 795   EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
 796 }
 797
 798 TEST_F(ClientLbEnd2endTest, PickFirstLocalSubchannelPool) {
 799   // Start one server.
 800   const int kNumServers = 1;
 801   StartServers(kNumServers);
 802   std::vector<int> ports = GetServersPorts();
 803   // Create two channels that use local subchannel pool.
 804   ChannelArguments args;
 805   args.SetInt(GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL, 1);
 806   auto response_generator1 = BuildResolverResponseGenerator();
 807   auto channel1 = BuildChannel("pick_first", response_generator1, args);
 808   auto stub1 = BuildStub(channel1);
 809   response_generator1.SetNextResolution(ports);
 810   auto response_generator2 = BuildResolverResponseGenerator();
 811   auto channel2 = BuildChannel("pick_first", response_generator2, args);
 812   auto stub2 = BuildStub(channel2);
 813   response_generator2.SetNextResolution(ports);
 814   WaitForServer(stub1, 0, DEBUG_LOCATION);
 815   // Send one RPC on each channel.
 816   CheckRpcSendOk(stub1, DEBUG_LOCATION);
 817   CheckRpcSendOk(stub2, DEBUG_LOCATION);
 818   // The server receives two requests.
 819   EXPECT_EQ(2, servers_[0]->service_.request_count());
 820   // The two requests are from two client ports, because the two channels didn't
 821   // share subchannels with each other.
 822   EXPECT_EQ(2UL, servers_[0]->service_.clients().size());
 823 }
 824
 825 TEST_F(ClientLbEnd2endTest, PickFirstManyUpdates) {
 826   const int kNumUpdates = 1000;
 827   const int kNumServers = 3;
 828   StartServers(kNumServers);
 829   auto response_generator = BuildResolverResponseGenerator();
 830   auto channel = BuildChannel("pick_first", response_generator);
 831   auto stub = BuildStub(channel);
 832   std::vector<int> ports = GetServersPorts();
 833   for (size_t i = 0; i < kNumUpdates; ++i) {
 834     std::shuffle(ports.begin(), ports.end(),
 835                  std::mt19937(std::random_device()()));
 836     response_generator.SetNextResolution(ports);
 837     // We should re-enter core at the end of the loop to give the resolution
 838     // setting closure a chance to run.
 839     if ((i + 1) % 10 == 0) CheckRpcSendOk(stub, DEBUG_LOCATION);
 840   }
 841   // Check LB policy name for the channel.
 842   EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
 843 }
 844
 845 TEST_F(ClientLbEnd2endTest, PickFirstReresolutionNoSelected) {
 846   // Prepare the ports for up servers and down servers.
 847   const int kNumServers = 3;
 848   const int kNumAliveServers = 1;
 849   StartServers(kNumAliveServers);
 850   std::vector<int> alive_ports, dead_ports;
 851   for (size_t i = 0; i < kNumServers; ++i) {
 852     if (i < kNumAliveServers) {
 853       alive_ports.emplace_back(servers_[i]->port_);
 854     } else {
 855       dead_ports.emplace_back(grpc_pick_unused_port_or_die());
 856     }
 857   }
 858   auto response_generator = BuildResolverResponseGenerator();
 859   auto channel = BuildChannel("pick_first", response_generator);
 860   auto stub = BuildStub(channel);
 861   // The initial resolution only contains dead ports. There won't be any
 862   // selected subchannel. Re-resolution will return the same result.
 863   response_generator.SetNextResolution(dead_ports);
 864   gpr_log(GPR_INFO, "****** INITIAL RESOLUTION SET *******");
 865   for (size_t i = 0; i < 10; ++i) CheckRpcSendFailure(stub);
 866   // Set a re-resolution result that contains reachable ports, so that the
 867   // pick_first LB policy can recover soon.
 868   response_generator.SetNextResolutionUponError(alive_ports);
 869   gpr_log(GPR_INFO, "****** RE-RESOLUTION SET *******");
 870   WaitForServer(stub, 0, DEBUG_LOCATION, true /* ignore_failure */);
 871   CheckRpcSendOk(stub, DEBUG_LOCATION);
 872   EXPECT_EQ(servers_[0]->service_.request_count(), 1);
 873   // Check LB policy name for the channel.
 874   EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
 875 }
 876
 877 TEST_F(ClientLbEnd2endTest, PickFirstReconnectWithoutNewResolverResult) {
 878   std::vector<int> ports = {grpc_pick_unused_port_or_die()};
 879   StartServers(1, ports);
 880   auto response_generator = BuildResolverResponseGenerator();
 881   auto channel = BuildChannel("pick_first", response_generator);
 882   auto stub = BuildStub(channel);
 883   response_generator.SetNextResolution(ports);
 884   gpr_log(GPR_INFO, "****** INITIAL CONNECTION *******");
 885   WaitForServer(stub, 0, DEBUG_LOCATION);
 886   gpr_log(GPR_INFO, "****** STOPPING SERVER ******");
 887   servers_[0]->Shutdown();
 888   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
 889   gpr_log(GPR_INFO, "****** RESTARTING SERVER ******");
 890   StartServers(1, ports);
 891   WaitForServer(stub, 0, DEBUG_LOCATION);
 892 }
 893
 894 TEST_F(ClientLbEnd2endTest,
 895        PickFirstReconnectWithoutNewResolverResultStartsFromTopOfList) {
 896   std::vector<int> ports = {grpc_pick_unused_port_or_die(),
 897                             grpc_pick_unused_port_or_die()};
 898   CreateServers(2, ports);
 899   StartServer(1);
 900   auto response_generator = BuildResolverResponseGenerator();
 901   auto channel = BuildChannel("pick_first", response_generator);
 902   auto stub = BuildStub(channel);
 903   response_generator.SetNextResolution(ports);
 904   gpr_log(GPR_INFO, "****** INITIAL CONNECTION *******");
 905   WaitForServer(stub, 1, DEBUG_LOCATION);
 906   gpr_log(GPR_INFO, "****** STOPPING SERVER ******");
 907   servers_[1]->Shutdown();
 908   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
 909   gpr_log(GPR_INFO, "****** STARTING BOTH SERVERS ******");
 910   StartServers(2, ports);
 911   WaitForServer(stub, 0, DEBUG_LOCATION);
 912 }
 913
 914 TEST_F(ClientLbEnd2endTest, PickFirstCheckStateBeforeStartWatch) {
 915   std::vector<int> ports = {grpc_pick_unused_port_or_die()};
 916   StartServers(1, ports);
 917   auto response_generator = BuildResolverResponseGenerator();
 918   auto channel_1 = BuildChannel("pick_first", response_generator);
 919   auto stub_1 = BuildStub(channel_1);
 920   response_generator.SetNextResolution(ports);
 921   gpr_log(GPR_INFO, "****** RESOLUTION SET FOR CHANNEL 1 *******");
 922   WaitForServer(stub_1, 0, DEBUG_LOCATION);
 923   gpr_log(GPR_INFO, "****** CHANNEL 1 CONNECTED *******");
 924   servers_[0]->Shutdown();
 925   // Channel 1 will receive a re-resolution containing the same server. It will
 926   // create a new subchannel and hold a ref to it.
 927   StartServers(1, ports);
 928   gpr_log(GPR_INFO, "****** SERVER RESTARTED *******");
 929   auto response_generator_2 = BuildResolverResponseGenerator();
 930   auto channel_2 = BuildChannel("pick_first", response_generator_2);
 931   auto stub_2 = BuildStub(channel_2);
 932   response_generator_2.SetNextResolution(ports);
 933   gpr_log(GPR_INFO, "****** RESOLUTION SET FOR CHANNEL 2 *******");
 934   WaitForServer(stub_2, 0, DEBUG_LOCATION, true);
 935   gpr_log(GPR_INFO, "****** CHANNEL 2 CONNECTED *******");
 936   servers_[0]->Shutdown();
 937   // Wait until the disconnection has triggered the connectivity notification.
 938   // Otherwise, the subchannel may be picked for next call but will fail soon.
 939   EXPECT_TRUE(WaitForChannelNotReady(channel_2.get()));
 940   // Channel 2 will also receive a re-resolution containing the same server.
 941   // Both channels will ref the same subchannel that failed.
 942   StartServers(1, ports);
 943   gpr_log(GPR_INFO, "****** SERVER RESTARTED AGAIN *******");
 944   gpr_log(GPR_INFO, "****** CHANNEL 2 STARTING A CALL *******");
 945   // The first call after the server restart will succeed.
 946   CheckRpcSendOk(stub_2, DEBUG_LOCATION);
 947   gpr_log(GPR_INFO, "****** CHANNEL 2 FINISHED A CALL *******");
 948   // Check LB policy name for the channel.
 949   EXPECT_EQ("pick_first", channel_1->GetLoadBalancingPolicyName());
 950   // Check LB policy name for the channel.
 951   EXPECT_EQ("pick_first", channel_2->GetLoadBalancingPolicyName());
 952 }
 953
 954 TEST_F(ClientLbEnd2endTest, PickFirstIdleOnDisconnect) {
 955   // Start server, send RPC, and make sure channel is READY.
 956   const int kNumServers = 1;
 957   StartServers(kNumServers);
 958   auto response_generator = BuildResolverResponseGenerator();
 959   auto channel =
 960       BuildChannel("", response_generator);  // pick_first is the default.
 961   auto stub = BuildStub(channel);
 962   response_generator.SetNextResolution(GetServersPorts());
 963   CheckRpcSendOk(stub, DEBUG_LOCATION);
 964   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 965   // Stop server.  Channel should go into state IDLE.
 966   response_generator.SetFailureOnReresolution();
 967   servers_[0]->Shutdown();
 968   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
 969   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
 970   servers_.clear();
 971 }
 972
 973 TEST_F(ClientLbEnd2endTest, PickFirstPendingUpdateAndSelectedSubchannelFails) {
 974   auto response_generator = BuildResolverResponseGenerator();
 975   auto channel =
 976       BuildChannel("", response_generator);  // pick_first is the default.
 977   auto stub = BuildStub(channel);
 978   // Create a number of servers, but only start 1 of them.
 979   CreateServers(10);
 980   StartServer(0);
 981   // Initially resolve to first server and make sure it connects.
 982   gpr_log(GPR_INFO, "Phase 1: Connect to first server.");
 983   response_generator.SetNextResolution({servers_[0]->port_});
 984   CheckRpcSendOk(stub, DEBUG_LOCATION, true /* wait_for_ready */);
 985   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 986   // Send a resolution update with the remaining servers, none of which are
 987   // running yet, so the update will stay pending.  Note that it's important
 988   // to have multiple servers here, or else the test will be flaky; with only
 989   // one server, the pending subchannel list has already gone into
 990   // TRANSIENT_FAILURE due to hitting the end of the list by the time we
 991   // check the state.
 992   gpr_log(GPR_INFO,
 993           "Phase 2: Resolver update pointing to remaining "
 994           "(not started) servers.");
 995   response_generator.SetNextResolution(GetServersPorts(1 /* start_index */));
 996   // RPCs will continue to be sent to the first server.
 997   CheckRpcSendOk(stub, DEBUG_LOCATION);
 998   // Now stop the first server, so that the current subchannel list
 999   // fails.  This should cause us to immediately swap over to the
1000   // pending list, even though it's not yet connected.  The state should
1001   // be set to CONNECTING, since that's what the pending subchannel list
1002   // was doing when we swapped over.
1003   gpr_log(GPR_INFO, "Phase 3: Stopping first server.");
1004   servers_[0]->Shutdown();
1005   WaitForChannelNotReady(channel.get());
1006   // TODO(roth): This should always return CONNECTING, but it's flaky
1007   // between that and TRANSIENT_FAILURE.  I suspect that this problem
1008   // will go away once we move the backoff code out of the subchannel
1009   // and into the LB policies.
1010   EXPECT_THAT(channel->GetState(false),
1011               ::testing::AnyOf(GRPC_CHANNEL_CONNECTING,
1012                                GRPC_CHANNEL_TRANSIENT_FAILURE));
1013   // Now start the second server.
1014   gpr_log(GPR_INFO, "Phase 4: Starting second server.");
1015   StartServer(1);
1016   // The channel should go to READY state and RPCs should go to the
1017   // second server.
1018   WaitForChannelReady(channel.get());
1019   WaitForServer(stub, 1, DEBUG_LOCATION, true /* ignore_failure */);
1020 }
1021
1022 TEST_F(ClientLbEnd2endTest, PickFirstStaysIdleUponEmptyUpdate) {
1023   // Start server, send RPC, and make sure channel is READY.
1024   const int kNumServers = 1;
1025   StartServers(kNumServers);
1026   auto response_generator = BuildResolverResponseGenerator();
1027   auto channel =
1028       BuildChannel("", response_generator);  // pick_first is the default.
1029   auto stub = BuildStub(channel);
1030   response_generator.SetNextResolution(GetServersPorts());
1031   CheckRpcSendOk(stub, DEBUG_LOCATION);
1032   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1033   // Stop server.  Channel should go into state IDLE.
1034   servers_[0]->Shutdown();
1035   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
1036   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
1037   // Now send resolver update that includes no addresses.  Channel
1038   // should stay in state IDLE.
1039   response_generator.SetNextResolution({});
1040   EXPECT_FALSE(channel->WaitForStateChange(
1041       GRPC_CHANNEL_IDLE, grpc_timeout_seconds_to_deadline(3)));
1042   // Now bring the backend back up and send a non-empty resolver update,
1043   // and then try to send an RPC.  Channel should go back into state READY.
1044   StartServer(0);
1045   response_generator.SetNextResolution(GetServersPorts());
1046   CheckRpcSendOk(stub, DEBUG_LOCATION);
1047   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1048 }
1049
1050 TEST_F(ClientLbEnd2endTest, RoundRobin) {
1051   // Start servers and send one RPC per server.
1052   const int kNumServers = 3;
1053   StartServers(kNumServers);
1054   auto response_generator = BuildResolverResponseGenerator();
1055   auto channel = BuildChannel("round_robin", response_generator);
1056   auto stub = BuildStub(channel);
1057   response_generator.SetNextResolution(GetServersPorts());
1058   // Wait until all backends are ready.
1059   do {
1060     CheckRpcSendOk(stub, DEBUG_LOCATION);
1061   } while (!SeenAllServers());
1062   ResetCounters();
1063   // "Sync" to the end of the list. Next sequence of picks will start at the
1064   // first server (index 0).
1065   WaitForServer(stub, servers_.size() - 1, DEBUG_LOCATION);
1066   std::vector<int> connection_order;
1067   for (size_t i = 0; i < servers_.size(); ++i) {
1068     CheckRpcSendOk(stub, DEBUG_LOCATION);
1069     UpdateConnectionOrder(servers_, &connection_order);
1070   }
1071   // Backends should be iterated over in the order in which the addresses were
1072   // given.
1073   const auto expected = std::vector<int>{0, 1, 2};
1074   EXPECT_EQ(expected, connection_order);
1075   // Check LB policy name for the channel.
1076   EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
1077 }
1078
1079 TEST_F(ClientLbEnd2endTest, RoundRobinProcessPending) {
1080   StartServers(1);  // Single server
1081   auto response_generator = BuildResolverResponseGenerator();
1082   auto channel = BuildChannel("round_robin", response_generator);
1083   auto stub = BuildStub(channel);
1084   response_generator.SetNextResolution({servers_[0]->port_});
1085   WaitForServer(stub, 0, DEBUG_LOCATION);
1086   // Create a new channel and its corresponding RR LB policy, which will pick
1087   // the subchannels in READY state from the previous RPC against the same
1088   // target (even if it happened over a different channel, because subchannels
1089   // are globally reused). Progress should happen without any transition from
1090   // this READY state.
1091   auto second_response_generator = BuildResolverResponseGenerator();
1092   auto second_channel = BuildChannel("round_robin", second_response_generator);
1093   auto second_stub = BuildStub(second_channel);
1094   second_response_generator.SetNextResolution({servers_[0]->port_});
1095   CheckRpcSendOk(second_stub, DEBUG_LOCATION);
1096 }
1097
1098 TEST_F(ClientLbEnd2endTest, RoundRobinUpdates) {
1099   // Start servers and send one RPC per server.
1100   const int kNumServers = 3;
1101   StartServers(kNumServers);
1102   auto response_generator = BuildResolverResponseGenerator();
1103   auto channel = BuildChannel("round_robin", response_generator);
1104   auto stub = BuildStub(channel);
1105   std::vector<int> ports;
1106   // Start with a single server.
1107   gpr_log(GPR_INFO, "*** FIRST BACKEND ***");
1108   ports.emplace_back(servers_[0]->port_);
1109   response_generator.SetNextResolution(ports);
1110   WaitForServer(stub, 0, DEBUG_LOCATION);
1111   // Send RPCs. They should all go servers_[0]
1112   for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1113   EXPECT_EQ(10, servers_[0]->service_.request_count());
1114   EXPECT_EQ(0, servers_[1]->service_.request_count());
1115   EXPECT_EQ(0, servers_[2]->service_.request_count());
1116   servers_[0]->service_.ResetCounters();
1117   // And now for the second server.
1118   gpr_log(GPR_INFO, "*** SECOND BACKEND ***");
1119   ports.clear();
1120   ports.emplace_back(servers_[1]->port_);
1121   response_generator.SetNextResolution(ports);
1122   // Wait until update has been processed, as signaled by the second backend
1123   // receiving a request.
1124   EXPECT_EQ(0, servers_[1]->service_.request_count());
1125   WaitForServer(stub, 1, DEBUG_LOCATION);
1126   for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1127   EXPECT_EQ(0, servers_[0]->service_.request_count());
1128   EXPECT_EQ(10, servers_[1]->service_.request_count());
1129   EXPECT_EQ(0, servers_[2]->service_.request_count());
1130   servers_[1]->service_.ResetCounters();
1131   // ... and for the last server.
1132   gpr_log(GPR_INFO, "*** THIRD BACKEND ***");
1133   ports.clear();
1134   ports.emplace_back(servers_[2]->port_);
1135   response_generator.SetNextResolution(ports);
1136   WaitForServer(stub, 2, DEBUG_LOCATION);
1137   for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1138   EXPECT_EQ(0, servers_[0]->service_.request_count());
1139   EXPECT_EQ(0, servers_[1]->service_.request_count());
1140   EXPECT_EQ(10, servers_[2]->service_.request_count());
1141   servers_[2]->service_.ResetCounters();
1142   // Back to all servers.
1143   gpr_log(GPR_INFO, "*** ALL BACKENDS ***");
1144   ports.clear();
1145   ports.emplace_back(servers_[0]->port_);
1146   ports.emplace_back(servers_[1]->port_);
1147   ports.emplace_back(servers_[2]->port_);
1148   response_generator.SetNextResolution(ports);
1149   WaitForServer(stub, 0, DEBUG_LOCATION);
1150   WaitForServer(stub, 1, DEBUG_LOCATION);
1151   WaitForServer(stub, 2, DEBUG_LOCATION);
1152   // Send three RPCs, one per server.
1153   for (size_t i = 0; i < 3; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1154   EXPECT_EQ(1, servers_[0]->service_.request_count());
1155   EXPECT_EQ(1, servers_[1]->service_.request_count());
1156   EXPECT_EQ(1, servers_[2]->service_.request_count());
1157   // An empty update will result in the channel going into TRANSIENT_FAILURE.
1158   gpr_log(GPR_INFO, "*** NO BACKENDS ***");
1159   ports.clear();
1160   response_generator.SetNextResolution(ports);
1161   grpc_connectivity_state channel_state;
1162   do {
1163     channel_state = channel->GetState(true /* try to connect */);
1164   } while (channel_state == GRPC_CHANNEL_READY);
1165   ASSERT_NE(channel_state, GRPC_CHANNEL_READY);
1166   servers_[0]->service_.ResetCounters();
1167   // Next update introduces servers_[1], making the channel recover.
1168   gpr_log(GPR_INFO, "*** BACK TO SECOND BACKEND ***");
1169   ports.clear();
1170   ports.emplace_back(servers_[1]->port_);
1171   response_generator.SetNextResolution(ports);
1172   WaitForServer(stub, 1, DEBUG_LOCATION);
1173   channel_state = channel->GetState(false /* try to connect */);
1174   ASSERT_EQ(channel_state, GRPC_CHANNEL_READY);
1175   // Check LB policy name for the channel.
1176   EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
1177 }
1178
1179 TEST_F(ClientLbEnd2endTest, RoundRobinUpdateInError) {
1180   const int kNumServers = 3;
1181   StartServers(kNumServers);
1182   auto response_generator = BuildResolverResponseGenerator();
1183   auto channel = BuildChannel("round_robin", response_generator);
1184   auto stub = BuildStub(channel);
1185   std::vector<int> ports;
1186   // Start with a single server.
1187   ports.emplace_back(servers_[0]->port_);
1188   response_generator.SetNextResolution(ports);
1189   WaitForServer(stub, 0, DEBUG_LOCATION);
1190   // Send RPCs. They should all go to servers_[0]
1191   for (size_t i = 0; i < 10; ++i) SendRpc(stub);
1192   EXPECT_EQ(10, servers_[0]->service_.request_count());
1193   EXPECT_EQ(0, servers_[1]->service_.request_count());
1194   EXPECT_EQ(0, servers_[2]->service_.request_count());
1195   servers_[0]->service_.ResetCounters();
1196   // Shutdown one of the servers to be sent in the update.
1197   servers_[1]->Shutdown();
1198   ports.emplace_back(servers_[1]->port_);
1199   ports.emplace_back(servers_[2]->port_);
1200   response_generator.SetNextResolution(ports);
1201   WaitForServer(stub, 0, DEBUG_LOCATION);
1202   WaitForServer(stub, 2, DEBUG_LOCATION);
1203   // Send three RPCs, one per server.
1204   for (size_t i = 0; i < kNumServers; ++i) SendRpc(stub);
1205   // The server in shutdown shouldn't receive any.
1206   EXPECT_EQ(0, servers_[1]->service_.request_count());
1207 }
1208
1209 TEST_F(ClientLbEnd2endTest, RoundRobinManyUpdates) {
1210   // Start servers and send one RPC per server.
1211   const int kNumServers = 3;
1212   StartServers(kNumServers);
1213   auto response_generator = BuildResolverResponseGenerator();
1214   auto channel = BuildChannel("round_robin", response_generator);
1215   auto stub = BuildStub(channel);
1216   std::vector<int> ports = GetServersPorts();
1217   for (size_t i = 0; i < 1000; ++i) {
1218     std::shuffle(ports.begin(), ports.end(),
1219                  std::mt19937(std::random_device()()));
1220     response_generator.SetNextResolution(ports);
1221     if (i % 10 == 0) CheckRpcSendOk(stub, DEBUG_LOCATION);
1222   }
1223   // Check LB policy name for the channel.
1224   EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
1225 }
1226
1227 TEST_F(ClientLbEnd2endTest, RoundRobinConcurrentUpdates) {
1228   // TODO(dgq): replicate the way internal testing exercises the concurrent
1229   // update provisions of RR.
1230 }
1231
1232 TEST_F(ClientLbEnd2endTest, RoundRobinReresolve) {
1233   // Start servers and send one RPC per server.
1234   const int kNumServers = 3;
1235   std::vector<int> first_ports;
1236   std::vector<int> second_ports;
1237   first_ports.reserve(kNumServers);
1238   for (int i = 0; i < kNumServers; ++i) {
1239     first_ports.push_back(grpc_pick_unused_port_or_die());
1240   }
1241   second_ports.reserve(kNumServers);
1242   for (int i = 0; i < kNumServers; ++i) {
1243     second_ports.push_back(grpc_pick_unused_port_or_die());
1244   }
1245   StartServers(kNumServers, first_ports);
1246   auto response_generator = BuildResolverResponseGenerator();
1247   auto channel = BuildChannel("round_robin", response_generator);
1248   auto stub = BuildStub(channel);
1249   response_generator.SetNextResolution(first_ports);
1250   // Send a number of RPCs, which succeed.
1251   for (size_t i = 0; i < 100; ++i) {
1252     CheckRpcSendOk(stub, DEBUG_LOCATION);
1253   }
1254   // Kill all servers
1255   gpr_log(GPR_INFO, "****** ABOUT TO KILL SERVERS *******");
1256   for (size_t i = 0; i < servers_.size(); ++i) {
1257     servers_[i]->Shutdown();
1258   }
1259   gpr_log(GPR_INFO, "****** SERVERS KILLED *******");
1260   gpr_log(GPR_INFO, "****** SENDING DOOMED REQUESTS *******");
1261   // Client requests should fail. Send enough to tickle all subchannels.
1262   for (size_t i = 0; i < servers_.size(); ++i) CheckRpcSendFailure(stub);
1263   gpr_log(GPR_INFO, "****** DOOMED REQUESTS SENT *******");
1264   // Bring servers back up on a different set of ports. We need to do this to be
1265   // sure that the eventual success is *not* due to subchannel reconnection
1266   // attempts and that an actual re-resolution has happened as a result of the
1267   // RR policy going into transient failure when all its subchannels become
1268   // unavailable (in transient failure as well).
1269   gpr_log(GPR_INFO, "****** RESTARTING SERVERS *******");
1270   StartServers(kNumServers, second_ports);
1271   // Don't notify of the update. Wait for the LB policy's re-resolution to
1272   // "pull" the new ports.
1273   response_generator.SetNextResolutionUponError(second_ports);
1274   gpr_log(GPR_INFO, "****** SERVERS RESTARTED *******");
1275   gpr_log(GPR_INFO, "****** SENDING REQUEST TO SUCCEED *******");
1276   // Client request should eventually (but still fairly soon) succeed.
1277   const gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
1278   gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
1279   while (gpr_time_cmp(deadline, now) > 0) {
1280     if (SendRpc(stub)) break;
1281     now = gpr_now(GPR_CLOCK_MONOTONIC);
1282   }
1283   ASSERT_GT(gpr_time_cmp(deadline, now), 0);
1284 }
1285
1286 TEST_F(ClientLbEnd2endTest, RoundRobinTransientFailure) {
1287   // Start servers and create channel.  Channel should go to READY state.
1288   const int kNumServers = 3;
1289   StartServers(kNumServers);
1290   auto response_generator = BuildResolverResponseGenerator();
1291   auto channel = BuildChannel("round_robin", response_generator);
1292   auto stub = BuildStub(channel);
1293   response_generator.SetNextResolution(GetServersPorts());
1294   EXPECT_TRUE(WaitForChannelReady(channel.get()));
1295   // Now kill the servers.  The channel should transition to TRANSIENT_FAILURE.
1296   // TODO(roth): This test should ideally check that even when the
1297   // subchannels are in state CONNECTING for an extended period of time,
1298   // we will still report TRANSIENT_FAILURE.  Unfortunately, we don't
1299   // currently have a good way to get a subchannel to report CONNECTING
1300   // for a long period of time, since the servers in this test framework
1301   // are on the loopback interface, which will immediately return a
1302   // "Connection refused" error, so the subchannels will only be in
1303   // CONNECTING state very briefly.  When we have time, see if we can
1304   // find a way to fix this.
1305   for (size_t i = 0; i < servers_.size(); ++i) {
1306     servers_[i]->Shutdown();
1307   }
1308   auto predicate = [](grpc_connectivity_state state) {
1309     return state == GRPC_CHANNEL_TRANSIENT_FAILURE;
1310   };
1311   EXPECT_TRUE(WaitForChannelState(channel.get(), predicate));
1312 }
1313
1314 TEST_F(ClientLbEnd2endTest, RoundRobinTransientFailureAtStartup) {
1315   // Create channel and return servers that don't exist.  Channel should
1316   // quickly transition into TRANSIENT_FAILURE.
1317   // TODO(roth): This test should ideally check that even when the
1318   // subchannels are in state CONNECTING for an extended period of time,
1319   // we will still report TRANSIENT_FAILURE.  Unfortunately, we don't
1320   // currently have a good way to get a subchannel to report CONNECTING
1321   // for a long period of time, since the servers in this test framework
1322   // are on the loopback interface, which will immediately return a
1323   // "Connection refused" error, so the subchannels will only be in
1324   // CONNECTING state very briefly.  When we have time, see if we can
1325   // find a way to fix this.
1326   auto response_generator = BuildResolverResponseGenerator();
1327   auto channel = BuildChannel("round_robin", response_generator);
1328   auto stub = BuildStub(channel);
1329   response_generator.SetNextResolution({
1330       grpc_pick_unused_port_or_die(),
1331       grpc_pick_unused_port_or_die(),
1332       grpc_pick_unused_port_or_die(),
1333   });
1334   for (size_t i = 0; i < servers_.size(); ++i) {
1335     servers_[i]->Shutdown();
1336   }
1337   auto predicate = [](grpc_connectivity_state state) {
1338     return state == GRPC_CHANNEL_TRANSIENT_FAILURE;
1339   };
1340   EXPECT_TRUE(WaitForChannelState(channel.get(), predicate, true));
1341 }
1342
1343 TEST_F(ClientLbEnd2endTest, RoundRobinSingleReconnect) {
1344   const int kNumServers = 3;
1345   StartServers(kNumServers);
1346   const auto ports = GetServersPorts();
1347   auto response_generator = BuildResolverResponseGenerator();
1348   auto channel = BuildChannel("round_robin", response_generator);
1349   auto stub = BuildStub(channel);
1350   response_generator.SetNextResolution(ports);
1351   for (size_t i = 0; i < kNumServers; ++i) {
1352     WaitForServer(stub, i, DEBUG_LOCATION);
1353   }
1354   for (size_t i = 0; i < servers_.size(); ++i) {
1355     CheckRpcSendOk(stub, DEBUG_LOCATION);
1356     EXPECT_EQ(1, servers_[i]->service_.request_count()) << "for backend #" << i;
1357   }
1358   // One request should have gone to each server.
1359   for (size_t i = 0; i < servers_.size(); ++i) {
1360     EXPECT_EQ(1, servers_[i]->service_.request_count());
1361   }
1362   const auto pre_death = servers_[0]->service_.request_count();
1363   // Kill the first server.
1364   servers_[0]->Shutdown();
1365   // Client request still succeed. May need retrying if RR had returned a pick
1366   // before noticing the change in the server's connectivity.
1367   while (!SendRpc(stub)) {
1368   }  // Retry until success.
1369   // Send a bunch of RPCs that should succeed.
1370   for (int i = 0; i < 10 * kNumServers; ++i) {
1371     CheckRpcSendOk(stub, DEBUG_LOCATION);
1372   }
1373   const auto post_death = servers_[0]->service_.request_count();
1374   // No requests have gone to the deceased server.
1375   EXPECT_EQ(pre_death, post_death);
1376   // Bring the first server back up.
1377   StartServer(0);
1378   // Requests should start arriving at the first server either right away (if
1379   // the server managed to start before the RR policy retried the subchannel) or
1380   // after the subchannel retry delay otherwise (RR's subchannel retried before
1381   // the server was fully back up).
1382   WaitForServer(stub, 0, DEBUG_LOCATION);
1383 }
1384
1385 // If health checking is required by client but health checking service
1386 // is not running on the server, the channel should be treated as healthy.
1387 TEST_F(ClientLbEnd2endTest,
1388        RoundRobinServersHealthCheckingUnimplementedTreatedAsHealthy) {
1389   StartServers(1);  // Single server
1390   ChannelArguments args;
1391   args.SetServiceConfigJSON(
1392       "{\"healthCheckConfig\": "
1393       "{\"serviceName\": \"health_check_service_name\"}}");
1394   auto response_generator = BuildResolverResponseGenerator();
1395   auto channel = BuildChannel("round_robin", response_generator, args);
1396   auto stub = BuildStub(channel);
1397   response_generator.SetNextResolution({servers_[0]->port_});
1398   EXPECT_TRUE(WaitForChannelReady(channel.get()));
1399   CheckRpcSendOk(stub, DEBUG_LOCATION);
1400 }
1401
1402 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthChecking) {
1403   EnableDefaultHealthCheckService(true);
1404   // Start servers.
1405   const int kNumServers = 3;
1406   StartServers(kNumServers);
1407   ChannelArguments args;
1408   args.SetServiceConfigJSON(
1409       "{\"healthCheckConfig\": "
1410       "{\"serviceName\": \"health_check_service_name\"}}");
1411   auto response_generator = BuildResolverResponseGenerator();
1412   auto channel = BuildChannel("round_robin", response_generator, args);
1413   auto stub = BuildStub(channel);
1414   response_generator.SetNextResolution(GetServersPorts());
1415   // Channel should not become READY, because health checks should be failing.
1416   gpr_log(GPR_INFO,
1417           "*** initial state: unknown health check service name for "
1418           "all servers");
1419   EXPECT_FALSE(WaitForChannelReady(channel.get(), 1));
1420   // Now set one of the servers to be healthy.
1421   // The channel should become healthy and all requests should go to
1422   // the healthy server.
1423   gpr_log(GPR_INFO, "*** server 0 healthy");
1424   servers_[0]->SetServingStatus("health_check_service_name", true);
1425   EXPECT_TRUE(WaitForChannelReady(channel.get()));
1426   for (int i = 0; i < 10; ++i) {
1427     CheckRpcSendOk(stub, DEBUG_LOCATION);
1428   }
1429   EXPECT_EQ(10, servers_[0]->service_.request_count());
1430   EXPECT_EQ(0, servers_[1]->service_.request_count());
1431   EXPECT_EQ(0, servers_[2]->service_.request_count());
1432   // Now set a second server to be healthy.
1433   gpr_log(GPR_INFO, "*** server 2 healthy");
1434   servers_[2]->SetServingStatus("health_check_service_name", true);
1435   WaitForServer(stub, 2, DEBUG_LOCATION);
1436   for (int i = 0; i < 10; ++i) {
1437     CheckRpcSendOk(stub, DEBUG_LOCATION);
1438   }
1439   EXPECT_EQ(5, servers_[0]->service_.request_count());
1440   EXPECT_EQ(0, servers_[1]->service_.request_count());
1441   EXPECT_EQ(5, servers_[2]->service_.request_count());
1442   // Now set the remaining server to be healthy.
1443   gpr_log(GPR_INFO, "*** server 1 healthy");
1444   servers_[1]->SetServingStatus("health_check_service_name", true);
1445   WaitForServer(stub, 1, DEBUG_LOCATION);
1446   for (int i = 0; i < 9; ++i) {
1447     CheckRpcSendOk(stub, DEBUG_LOCATION);
1448   }
1449   EXPECT_EQ(3, servers_[0]->service_.request_count());
1450   EXPECT_EQ(3, servers_[1]->service_.request_count());
1451   EXPECT_EQ(3, servers_[2]->service_.request_count());
1452   // Now set one server to be unhealthy again.  Then wait until the
1453   // unhealthiness has hit the client.  We know that the client will see
1454   // this when we send kNumServers requests and one of the remaining servers
1455   // sees two of the requests.
1456   gpr_log(GPR_INFO, "*** server 0 unhealthy");
1457   servers_[0]->SetServingStatus("health_check_service_name", false);
1458   do {
1459     ResetCounters();
1460     for (int i = 0; i < kNumServers; ++i) {
1461       CheckRpcSendOk(stub, DEBUG_LOCATION);
1462     }
1463   } while (servers_[1]->service_.request_count() != 2 &&
1464            servers_[2]->service_.request_count() != 2);
1465   // Now set the remaining two servers to be unhealthy.  Make sure the
1466   // channel leaves READY state and that RPCs fail.
1467   gpr_log(GPR_INFO, "*** all servers unhealthy");
1468   servers_[1]->SetServingStatus("health_check_service_name", false);
1469   servers_[2]->SetServingStatus("health_check_service_name", false);
1470   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
1471   CheckRpcSendFailure(stub);
1472   // Clean up.
1473   EnableDefaultHealthCheckService(false);
1474 }
1475
1476 TEST_F(ClientLbEnd2endTest,
1477        RoundRobinWithHealthCheckingHandlesSubchannelFailure) {
1478   EnableDefaultHealthCheckService(true);
1479   // Start servers.
1480   const int kNumServers = 3;
1481   StartServers(kNumServers);
1482   servers_[0]->SetServingStatus("health_check_service_name", true);
1483   servers_[1]->SetServingStatus("health_check_service_name", true);
1484   servers_[2]->SetServingStatus("health_check_service_name", true);
1485   ChannelArguments args;
1486   args.SetServiceConfigJSON(
1487       "{\"healthCheckConfig\": "
1488       "{\"serviceName\": \"health_check_service_name\"}}");
1489   auto response_generator = BuildResolverResponseGenerator();
1490   auto channel = BuildChannel("round_robin", response_generator, args);
1491   auto stub = BuildStub(channel);
1492   response_generator.SetNextResolution(GetServersPorts());
1493   WaitForServer(stub, 0, DEBUG_LOCATION);
1494   // Stop server 0 and send a new resolver result to ensure that RR
1495   // checks each subchannel's state.
1496   servers_[0]->Shutdown();
1497   response_generator.SetNextResolution(GetServersPorts());
1498   // Send a bunch more RPCs.
1499   for (size_t i = 0; i < 100; i++) {
1500     SendRpc(stub);
1501   }
1502 }
1503
1504 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthCheckingInhibitPerChannel) {
1505   EnableDefaultHealthCheckService(true);
1506   // Start server.
1507   const int kNumServers = 1;
1508   StartServers(kNumServers);
1509   // Create a channel with health-checking enabled.
1510   ChannelArguments args;
1511   args.SetServiceConfigJSON(
1512       "{\"healthCheckConfig\": "
1513       "{\"serviceName\": \"health_check_service_name\"}}");
1514   auto response_generator1 = BuildResolverResponseGenerator();
1515   auto channel1 = BuildChannel("round_robin", response_generator1, args);
1516   auto stub1 = BuildStub(channel1);
1517   std::vector<int> ports = GetServersPorts();
1518   response_generator1.SetNextResolution(ports);
1519   // Create a channel with health checking enabled but inhibited.
1520   args.SetInt(GRPC_ARG_INHIBIT_HEALTH_CHECKING, 1);
1521   auto response_generator2 = BuildResolverResponseGenerator();
1522   auto channel2 = BuildChannel("round_robin", response_generator2, args);
1523   auto stub2 = BuildStub(channel2);
1524   response_generator2.SetNextResolution(ports);
1525   // First channel should not become READY, because health checks should be
1526   // failing.
1527   EXPECT_FALSE(WaitForChannelReady(channel1.get(), 1));
1528   CheckRpcSendFailure(stub1);
1529   // Second channel should be READY.
1530   EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1));
1531   CheckRpcSendOk(stub2, DEBUG_LOCATION);
1532   // Enable health checks on the backend and wait for channel 1 to succeed.
1533   servers_[0]->SetServingStatus("health_check_service_name", true);
1534   CheckRpcSendOk(stub1, DEBUG_LOCATION, true /* wait_for_ready */);
1535   // Check that we created only one subchannel to the backend.
1536   EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
1537   // Clean up.
1538   EnableDefaultHealthCheckService(false);
1539 }
1540
1541 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthCheckingServiceNamePerChannel) {
1542   EnableDefaultHealthCheckService(true);
1543   // Start server.
1544   const int kNumServers = 1;
1545   StartServers(kNumServers);
1546   // Create a channel with health-checking enabled.
1547   ChannelArguments args;
1548   args.SetServiceConfigJSON(
1549       "{\"healthCheckConfig\": "
1550       "{\"serviceName\": \"health_check_service_name\"}}");
1551   auto response_generator1 = BuildResolverResponseGenerator();
1552   auto channel1 = BuildChannel("round_robin", response_generator1, args);
1553   auto stub1 = BuildStub(channel1);
1554   std::vector<int> ports = GetServersPorts();
1555   response_generator1.SetNextResolution(ports);
1556   // Create a channel with health-checking enabled with a different
1557   // service name.
1558   ChannelArguments args2;
1559   args2.SetServiceConfigJSON(
1560       "{\"healthCheckConfig\": "
1561       "{\"serviceName\": \"health_check_service_name2\"}}");
1562   auto response_generator2 = BuildResolverResponseGenerator();
1563   auto channel2 = BuildChannel("round_robin", response_generator2, args2);
1564   auto stub2 = BuildStub(channel2);
1565   response_generator2.SetNextResolution(ports);
1566   // Allow health checks from channel 2 to succeed.
1567   servers_[0]->SetServingStatus("health_check_service_name2", true);
1568   // First channel should not become READY, because health checks should be
1569   // failing.
1570   EXPECT_FALSE(WaitForChannelReady(channel1.get(), 1));
1571   CheckRpcSendFailure(stub1);
1572   // Second channel should be READY.
1573   EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1));
1574   CheckRpcSendOk(stub2, DEBUG_LOCATION);
1575   // Enable health checks for channel 1 and wait for it to succeed.
1576   servers_[0]->SetServingStatus("health_check_service_name", true);
1577   CheckRpcSendOk(stub1, DEBUG_LOCATION, true /* wait_for_ready */);
1578   // Check that we created only one subchannel to the backend.
1579   EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
1580   // Clean up.
1581   EnableDefaultHealthCheckService(false);
1582 }
1583
1584 TEST_F(ClientLbEnd2endTest,
1585        RoundRobinWithHealthCheckingServiceNameChangesAfterSubchannelsCreated) {
1586   EnableDefaultHealthCheckService(true);
1587   // Start server.
1588   const int kNumServers = 1;
1589   StartServers(kNumServers);
1590   // Create a channel with health-checking enabled.
1591   const char* kServiceConfigJson =
1592       "{\"healthCheckConfig\": "
1593       "{\"serviceName\": \"health_check_service_name\"}}";
1594   auto response_generator = BuildResolverResponseGenerator();
1595   auto channel = BuildChannel("round_robin", response_generator);
1596   auto stub = BuildStub(channel);
1597   std::vector<int> ports = GetServersPorts();
1598   response_generator.SetNextResolution(ports, kServiceConfigJson);
1599   servers_[0]->SetServingStatus("health_check_service_name", true);
1600   EXPECT_TRUE(WaitForChannelReady(channel.get(), 1 /* timeout_seconds */));
1601   // Send an update on the channel to change it to use a health checking
1602   // service name that is not being reported as healthy.
1603   const char* kServiceConfigJson2 =
1604       "{\"healthCheckConfig\": "
1605       "{\"serviceName\": \"health_check_service_name2\"}}";
1606   response_generator.SetNextResolution(ports, kServiceConfigJson2);
1607   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
1608   // Clean up.
1609   EnableDefaultHealthCheckService(false);
1610 }
1611
1612 TEST_F(ClientLbEnd2endTest, ChannelIdleness) {
1613   // Start server.
1614   const int kNumServers = 1;
1615   StartServers(kNumServers);
1616   // Set max idle time and build the channel.
1617   ChannelArguments args;
1618   args.SetInt(GRPC_ARG_CLIENT_IDLE_TIMEOUT_MS, 1000);
1619   auto response_generator = BuildResolverResponseGenerator();
1620   auto channel = BuildChannel("", response_generator, args);
1621   auto stub = BuildStub(channel);
1622   // The initial channel state should be IDLE.
1623   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
1624   // After sending RPC, channel state should be READY.
1625   response_generator.SetNextResolution(GetServersPorts());
1626   CheckRpcSendOk(stub, DEBUG_LOCATION);
1627   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1628   // After a period time not using the channel, the channel state should switch
1629   // to IDLE.
1630   gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(1200));
1631   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
1632   // Sending a new RPC should awake the IDLE channel.
1633   response_generator.SetNextResolution(GetServersPorts());
1634   CheckRpcSendOk(stub, DEBUG_LOCATION);
1635   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1636 }
1637
1638 class ClientLbInterceptTrailingMetadataTest : public ClientLbEnd2endTest {
1639  protected:
1640   void SetUp() override {
1641     ClientLbEnd2endTest::SetUp();
1642     grpc_core::RegisterInterceptRecvTrailingMetadataLoadBalancingPolicy(
1643         ReportTrailerIntercepted, this);
1644   }
1645
1646   void TearDown() override { ClientLbEnd2endTest::TearDown(); }
1647
1648   int trailers_intercepted() {
1649     grpc::internal::MutexLock lock(&mu_);
1650     return trailers_intercepted_;
1651   }
1652
1653   const udpa::data::orca::v1::OrcaLoadReport* backend_load_report() {
1654     grpc::internal::MutexLock lock(&mu_);
1655     return load_report_.get();
1656   }
1657
1658  private:
1659   static void ReportTrailerIntercepted(
1660       void* arg, const grpc_core::LoadBalancingPolicy::BackendMetricData*
1661                      backend_metric_data) {
1662     ClientLbInterceptTrailingMetadataTest* self =
1663         static_cast<ClientLbInterceptTrailingMetadataTest*>(arg);
1664     grpc::internal::MutexLock lock(&self->mu_);
1665     self->trailers_intercepted_++;
1666     if (backend_metric_data != nullptr) {
1667       self->load_report_.reset(new udpa::data::orca::v1::OrcaLoadReport);
1668       self->load_report_->set_cpu_utilization(
1669           backend_metric_data->cpu_utilization);
1670       self->load_report_->set_mem_utilization(
1671           backend_metric_data->mem_utilization);
1672       self->load_report_->set_rps(backend_metric_data->requests_per_second);
1673       for (const auto& p : backend_metric_data->request_cost) {
1674         grpc_core::UniquePtr<char> name =
1675             grpc_core::StringViewToCString(p.first);
1676         (*self->load_report_->mutable_request_cost())[name.get()] = p.second;
1677       }
1678       for (const auto& p : backend_metric_data->utilization) {
1679         grpc_core::UniquePtr<char> name =
1680             grpc_core::StringViewToCString(p.first);
1681         (*self->load_report_->mutable_utilization())[name.get()] = p.second;
1682       }
1683     }
1684   }
1685
1686   grpc::internal::Mutex mu_;
1687   int trailers_intercepted_ = 0;
1688   std::unique_ptr<udpa::data::orca::v1::OrcaLoadReport> load_report_;
1689 };
1690
1691 TEST_F(ClientLbInterceptTrailingMetadataTest, InterceptsRetriesDisabled) {
1692   const int kNumServers = 1;
1693   const int kNumRpcs = 10;
1694   StartServers(kNumServers);
1695   auto response_generator = BuildResolverResponseGenerator();
1696   auto channel =
1697       BuildChannel("intercept_trailing_metadata_lb", response_generator);
1698   auto stub = BuildStub(channel);
1699   response_generator.SetNextResolution(GetServersPorts());
1700   for (size_t i = 0; i < kNumRpcs; ++i) {
1701     CheckRpcSendOk(stub, DEBUG_LOCATION);
1702   }
1703   // Check LB policy name for the channel.
1704   EXPECT_EQ("intercept_trailing_metadata_lb",
1705             channel->GetLoadBalancingPolicyName());
1706   EXPECT_EQ(kNumRpcs, trailers_intercepted());
1707   EXPECT_EQ(nullptr, backend_load_report());
1708 }
1709
1710 TEST_F(ClientLbInterceptTrailingMetadataTest, InterceptsRetriesEnabled) {
1711   const int kNumServers = 1;
1712   const int kNumRpcs = 10;
1713   StartServers(kNumServers);
1714   ChannelArguments args;
1715   args.SetServiceConfigJSON(
1716       "{\n"
1717       "  \"methodConfig\": [ {\n"
1718       "    \"name\": [\n"
1719       "      { \"service\": \"grpc.testing.EchoTestService\" }\n"
1720       "    ],\n"
1721       "    \"retryPolicy\": {\n"
1722       "      \"maxAttempts\": 3,\n"
1723       "      \"initialBackoff\": \"1s\",\n"
1724       "      \"maxBackoff\": \"120s\",\n"
1725       "      \"backoffMultiplier\": 1.6,\n"
1726       "      \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
1727       "    }\n"
1728       "  } ]\n"
1729       "}");
1730   auto response_generator = BuildResolverResponseGenerator();
1731   auto channel =
1732       BuildChannel("intercept_trailing_metadata_lb", response_generator, args);
1733   auto stub = BuildStub(channel);
1734   response_generator.SetNextResolution(GetServersPorts());
1735   for (size_t i = 0; i < kNumRpcs; ++i) {
1736     CheckRpcSendOk(stub, DEBUG_LOCATION);
1737   }
1738   // Check LB policy name for the channel.
1739   EXPECT_EQ("intercept_trailing_metadata_lb",
1740             channel->GetLoadBalancingPolicyName());
1741   EXPECT_EQ(kNumRpcs, trailers_intercepted());
1742   EXPECT_EQ(nullptr, backend_load_report());
1743 }
1744
1745 TEST_F(ClientLbInterceptTrailingMetadataTest, BackendMetricData) {
1746   const int kNumServers = 1;
1747   const int kNumRpcs = 10;
1748   StartServers(kNumServers);
1749   udpa::data::orca::v1::OrcaLoadReport load_report;
1750   load_report.set_cpu_utilization(0.5);
1751   load_report.set_mem_utilization(0.75);
1752   load_report.set_rps(25);
1753   auto* request_cost = load_report.mutable_request_cost();
1754   (*request_cost)["foo"] = 0.8;
1755   (*request_cost)["bar"] = 1.4;
1756   auto* utilization = load_report.mutable_utilization();
1757   (*utilization)["baz"] = 1.1;
1758   (*utilization)["quux"] = 0.9;
1759   for (const auto& server : servers_) {
1760     server->service_.set_load_report(&load_report);
1761   }
1762   auto response_generator = BuildResolverResponseGenerator();
1763   auto channel =
1764       BuildChannel("intercept_trailing_metadata_lb", response_generator);
1765   auto stub = BuildStub(channel);
1766   response_generator.SetNextResolution(GetServersPorts());
1767   for (size_t i = 0; i < kNumRpcs; ++i) {
1768     CheckRpcSendOk(stub, DEBUG_LOCATION);
1769     auto* actual = backend_load_report();
1770     ASSERT_NE(actual, nullptr);
1771     // TODO(roth): Change this to use EqualsProto() once that becomes
1772     // available in OSS.
1773     EXPECT_EQ(actual->cpu_utilization(), load_report.cpu_utilization());
1774     EXPECT_EQ(actual->mem_utilization(), load_report.mem_utilization());
1775     EXPECT_EQ(actual->rps(), load_report.rps());
1776     EXPECT_EQ(actual->request_cost().size(), load_report.request_cost().size());
1777     for (const auto& p : actual->request_cost()) {
1778       auto it = load_report.request_cost().find(p.first);
1779       ASSERT_NE(it, load_report.request_cost().end());
1780       EXPECT_EQ(it->second, p.second);
1781     }
1782     EXPECT_EQ(actual->utilization().size(), load_report.utilization().size());
1783     for (const auto& p : actual->utilization()) {
1784       auto it = load_report.utilization().find(p.first);
1785       ASSERT_NE(it, load_report.utilization().end());
1786       EXPECT_EQ(it->second, p.second);
1787     }
1788   }
1789   // Check LB policy name for the channel.
1790   EXPECT_EQ("intercept_trailing_metadata_lb",
1791             channel->GetLoadBalancingPolicyName());
1792   EXPECT_EQ(kNumRpcs, trailers_intercepted());
1793 }
1794
1795 }  // namespace
1796 }  // namespace testing
1797 }  // namespace grpc
1798
1799 int main(int argc, char** argv) {
1800   ::testing::InitGoogleTest(&argc, argv);
1801   grpc::testing::TestEnvironment env(argc, argv);
1802   const auto result = RUN_ALL_TESTS();
1803   return result;
1804 }