1 //
2 //
3 // Copyright 2019 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 <arpa/inet.h>
20 #include <fcntl.h>
21 #include <inttypes.h>
22 #include <netinet/in.h>
23 #include <netinet/tcp.h>
24 #include <sys/wait.h>
25 #include <unistd.h>
26
27 #include <chrono>
28 #include <cstdlib>
29 #include <memory>
30 #include <string>
31 #include <thread>
32
33 #include "absl/flags/flag.h"
34 #include "absl/strings/str_format.h"
35 #include "absl/time/time.h"
36
37 #include <grpc/support/alloc.h>
38 #include <grpc/support/log.h>
39 #include <grpc/support/port_platform.h>
40 #include <grpcpp/channel.h>
41 #include <grpcpp/client_context.h>
42 #include <grpcpp/grpcpp.h>
43 #include <grpcpp/support/channel_arguments.h>
44
45 #include "src/core/lib/gpr/string.h"
46 #include "src/core/lib/gprpp/crash.h"
47 #include "src/core/lib/iomgr/port.h"
48 #include "src/core/lib/iomgr/socket_mutator.h"
49 #include "src/proto/grpc/testing/empty.pb.h"
50 #include "src/proto/grpc/testing/messages.pb.h"
51 #include "src/proto/grpc/testing/test.grpc.pb.h"
52 #include "src/proto/grpc/testing/test.pb.h"
53 #include "test/cpp/util/test_config.h"
54 #include "test/cpp/util/test_credentials_provider.h"
55
56 ABSL_FLAG(std::string, custom_credentials_type, "",
57 "User provided credentials type.");
58 ABSL_FLAG(std::string, server_uri, "localhost:1000", "Server URI target");
59 ABSL_FLAG(std::string, induce_fallback_cmd, "exit 1",
60 "Shell command to induce fallback, e.g. by unrouting addresses");
61 ABSL_FLAG(int, fallback_deadline_seconds, 1,
62 "Number of seconds to wait for fallback to occur after inducing it");
63 ABSL_FLAG(std::string, test_case, "",
64 "Test case to run. Valid options are:\n\n"
65 "fallback_before_startup : fallback before making RPCs to backends"
66 "fallback_after_startup : fallback after making RPCs to backends");
67
68 #ifdef LINUX_VERSION_CODE
69 #if LINUX_VERSION_CODE >= KERNEL_VERSION(2, 6, 37)
70 #define SOCKET_SUPPORTS_TCP_USER_TIMEOUT
71 #endif
72 #endif
73
74 #ifdef SOCKET_SUPPORTS_TCP_USER_TIMEOUT
75 using grpc::testing::GrpclbRouteType;
76 using grpc::testing::SimpleRequest;
77 using grpc::testing::SimpleResponse;
78 using grpc::testing::TestService;
79
80 namespace {
81
82 enum RpcMode {
83 FailFast,
84 WaitForReady,
85 };
86
DoRPCAndGetPath(TestService::Stub * stub,int deadline_seconds,RpcMode rpc_mode)87 GrpclbRouteType DoRPCAndGetPath(TestService::Stub* stub, int deadline_seconds,
88 RpcMode rpc_mode) {
89 gpr_log(GPR_INFO, "DoRPCAndGetPath deadline_seconds:%d rpc_mode:%d",
90 deadline_seconds, rpc_mode);
91 SimpleRequest request;
92 SimpleResponse response;
93 grpc::ClientContext context;
94 if (rpc_mode == WaitForReady) {
95 context.set_wait_for_ready(true);
96 }
97 request.set_fill_grpclb_route_type(true);
98 std::chrono::system_clock::time_point deadline =
99 std::chrono::system_clock::now() + std::chrono::seconds(deadline_seconds);
100 context.set_deadline(deadline);
101 grpc::Status s = stub->UnaryCall(&context, request, &response);
102 if (!s.ok()) {
103 gpr_log(GPR_INFO, "DoRPCAndGetPath failed. status-message: %s",
104 s.error_message().c_str());
105 return GrpclbRouteType::GRPCLB_ROUTE_TYPE_UNKNOWN;
106 }
107 GPR_ASSERT(response.grpclb_route_type() ==
108 GrpclbRouteType::GRPCLB_ROUTE_TYPE_BACKEND ||
109 response.grpclb_route_type() ==
110 GrpclbRouteType::GRPCLB_ROUTE_TYPE_FALLBACK);
111 gpr_log(GPR_INFO, "DoRPCAndGetPath done. grpclb_route_type:%d",
112 response.grpclb_route_type());
113 return response.grpclb_route_type();
114 }
115
DoRPCAndGetPath(TestService::Stub * stub,int deadline_seconds)116 GrpclbRouteType DoRPCAndGetPath(TestService::Stub* stub, int deadline_seconds) {
117 return DoRPCAndGetPath(stub, deadline_seconds, FailFast);
118 }
119
TcpUserTimeoutMutateFd(int fd,grpc_socket_mutator *)120 bool TcpUserTimeoutMutateFd(int fd, grpc_socket_mutator* /*mutator*/) {
121 int timeout = 20000; // 20 seconds
122 gpr_log(GPR_INFO, "Setting socket option TCP_USER_TIMEOUT on fd: %d", fd);
123 if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &timeout,
124 sizeof(timeout))) {
125 grpc_core::Crash("Failed to set socket option TCP_USER_TIMEOUT");
126 }
127 int newval;
128 socklen_t len = sizeof(newval);
129 if (0 != getsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &newval, &len) ||
130 newval != timeout) {
131 grpc_core::Crash("Failed to get expected socket option TCP_USER_TIMEOUT");
132 }
133 return true;
134 }
135
TcpUserTimeoutCompare(grpc_socket_mutator *,grpc_socket_mutator *)136 int TcpUserTimeoutCompare(grpc_socket_mutator* /*a*/,
137 grpc_socket_mutator* /*b*/) {
138 return 0;
139 }
140
TcpUserTimeoutDestroy(grpc_socket_mutator * mutator)141 void TcpUserTimeoutDestroy(grpc_socket_mutator* mutator) { delete mutator; }
142
143 const grpc_socket_mutator_vtable kTcpUserTimeoutMutatorVtable =
144 grpc_socket_mutator_vtable{TcpUserTimeoutMutateFd, TcpUserTimeoutCompare,
145 TcpUserTimeoutDestroy, nullptr};
146
CreateFallbackTestStub()147 std::unique_ptr<TestService::Stub> CreateFallbackTestStub() {
148 grpc::ChannelArguments channel_args;
149 grpc_socket_mutator* tcp_user_timeout_mutator = new grpc_socket_mutator();
150 grpc_socket_mutator_init(tcp_user_timeout_mutator,
151 &kTcpUserTimeoutMutatorVtable);
152 channel_args.SetSocketMutator(tcp_user_timeout_mutator);
153 // Allow LB policy to be configured by service config
154 channel_args.SetInt(GRPC_ARG_SERVICE_CONFIG_DISABLE_RESOLUTION, 0);
155 std::shared_ptr<grpc::ChannelCredentials> channel_creds =
156 grpc::testing::GetCredentialsProvider()->GetChannelCredentials(
157 absl::GetFlag(FLAGS_custom_credentials_type), &channel_args);
158 return TestService::NewStub(grpc::CreateCustomChannel(
159 absl::GetFlag(FLAGS_server_uri), channel_creds, channel_args));
160 }
161
RunCommand(const std::string & command)162 void RunCommand(const std::string& command) {
163 gpr_log(GPR_INFO, "RunCommand: |%s|", command.c_str());
164 int out = std::system(command.c_str());
165 if (WIFEXITED(out)) {
166 int code = WEXITSTATUS(out);
167 if (code != 0) {
168 grpc_core::Crash(
169 absl::StrFormat("RunCommand failed exit code:%d command:|%s|", code,
170 command.c_str()));
171 }
172 } else {
173 grpc_core::Crash(
174 absl::StrFormat("RunCommand failed command:|%s|", command.c_str()));
175 }
176 }
177
WaitForFallbackAndDoRPCs(TestService::Stub * stub)178 void WaitForFallbackAndDoRPCs(TestService::Stub* stub) {
179 int fallback_retry_count = 0;
180 bool fallback = false;
181 absl::Time fallback_deadline =
182 absl::Now() +
183 absl::Seconds(absl::GetFlag(FLAGS_fallback_deadline_seconds));
184 while (absl::Now() < fallback_deadline) {
185 GrpclbRouteType grpclb_route_type = DoRPCAndGetPath(stub, 1);
186 if (grpclb_route_type == GrpclbRouteType::GRPCLB_ROUTE_TYPE_BACKEND) {
187 gpr_log(GPR_ERROR,
188 "Got grpclb route type backend. Backends are "
189 "supposed to be unreachable, so this test is broken");
190 GPR_ASSERT(0);
191 }
192 if (grpclb_route_type == GrpclbRouteType::GRPCLB_ROUTE_TYPE_FALLBACK) {
193 gpr_log(GPR_INFO,
194 "Made one successful RPC to a fallback. Now expect the same for "
195 "the rest.");
196 fallback = true;
197 break;
198 } else {
199 gpr_log(GPR_ERROR, "Retryable RPC failure on iteration: %d",
200 fallback_retry_count);
201 }
202 fallback_retry_count++;
203 }
204 if (!fallback) {
205 gpr_log(GPR_ERROR, "Didn't fall back within deadline");
206 GPR_ASSERT(0);
207 }
208 for (int i = 0; i < 30; i++) {
209 GrpclbRouteType grpclb_route_type = DoRPCAndGetPath(stub, 20);
210 GPR_ASSERT(grpclb_route_type ==
211 GrpclbRouteType::GRPCLB_ROUTE_TYPE_FALLBACK);
212 std::this_thread::sleep_for(std::chrono::seconds(1));
213 }
214 }
215
DoFallbackBeforeStartupTest()216 void DoFallbackBeforeStartupTest() {
217 std::unique_ptr<TestService::Stub> stub = CreateFallbackTestStub();
218 RunCommand(absl::GetFlag(FLAGS_induce_fallback_cmd));
219 WaitForFallbackAndDoRPCs(stub.get());
220 }
221
DoFallbackAfterStartupTest()222 void DoFallbackAfterStartupTest() {
223 std::unique_ptr<TestService::Stub> stub = CreateFallbackTestStub();
224 GrpclbRouteType grpclb_route_type = DoRPCAndGetPath(stub.get(), 20);
225 GPR_ASSERT(grpclb_route_type == GrpclbRouteType::GRPCLB_ROUTE_TYPE_BACKEND);
226 RunCommand(absl::GetFlag(FLAGS_induce_fallback_cmd));
227 WaitForFallbackAndDoRPCs(stub.get());
228 }
229
230 } // namespace
231
main(int argc,char ** argv)232 int main(int argc, char** argv) {
233 grpc::testing::InitTest(&argc, &argv, true);
234 gpr_log(GPR_INFO, "Testing: %s", absl::GetFlag(FLAGS_test_case).c_str());
235 if (absl::GetFlag(FLAGS_test_case) == "fallback_before_startup") {
236 DoFallbackBeforeStartupTest();
237 gpr_log(GPR_INFO, "DoFallbackBeforeStartup done!");
238 } else if (absl::GetFlag(FLAGS_test_case) == "fallback_after_startup") {
239 DoFallbackAfterStartupTest();
240 gpr_log(GPR_INFO, "DoFallbackBeforeStartup done!");
241 } else {
242 grpc_core::Crash(absl::StrFormat("Invalid test case: %s",
243 absl::GetFlag(FLAGS_test_case).c_str()));
244 }
245 }
246
247 #else
248
main(int argc,char ** argv)249 int main(int argc, char** argv) {
250 grpc::testing::InitTest(&argc, &argv, true);
251 grpc_core::Crash(
252 "This test requires TCP_USER_TIMEOUT, which isn't available");
253 }
254
255 #endif // SOCKET_SUPPORTS_TCP_USER_TIMEOUT
256