test/cpp/interop/grpclb_fallback_test.cc

   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 <grpc/support/alloc.h>
  22 #include <grpc/support/log.h>
  23 #include <grpc/support/port_platform.h>
  24 #include <grpcpp/channel.h>
  25 #include <grpcpp/client_context.h>
  26 #include <grpcpp/grpcpp.h>
  27 #include <grpcpp/support/channel_arguments.h>
  28 #include <inttypes.h>
  29 #include <netinet/in.h>
  30 #include <netinet/tcp.h>
  31 #include <sys/wait.h>
  32 #include <unistd.h>
  33
  34 #include <chrono>
  35 #include <cstdlib>
  36 #include <memory>
  37 #include <string>
  38 #include <thread>
  39
  40 #include "absl/flags/flag.h"
  41 #include "src/core/lib/gpr/string.h"
  42 #include "src/core/lib/iomgr/port.h"
  43 #include "src/core/lib/iomgr/socket_mutator.h"
  44 #include "src/proto/grpc/testing/empty.pb.h"
  45 #include "src/proto/grpc/testing/messages.pb.h"
  46 #include "src/proto/grpc/testing/test.grpc.pb.h"
  47 #include "src/proto/grpc/testing/test.pb.h"
  48 #include "test/cpp/util/test_config.h"
  49 #include "test/cpp/util/test_credentials_provider.h"
  50
  51 ABSL_FLAG(std::string, custom_credentials_type, "",
  52           "User provided credentials type.");
  53 ABSL_FLAG(std::string, server_uri, "localhost:1000", "Server URI target");
  54 ABSL_FLAG(std::string, unroute_lb_and_backend_addrs_cmd, "exit 1",
  55           "Shell command used to make LB and backend addresses unroutable");
  56 ABSL_FLAG(std::string, blackhole_lb_and_backend_addrs_cmd, "exit 1",
  57           "Shell command used to make LB and backend addresses blackholed");
  58 ABSL_FLAG(
  59     std::string, test_case, "",
  60     "Test case to run. Valid options are:\n\n"
  61     "fast_fallback_before_startup : fallback before establishing connection to "
  62     "LB;\n"
  63     "fast_fallback_after_startup : fallback after startup due to LB/backend "
  64     "addresses becoming unroutable;\n"
  65     "slow_fallback_before_startup : fallback before startup due to LB address "
  66     "being blackholed;\n"
  67     "slow_fallback_after_startup : fallback after startup due to LB/backend "
  68     "addresses becoming blackholed;\n");
  69
  70 #ifdef LINUX_VERSION_CODE
  71 #if LINUX_VERSION_CODE >= KERNEL_VERSION(2, 6, 37)
  72 #define SOCKET_SUPPORTS_TCP_USER_TIMEOUT
  73 #endif
  74 #endif
  75
  76 #ifdef SOCKET_SUPPORTS_TCP_USER_TIMEOUT
  77 using grpc::testing::GrpclbRouteType;
  78 using grpc::testing::SimpleRequest;
  79 using grpc::testing::SimpleResponse;
  80 using grpc::testing::TestService;
  81
  82 namespace {
  83
  84 enum RpcMode {
  85   FailFast,
  86   WaitForReady,
  87 };
  88
  89 GrpclbRouteType DoRPCAndGetPath(TestService::Stub* stub, int deadline_seconds,
  90                                 RpcMode rpc_mode) {
  91   gpr_log(GPR_INFO, "DoRPCAndGetPath deadline_seconds:%d rpc_mode:%d",
  92           deadline_seconds, rpc_mode);
  93   SimpleRequest request;
  94   SimpleResponse response;
  95   grpc::ClientContext context;
  96   if (rpc_mode == WaitForReady) {
  97     context.set_wait_for_ready(true);
  98   }
  99   request.set_fill_grpclb_route_type(true);
 100   std::chrono::system_clock::time_point deadline =
 101       std::chrono::system_clock::now() + std::chrono::seconds(deadline_seconds);
 102   context.set_deadline(deadline);
 103   grpc::Status s = stub->UnaryCall(&context, request, &response);
 104   if (!s.ok()) {
 105     gpr_log(GPR_INFO, "DoRPCAndGetPath failed. status-message: %s",
 106             s.error_message().c_str());
 107     return GrpclbRouteType::GRPCLB_ROUTE_TYPE_UNKNOWN;
 108   }
 109   GPR_ASSERT(response.grpclb_route_type() ==
 110                  GrpclbRouteType::GRPCLB_ROUTE_TYPE_BACKEND ||
 111              response.grpclb_route_type() ==
 112                  GrpclbRouteType::GRPCLB_ROUTE_TYPE_FALLBACK);
 113   gpr_log(GPR_INFO, "DoRPCAndGetPath done. grpclb_route_type:%d",
 114           response.grpclb_route_type());
 115   return response.grpclb_route_type();
 116 }
 117
 118 GrpclbRouteType DoRPCAndGetPath(TestService::Stub* stub, int deadline_seconds) {
 119   return DoRPCAndGetPath(stub, deadline_seconds, FailFast);
 120 }
 121
 122 GrpclbRouteType DoWaitForReadyRPCAndGetPath(TestService::Stub* stub,
 123                                             int deadline_seconds) {
 124   return DoRPCAndGetPath(stub, deadline_seconds, WaitForReady);
 125 }
 126
 127 bool TcpUserTimeoutMutateFd(int fd, grpc_socket_mutator* /*mutator*/) {
 128   int timeout = 20000;  // 20 seconds
 129   gpr_log(GPR_INFO, "Setting socket option TCP_USER_TIMEOUT on fd: %d", fd);
 130   if (0 != setsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &timeout,
 131                       sizeof(timeout))) {
 132     gpr_log(GPR_ERROR, "Failed to set socket option TCP_USER_TIMEOUT");
 133     abort();
 134   }
 135   int newval;
 136   socklen_t len = sizeof(newval);
 137   if (0 != getsockopt(fd, IPPROTO_TCP, TCP_USER_TIMEOUT, &newval, &len) ||
 138       newval != timeout) {
 139     gpr_log(GPR_ERROR, "Failed to get expected socket option TCP_USER_TIMEOUT");
 140     abort();
 141   }
 142   return true;
 143 }
 144
 145 int TcpUserTimeoutCompare(grpc_socket_mutator* /*a*/,
 146                           grpc_socket_mutator* /*b*/) {
 147   return 0;
 148 }
 149
 150 void TcpUserTimeoutDestroy(grpc_socket_mutator* mutator) { gpr_free(mutator); }
 151
 152 const grpc_socket_mutator_vtable kTcpUserTimeoutMutatorVtable =
 153     grpc_socket_mutator_vtable{TcpUserTimeoutMutateFd, TcpUserTimeoutCompare,
 154                                TcpUserTimeoutDestroy, nullptr};
 155
 156 std::unique_ptr<TestService::Stub> CreateFallbackTestStub() {
 157   grpc::ChannelArguments channel_args;
 158   grpc_socket_mutator* tcp_user_timeout_mutator =
 159       static_cast<grpc_socket_mutator*>(
 160           gpr_malloc(sizeof(tcp_user_timeout_mutator)));
 161   grpc_socket_mutator_init(tcp_user_timeout_mutator,
 162                            &kTcpUserTimeoutMutatorVtable);
 163   channel_args.SetSocketMutator(tcp_user_timeout_mutator);
 164   // Allow LB policy to be configured by service config
 165   channel_args.SetInt(GRPC_ARG_SERVICE_CONFIG_DISABLE_RESOLUTION, 0);
 166   std::shared_ptr<grpc::ChannelCredentials> channel_creds =
 167       grpc::testing::GetCredentialsProvider()->GetChannelCredentials(
 168           absl::GetFlag(FLAGS_custom_credentials_type), &channel_args);
 169   return TestService::NewStub(grpc::CreateCustomChannel(
 170       absl::GetFlag(FLAGS_server_uri), channel_creds, channel_args));
 171 }
 172
 173 void RunCommand(const std::string& command) {
 174   gpr_log(GPR_INFO, "RunCommand: |%s|", command.c_str());
 175   int out = std::system(command.c_str());
 176   if (WIFEXITED(out)) {
 177     int code = WEXITSTATUS(out);
 178     if (code != 0) {
 179       gpr_log(GPR_ERROR, "RunCommand failed exit code:%d command:|%s|", code,
 180               command.c_str());
 181       abort();
 182     }
 183   } else {
 184     gpr_log(GPR_ERROR, "RunCommand failed command:|%s|", command.c_str());
 185     abort();
 186   }
 187 }
 188
 189 void RunFallbackBeforeStartupTest(
 190     const std::string& break_lb_and_backend_conns_cmd,
 191     int per_rpc_deadline_seconds) {
 192   std::unique_ptr<TestService::Stub> stub = CreateFallbackTestStub();
 193   RunCommand(break_lb_and_backend_conns_cmd);
 194   for (size_t i = 0; i < 30; i++) {
 195     GrpclbRouteType grpclb_route_type =
 196         DoRPCAndGetPath(stub.get(), per_rpc_deadline_seconds);
 197     if (grpclb_route_type != GrpclbRouteType::GRPCLB_ROUTE_TYPE_FALLBACK) {
 198       gpr_log(GPR_ERROR, "Expected grpclb route type: FALLBACK. Got: %d",
 199               grpclb_route_type);
 200       abort();
 201     }
 202     std::this_thread::sleep_for(std::chrono::seconds(1));
 203   }
 204 }
 205
 206 void DoFastFallbackBeforeStartup() {
 207   RunFallbackBeforeStartupTest(
 208       absl::GetFlag(FLAGS_unroute_lb_and_backend_addrs_cmd), 9);
 209 }
 210
 211 void DoSlowFallbackBeforeStartup() {
 212   RunFallbackBeforeStartupTest(
 213       absl::GetFlag(FLAGS_blackhole_lb_and_backend_addrs_cmd), 20);
 214 }
 215
 216 void RunFallbackAfterStartupTest(
 217     const std::string& break_lb_and_backend_conns_cmd) {
 218   std::unique_ptr<TestService::Stub> stub = CreateFallbackTestStub();
 219   GrpclbRouteType grpclb_route_type = DoRPCAndGetPath(stub.get(), 20);
 220   if (grpclb_route_type != GrpclbRouteType::GRPCLB_ROUTE_TYPE_BACKEND) {
 221     gpr_log(GPR_ERROR, "Expected grpclb route type: BACKEND. Got: %d",
 222             grpclb_route_type);
 223     abort();
 224   }
 225   RunCommand(break_lb_and_backend_conns_cmd);
 226   for (size_t i = 0; i < 40; i++) {
 227     GrpclbRouteType grpclb_route_type =
 228         DoWaitForReadyRPCAndGetPath(stub.get(), 1);
 229     // Backends should be unreachable by now, otherwise the test is broken.
 230     GPR_ASSERT(grpclb_route_type != GrpclbRouteType::GRPCLB_ROUTE_TYPE_BACKEND);
 231     if (grpclb_route_type == GrpclbRouteType::GRPCLB_ROUTE_TYPE_FALLBACK) {
 232       gpr_log(GPR_INFO,
 233               "Made one successul RPC to a fallback. Now expect the same for "
 234               "the rest.");
 235       break;
 236     } else {
 237       gpr_log(GPR_ERROR, "Retryable RPC failure on iteration: %" PRIdPTR, i);
 238     }
 239   }
 240   for (size_t i = 0; i < 30; i++) {
 241     GrpclbRouteType grpclb_route_type = DoRPCAndGetPath(stub.get(), 20);
 242     if (grpclb_route_type != GrpclbRouteType::GRPCLB_ROUTE_TYPE_FALLBACK) {
 243       gpr_log(GPR_ERROR, "Expected grpclb route type: FALLBACK. Got: %d",
 244               grpclb_route_type);
 245       abort();
 246     }
 247     std::this_thread::sleep_for(std::chrono::seconds(1));
 248   }
 249 }
 250
 251 void DoFastFallbackAfterStartup() {
 252   RunFallbackAfterStartupTest(
 253       absl::GetFlag(FLAGS_unroute_lb_and_backend_addrs_cmd));
 254 }
 255
 256 void DoSlowFallbackAfterStartup() {
 257   RunFallbackAfterStartupTest(
 258       absl::GetFlag(FLAGS_blackhole_lb_and_backend_addrs_cmd));
 259 }
 260 }  // namespace
 261
 262 int main(int argc, char** argv) {
 263   grpc::testing::InitTest(&argc, &argv, true);
 264   gpr_log(GPR_INFO, "Testing: %s", absl::GetFlag(FLAGS_test_case).c_str());
 265   if (absl::GetFlag(FLAGS_test_case) == "fast_fallback_before_startup") {
 266     DoFastFallbackBeforeStartup();
 267     gpr_log(GPR_INFO, "DoFastFallbackBeforeStartup done!");
 268   } else if (absl::GetFlag(FLAGS_test_case) == "slow_fallback_before_startup") {
 269     DoSlowFallbackBeforeStartup();
 270     gpr_log(GPR_INFO, "DoSlowFallbackBeforeStartup done!");
 271   } else if (absl::GetFlag(FLAGS_test_case) == "fast_fallback_after_startup") {
 272     DoFastFallbackAfterStartup();
 273     gpr_log(GPR_INFO, "DoFastFallbackAfterStartup done!");
 274   } else if (absl::GetFlag(FLAGS_test_case) == "slow_fallback_after_startup") {
 275     DoSlowFallbackAfterStartup();
 276     gpr_log(GPR_INFO, "DoSlowFallbackAfterStartup done!");
 277   } else {
 278     gpr_log(GPR_ERROR, "Invalid test case: %s",
 279             absl::GetFlag(FLAGS_test_case).c_str());
 280     abort();
 281   }
 282 }
 283
 284 #else
 285
 286 int main(int argc, char** argv) {
 287   grpc::testing::InitTest(&argc, &argv, true);
 288   gpr_log(GPR_ERROR,
 289           "This test requires TCP_USER_TIMEOUT, which isn't available");
 290   abort();
 291 }
 292
 293 #endif  // SOCKET_SUPPORTS_TCP_USER_TIMEOUT