3 * Copyright 2016 gRPC authors.
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
9 * http://www.apache.org/licenses/LICENSE-2.0
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.
27 #include "absl/memory/memory.h"
28 #include "absl/strings/str_cat.h"
30 #include <grpc/grpc.h>
31 #include <grpc/support/alloc.h>
32 #include <grpc/support/atm.h>
33 #include <grpc/support/log.h>
34 #include <grpc/support/time.h>
35 #include <grpcpp/channel.h>
36 #include <grpcpp/client_context.h>
37 #include <grpcpp/create_channel.h>
38 #include <grpcpp/health_check_service_interface.h>
39 #include <grpcpp/impl/codegen/sync.h>
40 #include <grpcpp/server.h>
41 #include <grpcpp/server_builder.h>
43 #include "src/core/ext/filters/client_channel/backup_poller.h"
44 #include "src/core/ext/filters/client_channel/global_subchannel_pool.h"
45 #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
46 #include "src/core/ext/filters/client_channel/server_address.h"
47 #include "src/core/ext/filters/client_channel/service_config.h"
48 #include "src/core/lib/backoff/backoff.h"
49 #include "src/core/lib/channel/channel_args.h"
50 #include "src/core/lib/gpr/env.h"
51 #include "src/core/lib/gprpp/debug_location.h"
52 #include "src/core/lib/gprpp/ref_counted_ptr.h"
53 #include "src/core/lib/iomgr/parse_address.h"
54 #include "src/core/lib/iomgr/tcp_client.h"
55 #include "src/core/lib/security/credentials/fake/fake_credentials.h"
56 #include "src/cpp/client/secure_credentials.h"
57 #include "src/cpp/server/secure_server_credentials.h"
59 #include "src/proto/grpc/testing/echo.grpc.pb.h"
60 #include "src/proto/grpc/testing/xds/orca_load_report_for_test.pb.h"
61 #include "test/core/util/port.h"
62 #include "test/core/util/resolve_localhost_ip46.h"
63 #include "test/core/util/test_config.h"
64 #include "test/core/util/test_lb_policies.h"
65 #include "test/cpp/end2end/test_service_impl.h"
67 #include <gmock/gmock.h>
68 #include <gtest/gtest.h>
70 using grpc::testing::EchoRequest;
71 using grpc::testing::EchoResponse;
73 // defined in tcp_client.cc
74 extern grpc_tcp_client_vtable* grpc_tcp_client_impl;
76 static grpc_tcp_client_vtable* default_client_impl;
82 gpr_atm g_connection_delay_ms;
84 void tcp_client_connect_with_delay(grpc_closure* closure, grpc_endpoint** ep,
85 grpc_pollset_set* interested_parties,
86 const grpc_channel_args* channel_args,
87 const grpc_resolved_address* addr,
88 grpc_millis deadline) {
89 const int delay_ms = gpr_atm_acq_load(&g_connection_delay_ms);
91 gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(delay_ms));
93 default_client_impl->connect(closure, ep, interested_parties, channel_args,
94 addr, deadline + delay_ms);
97 grpc_tcp_client_vtable delayed_connect = {tcp_client_connect_with_delay};
99 // Subclass of TestServiceImpl that increments a request counter for
100 // every call to the Echo RPC.
101 class MyTestServiceImpl : public TestServiceImpl {
103 Status Echo(ServerContext* context, const EchoRequest* request,
104 EchoResponse* response) override {
105 const udpa::data::orca::v1::OrcaLoadReport* load_report = nullptr;
107 grpc::internal::MutexLock lock(&mu_);
109 load_report = load_report_;
111 AddClient(context->peer());
112 if (load_report != nullptr) {
113 // TODO(roth): Once we provide a more standard server-side API for
114 // populating this data, use that API here.
115 context->AddTrailingMetadata("x-endpoint-load-metrics-bin",
116 load_report->SerializeAsString());
118 return TestServiceImpl::Echo(context, request, response);
121 int request_count() {
122 grpc::internal::MutexLock lock(&mu_);
123 return request_count_;
126 void ResetCounters() {
127 grpc::internal::MutexLock lock(&mu_);
131 std::set<std::string> clients() {
132 grpc::internal::MutexLock lock(&clients_mu_);
136 void set_load_report(udpa::data::orca::v1::OrcaLoadReport* load_report) {
137 grpc::internal::MutexLock lock(&mu_);
138 load_report_ = load_report;
142 void AddClient(const std::string& client) {
143 grpc::internal::MutexLock lock(&clients_mu_);
144 clients_.insert(client);
147 grpc::internal::Mutex mu_;
148 int request_count_ = 0;
149 const udpa::data::orca::v1::OrcaLoadReport* load_report_ = nullptr;
150 grpc::internal::Mutex clients_mu_;
151 std::set<std::string> clients_;
154 class FakeResolverResponseGeneratorWrapper {
156 explicit FakeResolverResponseGeneratorWrapper(bool ipv6_only)
157 : ipv6_only_(ipv6_only),
158 response_generator_(grpc_core::MakeRefCounted<
159 grpc_core::FakeResolverResponseGenerator>()) {}
161 FakeResolverResponseGeneratorWrapper(
162 FakeResolverResponseGeneratorWrapper&& other) noexcept {
163 ipv6_only_ = other.ipv6_only_;
164 response_generator_ = std::move(other.response_generator_);
167 void SetNextResolution(
168 const std::vector<int>& ports, const char* service_config_json = nullptr,
169 const char* attribute_key = nullptr,
170 std::unique_ptr<grpc_core::ServerAddress::AttributeInterface> attribute =
172 grpc_core::ExecCtx exec_ctx;
173 response_generator_->SetResponse(
174 BuildFakeResults(ipv6_only_, ports, service_config_json, attribute_key,
175 std::move(attribute)));
178 void SetNextResolutionUponError(const std::vector<int>& ports) {
179 grpc_core::ExecCtx exec_ctx;
180 response_generator_->SetReresolutionResponse(
181 BuildFakeResults(ipv6_only_, ports));
184 void SetFailureOnReresolution() {
185 grpc_core::ExecCtx exec_ctx;
186 response_generator_->SetFailureOnReresolution();
189 grpc_core::FakeResolverResponseGenerator* Get() const {
190 return response_generator_.get();
194 static grpc_core::Resolver::Result BuildFakeResults(
195 bool ipv6_only, const std::vector<int>& ports,
196 const char* service_config_json = nullptr,
197 const char* attribute_key = nullptr,
198 std::unique_ptr<grpc_core::ServerAddress::AttributeInterface> attribute =
200 grpc_core::Resolver::Result result;
201 for (const int& port : ports) {
202 std::string lb_uri_str =
203 absl::StrCat(ipv6_only ? "ipv6:[::1]:" : "ipv4:127.0.0.1:", port);
204 grpc_uri* lb_uri = grpc_uri_parse(lb_uri_str.c_str(), true);
205 GPR_ASSERT(lb_uri != nullptr);
206 grpc_resolved_address address;
207 GPR_ASSERT(grpc_parse_uri(lb_uri, &address));
208 std::map<const char*,
209 std::unique_ptr<grpc_core::ServerAddress::AttributeInterface>>
211 if (attribute != nullptr) {
212 attributes[attribute_key] = attribute->Copy();
214 result.addresses.emplace_back(address.addr, address.len,
215 nullptr /* args */, std::move(attributes));
216 grpc_uri_destroy(lb_uri);
218 if (service_config_json != nullptr) {
219 result.service_config = grpc_core::ServiceConfig::Create(
220 nullptr, service_config_json, &result.service_config_error);
221 GPR_ASSERT(result.service_config != nullptr);
226 bool ipv6_only_ = false;
227 grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
231 class ClientLbEnd2endTest : public ::testing::Test {
233 ClientLbEnd2endTest()
234 : server_host_("localhost"),
235 kRequestMessage_("Live long and prosper."),
236 creds_(new SecureChannelCredentials(
237 grpc_fake_transport_security_credentials_create())) {}
239 static void SetUpTestCase() {
240 // Make the backup poller poll very frequently in order to pick up
241 // updates from all the subchannels's FDs.
242 GPR_GLOBAL_CONFIG_SET(grpc_client_channel_backup_poll_interval_ms, 1);
244 // Workaround Apple CFStream bug
245 gpr_setenv("grpc_cfstream", "0");
249 void SetUp() override {
251 bool localhost_resolves_to_ipv4 = false;
252 bool localhost_resolves_to_ipv6 = false;
253 grpc_core::LocalhostResolves(&localhost_resolves_to_ipv4,
254 &localhost_resolves_to_ipv6);
255 ipv6_only_ = !localhost_resolves_to_ipv4 && localhost_resolves_to_ipv6;
258 void TearDown() override {
259 for (size_t i = 0; i < servers_.size(); ++i) {
260 servers_[i]->Shutdown();
267 void CreateServers(size_t num_servers,
268 std::vector<int> ports = std::vector<int>()) {
270 for (size_t i = 0; i < num_servers; ++i) {
272 if (ports.size() == num_servers) port = ports[i];
273 servers_.emplace_back(new ServerData(port));
277 void StartServer(size_t index) { servers_[index]->Start(server_host_); }
279 void StartServers(size_t num_servers,
280 std::vector<int> ports = std::vector<int>()) {
281 CreateServers(num_servers, std::move(ports));
282 for (size_t i = 0; i < num_servers; ++i) {
287 std::vector<int> GetServersPorts(size_t start_index = 0) {
288 std::vector<int> ports;
289 for (size_t i = start_index; i < servers_.size(); ++i) {
290 ports.push_back(servers_[i]->port_);
295 FakeResolverResponseGeneratorWrapper BuildResolverResponseGenerator() {
296 return FakeResolverResponseGeneratorWrapper(ipv6_only_);
299 std::unique_ptr<grpc::testing::EchoTestService::Stub> BuildStub(
300 const std::shared_ptr<Channel>& channel) {
301 return grpc::testing::EchoTestService::NewStub(channel);
304 std::shared_ptr<Channel> BuildChannel(
305 const std::string& lb_policy_name,
306 const FakeResolverResponseGeneratorWrapper& response_generator,
307 ChannelArguments args = ChannelArguments()) {
308 if (!lb_policy_name.empty()) {
309 args.SetLoadBalancingPolicyName(lb_policy_name);
310 } // else, default to pick first
311 args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
312 response_generator.Get());
313 return ::grpc::CreateCustomChannel("fake:///", creds_, args);
317 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
318 EchoResponse* response = nullptr, int timeout_ms = 1000,
319 Status* result = nullptr, bool wait_for_ready = false) {
320 const bool local_response = (response == nullptr);
321 if (local_response) response = new EchoResponse;
323 request.set_message(kRequestMessage_);
324 request.mutable_param()->set_echo_metadata(true);
325 ClientContext context;
326 context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms));
327 if (wait_for_ready) context.set_wait_for_ready(true);
328 context.AddMetadata("foo", "1");
329 context.AddMetadata("bar", "2");
330 context.AddMetadata("baz", "3");
331 Status status = stub->Echo(&context, request, response);
332 if (result != nullptr) *result = status;
333 if (local_response) delete response;
338 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
339 const grpc_core::DebugLocation& location, bool wait_for_ready = false) {
340 EchoResponse response;
343 SendRpc(stub, &response, 2000, &status, wait_for_ready);
344 ASSERT_TRUE(success) << "From " << location.file() << ":" << location.line()
346 << "Error: " << status.error_message() << " "
347 << status.error_details();
348 ASSERT_EQ(response.message(), kRequestMessage_)
349 << "From " << location.file() << ":" << location.line();
350 if (!success) abort();
353 void CheckRpcSendFailure(
354 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub) {
355 const bool success = SendRpc(stub);
356 EXPECT_FALSE(success);
361 std::unique_ptr<Server> server_;
362 MyTestServiceImpl service_;
363 std::unique_ptr<std::thread> thread_;
364 bool server_ready_ = false;
365 bool started_ = false;
367 explicit ServerData(int port = 0) {
368 port_ = port > 0 ? port : grpc_pick_unused_port_or_die();
371 void Start(const std::string& server_host) {
372 gpr_log(GPR_INFO, "starting server on port %d", port_);
374 grpc::internal::Mutex mu;
375 grpc::internal::MutexLock lock(&mu);
376 grpc::internal::CondVar cond;
377 thread_ = absl::make_unique<std::thread>(
378 std::bind(&ServerData::Serve, this, server_host, &mu, &cond));
379 cond.WaitUntil(&mu, [this] { return server_ready_; });
380 server_ready_ = false;
381 gpr_log(GPR_INFO, "server startup complete");
384 void Serve(const std::string& server_host, grpc::internal::Mutex* mu,
385 grpc::internal::CondVar* cond) {
386 std::ostringstream server_address;
387 server_address << server_host << ":" << port_;
388 ServerBuilder builder;
389 std::shared_ptr<ServerCredentials> creds(new SecureServerCredentials(
390 grpc_fake_transport_security_server_credentials_create()));
391 builder.AddListeningPort(server_address.str(), std::move(creds));
392 builder.RegisterService(&service_);
393 server_ = builder.BuildAndStart();
394 grpc::internal::MutexLock lock(mu);
395 server_ready_ = true;
400 if (!started_) return;
401 server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0));
406 void SetServingStatus(const std::string& service, bool serving) {
407 server_->GetHealthCheckService()->SetServingStatus(service, serving);
411 void ResetCounters() {
412 for (const auto& server : servers_) server->service_.ResetCounters();
416 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
417 size_t server_idx, const grpc_core::DebugLocation& location,
418 bool ignore_failure = false) {
420 if (ignore_failure) {
423 CheckRpcSendOk(stub, location, true);
425 } while (servers_[server_idx]->service_.request_count() == 0);
429 bool WaitForChannelState(
431 const std::function<bool(grpc_connectivity_state)>& predicate,
432 bool try_to_connect = false, int timeout_seconds = 5) {
433 const gpr_timespec deadline =
434 grpc_timeout_seconds_to_deadline(timeout_seconds);
436 grpc_connectivity_state state = channel->GetState(try_to_connect);
437 if (predicate(state)) break;
438 if (!channel->WaitForStateChange(state, deadline)) return false;
443 bool WaitForChannelNotReady(Channel* channel, int timeout_seconds = 5) {
444 auto predicate = [](grpc_connectivity_state state) {
445 return state != GRPC_CHANNEL_READY;
447 return WaitForChannelState(channel, predicate, false, timeout_seconds);
450 bool WaitForChannelReady(Channel* channel, int timeout_seconds = 5) {
451 auto predicate = [](grpc_connectivity_state state) {
452 return state == GRPC_CHANNEL_READY;
454 return WaitForChannelState(channel, predicate, true, timeout_seconds);
457 bool SeenAllServers() {
458 for (const auto& server : servers_) {
459 if (server->service_.request_count() == 0) return false;
464 // Updates \a connection_order by appending to it the index of the newly
465 // connected server. Must be called after every single RPC.
466 void UpdateConnectionOrder(
467 const std::vector<std::unique_ptr<ServerData>>& servers,
468 std::vector<int>* connection_order) {
469 for (size_t i = 0; i < servers.size(); ++i) {
470 if (servers[i]->service_.request_count() == 1) {
471 // Was the server index known? If not, update connection_order.
473 std::find(connection_order->begin(), connection_order->end(), i);
474 if (it == connection_order->end()) {
475 connection_order->push_back(i);
482 const std::string server_host_;
483 std::vector<std::unique_ptr<ServerData>> servers_;
484 const std::string kRequestMessage_;
485 std::shared_ptr<ChannelCredentials> creds_;
486 bool ipv6_only_ = false;
489 TEST_F(ClientLbEnd2endTest, ChannelStateConnectingWhenResolving) {
490 const int kNumServers = 3;
491 StartServers(kNumServers);
492 auto response_generator = BuildResolverResponseGenerator();
493 auto channel = BuildChannel("", response_generator);
494 auto stub = BuildStub(channel);
495 // Initial state should be IDLE.
496 EXPECT_EQ(channel->GetState(false /* try_to_connect */), GRPC_CHANNEL_IDLE);
497 // Tell the channel to try to connect.
498 // Note that this call also returns IDLE, since the state change has
499 // not yet occurred; it just gets triggered by this call.
500 EXPECT_EQ(channel->GetState(true /* try_to_connect */), GRPC_CHANNEL_IDLE);
501 // Now that the channel is trying to connect, we should be in state
503 EXPECT_EQ(channel->GetState(false /* try_to_connect */),
504 GRPC_CHANNEL_CONNECTING);
505 // Return a resolver result, which allows the connection attempt to proceed.
506 response_generator.SetNextResolution(GetServersPorts());
507 // We should eventually transition into state READY.
508 EXPECT_TRUE(WaitForChannelReady(channel.get()));
511 TEST_F(ClientLbEnd2endTest, PickFirst) {
512 // Start servers and send one RPC per server.
513 const int kNumServers = 3;
514 StartServers(kNumServers);
515 auto response_generator = BuildResolverResponseGenerator();
516 auto channel = BuildChannel(
517 "", response_generator); // test that pick first is the default.
518 auto stub = BuildStub(channel);
519 response_generator.SetNextResolution(GetServersPorts());
520 for (size_t i = 0; i < servers_.size(); ++i) {
521 CheckRpcSendOk(stub, DEBUG_LOCATION);
523 // All requests should have gone to a single server.
525 for (size_t i = 0; i < servers_.size(); ++i) {
526 const int request_count = servers_[i]->service_.request_count();
527 if (request_count == kNumServers) {
530 EXPECT_EQ(0, request_count);
534 // Check LB policy name for the channel.
535 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
538 TEST_F(ClientLbEnd2endTest, PickFirstProcessPending) {
539 StartServers(1); // Single server
540 auto response_generator = BuildResolverResponseGenerator();
541 auto channel = BuildChannel(
542 "", response_generator); // test that pick first is the default.
543 auto stub = BuildStub(channel);
544 response_generator.SetNextResolution({servers_[0]->port_});
545 WaitForServer(stub, 0, DEBUG_LOCATION);
546 // Create a new channel and its corresponding PF LB policy, which will pick
547 // the subchannels in READY state from the previous RPC against the same
548 // target (even if it happened over a different channel, because subchannels
549 // are globally reused). Progress should happen without any transition from
551 auto second_response_generator = BuildResolverResponseGenerator();
552 auto second_channel = BuildChannel("", second_response_generator);
553 auto second_stub = BuildStub(second_channel);
554 second_response_generator.SetNextResolution({servers_[0]->port_});
555 CheckRpcSendOk(second_stub, DEBUG_LOCATION);
558 TEST_F(ClientLbEnd2endTest, PickFirstSelectsReadyAtStartup) {
559 ChannelArguments args;
560 constexpr int kInitialBackOffMs = 5000;
561 args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
562 // Create 2 servers, but start only the second one.
563 std::vector<int> ports = {grpc_pick_unused_port_or_die(),
564 grpc_pick_unused_port_or_die()};
565 CreateServers(2, ports);
567 auto response_generator1 = BuildResolverResponseGenerator();
568 auto channel1 = BuildChannel("pick_first", response_generator1, args);
569 auto stub1 = BuildStub(channel1);
570 response_generator1.SetNextResolution(ports);
571 // Wait for second server to be ready.
572 WaitForServer(stub1, 1, DEBUG_LOCATION);
573 // Create a second channel with the same addresses. Its PF instance
574 // should immediately pick the second subchannel, since it's already
576 auto response_generator2 = BuildResolverResponseGenerator();
577 auto channel2 = BuildChannel("pick_first", response_generator2, args);
578 response_generator2.SetNextResolution(ports);
579 // Check that the channel reports READY without waiting for the
581 EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1 /* timeout_seconds */));
584 TEST_F(ClientLbEnd2endTest, PickFirstBackOffInitialReconnect) {
585 ChannelArguments args;
586 constexpr int kInitialBackOffMs = 100;
587 args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
588 const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
589 const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
590 auto response_generator = BuildResolverResponseGenerator();
591 auto channel = BuildChannel("pick_first", response_generator, args);
592 auto stub = BuildStub(channel);
593 response_generator.SetNextResolution(ports);
594 // The channel won't become connected (there's no server).
595 ASSERT_FALSE(channel->WaitForConnected(
596 grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 2)));
597 // Bring up a server on the chosen port.
598 StartServers(1, ports);
600 ASSERT_TRUE(channel->WaitForConnected(
601 grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 2)));
602 const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
603 const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
604 gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
605 // We should have waited at least kInitialBackOffMs. We substract one to
606 // account for test and precision accuracy drift.
607 EXPECT_GE(waited_ms, kInitialBackOffMs - 1);
608 // But not much more.
611 grpc_timeout_milliseconds_to_deadline(kInitialBackOffMs * 1.10), t1),
615 TEST_F(ClientLbEnd2endTest, PickFirstBackOffMinReconnect) {
616 ChannelArguments args;
617 constexpr int kMinReconnectBackOffMs = 1000;
618 args.SetInt(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, kMinReconnectBackOffMs);
619 const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
620 auto response_generator = BuildResolverResponseGenerator();
621 auto channel = BuildChannel("pick_first", response_generator, args);
622 auto stub = BuildStub(channel);
623 response_generator.SetNextResolution(ports);
624 // Make connection delay a 10% longer than it's willing to in order to make
625 // sure we are hitting the codepath that waits for the min reconnect backoff.
626 gpr_atm_rel_store(&g_connection_delay_ms, kMinReconnectBackOffMs * 1.10);
627 default_client_impl = grpc_tcp_client_impl;
628 grpc_set_tcp_client_impl(&delayed_connect);
629 const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
630 channel->WaitForConnected(
631 grpc_timeout_milliseconds_to_deadline(kMinReconnectBackOffMs * 2));
632 const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
633 const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
634 gpr_log(GPR_DEBUG, "Waited %" PRId64 " ms", waited_ms);
635 // We should have waited at least kMinReconnectBackOffMs. We substract one to
636 // account for test and precision accuracy drift.
637 EXPECT_GE(waited_ms, kMinReconnectBackOffMs - 1);
638 gpr_atm_rel_store(&g_connection_delay_ms, 0);
641 TEST_F(ClientLbEnd2endTest, PickFirstResetConnectionBackoff) {
642 ChannelArguments args;
643 constexpr int kInitialBackOffMs = 1000;
644 args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
645 const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
646 auto response_generator = BuildResolverResponseGenerator();
647 auto channel = BuildChannel("pick_first", response_generator, args);
648 auto stub = BuildStub(channel);
649 response_generator.SetNextResolution(ports);
650 // The channel won't become connected (there's no server).
652 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
653 // Bring up a server on the chosen port.
654 StartServers(1, ports);
655 const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
656 // Wait for connect, but not long enough. This proves that we're
657 // being throttled by initial backoff.
659 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
660 // Reset connection backoff.
661 experimental::ChannelResetConnectionBackoff(channel.get());
662 // Wait for connect. Should happen as soon as the client connects to
663 // the newly started server, which should be before the initial
664 // backoff timeout elapses.
666 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(20)));
667 const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
668 const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
669 gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
670 // We should have waited less than kInitialBackOffMs.
671 EXPECT_LT(waited_ms, kInitialBackOffMs);
674 TEST_F(ClientLbEnd2endTest,
675 PickFirstResetConnectionBackoffNextAttemptStartsImmediately) {
676 ChannelArguments args;
677 constexpr int kInitialBackOffMs = 1000;
678 args.SetInt(GRPC_ARG_INITIAL_RECONNECT_BACKOFF_MS, kInitialBackOffMs);
679 const std::vector<int> ports = {grpc_pick_unused_port_or_die()};
680 auto response_generator = BuildResolverResponseGenerator();
681 auto channel = BuildChannel("pick_first", response_generator, args);
682 auto stub = BuildStub(channel);
683 response_generator.SetNextResolution(ports);
684 // Wait for connect, which should fail ~immediately, because the server
686 gpr_log(GPR_INFO, "=== INITIAL CONNECTION ATTEMPT");
688 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
689 // Reset connection backoff.
690 // Note that the time at which the third attempt will be started is
691 // actually computed at this point, so we record the start time here.
692 gpr_log(GPR_INFO, "=== RESETTING BACKOFF");
693 const gpr_timespec t0 = gpr_now(GPR_CLOCK_MONOTONIC);
694 experimental::ChannelResetConnectionBackoff(channel.get());
695 // Trigger a second connection attempt. This should also fail
696 // ~immediately, but the retry should be scheduled for
697 // kInitialBackOffMs instead of applying the multiplier.
698 gpr_log(GPR_INFO, "=== POLLING FOR SECOND CONNECTION ATTEMPT");
700 channel->WaitForConnected(grpc_timeout_milliseconds_to_deadline(10)));
701 // Bring up a server on the chosen port.
702 gpr_log(GPR_INFO, "=== STARTING BACKEND");
703 StartServers(1, ports);
704 // Wait for connect. Should happen within kInitialBackOffMs.
705 // Give an extra 100ms to account for the time spent in the second and
706 // third connection attempts themselves (since what we really want to
707 // measure is the time between the two). As long as this is less than
708 // the 1.6x increase we would see if the backoff state was not reset
709 // properly, the test is still proving that the backoff was reset.
710 constexpr int kWaitMs = kInitialBackOffMs + 100;
711 gpr_log(GPR_INFO, "=== POLLING FOR THIRD CONNECTION ATTEMPT");
712 EXPECT_TRUE(channel->WaitForConnected(
713 grpc_timeout_milliseconds_to_deadline(kWaitMs)));
714 const gpr_timespec t1 = gpr_now(GPR_CLOCK_MONOTONIC);
715 const grpc_millis waited_ms = gpr_time_to_millis(gpr_time_sub(t1, t0));
716 gpr_log(GPR_DEBUG, "Waited %" PRId64 " milliseconds", waited_ms);
717 EXPECT_LT(waited_ms, kWaitMs);
720 TEST_F(ClientLbEnd2endTest, PickFirstUpdates) {
721 // Start servers and send one RPC per server.
722 const int kNumServers = 3;
723 StartServers(kNumServers);
724 auto response_generator = BuildResolverResponseGenerator();
725 auto channel = BuildChannel("pick_first", response_generator);
726 auto stub = BuildStub(channel);
728 std::vector<int> ports;
730 // Perform one RPC against the first server.
731 ports.emplace_back(servers_[0]->port_);
732 response_generator.SetNextResolution(ports);
733 gpr_log(GPR_INFO, "****** SET [0] *******");
734 CheckRpcSendOk(stub, DEBUG_LOCATION);
735 EXPECT_EQ(servers_[0]->service_.request_count(), 1);
737 // An empty update will result in the channel going into TRANSIENT_FAILURE.
739 response_generator.SetNextResolution(ports);
740 gpr_log(GPR_INFO, "****** SET none *******");
741 grpc_connectivity_state channel_state;
743 channel_state = channel->GetState(true /* try to connect */);
744 } while (channel_state == GRPC_CHANNEL_READY);
745 ASSERT_NE(channel_state, GRPC_CHANNEL_READY);
746 servers_[0]->service_.ResetCounters();
748 // Next update introduces servers_[1], making the channel recover.
750 ports.emplace_back(servers_[1]->port_);
751 response_generator.SetNextResolution(ports);
752 gpr_log(GPR_INFO, "****** SET [1] *******");
753 WaitForServer(stub, 1, DEBUG_LOCATION);
754 EXPECT_EQ(servers_[0]->service_.request_count(), 0);
756 // And again for servers_[2]
758 ports.emplace_back(servers_[2]->port_);
759 response_generator.SetNextResolution(ports);
760 gpr_log(GPR_INFO, "****** SET [2] *******");
761 WaitForServer(stub, 2, DEBUG_LOCATION);
762 EXPECT_EQ(servers_[0]->service_.request_count(), 0);
763 EXPECT_EQ(servers_[1]->service_.request_count(), 0);
765 // Check LB policy name for the channel.
766 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
769 TEST_F(ClientLbEnd2endTest, PickFirstUpdateSuperset) {
770 // Start servers and send one RPC per server.
771 const int kNumServers = 3;
772 StartServers(kNumServers);
773 auto response_generator = BuildResolverResponseGenerator();
774 auto channel = BuildChannel("pick_first", response_generator);
775 auto stub = BuildStub(channel);
777 std::vector<int> ports;
779 // Perform one RPC against the first server.
780 ports.emplace_back(servers_[0]->port_);
781 response_generator.SetNextResolution(ports);
782 gpr_log(GPR_INFO, "****** SET [0] *******");
783 CheckRpcSendOk(stub, DEBUG_LOCATION);
784 EXPECT_EQ(servers_[0]->service_.request_count(), 1);
785 servers_[0]->service_.ResetCounters();
787 // Send and superset update
789 ports.emplace_back(servers_[1]->port_);
790 ports.emplace_back(servers_[0]->port_);
791 response_generator.SetNextResolution(ports);
792 gpr_log(GPR_INFO, "****** SET superset *******");
793 CheckRpcSendOk(stub, DEBUG_LOCATION);
794 // We stick to the previously connected server.
795 WaitForServer(stub, 0, DEBUG_LOCATION);
796 EXPECT_EQ(0, servers_[1]->service_.request_count());
798 // Check LB policy name for the channel.
799 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
802 TEST_F(ClientLbEnd2endTest, PickFirstGlobalSubchannelPool) {
804 const int kNumServers = 1;
805 StartServers(kNumServers);
806 std::vector<int> ports = GetServersPorts();
807 // Create two channels that (by default) use the global subchannel pool.
808 auto response_generator1 = BuildResolverResponseGenerator();
809 auto channel1 = BuildChannel("pick_first", response_generator1);
810 auto stub1 = BuildStub(channel1);
811 response_generator1.SetNextResolution(ports);
812 auto response_generator2 = BuildResolverResponseGenerator();
813 auto channel2 = BuildChannel("pick_first", response_generator2);
814 auto stub2 = BuildStub(channel2);
815 response_generator2.SetNextResolution(ports);
816 WaitForServer(stub1, 0, DEBUG_LOCATION);
817 // Send one RPC on each channel.
818 CheckRpcSendOk(stub1, DEBUG_LOCATION);
819 CheckRpcSendOk(stub2, DEBUG_LOCATION);
820 // The server receives two requests.
821 EXPECT_EQ(2, servers_[0]->service_.request_count());
822 // The two requests are from the same client port, because the two channels
823 // share subchannels via the global subchannel pool.
824 EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
827 TEST_F(ClientLbEnd2endTest, PickFirstLocalSubchannelPool) {
829 const int kNumServers = 1;
830 StartServers(kNumServers);
831 std::vector<int> ports = GetServersPorts();
832 // Create two channels that use local subchannel pool.
833 ChannelArguments args;
834 args.SetInt(GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL, 1);
835 auto response_generator1 = BuildResolverResponseGenerator();
836 auto channel1 = BuildChannel("pick_first", response_generator1, args);
837 auto stub1 = BuildStub(channel1);
838 response_generator1.SetNextResolution(ports);
839 auto response_generator2 = BuildResolverResponseGenerator();
840 auto channel2 = BuildChannel("pick_first", response_generator2, args);
841 auto stub2 = BuildStub(channel2);
842 response_generator2.SetNextResolution(ports);
843 WaitForServer(stub1, 0, DEBUG_LOCATION);
844 // Send one RPC on each channel.
845 CheckRpcSendOk(stub1, DEBUG_LOCATION);
846 CheckRpcSendOk(stub2, DEBUG_LOCATION);
847 // The server receives two requests.
848 EXPECT_EQ(2, servers_[0]->service_.request_count());
849 // The two requests are from two client ports, because the two channels didn't
850 // share subchannels with each other.
851 EXPECT_EQ(2UL, servers_[0]->service_.clients().size());
854 TEST_F(ClientLbEnd2endTest, PickFirstManyUpdates) {
855 const int kNumUpdates = 1000;
856 const int kNumServers = 3;
857 StartServers(kNumServers);
858 auto response_generator = BuildResolverResponseGenerator();
859 auto channel = BuildChannel("pick_first", response_generator);
860 auto stub = BuildStub(channel);
861 std::vector<int> ports = GetServersPorts();
862 for (size_t i = 0; i < kNumUpdates; ++i) {
863 std::shuffle(ports.begin(), ports.end(),
864 std::mt19937(std::random_device()()));
865 response_generator.SetNextResolution(ports);
866 // We should re-enter core at the end of the loop to give the resolution
867 // setting closure a chance to run.
868 if ((i + 1) % 10 == 0) CheckRpcSendOk(stub, DEBUG_LOCATION);
870 // Check LB policy name for the channel.
871 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
874 TEST_F(ClientLbEnd2endTest, PickFirstReresolutionNoSelected) {
875 // Prepare the ports for up servers and down servers.
876 const int kNumServers = 3;
877 const int kNumAliveServers = 1;
878 StartServers(kNumAliveServers);
879 std::vector<int> alive_ports, dead_ports;
880 for (size_t i = 0; i < kNumServers; ++i) {
881 if (i < kNumAliveServers) {
882 alive_ports.emplace_back(servers_[i]->port_);
884 dead_ports.emplace_back(grpc_pick_unused_port_or_die());
887 auto response_generator = BuildResolverResponseGenerator();
888 auto channel = BuildChannel("pick_first", response_generator);
889 auto stub = BuildStub(channel);
890 // The initial resolution only contains dead ports. There won't be any
891 // selected subchannel. Re-resolution will return the same result.
892 response_generator.SetNextResolution(dead_ports);
893 gpr_log(GPR_INFO, "****** INITIAL RESOLUTION SET *******");
894 for (size_t i = 0; i < 10; ++i) CheckRpcSendFailure(stub);
895 // Set a re-resolution result that contains reachable ports, so that the
896 // pick_first LB policy can recover soon.
897 response_generator.SetNextResolutionUponError(alive_ports);
898 gpr_log(GPR_INFO, "****** RE-RESOLUTION SET *******");
899 WaitForServer(stub, 0, DEBUG_LOCATION, true /* ignore_failure */);
900 CheckRpcSendOk(stub, DEBUG_LOCATION);
901 EXPECT_EQ(servers_[0]->service_.request_count(), 1);
902 // Check LB policy name for the channel.
903 EXPECT_EQ("pick_first", channel->GetLoadBalancingPolicyName());
906 TEST_F(ClientLbEnd2endTest, PickFirstReconnectWithoutNewResolverResult) {
907 std::vector<int> ports = {grpc_pick_unused_port_or_die()};
908 StartServers(1, ports);
909 auto response_generator = BuildResolverResponseGenerator();
910 auto channel = BuildChannel("pick_first", response_generator);
911 auto stub = BuildStub(channel);
912 response_generator.SetNextResolution(ports);
913 gpr_log(GPR_INFO, "****** INITIAL CONNECTION *******");
914 WaitForServer(stub, 0, DEBUG_LOCATION);
915 gpr_log(GPR_INFO, "****** STOPPING SERVER ******");
916 servers_[0]->Shutdown();
917 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
918 gpr_log(GPR_INFO, "****** RESTARTING SERVER ******");
919 StartServers(1, ports);
920 WaitForServer(stub, 0, DEBUG_LOCATION);
923 TEST_F(ClientLbEnd2endTest,
924 PickFirstReconnectWithoutNewResolverResultStartsFromTopOfList) {
925 std::vector<int> ports = {grpc_pick_unused_port_or_die(),
926 grpc_pick_unused_port_or_die()};
927 CreateServers(2, ports);
929 auto response_generator = BuildResolverResponseGenerator();
930 auto channel = BuildChannel("pick_first", response_generator);
931 auto stub = BuildStub(channel);
932 response_generator.SetNextResolution(ports);
933 gpr_log(GPR_INFO, "****** INITIAL CONNECTION *******");
934 WaitForServer(stub, 1, DEBUG_LOCATION);
935 gpr_log(GPR_INFO, "****** STOPPING SERVER ******");
936 servers_[1]->Shutdown();
937 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
938 gpr_log(GPR_INFO, "****** STARTING BOTH SERVERS ******");
939 StartServers(2, ports);
940 WaitForServer(stub, 0, DEBUG_LOCATION);
943 TEST_F(ClientLbEnd2endTest, PickFirstCheckStateBeforeStartWatch) {
944 std::vector<int> ports = {grpc_pick_unused_port_or_die()};
945 StartServers(1, ports);
946 auto response_generator = BuildResolverResponseGenerator();
947 auto channel_1 = BuildChannel("pick_first", response_generator);
948 auto stub_1 = BuildStub(channel_1);
949 response_generator.SetNextResolution(ports);
950 gpr_log(GPR_INFO, "****** RESOLUTION SET FOR CHANNEL 1 *******");
951 WaitForServer(stub_1, 0, DEBUG_LOCATION);
952 gpr_log(GPR_INFO, "****** CHANNEL 1 CONNECTED *******");
953 servers_[0]->Shutdown();
954 // Channel 1 will receive a re-resolution containing the same server. It will
955 // create a new subchannel and hold a ref to it.
956 StartServers(1, ports);
957 gpr_log(GPR_INFO, "****** SERVER RESTARTED *******");
958 auto response_generator_2 = BuildResolverResponseGenerator();
959 auto channel_2 = BuildChannel("pick_first", response_generator_2);
960 auto stub_2 = BuildStub(channel_2);
961 response_generator_2.SetNextResolution(ports);
962 gpr_log(GPR_INFO, "****** RESOLUTION SET FOR CHANNEL 2 *******");
963 WaitForServer(stub_2, 0, DEBUG_LOCATION, true);
964 gpr_log(GPR_INFO, "****** CHANNEL 2 CONNECTED *******");
965 servers_[0]->Shutdown();
966 // Wait until the disconnection has triggered the connectivity notification.
967 // Otherwise, the subchannel may be picked for next call but will fail soon.
968 EXPECT_TRUE(WaitForChannelNotReady(channel_2.get()));
969 // Channel 2 will also receive a re-resolution containing the same server.
970 // Both channels will ref the same subchannel that failed.
971 StartServers(1, ports);
972 gpr_log(GPR_INFO, "****** SERVER RESTARTED AGAIN *******");
973 gpr_log(GPR_INFO, "****** CHANNEL 2 STARTING A CALL *******");
974 // The first call after the server restart will succeed.
975 CheckRpcSendOk(stub_2, DEBUG_LOCATION);
976 gpr_log(GPR_INFO, "****** CHANNEL 2 FINISHED A CALL *******");
977 // Check LB policy name for the channel.
978 EXPECT_EQ("pick_first", channel_1->GetLoadBalancingPolicyName());
979 // Check LB policy name for the channel.
980 EXPECT_EQ("pick_first", channel_2->GetLoadBalancingPolicyName());
983 TEST_F(ClientLbEnd2endTest, PickFirstIdleOnDisconnect) {
984 // Start server, send RPC, and make sure channel is READY.
985 const int kNumServers = 1;
986 StartServers(kNumServers);
987 auto response_generator = BuildResolverResponseGenerator();
989 BuildChannel("", response_generator); // pick_first is the default.
990 auto stub = BuildStub(channel);
991 response_generator.SetNextResolution(GetServersPorts());
992 CheckRpcSendOk(stub, DEBUG_LOCATION);
993 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
994 // Stop server. Channel should go into state IDLE.
995 response_generator.SetFailureOnReresolution();
996 servers_[0]->Shutdown();
997 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
998 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
1002 TEST_F(ClientLbEnd2endTest, PickFirstPendingUpdateAndSelectedSubchannelFails) {
1003 auto response_generator = BuildResolverResponseGenerator();
1005 BuildChannel("", response_generator); // pick_first is the default.
1006 auto stub = BuildStub(channel);
1007 // Create a number of servers, but only start 1 of them.
1010 // Initially resolve to first server and make sure it connects.
1011 gpr_log(GPR_INFO, "Phase 1: Connect to first server.");
1012 response_generator.SetNextResolution({servers_[0]->port_});
1013 CheckRpcSendOk(stub, DEBUG_LOCATION, true /* wait_for_ready */);
1014 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1015 // Send a resolution update with the remaining servers, none of which are
1016 // running yet, so the update will stay pending. Note that it's important
1017 // to have multiple servers here, or else the test will be flaky; with only
1018 // one server, the pending subchannel list has already gone into
1019 // TRANSIENT_FAILURE due to hitting the end of the list by the time we
1022 "Phase 2: Resolver update pointing to remaining "
1023 "(not started) servers.");
1024 response_generator.SetNextResolution(GetServersPorts(1 /* start_index */));
1025 // RPCs will continue to be sent to the first server.
1026 CheckRpcSendOk(stub, DEBUG_LOCATION);
1027 // Now stop the first server, so that the current subchannel list
1028 // fails. This should cause us to immediately swap over to the
1029 // pending list, even though it's not yet connected. The state should
1030 // be set to CONNECTING, since that's what the pending subchannel list
1031 // was doing when we swapped over.
1032 gpr_log(GPR_INFO, "Phase 3: Stopping first server.");
1033 servers_[0]->Shutdown();
1034 WaitForChannelNotReady(channel.get());
1035 // TODO(roth): This should always return CONNECTING, but it's flaky
1036 // between that and TRANSIENT_FAILURE. I suspect that this problem
1037 // will go away once we move the backoff code out of the subchannel
1038 // and into the LB policies.
1039 EXPECT_THAT(channel->GetState(false),
1040 ::testing::AnyOf(GRPC_CHANNEL_CONNECTING,
1041 GRPC_CHANNEL_TRANSIENT_FAILURE));
1042 // Now start the second server.
1043 gpr_log(GPR_INFO, "Phase 4: Starting second server.");
1045 // The channel should go to READY state and RPCs should go to the
1047 WaitForChannelReady(channel.get());
1048 WaitForServer(stub, 1, DEBUG_LOCATION, true /* ignore_failure */);
1051 TEST_F(ClientLbEnd2endTest, PickFirstStaysIdleUponEmptyUpdate) {
1052 // Start server, send RPC, and make sure channel is READY.
1053 const int kNumServers = 1;
1054 StartServers(kNumServers);
1055 auto response_generator = BuildResolverResponseGenerator();
1057 BuildChannel("", response_generator); // pick_first is the default.
1058 auto stub = BuildStub(channel);
1059 response_generator.SetNextResolution(GetServersPorts());
1060 CheckRpcSendOk(stub, DEBUG_LOCATION);
1061 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1062 // Stop server. Channel should go into state IDLE.
1063 servers_[0]->Shutdown();
1064 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
1065 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
1066 // Now send resolver update that includes no addresses. Channel
1067 // should stay in state IDLE.
1068 response_generator.SetNextResolution({});
1069 EXPECT_FALSE(channel->WaitForStateChange(
1070 GRPC_CHANNEL_IDLE, grpc_timeout_seconds_to_deadline(3)));
1071 // Now bring the backend back up and send a non-empty resolver update,
1072 // and then try to send an RPC. Channel should go back into state READY.
1074 response_generator.SetNextResolution(GetServersPorts());
1075 CheckRpcSendOk(stub, DEBUG_LOCATION);
1076 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1079 TEST_F(ClientLbEnd2endTest, RoundRobin) {
1080 // Start servers and send one RPC per server.
1081 const int kNumServers = 3;
1082 StartServers(kNumServers);
1083 auto response_generator = BuildResolverResponseGenerator();
1084 auto channel = BuildChannel("round_robin", response_generator);
1085 auto stub = BuildStub(channel);
1086 response_generator.SetNextResolution(GetServersPorts());
1087 // Wait until all backends are ready.
1089 CheckRpcSendOk(stub, DEBUG_LOCATION);
1090 } while (!SeenAllServers());
1092 // "Sync" to the end of the list. Next sequence of picks will start at the
1093 // first server (index 0).
1094 WaitForServer(stub, servers_.size() - 1, DEBUG_LOCATION);
1095 std::vector<int> connection_order;
1096 for (size_t i = 0; i < servers_.size(); ++i) {
1097 CheckRpcSendOk(stub, DEBUG_LOCATION);
1098 UpdateConnectionOrder(servers_, &connection_order);
1100 // Backends should be iterated over in the order in which the addresses were
1102 const auto expected = std::vector<int>{0, 1, 2};
1103 EXPECT_EQ(expected, connection_order);
1104 // Check LB policy name for the channel.
1105 EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
1108 TEST_F(ClientLbEnd2endTest, RoundRobinProcessPending) {
1109 StartServers(1); // Single server
1110 auto response_generator = BuildResolverResponseGenerator();
1111 auto channel = BuildChannel("round_robin", response_generator);
1112 auto stub = BuildStub(channel);
1113 response_generator.SetNextResolution({servers_[0]->port_});
1114 WaitForServer(stub, 0, DEBUG_LOCATION);
1115 // Create a new channel and its corresponding RR LB policy, which will pick
1116 // the subchannels in READY state from the previous RPC against the same
1117 // target (even if it happened over a different channel, because subchannels
1118 // are globally reused). Progress should happen without any transition from
1119 // this READY state.
1120 auto second_response_generator = BuildResolverResponseGenerator();
1121 auto second_channel = BuildChannel("round_robin", second_response_generator);
1122 auto second_stub = BuildStub(second_channel);
1123 second_response_generator.SetNextResolution({servers_[0]->port_});
1124 CheckRpcSendOk(second_stub, DEBUG_LOCATION);
1127 TEST_F(ClientLbEnd2endTest, RoundRobinUpdates) {
1128 // Start servers and send one RPC per server.
1129 const int kNumServers = 3;
1130 StartServers(kNumServers);
1131 auto response_generator = BuildResolverResponseGenerator();
1132 auto channel = BuildChannel("round_robin", response_generator);
1133 auto stub = BuildStub(channel);
1134 std::vector<int> ports;
1135 // Start with a single server.
1136 gpr_log(GPR_INFO, "*** FIRST BACKEND ***");
1137 ports.emplace_back(servers_[0]->port_);
1138 response_generator.SetNextResolution(ports);
1139 WaitForServer(stub, 0, DEBUG_LOCATION);
1140 // Send RPCs. They should all go servers_[0]
1141 for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1142 EXPECT_EQ(10, servers_[0]->service_.request_count());
1143 EXPECT_EQ(0, servers_[1]->service_.request_count());
1144 EXPECT_EQ(0, servers_[2]->service_.request_count());
1145 servers_[0]->service_.ResetCounters();
1146 // And now for the second server.
1147 gpr_log(GPR_INFO, "*** SECOND BACKEND ***");
1149 ports.emplace_back(servers_[1]->port_);
1150 response_generator.SetNextResolution(ports);
1151 // Wait until update has been processed, as signaled by the second backend
1152 // receiving a request.
1153 EXPECT_EQ(0, servers_[1]->service_.request_count());
1154 WaitForServer(stub, 1, DEBUG_LOCATION);
1155 for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1156 EXPECT_EQ(0, servers_[0]->service_.request_count());
1157 EXPECT_EQ(10, servers_[1]->service_.request_count());
1158 EXPECT_EQ(0, servers_[2]->service_.request_count());
1159 servers_[1]->service_.ResetCounters();
1160 // ... and for the last server.
1161 gpr_log(GPR_INFO, "*** THIRD BACKEND ***");
1163 ports.emplace_back(servers_[2]->port_);
1164 response_generator.SetNextResolution(ports);
1165 WaitForServer(stub, 2, DEBUG_LOCATION);
1166 for (size_t i = 0; i < 10; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1167 EXPECT_EQ(0, servers_[0]->service_.request_count());
1168 EXPECT_EQ(0, servers_[1]->service_.request_count());
1169 EXPECT_EQ(10, servers_[2]->service_.request_count());
1170 servers_[2]->service_.ResetCounters();
1171 // Back to all servers.
1172 gpr_log(GPR_INFO, "*** ALL BACKENDS ***");
1174 ports.emplace_back(servers_[0]->port_);
1175 ports.emplace_back(servers_[1]->port_);
1176 ports.emplace_back(servers_[2]->port_);
1177 response_generator.SetNextResolution(ports);
1178 WaitForServer(stub, 0, DEBUG_LOCATION);
1179 WaitForServer(stub, 1, DEBUG_LOCATION);
1180 WaitForServer(stub, 2, DEBUG_LOCATION);
1181 // Send three RPCs, one per server.
1182 for (size_t i = 0; i < 3; ++i) CheckRpcSendOk(stub, DEBUG_LOCATION);
1183 EXPECT_EQ(1, servers_[0]->service_.request_count());
1184 EXPECT_EQ(1, servers_[1]->service_.request_count());
1185 EXPECT_EQ(1, servers_[2]->service_.request_count());
1186 // An empty update will result in the channel going into TRANSIENT_FAILURE.
1187 gpr_log(GPR_INFO, "*** NO BACKENDS ***");
1189 response_generator.SetNextResolution(ports);
1190 grpc_connectivity_state channel_state;
1192 channel_state = channel->GetState(true /* try to connect */);
1193 } while (channel_state == GRPC_CHANNEL_READY);
1194 ASSERT_NE(channel_state, GRPC_CHANNEL_READY);
1195 servers_[0]->service_.ResetCounters();
1196 // Next update introduces servers_[1], making the channel recover.
1197 gpr_log(GPR_INFO, "*** BACK TO SECOND BACKEND ***");
1199 ports.emplace_back(servers_[1]->port_);
1200 response_generator.SetNextResolution(ports);
1201 WaitForServer(stub, 1, DEBUG_LOCATION);
1202 channel_state = channel->GetState(false /* try to connect */);
1203 ASSERT_EQ(channel_state, GRPC_CHANNEL_READY);
1204 // Check LB policy name for the channel.
1205 EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
1208 TEST_F(ClientLbEnd2endTest, RoundRobinUpdateInError) {
1209 const int kNumServers = 3;
1210 StartServers(kNumServers);
1211 auto response_generator = BuildResolverResponseGenerator();
1212 auto channel = BuildChannel("round_robin", response_generator);
1213 auto stub = BuildStub(channel);
1214 std::vector<int> ports;
1215 // Start with a single server.
1216 ports.emplace_back(servers_[0]->port_);
1217 response_generator.SetNextResolution(ports);
1218 WaitForServer(stub, 0, DEBUG_LOCATION);
1219 // Send RPCs. They should all go to servers_[0]
1220 for (size_t i = 0; i < 10; ++i) SendRpc(stub);
1221 EXPECT_EQ(10, servers_[0]->service_.request_count());
1222 EXPECT_EQ(0, servers_[1]->service_.request_count());
1223 EXPECT_EQ(0, servers_[2]->service_.request_count());
1224 servers_[0]->service_.ResetCounters();
1225 // Shutdown one of the servers to be sent in the update.
1226 servers_[1]->Shutdown();
1227 ports.emplace_back(servers_[1]->port_);
1228 ports.emplace_back(servers_[2]->port_);
1229 response_generator.SetNextResolution(ports);
1230 WaitForServer(stub, 0, DEBUG_LOCATION);
1231 WaitForServer(stub, 2, DEBUG_LOCATION);
1232 // Send three RPCs, one per server.
1233 for (size_t i = 0; i < kNumServers; ++i) SendRpc(stub);
1234 // The server in shutdown shouldn't receive any.
1235 EXPECT_EQ(0, servers_[1]->service_.request_count());
1238 TEST_F(ClientLbEnd2endTest, RoundRobinManyUpdates) {
1239 // Start servers and send one RPC per server.
1240 const int kNumServers = 3;
1241 StartServers(kNumServers);
1242 auto response_generator = BuildResolverResponseGenerator();
1243 auto channel = BuildChannel("round_robin", response_generator);
1244 auto stub = BuildStub(channel);
1245 std::vector<int> ports = GetServersPorts();
1246 for (size_t i = 0; i < 1000; ++i) {
1247 std::shuffle(ports.begin(), ports.end(),
1248 std::mt19937(std::random_device()()));
1249 response_generator.SetNextResolution(ports);
1250 if (i % 10 == 0) CheckRpcSendOk(stub, DEBUG_LOCATION);
1252 // Check LB policy name for the channel.
1253 EXPECT_EQ("round_robin", channel->GetLoadBalancingPolicyName());
1256 TEST_F(ClientLbEnd2endTest, RoundRobinConcurrentUpdates) {
1257 // TODO(dgq): replicate the way internal testing exercises the concurrent
1258 // update provisions of RR.
1261 TEST_F(ClientLbEnd2endTest, RoundRobinReresolve) {
1262 // Start servers and send one RPC per server.
1263 const int kNumServers = 3;
1264 std::vector<int> first_ports;
1265 std::vector<int> second_ports;
1266 first_ports.reserve(kNumServers);
1267 for (int i = 0; i < kNumServers; ++i) {
1268 first_ports.push_back(grpc_pick_unused_port_or_die());
1270 second_ports.reserve(kNumServers);
1271 for (int i = 0; i < kNumServers; ++i) {
1272 second_ports.push_back(grpc_pick_unused_port_or_die());
1274 StartServers(kNumServers, first_ports);
1275 auto response_generator = BuildResolverResponseGenerator();
1276 auto channel = BuildChannel("round_robin", response_generator);
1277 auto stub = BuildStub(channel);
1278 response_generator.SetNextResolution(first_ports);
1279 // Send a number of RPCs, which succeed.
1280 for (size_t i = 0; i < 100; ++i) {
1281 CheckRpcSendOk(stub, DEBUG_LOCATION);
1284 gpr_log(GPR_INFO, "****** ABOUT TO KILL SERVERS *******");
1285 for (size_t i = 0; i < servers_.size(); ++i) {
1286 servers_[i]->Shutdown();
1288 gpr_log(GPR_INFO, "****** SERVERS KILLED *******");
1289 gpr_log(GPR_INFO, "****** SENDING DOOMED REQUESTS *******");
1290 // Client requests should fail. Send enough to tickle all subchannels.
1291 for (size_t i = 0; i < servers_.size(); ++i) CheckRpcSendFailure(stub);
1292 gpr_log(GPR_INFO, "****** DOOMED REQUESTS SENT *******");
1293 // Bring servers back up on a different set of ports. We need to do this to be
1294 // sure that the eventual success is *not* due to subchannel reconnection
1295 // attempts and that an actual re-resolution has happened as a result of the
1296 // RR policy going into transient failure when all its subchannels become
1297 // unavailable (in transient failure as well).
1298 gpr_log(GPR_INFO, "****** RESTARTING SERVERS *******");
1299 StartServers(kNumServers, second_ports);
1300 // Don't notify of the update. Wait for the LB policy's re-resolution to
1301 // "pull" the new ports.
1302 response_generator.SetNextResolutionUponError(second_ports);
1303 gpr_log(GPR_INFO, "****** SERVERS RESTARTED *******");
1304 gpr_log(GPR_INFO, "****** SENDING REQUEST TO SUCCEED *******");
1305 // Client request should eventually (but still fairly soon) succeed.
1306 const gpr_timespec deadline = grpc_timeout_seconds_to_deadline(5);
1307 gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC);
1308 while (gpr_time_cmp(deadline, now) > 0) {
1309 if (SendRpc(stub)) break;
1310 now = gpr_now(GPR_CLOCK_MONOTONIC);
1312 ASSERT_GT(gpr_time_cmp(deadline, now), 0);
1315 TEST_F(ClientLbEnd2endTest, RoundRobinTransientFailure) {
1316 // Start servers and create channel. Channel should go to READY state.
1317 const int kNumServers = 3;
1318 StartServers(kNumServers);
1319 auto response_generator = BuildResolverResponseGenerator();
1320 auto channel = BuildChannel("round_robin", response_generator);
1321 auto stub = BuildStub(channel);
1322 response_generator.SetNextResolution(GetServersPorts());
1323 EXPECT_TRUE(WaitForChannelReady(channel.get()));
1324 // Now kill the servers. The channel should transition to TRANSIENT_FAILURE.
1325 // TODO(roth): This test should ideally check that even when the
1326 // subchannels are in state CONNECTING for an extended period of time,
1327 // we will still report TRANSIENT_FAILURE. Unfortunately, we don't
1328 // currently have a good way to get a subchannel to report CONNECTING
1329 // for a long period of time, since the servers in this test framework
1330 // are on the loopback interface, which will immediately return a
1331 // "Connection refused" error, so the subchannels will only be in
1332 // CONNECTING state very briefly. When we have time, see if we can
1333 // find a way to fix this.
1334 for (size_t i = 0; i < servers_.size(); ++i) {
1335 servers_[i]->Shutdown();
1337 auto predicate = [](grpc_connectivity_state state) {
1338 return state == GRPC_CHANNEL_TRANSIENT_FAILURE;
1340 EXPECT_TRUE(WaitForChannelState(channel.get(), predicate));
1343 TEST_F(ClientLbEnd2endTest, RoundRobinTransientFailureAtStartup) {
1344 // Create channel and return servers that don't exist. Channel should
1345 // quickly transition into TRANSIENT_FAILURE.
1346 // TODO(roth): This test should ideally check that even when the
1347 // subchannels are in state CONNECTING for an extended period of time,
1348 // we will still report TRANSIENT_FAILURE. Unfortunately, we don't
1349 // currently have a good way to get a subchannel to report CONNECTING
1350 // for a long period of time, since the servers in this test framework
1351 // are on the loopback interface, which will immediately return a
1352 // "Connection refused" error, so the subchannels will only be in
1353 // CONNECTING state very briefly. When we have time, see if we can
1354 // find a way to fix this.
1355 auto response_generator = BuildResolverResponseGenerator();
1356 auto channel = BuildChannel("round_robin", response_generator);
1357 auto stub = BuildStub(channel);
1358 response_generator.SetNextResolution({
1359 grpc_pick_unused_port_or_die(),
1360 grpc_pick_unused_port_or_die(),
1361 grpc_pick_unused_port_or_die(),
1363 for (size_t i = 0; i < servers_.size(); ++i) {
1364 servers_[i]->Shutdown();
1366 auto predicate = [](grpc_connectivity_state state) {
1367 return state == GRPC_CHANNEL_TRANSIENT_FAILURE;
1369 EXPECT_TRUE(WaitForChannelState(channel.get(), predicate, true));
1372 TEST_F(ClientLbEnd2endTest, RoundRobinSingleReconnect) {
1373 const int kNumServers = 3;
1374 StartServers(kNumServers);
1375 const auto ports = GetServersPorts();
1376 auto response_generator = BuildResolverResponseGenerator();
1377 auto channel = BuildChannel("round_robin", response_generator);
1378 auto stub = BuildStub(channel);
1379 response_generator.SetNextResolution(ports);
1380 for (size_t i = 0; i < kNumServers; ++i) {
1381 WaitForServer(stub, i, DEBUG_LOCATION);
1383 for (size_t i = 0; i < servers_.size(); ++i) {
1384 CheckRpcSendOk(stub, DEBUG_LOCATION);
1385 EXPECT_EQ(1, servers_[i]->service_.request_count()) << "for backend #" << i;
1387 // One request should have gone to each server.
1388 for (size_t i = 0; i < servers_.size(); ++i) {
1389 EXPECT_EQ(1, servers_[i]->service_.request_count());
1391 const auto pre_death = servers_[0]->service_.request_count();
1392 // Kill the first server.
1393 servers_[0]->Shutdown();
1394 // Client request still succeed. May need retrying if RR had returned a pick
1395 // before noticing the change in the server's connectivity.
1396 while (!SendRpc(stub)) {
1397 } // Retry until success.
1398 // Send a bunch of RPCs that should succeed.
1399 for (int i = 0; i < 10 * kNumServers; ++i) {
1400 CheckRpcSendOk(stub, DEBUG_LOCATION);
1402 const auto post_death = servers_[0]->service_.request_count();
1403 // No requests have gone to the deceased server.
1404 EXPECT_EQ(pre_death, post_death);
1405 // Bring the first server back up.
1407 // Requests should start arriving at the first server either right away (if
1408 // the server managed to start before the RR policy retried the subchannel) or
1409 // after the subchannel retry delay otherwise (RR's subchannel retried before
1410 // the server was fully back up).
1411 WaitForServer(stub, 0, DEBUG_LOCATION);
1414 // If health checking is required by client but health checking service
1415 // is not running on the server, the channel should be treated as healthy.
1416 TEST_F(ClientLbEnd2endTest,
1417 RoundRobinServersHealthCheckingUnimplementedTreatedAsHealthy) {
1418 StartServers(1); // Single server
1419 ChannelArguments args;
1420 args.SetServiceConfigJSON(
1421 "{\"healthCheckConfig\": "
1422 "{\"serviceName\": \"health_check_service_name\"}}");
1423 auto response_generator = BuildResolverResponseGenerator();
1424 auto channel = BuildChannel("round_robin", response_generator, args);
1425 auto stub = BuildStub(channel);
1426 response_generator.SetNextResolution({servers_[0]->port_});
1427 EXPECT_TRUE(WaitForChannelReady(channel.get()));
1428 CheckRpcSendOk(stub, DEBUG_LOCATION);
1431 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthChecking) {
1432 EnableDefaultHealthCheckService(true);
1434 const int kNumServers = 3;
1435 StartServers(kNumServers);
1436 ChannelArguments args;
1437 args.SetServiceConfigJSON(
1438 "{\"healthCheckConfig\": "
1439 "{\"serviceName\": \"health_check_service_name\"}}");
1440 auto response_generator = BuildResolverResponseGenerator();
1441 auto channel = BuildChannel("round_robin", response_generator, args);
1442 auto stub = BuildStub(channel);
1443 response_generator.SetNextResolution(GetServersPorts());
1444 // Channel should not become READY, because health checks should be failing.
1446 "*** initial state: unknown health check service name for "
1448 EXPECT_FALSE(WaitForChannelReady(channel.get(), 1));
1449 // Now set one of the servers to be healthy.
1450 // The channel should become healthy and all requests should go to
1451 // the healthy server.
1452 gpr_log(GPR_INFO, "*** server 0 healthy");
1453 servers_[0]->SetServingStatus("health_check_service_name", true);
1454 EXPECT_TRUE(WaitForChannelReady(channel.get()));
1455 for (int i = 0; i < 10; ++i) {
1456 CheckRpcSendOk(stub, DEBUG_LOCATION);
1458 EXPECT_EQ(10, servers_[0]->service_.request_count());
1459 EXPECT_EQ(0, servers_[1]->service_.request_count());
1460 EXPECT_EQ(0, servers_[2]->service_.request_count());
1461 // Now set a second server to be healthy.
1462 gpr_log(GPR_INFO, "*** server 2 healthy");
1463 servers_[2]->SetServingStatus("health_check_service_name", true);
1464 WaitForServer(stub, 2, DEBUG_LOCATION);
1465 for (int i = 0; i < 10; ++i) {
1466 CheckRpcSendOk(stub, DEBUG_LOCATION);
1468 EXPECT_EQ(5, servers_[0]->service_.request_count());
1469 EXPECT_EQ(0, servers_[1]->service_.request_count());
1470 EXPECT_EQ(5, servers_[2]->service_.request_count());
1471 // Now set the remaining server to be healthy.
1472 gpr_log(GPR_INFO, "*** server 1 healthy");
1473 servers_[1]->SetServingStatus("health_check_service_name", true);
1474 WaitForServer(stub, 1, DEBUG_LOCATION);
1475 for (int i = 0; i < 9; ++i) {
1476 CheckRpcSendOk(stub, DEBUG_LOCATION);
1478 EXPECT_EQ(3, servers_[0]->service_.request_count());
1479 EXPECT_EQ(3, servers_[1]->service_.request_count());
1480 EXPECT_EQ(3, servers_[2]->service_.request_count());
1481 // Now set one server to be unhealthy again. Then wait until the
1482 // unhealthiness has hit the client. We know that the client will see
1483 // this when we send kNumServers requests and one of the remaining servers
1484 // sees two of the requests.
1485 gpr_log(GPR_INFO, "*** server 0 unhealthy");
1486 servers_[0]->SetServingStatus("health_check_service_name", false);
1489 for (int i = 0; i < kNumServers; ++i) {
1490 CheckRpcSendOk(stub, DEBUG_LOCATION);
1492 } while (servers_[1]->service_.request_count() != 2 &&
1493 servers_[2]->service_.request_count() != 2);
1494 // Now set the remaining two servers to be unhealthy. Make sure the
1495 // channel leaves READY state and that RPCs fail.
1496 gpr_log(GPR_INFO, "*** all servers unhealthy");
1497 servers_[1]->SetServingStatus("health_check_service_name", false);
1498 servers_[2]->SetServingStatus("health_check_service_name", false);
1499 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
1500 CheckRpcSendFailure(stub);
1502 EnableDefaultHealthCheckService(false);
1505 TEST_F(ClientLbEnd2endTest,
1506 RoundRobinWithHealthCheckingHandlesSubchannelFailure) {
1507 EnableDefaultHealthCheckService(true);
1509 const int kNumServers = 3;
1510 StartServers(kNumServers);
1511 servers_[0]->SetServingStatus("health_check_service_name", true);
1512 servers_[1]->SetServingStatus("health_check_service_name", true);
1513 servers_[2]->SetServingStatus("health_check_service_name", true);
1514 ChannelArguments args;
1515 args.SetServiceConfigJSON(
1516 "{\"healthCheckConfig\": "
1517 "{\"serviceName\": \"health_check_service_name\"}}");
1518 auto response_generator = BuildResolverResponseGenerator();
1519 auto channel = BuildChannel("round_robin", response_generator, args);
1520 auto stub = BuildStub(channel);
1521 response_generator.SetNextResolution(GetServersPorts());
1522 WaitForServer(stub, 0, DEBUG_LOCATION);
1523 // Stop server 0 and send a new resolver result to ensure that RR
1524 // checks each subchannel's state.
1525 servers_[0]->Shutdown();
1526 response_generator.SetNextResolution(GetServersPorts());
1527 // Send a bunch more RPCs.
1528 for (size_t i = 0; i < 100; i++) {
1533 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthCheckingInhibitPerChannel) {
1534 EnableDefaultHealthCheckService(true);
1536 const int kNumServers = 1;
1537 StartServers(kNumServers);
1538 // Create a channel with health-checking enabled.
1539 ChannelArguments args;
1540 args.SetServiceConfigJSON(
1541 "{\"healthCheckConfig\": "
1542 "{\"serviceName\": \"health_check_service_name\"}}");
1543 auto response_generator1 = BuildResolverResponseGenerator();
1544 auto channel1 = BuildChannel("round_robin", response_generator1, args);
1545 auto stub1 = BuildStub(channel1);
1546 std::vector<int> ports = GetServersPorts();
1547 response_generator1.SetNextResolution(ports);
1548 // Create a channel with health checking enabled but inhibited.
1549 args.SetInt(GRPC_ARG_INHIBIT_HEALTH_CHECKING, 1);
1550 auto response_generator2 = BuildResolverResponseGenerator();
1551 auto channel2 = BuildChannel("round_robin", response_generator2, args);
1552 auto stub2 = BuildStub(channel2);
1553 response_generator2.SetNextResolution(ports);
1554 // First channel should not become READY, because health checks should be
1556 EXPECT_FALSE(WaitForChannelReady(channel1.get(), 1));
1557 CheckRpcSendFailure(stub1);
1558 // Second channel should be READY.
1559 EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1));
1560 CheckRpcSendOk(stub2, DEBUG_LOCATION);
1561 // Enable health checks on the backend and wait for channel 1 to succeed.
1562 servers_[0]->SetServingStatus("health_check_service_name", true);
1563 CheckRpcSendOk(stub1, DEBUG_LOCATION, true /* wait_for_ready */);
1564 // Check that we created only one subchannel to the backend.
1565 EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
1567 EnableDefaultHealthCheckService(false);
1570 TEST_F(ClientLbEnd2endTest, RoundRobinWithHealthCheckingServiceNamePerChannel) {
1571 EnableDefaultHealthCheckService(true);
1573 const int kNumServers = 1;
1574 StartServers(kNumServers);
1575 // Create a channel with health-checking enabled.
1576 ChannelArguments args;
1577 args.SetServiceConfigJSON(
1578 "{\"healthCheckConfig\": "
1579 "{\"serviceName\": \"health_check_service_name\"}}");
1580 auto response_generator1 = BuildResolverResponseGenerator();
1581 auto channel1 = BuildChannel("round_robin", response_generator1, args);
1582 auto stub1 = BuildStub(channel1);
1583 std::vector<int> ports = GetServersPorts();
1584 response_generator1.SetNextResolution(ports);
1585 // Create a channel with health-checking enabled with a different
1587 ChannelArguments args2;
1588 args2.SetServiceConfigJSON(
1589 "{\"healthCheckConfig\": "
1590 "{\"serviceName\": \"health_check_service_name2\"}}");
1591 auto response_generator2 = BuildResolverResponseGenerator();
1592 auto channel2 = BuildChannel("round_robin", response_generator2, args2);
1593 auto stub2 = BuildStub(channel2);
1594 response_generator2.SetNextResolution(ports);
1595 // Allow health checks from channel 2 to succeed.
1596 servers_[0]->SetServingStatus("health_check_service_name2", true);
1597 // First channel should not become READY, because health checks should be
1599 EXPECT_FALSE(WaitForChannelReady(channel1.get(), 1));
1600 CheckRpcSendFailure(stub1);
1601 // Second channel should be READY.
1602 EXPECT_TRUE(WaitForChannelReady(channel2.get(), 1));
1603 CheckRpcSendOk(stub2, DEBUG_LOCATION);
1604 // Enable health checks for channel 1 and wait for it to succeed.
1605 servers_[0]->SetServingStatus("health_check_service_name", true);
1606 CheckRpcSendOk(stub1, DEBUG_LOCATION, true /* wait_for_ready */);
1607 // Check that we created only one subchannel to the backend.
1608 EXPECT_EQ(1UL, servers_[0]->service_.clients().size());
1610 EnableDefaultHealthCheckService(false);
1613 TEST_F(ClientLbEnd2endTest,
1614 RoundRobinWithHealthCheckingServiceNameChangesAfterSubchannelsCreated) {
1615 EnableDefaultHealthCheckService(true);
1617 const int kNumServers = 1;
1618 StartServers(kNumServers);
1619 // Create a channel with health-checking enabled.
1620 const char* kServiceConfigJson =
1621 "{\"healthCheckConfig\": "
1622 "{\"serviceName\": \"health_check_service_name\"}}";
1623 auto response_generator = BuildResolverResponseGenerator();
1624 auto channel = BuildChannel("round_robin", response_generator);
1625 auto stub = BuildStub(channel);
1626 std::vector<int> ports = GetServersPorts();
1627 response_generator.SetNextResolution(ports, kServiceConfigJson);
1628 servers_[0]->SetServingStatus("health_check_service_name", true);
1629 EXPECT_TRUE(WaitForChannelReady(channel.get(), 1 /* timeout_seconds */));
1630 // Send an update on the channel to change it to use a health checking
1631 // service name that is not being reported as healthy.
1632 const char* kServiceConfigJson2 =
1633 "{\"healthCheckConfig\": "
1634 "{\"serviceName\": \"health_check_service_name2\"}}";
1635 response_generator.SetNextResolution(ports, kServiceConfigJson2);
1636 EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
1638 EnableDefaultHealthCheckService(false);
1641 TEST_F(ClientLbEnd2endTest, ChannelIdleness) {
1643 const int kNumServers = 1;
1644 StartServers(kNumServers);
1645 // Set max idle time and build the channel.
1646 ChannelArguments args;
1647 args.SetInt(GRPC_ARG_CLIENT_IDLE_TIMEOUT_MS, 1000);
1648 auto response_generator = BuildResolverResponseGenerator();
1649 auto channel = BuildChannel("", response_generator, args);
1650 auto stub = BuildStub(channel);
1651 // The initial channel state should be IDLE.
1652 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
1653 // After sending RPC, channel state should be READY.
1654 response_generator.SetNextResolution(GetServersPorts());
1655 CheckRpcSendOk(stub, DEBUG_LOCATION);
1656 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1657 // After a period time not using the channel, the channel state should switch
1659 gpr_sleep_until(grpc_timeout_milliseconds_to_deadline(1200));
1660 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_IDLE);
1661 // Sending a new RPC should awake the IDLE channel.
1662 response_generator.SetNextResolution(GetServersPorts());
1663 CheckRpcSendOk(stub, DEBUG_LOCATION);
1664 EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
1667 class ClientLbPickArgsTest : public ClientLbEnd2endTest {
1669 void SetUp() override {
1670 ClientLbEnd2endTest::SetUp();
1671 current_test_instance_ = this;
1674 static void SetUpTestCase() {
1676 grpc_core::RegisterTestPickArgsLoadBalancingPolicy(SavePickArgs);
1679 static void TearDownTestCase() { grpc_shutdown(); }
1681 const std::vector<grpc_core::PickArgsSeen>& args_seen_list() {
1682 grpc::internal::MutexLock lock(&mu_);
1683 return args_seen_list_;
1687 static void SavePickArgs(const grpc_core::PickArgsSeen& args_seen) {
1688 ClientLbPickArgsTest* self = current_test_instance_;
1689 grpc::internal::MutexLock lock(&self->mu_);
1690 self->args_seen_list_.emplace_back(args_seen);
1693 static ClientLbPickArgsTest* current_test_instance_;
1694 grpc::internal::Mutex mu_;
1695 std::vector<grpc_core::PickArgsSeen> args_seen_list_;
1698 ClientLbPickArgsTest* ClientLbPickArgsTest::current_test_instance_ = nullptr;
1700 TEST_F(ClientLbPickArgsTest, Basic) {
1701 const int kNumServers = 1;
1702 StartServers(kNumServers);
1703 auto response_generator = BuildResolverResponseGenerator();
1704 auto channel = BuildChannel("test_pick_args_lb", response_generator);
1705 auto stub = BuildStub(channel);
1706 response_generator.SetNextResolution(GetServersPorts());
1707 CheckRpcSendOk(stub, DEBUG_LOCATION, /*wait_for_ready=*/true);
1708 // Check LB policy name for the channel.
1709 EXPECT_EQ("test_pick_args_lb", channel->GetLoadBalancingPolicyName());
1710 // There will be two entries, one for the pick tried in state
1711 // CONNECTING and another for the pick tried in state READY.
1712 EXPECT_THAT(args_seen_list(),
1713 ::testing::ElementsAre(
1715 ::testing::Field(&grpc_core::PickArgsSeen::path,
1716 "/grpc.testing.EchoTestService/Echo"),
1717 ::testing::Field(&grpc_core::PickArgsSeen::metadata,
1718 ::testing::UnorderedElementsAre(
1719 ::testing::Pair("foo", "1"),
1720 ::testing::Pair("bar", "2"),
1721 ::testing::Pair("baz", "3")))),
1723 ::testing::Field(&grpc_core::PickArgsSeen::path,
1724 "/grpc.testing.EchoTestService/Echo"),
1725 ::testing::Field(&grpc_core::PickArgsSeen::metadata,
1726 ::testing::UnorderedElementsAre(
1727 ::testing::Pair("foo", "1"),
1728 ::testing::Pair("bar", "2"),
1729 ::testing::Pair("baz", "3"))))));
1732 class ClientLbInterceptTrailingMetadataTest : public ClientLbEnd2endTest {
1734 void SetUp() override {
1735 ClientLbEnd2endTest::SetUp();
1736 current_test_instance_ = this;
1739 static void SetUpTestCase() {
1741 grpc_core::RegisterInterceptRecvTrailingMetadataLoadBalancingPolicy(
1742 ReportTrailerIntercepted);
1745 static void TearDownTestCase() { grpc_shutdown(); }
1747 int trailers_intercepted() {
1748 grpc::internal::MutexLock lock(&mu_);
1749 return trailers_intercepted_;
1752 const grpc_core::MetadataVector& trailing_metadata() {
1753 grpc::internal::MutexLock lock(&mu_);
1754 return trailing_metadata_;
1757 const udpa::data::orca::v1::OrcaLoadReport* backend_load_report() {
1758 grpc::internal::MutexLock lock(&mu_);
1759 return load_report_.get();
1763 static void ReportTrailerIntercepted(
1764 const grpc_core::TrailingMetadataArgsSeen& args_seen) {
1765 const auto* backend_metric_data = args_seen.backend_metric_data;
1766 ClientLbInterceptTrailingMetadataTest* self = current_test_instance_;
1767 grpc::internal::MutexLock lock(&self->mu_);
1768 self->trailers_intercepted_++;
1769 self->trailing_metadata_ = args_seen.metadata;
1770 if (backend_metric_data != nullptr) {
1771 self->load_report_ =
1772 absl::make_unique<udpa::data::orca::v1::OrcaLoadReport>();
1773 self->load_report_->set_cpu_utilization(
1774 backend_metric_data->cpu_utilization);
1775 self->load_report_->set_mem_utilization(
1776 backend_metric_data->mem_utilization);
1777 self->load_report_->set_rps(backend_metric_data->requests_per_second);
1778 for (const auto& p : backend_metric_data->request_cost) {
1779 std::string name = std::string(p.first);
1780 (*self->load_report_->mutable_request_cost())[name] = p.second;
1782 for (const auto& p : backend_metric_data->utilization) {
1783 std::string name = std::string(p.first);
1784 (*self->load_report_->mutable_utilization())[name] = p.second;
1789 static ClientLbInterceptTrailingMetadataTest* current_test_instance_;
1790 grpc::internal::Mutex mu_;
1791 int trailers_intercepted_ = 0;
1792 grpc_core::MetadataVector trailing_metadata_;
1793 std::unique_ptr<udpa::data::orca::v1::OrcaLoadReport> load_report_;
1796 ClientLbInterceptTrailingMetadataTest*
1797 ClientLbInterceptTrailingMetadataTest::current_test_instance_ = nullptr;
1799 TEST_F(ClientLbInterceptTrailingMetadataTest, InterceptsRetriesDisabled) {
1800 const int kNumServers = 1;
1801 const int kNumRpcs = 10;
1802 StartServers(kNumServers);
1803 auto response_generator = BuildResolverResponseGenerator();
1805 BuildChannel("intercept_trailing_metadata_lb", response_generator);
1806 auto stub = BuildStub(channel);
1807 response_generator.SetNextResolution(GetServersPorts());
1808 for (size_t i = 0; i < kNumRpcs; ++i) {
1809 CheckRpcSendOk(stub, DEBUG_LOCATION);
1811 // Check LB policy name for the channel.
1812 EXPECT_EQ("intercept_trailing_metadata_lb",
1813 channel->GetLoadBalancingPolicyName());
1814 EXPECT_EQ(kNumRpcs, trailers_intercepted());
1815 EXPECT_THAT(trailing_metadata(),
1816 ::testing::UnorderedElementsAre(
1817 // TODO(roth): Should grpc-status be visible here?
1818 ::testing::Pair("grpc-status", "0"),
1819 ::testing::Pair("user-agent", ::testing::_),
1820 ::testing::Pair("foo", "1"), ::testing::Pair("bar", "2"),
1821 ::testing::Pair("baz", "3")));
1822 EXPECT_EQ(nullptr, backend_load_report());
1825 TEST_F(ClientLbInterceptTrailingMetadataTest, InterceptsRetriesEnabled) {
1826 const int kNumServers = 1;
1827 const int kNumRpcs = 10;
1828 StartServers(kNumServers);
1829 ChannelArguments args;
1830 args.SetServiceConfigJSON(
1832 " \"methodConfig\": [ {\n"
1834 " { \"service\": \"grpc.testing.EchoTestService\" }\n"
1836 " \"retryPolicy\": {\n"
1837 " \"maxAttempts\": 3,\n"
1838 " \"initialBackoff\": \"1s\",\n"
1839 " \"maxBackoff\": \"120s\",\n"
1840 " \"backoffMultiplier\": 1.6,\n"
1841 " \"retryableStatusCodes\": [ \"ABORTED\" ]\n"
1845 auto response_generator = BuildResolverResponseGenerator();
1847 BuildChannel("intercept_trailing_metadata_lb", response_generator, args);
1848 auto stub = BuildStub(channel);
1849 response_generator.SetNextResolution(GetServersPorts());
1850 for (size_t i = 0; i < kNumRpcs; ++i) {
1851 CheckRpcSendOk(stub, DEBUG_LOCATION);
1853 // Check LB policy name for the channel.
1854 EXPECT_EQ("intercept_trailing_metadata_lb",
1855 channel->GetLoadBalancingPolicyName());
1856 EXPECT_EQ(kNumRpcs, trailers_intercepted());
1857 EXPECT_THAT(trailing_metadata(),
1858 ::testing::UnorderedElementsAre(
1859 // TODO(roth): Should grpc-status be visible here?
1860 ::testing::Pair("grpc-status", "0"),
1861 ::testing::Pair("user-agent", ::testing::_),
1862 ::testing::Pair("foo", "1"), ::testing::Pair("bar", "2"),
1863 ::testing::Pair("baz", "3")));
1864 EXPECT_EQ(nullptr, backend_load_report());
1867 TEST_F(ClientLbInterceptTrailingMetadataTest, BackendMetricData) {
1868 const int kNumServers = 1;
1869 const int kNumRpcs = 10;
1870 StartServers(kNumServers);
1871 udpa::data::orca::v1::OrcaLoadReport load_report;
1872 load_report.set_cpu_utilization(0.5);
1873 load_report.set_mem_utilization(0.75);
1874 load_report.set_rps(25);
1875 auto* request_cost = load_report.mutable_request_cost();
1876 (*request_cost)["foo"] = 0.8;
1877 (*request_cost)["bar"] = 1.4;
1878 auto* utilization = load_report.mutable_utilization();
1879 (*utilization)["baz"] = 1.1;
1880 (*utilization)["quux"] = 0.9;
1881 for (const auto& server : servers_) {
1882 server->service_.set_load_report(&load_report);
1884 auto response_generator = BuildResolverResponseGenerator();
1886 BuildChannel("intercept_trailing_metadata_lb", response_generator);
1887 auto stub = BuildStub(channel);
1888 response_generator.SetNextResolution(GetServersPorts());
1889 for (size_t i = 0; i < kNumRpcs; ++i) {
1890 CheckRpcSendOk(stub, DEBUG_LOCATION);
1891 auto* actual = backend_load_report();
1892 ASSERT_NE(actual, nullptr);
1893 // TODO(roth): Change this to use EqualsProto() once that becomes
1894 // available in OSS.
1895 EXPECT_EQ(actual->cpu_utilization(), load_report.cpu_utilization());
1896 EXPECT_EQ(actual->mem_utilization(), load_report.mem_utilization());
1897 EXPECT_EQ(actual->rps(), load_report.rps());
1898 EXPECT_EQ(actual->request_cost().size(), load_report.request_cost().size());
1899 for (const auto& p : actual->request_cost()) {
1900 auto it = load_report.request_cost().find(p.first);
1901 ASSERT_NE(it, load_report.request_cost().end());
1902 EXPECT_EQ(it->second, p.second);
1904 EXPECT_EQ(actual->utilization().size(), load_report.utilization().size());
1905 for (const auto& p : actual->utilization()) {
1906 auto it = load_report.utilization().find(p.first);
1907 ASSERT_NE(it, load_report.utilization().end());
1908 EXPECT_EQ(it->second, p.second);
1911 // Check LB policy name for the channel.
1912 EXPECT_EQ("intercept_trailing_metadata_lb",
1913 channel->GetLoadBalancingPolicyName());
1914 EXPECT_EQ(kNumRpcs, trailers_intercepted());
1917 class ClientLbAddressTest : public ClientLbEnd2endTest {
1919 static const char* kAttributeKey;
1921 class Attribute : public grpc_core::ServerAddress::AttributeInterface {
1923 explicit Attribute(const std::string& str) : str_(str) {}
1925 std::unique_ptr<AttributeInterface> Copy() const override {
1926 return absl::make_unique<Attribute>(str_);
1929 int Cmp(const AttributeInterface* other) const override {
1930 return str_.compare(static_cast<const Attribute*>(other)->str_);
1933 std::string ToString() const override { return str_; }
1939 void SetUp() override {
1940 ClientLbEnd2endTest::SetUp();
1941 current_test_instance_ = this;
1944 static void SetUpTestCase() {
1946 grpc_core::RegisterAddressTestLoadBalancingPolicy(SaveAddress);
1949 static void TearDownTestCase() { grpc_shutdown(); }
1951 const std::vector<std::string>& addresses_seen() {
1952 grpc::internal::MutexLock lock(&mu_);
1953 return addresses_seen_;
1957 static void SaveAddress(const grpc_core::ServerAddress& address) {
1958 ClientLbAddressTest* self = current_test_instance_;
1959 grpc::internal::MutexLock lock(&self->mu_);
1960 self->addresses_seen_.emplace_back(address.ToString());
1963 static ClientLbAddressTest* current_test_instance_;
1964 grpc::internal::Mutex mu_;
1965 std::vector<std::string> addresses_seen_;
1968 const char* ClientLbAddressTest::kAttributeKey = "attribute_key";
1970 ClientLbAddressTest* ClientLbAddressTest::current_test_instance_ = nullptr;
1972 TEST_F(ClientLbAddressTest, Basic) {
1973 const int kNumServers = 1;
1974 StartServers(kNumServers);
1975 auto response_generator = BuildResolverResponseGenerator();
1976 auto channel = BuildChannel("address_test_lb", response_generator);
1977 auto stub = BuildStub(channel);
1978 // Addresses returned by the resolver will have attached attributes.
1979 response_generator.SetNextResolution(GetServersPorts(), nullptr,
1981 absl::make_unique<Attribute>("foo"));
1982 CheckRpcSendOk(stub, DEBUG_LOCATION);
1983 // Check LB policy name for the channel.
1984 EXPECT_EQ("address_test_lb", channel->GetLoadBalancingPolicyName());
1985 // Make sure that the attributes wind up on the subchannels.
1986 std::vector<std::string> expected;
1987 for (const int port : GetServersPorts()) {
1988 expected.emplace_back(
1989 absl::StrCat(ipv6_only_ ? "[::1]:" : "127.0.0.1:", port,
1990 " args={} attributes={", kAttributeKey, "=foo}"));
1992 EXPECT_EQ(addresses_seen(), expected);
1996 } // namespace testing
1999 int main(int argc, char** argv) {
2000 ::testing::InitGoogleTest(&argc, argv);
2001 grpc::testing::TestEnvironment env(argc, argv);
2002 const auto result = RUN_ALL_TESTS();