3 * Copyright 2016 gRPC authors.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
27 #include "absl/memory/memory.h"
28 #include "absl/strings/str_cat.h"
30 #include <grpc/grpc.h>
31 #include <grpc/support/alloc.h>
32 #include <grpc/support/atm.h>
33 #include <grpc/support/log.h>
34 #include <grpc/support/time.h>
35 #include <grpcpp/channel.h>
36 #include <grpcpp/client_context.h>
37 #include <grpcpp/create_channel.h>
38 #include <grpcpp/health_check_service_interface.h>
39 #include <grpcpp/impl/codegen/sync.h>
40 #include <grpcpp/server.h>
41 #include <grpcpp/server_builder.h>
42 #include <grpcpp/support/validate_service_config.h>
44 #include "src/core/ext/filters/client_channel/backup_poller.h"
45 #include "src/core/ext/filters/client_channel/global_subchannel_pool.h"
46 #include "src/core/ext/filters/client_channel/resolver/fake/fake_resolver.h"
47 #include "src/core/ext/filters/client_channel/server_address.h"
48 #include "src/core/lib/backoff/backoff.h"
49 #include "src/core/lib/channel/channel_args.h"
50 #include "src/core/lib/gprpp/debug_location.h"
51 #include "src/core/lib/gprpp/ref_counted_ptr.h"
52 #include "src/core/lib/iomgr/parse_address.h"
53 #include "src/core/lib/iomgr/tcp_client.h"
54 #include "src/core/lib/security/credentials/fake/fake_credentials.h"
55 #include "src/cpp/client/secure_credentials.h"
56 #include "src/cpp/server/secure_server_credentials.h"
58 #include "src/proto/grpc/testing/echo.grpc.pb.h"
59 #include "test/core/util/port.h"
60 #include "test/core/util/resolve_localhost_ip46.h"
61 #include "test/core/util/test_config.h"
62 #include "test/cpp/end2end/test_service_impl.h"
64 #include <gmock/gmock.h>
65 #include <gtest/gtest.h>
67 using grpc::testing::EchoRequest;
68 using grpc::testing::EchoResponse;
74 // Subclass of TestServiceImpl that increments a request counter for
75 // every call to the Echo RPC.
76 class MyTestServiceImpl : public TestServiceImpl {
78 MyTestServiceImpl() : request_count_(0) {}
80 Status Echo(ServerContext* context, const EchoRequest* request,
81 EchoResponse* response) override {
83 grpc::internal::MutexLock lock(&mu_);
86 AddClient(context->peer());
87 return TestServiceImpl::Echo(context, request, response);
91 grpc::internal::MutexLock lock(&mu_);
92 return request_count_;
95 void ResetCounters() {
96 grpc::internal::MutexLock lock(&mu_);
100 std::set<std::string> clients() {
101 grpc::internal::MutexLock lock(&clients_mu_);
106 void AddClient(const std::string& client) {
107 grpc::internal::MutexLock lock(&clients_mu_);
108 clients_.insert(client);
111 grpc::internal::Mutex mu_;
113 grpc::internal::Mutex clients_mu_;
114 std::set<std::string> clients_;
117 class ServiceConfigEnd2endTest : public ::testing::Test {
119 ServiceConfigEnd2endTest()
120 : server_host_("localhost"),
121 kRequestMessage_("Live long and prosper."),
122 creds_(new SecureChannelCredentials(
123 grpc_fake_transport_security_credentials_create())) {}
125 static void SetUpTestCase() {
126 // Make the backup poller poll very frequently in order to pick up
127 // updates from all the subchannels's FDs.
128 GPR_GLOBAL_CONFIG_SET(grpc_client_channel_backup_poll_interval_ms, 1);
131 void SetUp() override {
133 response_generator_ =
134 grpc_core::MakeRefCounted<grpc_core::FakeResolverResponseGenerator>();
135 bool localhost_resolves_to_ipv4 = false;
136 bool localhost_resolves_to_ipv6 = false;
137 grpc_core::LocalhostResolves(&localhost_resolves_to_ipv4,
138 &localhost_resolves_to_ipv6);
139 ipv6_only_ = !localhost_resolves_to_ipv4 && localhost_resolves_to_ipv6;
142 void TearDown() override {
143 for (size_t i = 0; i < servers_.size(); ++i) {
144 servers_[i]->Shutdown();
146 // Explicitly destroy all the members so that we can make sure grpc_shutdown
147 // has finished by the end of this function, and thus all the registered
148 // LB policy factories are removed.
155 void CreateServers(size_t num_servers,
156 std::vector<int> ports = std::vector<int>()) {
158 for (size_t i = 0; i < num_servers; ++i) {
160 if (ports.size() == num_servers) port = ports[i];
161 servers_.emplace_back(new ServerData(port));
165 void StartServer(size_t index) { servers_[index]->Start(server_host_); }
167 void StartServers(size_t num_servers,
168 std::vector<int> ports = std::vector<int>()) {
169 CreateServers(num_servers, std::move(ports));
170 for (size_t i = 0; i < num_servers; ++i) {
175 grpc_core::Resolver::Result BuildFakeResults(const std::vector<int>& ports) {
176 grpc_core::Resolver::Result result;
177 for (const int& port : ports) {
178 std::string lb_uri_str =
179 absl::StrCat(ipv6_only_ ? "ipv6:[::1]:" : "ipv4:127.0.0.1:", port);
180 absl::StatusOr<grpc_core::URI> lb_uri = grpc_core::URI::Parse(lb_uri_str);
181 GPR_ASSERT(lb_uri.ok());
182 grpc_resolved_address address;
183 GPR_ASSERT(grpc_parse_uri(*lb_uri, &address));
184 result.addresses.emplace_back(address.addr, address.len,
190 void SetNextResolutionNoServiceConfig(const std::vector<int>& ports) {
191 grpc_core::ExecCtx exec_ctx;
192 grpc_core::Resolver::Result result = BuildFakeResults(ports);
193 response_generator_->SetResponse(result);
196 void SetNextResolutionValidServiceConfig(const std::vector<int>& ports) {
197 grpc_core::ExecCtx exec_ctx;
198 grpc_core::Resolver::Result result = BuildFakeResults(ports);
199 result.service_config = grpc_core::ServiceConfig::Create(
200 nullptr, "{}", &result.service_config_error);
201 response_generator_->SetResponse(result);
204 void SetNextResolutionInvalidServiceConfig(const std::vector<int>& ports) {
205 grpc_core::ExecCtx exec_ctx;
206 grpc_core::Resolver::Result result = BuildFakeResults(ports);
207 result.service_config = grpc_core::ServiceConfig::Create(
208 nullptr, "{", &result.service_config_error);
209 response_generator_->SetResponse(result);
212 void SetNextResolutionWithServiceConfig(const std::vector<int>& ports,
213 const char* svc_cfg) {
214 grpc_core::ExecCtx exec_ctx;
215 grpc_core::Resolver::Result result = BuildFakeResults(ports);
216 result.service_config = grpc_core::ServiceConfig::Create(
217 nullptr, svc_cfg, &result.service_config_error);
218 response_generator_->SetResponse(result);
221 std::vector<int> GetServersPorts(size_t start_index = 0) {
222 std::vector<int> ports;
223 for (size_t i = start_index; i < servers_.size(); ++i) {
224 ports.push_back(servers_[i]->port_);
229 std::unique_ptr<grpc::testing::EchoTestService::Stub> BuildStub(
230 const std::shared_ptr<Channel>& channel) {
231 return grpc::testing::EchoTestService::NewStub(channel);
234 std::shared_ptr<Channel> BuildChannel() {
235 ChannelArguments args;
236 args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
237 response_generator_.get());
238 return ::grpc::CreateCustomChannel("fake:///", creds_, args);
241 std::shared_ptr<Channel> BuildChannelWithDefaultServiceConfig() {
242 ChannelArguments args;
243 EXPECT_THAT(grpc::experimental::ValidateServiceConfigJSON(
244 ValidDefaultServiceConfig()),
245 ::testing::StrEq(""));
246 args.SetServiceConfigJSON(ValidDefaultServiceConfig());
247 args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
248 response_generator_.get());
249 return ::grpc::CreateCustomChannel("fake:///", creds_, args);
252 std::shared_ptr<Channel> BuildChannelWithInvalidDefaultServiceConfig() {
253 ChannelArguments args;
254 EXPECT_THAT(grpc::experimental::ValidateServiceConfigJSON(
255 InvalidDefaultServiceConfig()),
256 ::testing::HasSubstr("JSON parse error"));
257 args.SetServiceConfigJSON(InvalidDefaultServiceConfig());
258 args.SetPointer(GRPC_ARG_FAKE_RESOLVER_RESPONSE_GENERATOR,
259 response_generator_.get());
260 return ::grpc::CreateCustomChannel("fake:///", creds_, args);
264 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
265 EchoResponse* response = nullptr, int timeout_ms = 1000,
266 Status* result = nullptr, bool wait_for_ready = false) {
267 const bool local_response = (response == nullptr);
268 if (local_response) response = new EchoResponse;
270 request.set_message(kRequestMessage_);
271 ClientContext context;
272 context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms));
273 if (wait_for_ready) context.set_wait_for_ready(true);
274 Status status = stub->Echo(&context, request, response);
275 if (result != nullptr) *result = status;
276 if (local_response) delete response;
281 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
282 const grpc_core::DebugLocation& location, bool wait_for_ready = false) {
283 EchoResponse response;
286 SendRpc(stub, &response, 2000, &status, wait_for_ready);
287 ASSERT_TRUE(success) << "From " << location.file() << ":" << location.line()
289 << "Error: " << status.error_message() << " "
290 << status.error_details();
291 ASSERT_EQ(response.message(), kRequestMessage_)
292 << "From " << location.file() << ":" << location.line();
293 if (!success) abort();
296 void CheckRpcSendFailure(
297 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub) {
298 const bool success = SendRpc(stub);
299 EXPECT_FALSE(success);
304 std::unique_ptr<Server> server_;
305 MyTestServiceImpl service_;
306 std::unique_ptr<std::thread> thread_;
307 bool server_ready_ = false;
308 bool started_ = false;
310 explicit ServerData(int port = 0) {
311 port_ = port > 0 ? port : grpc_pick_unused_port_or_die();
314 void Start(const std::string& server_host) {
315 gpr_log(GPR_INFO, "starting server on port %d", port_);
317 grpc::internal::Mutex mu;
318 grpc::internal::MutexLock lock(&mu);
319 grpc::internal::CondVar cond;
320 thread_ = absl::make_unique<std::thread>(
321 std::bind(&ServerData::Serve, this, server_host, &mu, &cond));
322 grpc::internal::WaitUntil(&cond, &mu, [this] { return server_ready_; });
323 server_ready_ = false;
324 gpr_log(GPR_INFO, "server startup complete");
327 void Serve(const std::string& server_host, grpc::internal::Mutex* mu,
328 grpc::internal::CondVar* cond) {
329 std::ostringstream server_address;
330 server_address << server_host << ":" << port_;
331 ServerBuilder builder;
332 std::shared_ptr<ServerCredentials> creds(new SecureServerCredentials(
333 grpc_fake_transport_security_server_credentials_create()));
334 builder.AddListeningPort(server_address.str(), std::move(creds));
335 builder.RegisterService(&service_);
336 server_ = builder.BuildAndStart();
337 grpc::internal::MutexLock lock(mu);
338 server_ready_ = true;
343 if (!started_) return;
344 server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0));
349 void SetServingStatus(const std::string& service, bool serving) {
350 server_->GetHealthCheckService()->SetServingStatus(service, serving);
354 void ResetCounters() {
355 for (const auto& server : servers_) server->service_.ResetCounters();
359 const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
360 size_t server_idx, const grpc_core::DebugLocation& location,
361 bool ignore_failure = false) {
363 if (ignore_failure) {
366 CheckRpcSendOk(stub, location, true);
368 } while (servers_[server_idx]->service_.request_count() == 0);
372 bool WaitForChannelNotReady(Channel* channel, int timeout_seconds = 5) {
373 const gpr_timespec deadline =
374 grpc_timeout_seconds_to_deadline(timeout_seconds);
375 grpc_connectivity_state state;
376 while ((state = channel->GetState(false /* try_to_connect */)) ==
377 GRPC_CHANNEL_READY) {
378 if (!channel->WaitForStateChange(state, deadline)) return false;
383 bool WaitForChannelReady(Channel* channel, int timeout_seconds = 5) {
384 const gpr_timespec deadline =
385 grpc_timeout_seconds_to_deadline(timeout_seconds);
386 grpc_connectivity_state state;
387 while ((state = channel->GetState(true /* try_to_connect */)) !=
388 GRPC_CHANNEL_READY) {
389 if (!channel->WaitForStateChange(state, deadline)) return false;
394 bool SeenAllServers() {
395 for (const auto& server : servers_) {
396 if (server->service_.request_count() == 0) return false;
401 // Updates \a connection_order by appending to it the index of the newly
402 // connected server. Must be called after every single RPC.
403 void UpdateConnectionOrder(
404 const std::vector<std::unique_ptr<ServerData>>& servers,
405 std::vector<int>* connection_order) {
406 for (size_t i = 0; i < servers.size(); ++i) {
407 if (servers[i]->service_.request_count() == 1) {
408 // Was the server index known? If not, update connection_order.
410 std::find(connection_order->begin(), connection_order->end(), i);
411 if (it == connection_order->end()) {
412 connection_order->push_back(i);
419 const char* ValidServiceConfigV1() { return "{\"version\": \"1\"}"; }
421 const char* ValidServiceConfigV2() { return "{\"version\": \"2\"}"; }
423 const char* ValidDefaultServiceConfig() {
424 return "{\"version\": \"valid_default\"}";
427 const char* InvalidDefaultServiceConfig() {
428 return "{\"version\": \"invalid_default\"";
431 bool ipv6_only_ = false;
432 const std::string server_host_;
433 std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
434 std::vector<std::unique_ptr<ServerData>> servers_;
435 grpc_core::RefCountedPtr<grpc_core::FakeResolverResponseGenerator>
437 const std::string kRequestMessage_;
438 std::shared_ptr<ChannelCredentials> creds_;
441 TEST_F(ServiceConfigEnd2endTest, NoServiceConfigTest) {
443 auto channel = BuildChannel();
444 auto stub = BuildStub(channel);
445 SetNextResolutionNoServiceConfig(GetServersPorts());
446 CheckRpcSendOk(stub, DEBUG_LOCATION);
447 EXPECT_STREQ("{}", channel->GetServiceConfigJSON().c_str());
450 TEST_F(ServiceConfigEnd2endTest, NoServiceConfigWithDefaultConfigTest) {
452 auto channel = BuildChannelWithDefaultServiceConfig();
453 auto stub = BuildStub(channel);
454 SetNextResolutionNoServiceConfig(GetServersPorts());
455 CheckRpcSendOk(stub, DEBUG_LOCATION);
456 EXPECT_STREQ(ValidDefaultServiceConfig(),
457 channel->GetServiceConfigJSON().c_str());
460 TEST_F(ServiceConfigEnd2endTest, InvalidServiceConfigTest) {
462 auto channel = BuildChannel();
463 auto stub = BuildStub(channel);
464 SetNextResolutionInvalidServiceConfig(GetServersPorts());
465 CheckRpcSendFailure(stub);
468 TEST_F(ServiceConfigEnd2endTest, ValidServiceConfigUpdatesTest) {
470 auto channel = BuildChannel();
471 auto stub = BuildStub(channel);
472 SetNextResolutionWithServiceConfig(GetServersPorts(), ValidServiceConfigV1());
473 CheckRpcSendOk(stub, DEBUG_LOCATION);
474 EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str());
475 SetNextResolutionWithServiceConfig(GetServersPorts(), ValidServiceConfigV2());
476 CheckRpcSendOk(stub, DEBUG_LOCATION);
477 EXPECT_STREQ(ValidServiceConfigV2(), channel->GetServiceConfigJSON().c_str());
480 TEST_F(ServiceConfigEnd2endTest,
481 NoServiceConfigUpdateAfterValidServiceConfigTest) {
483 auto channel = BuildChannel();
484 auto stub = BuildStub(channel);
485 SetNextResolutionWithServiceConfig(GetServersPorts(), ValidServiceConfigV1());
486 CheckRpcSendOk(stub, DEBUG_LOCATION);
487 EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str());
488 SetNextResolutionNoServiceConfig(GetServersPorts());
489 CheckRpcSendOk(stub, DEBUG_LOCATION);
490 EXPECT_STREQ("{}", channel->GetServiceConfigJSON().c_str());
493 TEST_F(ServiceConfigEnd2endTest,
494 NoServiceConfigUpdateAfterValidServiceConfigWithDefaultConfigTest) {
496 auto channel = BuildChannelWithDefaultServiceConfig();
497 auto stub = BuildStub(channel);
498 SetNextResolutionWithServiceConfig(GetServersPorts(), ValidServiceConfigV1());
499 CheckRpcSendOk(stub, DEBUG_LOCATION);
500 EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str());
501 SetNextResolutionNoServiceConfig(GetServersPorts());
502 CheckRpcSendOk(stub, DEBUG_LOCATION);
503 EXPECT_STREQ(ValidDefaultServiceConfig(),
504 channel->GetServiceConfigJSON().c_str());
507 TEST_F(ServiceConfigEnd2endTest,
508 InvalidServiceConfigUpdateAfterValidServiceConfigTest) {
510 auto channel = BuildChannel();
511 auto stub = BuildStub(channel);
512 SetNextResolutionWithServiceConfig(GetServersPorts(), ValidServiceConfigV1());
513 CheckRpcSendOk(stub, DEBUG_LOCATION);
514 EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str());
515 SetNextResolutionInvalidServiceConfig(GetServersPorts());
516 CheckRpcSendOk(stub, DEBUG_LOCATION);
517 EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str());
520 TEST_F(ServiceConfigEnd2endTest,
521 InvalidServiceConfigUpdateAfterValidServiceConfigWithDefaultConfigTest) {
523 auto channel = BuildChannelWithDefaultServiceConfig();
524 auto stub = BuildStub(channel);
525 SetNextResolutionWithServiceConfig(GetServersPorts(), ValidServiceConfigV1());
526 CheckRpcSendOk(stub, DEBUG_LOCATION);
527 EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str());
528 SetNextResolutionInvalidServiceConfig(GetServersPorts());
529 CheckRpcSendOk(stub, DEBUG_LOCATION);
530 EXPECT_STREQ(ValidServiceConfigV1(), channel->GetServiceConfigJSON().c_str());
533 TEST_F(ServiceConfigEnd2endTest,
534 ValidServiceConfigAfterInvalidServiceConfigTest) {
536 auto channel = BuildChannel();
537 auto stub = BuildStub(channel);
538 SetNextResolutionInvalidServiceConfig(GetServersPorts());
539 CheckRpcSendFailure(stub);
540 SetNextResolutionValidServiceConfig(GetServersPorts());
541 CheckRpcSendOk(stub, DEBUG_LOCATION);
544 TEST_F(ServiceConfigEnd2endTest, NoServiceConfigAfterInvalidServiceConfigTest) {
546 auto channel = BuildChannel();
547 auto stub = BuildStub(channel);
548 SetNextResolutionInvalidServiceConfig(GetServersPorts());
549 CheckRpcSendFailure(stub);
550 SetNextResolutionNoServiceConfig(GetServersPorts());
551 CheckRpcSendOk(stub, DEBUG_LOCATION);
552 EXPECT_STREQ("{}", channel->GetServiceConfigJSON().c_str());
555 TEST_F(ServiceConfigEnd2endTest,
556 AnotherInvalidServiceConfigAfterInvalidServiceConfigTest) {
558 auto channel = BuildChannel();
559 auto stub = BuildStub(channel);
560 SetNextResolutionInvalidServiceConfig(GetServersPorts());
561 CheckRpcSendFailure(stub);
562 SetNextResolutionInvalidServiceConfig(GetServersPorts());
563 CheckRpcSendFailure(stub);
566 TEST_F(ServiceConfigEnd2endTest, InvalidDefaultServiceConfigTest) {
568 auto channel = BuildChannelWithInvalidDefaultServiceConfig();
569 auto stub = BuildStub(channel);
570 // An invalid default service config results in a lame channel which fails all
572 CheckRpcSendFailure(stub);
575 TEST_F(ServiceConfigEnd2endTest,
576 InvalidDefaultServiceConfigTestWithValidServiceConfig) {
578 auto channel = BuildChannelWithInvalidDefaultServiceConfig();
579 auto stub = BuildStub(channel);
580 CheckRpcSendFailure(stub);
581 // An invalid default service config results in a lame channel which fails all
583 SetNextResolutionValidServiceConfig(GetServersPorts());
584 CheckRpcSendFailure(stub);
587 TEST_F(ServiceConfigEnd2endTest,
588 InvalidDefaultServiceConfigTestWithInvalidServiceConfig) {
590 auto channel = BuildChannelWithInvalidDefaultServiceConfig();
591 auto stub = BuildStub(channel);
592 CheckRpcSendFailure(stub);
593 // An invalid default service config results in a lame channel which fails all
595 SetNextResolutionInvalidServiceConfig(GetServersPorts());
596 CheckRpcSendFailure(stub);
599 TEST_F(ServiceConfigEnd2endTest,
600 InvalidDefaultServiceConfigTestWithNoServiceConfig) {
602 auto channel = BuildChannelWithInvalidDefaultServiceConfig();
603 auto stub = BuildStub(channel);
604 CheckRpcSendFailure(stub);
605 // An invalid default service config results in a lame channel which fails all
607 SetNextResolutionNoServiceConfig(GetServersPorts());
608 CheckRpcSendFailure(stub);
612 } // namespace testing
615 int main(int argc, char** argv) {
616 ::testing::InitGoogleTest(&argc, argv);
617 grpc::testing::TestEnvironment env(argc, argv);
618 const auto result = RUN_ALL_TESTS();