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