test/cpp/end2end/flaky_network_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 <grpc/grpc.h>
  20 #include <grpc/support/alloc.h>
  21 #include <grpc/support/atm.h>
  22 #include <grpc/support/log.h>
  23 #include <grpc/support/port_platform.h>
  24 #include <grpc/support/string_util.h>
  25 #include <grpc/support/time.h>
  26 #include <grpcpp/channel.h>
  27 #include <grpcpp/client_context.h>
  28 #include <grpcpp/create_channel.h>
  29 #include <grpcpp/health_check_service_interface.h>
  30 #include <grpcpp/server.h>
  31 #include <grpcpp/server_builder.h>
  32 #include <gtest/gtest.h>
  33
  34 #include <algorithm>
  35 #include <memory>
  36 #include <mutex>
  37 #include <random>
  38 #include <thread>
  39
  40 #include "src/core/lib/backoff/backoff.h"
  41 #include "src/core/lib/gpr/env.h"
  42 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  43 #include "test/core/util/debugger_macros.h"
  44 #include "test/core/util/port.h"
  45 #include "test/core/util/test_config.h"
  46 #include "test/cpp/end2end/test_service_impl.h"
  47 #include "test/cpp/util/test_credentials_provider.h"
  48
  49 #ifdef GPR_LINUX
  50 using grpc::testing::EchoRequest;
  51 using grpc::testing::EchoResponse;
  52
  53 namespace grpc {
  54 namespace testing {
  55 namespace {
  56
  57 struct TestScenario {
  58   TestScenario(const grpc::string& creds_type, const grpc::string& content)
  59       : credentials_type(creds_type), message_content(content) {}
  60   const grpc::string credentials_type;
  61   const grpc::string message_content;
  62 };
  63
  64 class FlakyNetworkTest : public ::testing::TestWithParam<TestScenario> {
  65  protected:
  66   FlakyNetworkTest()
  67       : server_host_("grpctest"),
  68         interface_("lo:1"),
  69         ipv4_address_("10.0.0.1"),
  70         netmask_("/32") {}
  71
  72   void InterfaceUp() {
  73     std::ostringstream cmd;
  74     // create interface_ with address ipv4_address_
  75     cmd << "ip addr add " << ipv4_address_ << netmask_ << " dev " << interface_;
  76     std::system(cmd.str().c_str());
  77   }
  78
  79   void InterfaceDown() {
  80     std::ostringstream cmd;
  81     // remove interface_
  82     cmd << "ip addr del " << ipv4_address_ << netmask_ << " dev " << interface_;
  83     std::system(cmd.str().c_str());
  84   }
  85
  86   void DNSUp() {
  87     std::ostringstream cmd;
  88     // Add DNS entry for server_host_ in /etc/hosts
  89     cmd << "echo '" << ipv4_address_ << "      " << server_host_
  90         << "' >> /etc/hosts";
  91     std::system(cmd.str().c_str());
  92   }
  93
  94   void DNSDown() {
  95     std::ostringstream cmd;
  96     // Remove DNS entry for server_host_ from /etc/hosts
  97     // NOTE: we can't do this in one step with sed -i because when we are
  98     // running under docker, the file is mounted by docker so we can't change
  99     // its inode from within the container (sed -i creates a new file and
 100     // replaces the old file, which changes the inode)
 101     cmd << "sed  '/" << server_host_ << "/d' /etc/hosts > /etc/hosts.orig";
 102     std::system(cmd.str().c_str());
 103
 104     // clear the stream
 105     cmd.str("");
 106
 107     cmd << "cat /etc/hosts.orig > /etc/hosts";
 108     std::system(cmd.str().c_str());
 109   }
 110
 111   void DropPackets() {
 112     std::ostringstream cmd;
 113     // drop packets with src IP = ipv4_address_
 114     cmd << "iptables -A INPUT -s " << ipv4_address_ << " -j DROP";
 115
 116     std::system(cmd.str().c_str());
 117     // clear the stream
 118     cmd.str("");
 119
 120     // drop packets with dst IP = ipv4_address_
 121     cmd << "iptables -A INPUT -d " << ipv4_address_ << " -j DROP";
 122   }
 123
 124   void RestoreNetwork() {
 125     std::ostringstream cmd;
 126     // remove iptables rule to drop packets with src IP = ipv4_address_
 127     cmd << "iptables -D INPUT -s " << ipv4_address_ << " -j DROP";
 128     std::system(cmd.str().c_str());
 129     // clear the stream
 130     cmd.str("");
 131     // remove iptables rule to drop packets with dest IP = ipv4_address_
 132     cmd << "iptables -D INPUT -d " << ipv4_address_ << " -j DROP";
 133   }
 134
 135   void FlakeNetwork() {
 136     std::ostringstream cmd;
 137     // Emulate a flaky network connection over interface_. Add a delay of 100ms
 138     // +/- 20ms, 0.1% packet loss, 1% duplicates and 0.01% corrupt packets.
 139     cmd << "tc qdisc replace dev " << interface_
 140         << " root netem delay 100ms 20ms distribution normal loss 0.1% "
 141            "duplicate "
 142            "0.1% corrupt 0.01% ";
 143     std::system(cmd.str().c_str());
 144   }
 145
 146   void UnflakeNetwork() {
 147     // Remove simulated network flake on interface_
 148     std::ostringstream cmd;
 149     cmd << "tc qdisc del dev " << interface_ << " root netem";
 150     std::system(cmd.str().c_str());
 151   }
 152
 153   void NetworkUp() {
 154     InterfaceUp();
 155     DNSUp();
 156   }
 157
 158   void NetworkDown() {
 159     InterfaceDown();
 160     DNSDown();
 161   }
 162
 163   void SetUp() override {
 164     NetworkUp();
 165     grpc_init();
 166     StartServer();
 167   }
 168
 169   void TearDown() override {
 170     NetworkDown();
 171     StopServer();
 172     grpc_shutdown();
 173   }
 174
 175   void StartServer() {
 176     // TODO (pjaikumar): Ideally, we should allocate the port dynamically using
 177     // grpc_pick_unused_port_or_die(). That doesn't work inside some docker
 178     // containers because port_server listens on localhost which maps to
 179     // ip6-looopback, but ipv6 support is not enabled by default in docker.
 180     port_ = SERVER_PORT;
 181
 182     server_.reset(new ServerData(port_, GetParam().credentials_type));
 183     server_->Start(server_host_);
 184   }
 185   void StopServer() { server_->Shutdown(); }
 186
 187   std::unique_ptr<grpc::testing::EchoTestService::Stub> BuildStub(
 188       const std::shared_ptr<Channel>& channel) {
 189     return grpc::testing::EchoTestService::NewStub(channel);
 190   }
 191
 192   std::shared_ptr<Channel> BuildChannel(
 193       const grpc::string& lb_policy_name,
 194       ChannelArguments args = ChannelArguments()) {
 195     if (lb_policy_name.size() > 0) {
 196       args.SetLoadBalancingPolicyName(lb_policy_name);
 197     }  // else, default to pick first
 198     auto channel_creds = GetCredentialsProvider()->GetChannelCredentials(
 199         GetParam().credentials_type, &args);
 200     std::ostringstream server_address;
 201     server_address << server_host_ << ":" << port_;
 202     return CreateCustomChannel(server_address.str(), channel_creds, args);
 203   }
 204
 205   bool SendRpc(
 206       const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
 207       int timeout_ms = 0, bool wait_for_ready = false) {
 208     auto response = std::unique_ptr<EchoResponse>(new EchoResponse());
 209     EchoRequest request;
 210     auto& msg = GetParam().message_content;
 211     request.set_message(msg);
 212     ClientContext context;
 213     if (timeout_ms > 0) {
 214       context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms));
 215     }
 216     // See https://github.com/grpc/grpc/blob/master/doc/wait-for-ready.md for
 217     // details of wait-for-ready semantics
 218     if (wait_for_ready) {
 219       context.set_wait_for_ready(true);
 220     }
 221     Status status = stub->Echo(&context, request, response.get());
 222     auto ok = status.ok();
 223     int stream_id = 0;
 224     grpc_call* call = context.c_call();
 225     if (call) {
 226       grpc_chttp2_stream* stream = grpc_chttp2_stream_from_call(call);
 227       if (stream) {
 228         stream_id = stream->id;
 229       }
 230     }
 231     if (ok) {
 232       gpr_log(GPR_DEBUG, "RPC with stream_id %d succeeded", stream_id);
 233     } else {
 234       gpr_log(GPR_DEBUG, "RPC with stream_id %d failed: %s", stream_id,
 235               status.error_message().c_str());
 236     }
 237     return ok;
 238   }
 239
 240   struct ServerData {
 241     int port_;
 242     const grpc::string creds_;
 243     std::unique_ptr<Server> server_;
 244     TestServiceImpl service_;
 245     std::unique_ptr<std::thread> thread_;
 246     bool server_ready_ = false;
 247
 248     ServerData(int port, const grpc::string& creds)
 249         : port_(port), creds_(creds) {}
 250
 251     void Start(const grpc::string& server_host) {
 252       gpr_log(GPR_INFO, "starting server on port %d", port_);
 253       std::mutex mu;
 254       std::unique_lock<std::mutex> lock(mu);
 255       std::condition_variable cond;
 256       thread_.reset(new std::thread(
 257           std::bind(&ServerData::Serve, this, server_host, &mu, &cond)));
 258       cond.wait(lock, [this] { return server_ready_; });
 259       server_ready_ = false;
 260       gpr_log(GPR_INFO, "server startup complete");
 261     }
 262
 263     void Serve(const grpc::string& server_host, std::mutex* mu,
 264                std::condition_variable* cond) {
 265       std::ostringstream server_address;
 266       server_address << server_host << ":" << port_;
 267       ServerBuilder builder;
 268       auto server_creds =
 269           GetCredentialsProvider()->GetServerCredentials(creds_);
 270       builder.AddListeningPort(server_address.str(), server_creds);
 271       builder.RegisterService(&service_);
 272       server_ = builder.BuildAndStart();
 273       std::lock_guard<std::mutex> lock(*mu);
 274       server_ready_ = true;
 275       cond->notify_one();
 276     }
 277
 278     void Shutdown() {
 279       server_->Shutdown(grpc_timeout_milliseconds_to_deadline(0));
 280       thread_->join();
 281     }
 282   };
 283
 284   bool WaitForChannelNotReady(Channel* channel, int timeout_seconds = 5) {
 285     const gpr_timespec deadline =
 286         grpc_timeout_seconds_to_deadline(timeout_seconds);
 287     grpc_connectivity_state state;
 288     while ((state = channel->GetState(false /* try_to_connect */)) ==
 289            GRPC_CHANNEL_READY) {
 290       if (!channel->WaitForStateChange(state, deadline)) return false;
 291     }
 292     return true;
 293   }
 294
 295   bool WaitForChannelReady(Channel* channel, int timeout_seconds = 5) {
 296     const gpr_timespec deadline =
 297         grpc_timeout_seconds_to_deadline(timeout_seconds);
 298     grpc_connectivity_state state;
 299     while ((state = channel->GetState(true /* try_to_connect */)) !=
 300            GRPC_CHANNEL_READY) {
 301       if (!channel->WaitForStateChange(state, deadline)) return false;
 302     }
 303     return true;
 304   }
 305
 306  private:
 307   const grpc::string server_host_;
 308   const grpc::string interface_;
 309   const grpc::string ipv4_address_;
 310   const grpc::string netmask_;
 311   std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
 312   std::unique_ptr<ServerData> server_;
 313   const int SERVER_PORT = 32750;
 314   int port_;
 315 };
 316
 317 std::vector<TestScenario> CreateTestScenarios() {
 318   std::vector<TestScenario> scenarios;
 319   std::vector<grpc::string> credentials_types;
 320   std::vector<grpc::string> messages;
 321
 322   credentials_types.push_back(kInsecureCredentialsType);
 323   auto sec_list = GetCredentialsProvider()->GetSecureCredentialsTypeList();
 324   for (auto sec = sec_list.begin(); sec != sec_list.end(); sec++) {
 325     credentials_types.push_back(*sec);
 326   }
 327
 328   messages.push_back("🖖");
 329   for (size_t k = 1; k < GRPC_DEFAULT_MAX_RECV_MESSAGE_LENGTH / 1024; k *= 32) {
 330     grpc::string big_msg;
 331     for (size_t i = 0; i < k * 1024; ++i) {
 332       char c = 'a' + (i % 26);
 333       big_msg += c;
 334     }
 335     messages.push_back(big_msg);
 336   }
 337   for (auto cred = credentials_types.begin(); cred != credentials_types.end();
 338        ++cred) {
 339     for (auto msg = messages.begin(); msg != messages.end(); msg++) {
 340       scenarios.emplace_back(*cred, *msg);
 341     }
 342   }
 343
 344   return scenarios;
 345 }
 346
 347 INSTANTIATE_TEST_CASE_P(FlakyNetworkTest, FlakyNetworkTest,
 348                         ::testing::ValuesIn(CreateTestScenarios()));
 349
 350 // Network interface connected to server flaps
 351 TEST_P(FlakyNetworkTest, NetworkTransition) {
 352   const int kKeepAliveTimeMs = 1000;
 353   const int kKeepAliveTimeoutMs = 1000;
 354   ChannelArguments args;
 355   args.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, kKeepAliveTimeMs);
 356   args.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, kKeepAliveTimeoutMs);
 357   args.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
 358   args.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
 359
 360   auto channel = BuildChannel("pick_first", args);
 361   auto stub = BuildStub(channel);
 362   // Channel should be in READY state after we send an RPC
 363   EXPECT_TRUE(SendRpc(stub));
 364   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 365
 366   std::atomic_bool shutdown{false};
 367   std::thread sender = std::thread([this, &stub, &shutdown]() {
 368     while (true) {
 369       if (shutdown.load()) {
 370         return;
 371       }
 372       SendRpc(stub);
 373       std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 374     }
 375   });
 376
 377   // bring down network
 378   NetworkDown();
 379   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
 380   // bring network interface back up
 381   InterfaceUp();
 382   std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 383   // Restore DNS entry for server
 384   DNSUp();
 385   EXPECT_TRUE(WaitForChannelReady(channel.get()));
 386   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 387   shutdown.store(true);
 388   sender.join();
 389 }
 390
 391 // Traffic to server server is blackholed temporarily with keepalives enabled
 392 TEST_P(FlakyNetworkTest, ServerUnreachableWithKeepalive) {
 393   const int kKeepAliveTimeMs = 1000;
 394   const int kKeepAliveTimeoutMs = 1000;
 395   const int kReconnectBackoffMs = 1000;
 396   ChannelArguments args;
 397   args.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, kKeepAliveTimeMs);
 398   args.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, kKeepAliveTimeoutMs);
 399   args.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
 400   args.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
 401   // max time for a connection attempt
 402   args.SetInt(GRPC_ARG_MIN_RECONNECT_BACKOFF_MS, kReconnectBackoffMs);
 403   // max time between reconnect attempts
 404   args.SetInt(GRPC_ARG_MAX_RECONNECT_BACKOFF_MS, kReconnectBackoffMs);
 405
 406   gpr_log(GPR_DEBUG, "FlakyNetworkTest.ServerUnreachableWithKeepalive start");
 407   auto channel = BuildChannel("pick_first", args);
 408   auto stub = BuildStub(channel);
 409   // Channel should be in READY state after we send an RPC
 410   EXPECT_TRUE(SendRpc(stub));
 411   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 412
 413   std::atomic_bool shutdown{false};
 414   std::thread sender = std::thread([this, &stub, &shutdown]() {
 415     while (true) {
 416       if (shutdown.load()) {
 417         return;
 418       }
 419       SendRpc(stub);
 420       std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 421     }
 422   });
 423
 424   // break network connectivity
 425   gpr_log(GPR_DEBUG, "Adding iptables rule to drop packets");
 426   DropPackets();
 427   std::this_thread::sleep_for(std::chrono::milliseconds(10000));
 428   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
 429   // bring network interface back up
 430   RestoreNetwork();
 431   gpr_log(GPR_DEBUG, "Removed iptables rule to drop packets");
 432   EXPECT_TRUE(WaitForChannelReady(channel.get()));
 433   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 434   shutdown.store(true);
 435   sender.join();
 436   gpr_log(GPR_DEBUG, "FlakyNetworkTest.ServerUnreachableWithKeepalive end");
 437 }
 438
 439 //
 440 // Traffic to server server is blackholed temporarily with keepalives disabled
 441 TEST_P(FlakyNetworkTest, ServerUnreachableNoKeepalive) {
 442   auto channel = BuildChannel("pick_first", ChannelArguments());
 443   auto stub = BuildStub(channel);
 444   // Channel should be in READY state after we send an RPC
 445   EXPECT_TRUE(SendRpc(stub));
 446   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 447
 448   // break network connectivity
 449   DropPackets();
 450
 451   std::thread sender = std::thread([this, &stub]() {
 452     // RPC with deadline should timeout
 453     EXPECT_FALSE(SendRpc(stub, /*timeout_ms=*/500, /*wait_for_ready=*/true));
 454     // RPC without deadline forever until call finishes
 455     EXPECT_TRUE(SendRpc(stub, /*timeout_ms=*/0, /*wait_for_ready=*/true));
 456   });
 457
 458   std::this_thread::sleep_for(std::chrono::milliseconds(2000));
 459   // bring network interface back up
 460   RestoreNetwork();
 461
 462   // wait for RPC to finish
 463   sender.join();
 464 }
 465
 466 // Send RPCs over a flaky network connection
 467 TEST_P(FlakyNetworkTest, FlakyNetwork) {
 468   const int kKeepAliveTimeMs = 1000;
 469   const int kKeepAliveTimeoutMs = 1000;
 470   const int kMessageCount = 100;
 471   ChannelArguments args;
 472   args.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, kKeepAliveTimeMs);
 473   args.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, kKeepAliveTimeoutMs);
 474   args.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
 475   args.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
 476
 477   auto channel = BuildChannel("pick_first", args);
 478   auto stub = BuildStub(channel);
 479   // Channel should be in READY state after we send an RPC
 480   EXPECT_TRUE(SendRpc(stub));
 481   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 482
 483   // simulate flaky network (packet loss, corruption and delays)
 484   FlakeNetwork();
 485   for (int i = 0; i < kMessageCount; ++i) {
 486     SendRpc(stub);
 487   }
 488   // remove network flakiness
 489   UnflakeNetwork();
 490   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 491 }
 492
 493 // Server is shutdown gracefully and restarted. Client keepalives are enabled
 494 TEST_P(FlakyNetworkTest, ServerRestartKeepaliveEnabled) {
 495   const int kKeepAliveTimeMs = 1000;
 496   const int kKeepAliveTimeoutMs = 1000;
 497   ChannelArguments args;
 498   args.SetInt(GRPC_ARG_KEEPALIVE_TIME_MS, kKeepAliveTimeMs);
 499   args.SetInt(GRPC_ARG_KEEPALIVE_TIMEOUT_MS, kKeepAliveTimeoutMs);
 500   args.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
 501   args.SetInt(GRPC_ARG_HTTP2_MAX_PINGS_WITHOUT_DATA, 0);
 502
 503   auto channel = BuildChannel("pick_first", args);
 504   auto stub = BuildStub(channel);
 505   // Channel should be in READY state after we send an RPC
 506   EXPECT_TRUE(SendRpc(stub));
 507   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 508
 509   // server goes down, client should detect server going down and calls should
 510   // fail
 511   StopServer();
 512   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
 513   EXPECT_FALSE(SendRpc(stub));
 514
 515   std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 516
 517   // server restarts, calls succeed
 518   StartServer();
 519   EXPECT_TRUE(WaitForChannelReady(channel.get()));
 520   // EXPECT_TRUE(SendRpc(stub));
 521 }
 522
 523 // Server is shutdown gracefully and restarted. Client keepalives are enabled
 524 TEST_P(FlakyNetworkTest, ServerRestartKeepaliveDisabled) {
 525   auto channel = BuildChannel("pick_first", ChannelArguments());
 526   auto stub = BuildStub(channel);
 527   // Channel should be in READY state after we send an RPC
 528   EXPECT_TRUE(SendRpc(stub));
 529   EXPECT_EQ(channel->GetState(false), GRPC_CHANNEL_READY);
 530
 531   // server sends GOAWAY when it's shutdown, so client attempts to reconnect
 532   StopServer();
 533   std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 534
 535   EXPECT_TRUE(WaitForChannelNotReady(channel.get()));
 536
 537   std::this_thread::sleep_for(std::chrono::milliseconds(1000));
 538
 539   // server restarts, calls succeed
 540   StartServer();
 541   EXPECT_TRUE(WaitForChannelReady(channel.get()));
 542 }
 543
 544 }  // namespace
 545 }  // namespace testing
 546 }  // namespace grpc
 547 #endif  // GPR_LINUX
 548
 549 int main(int argc, char** argv) {
 550   ::testing::InitGoogleTest(&argc, argv);
 551   grpc_test_init(argc, argv);
 552   auto result = RUN_ALL_TESTS();
 553   return result;
 554 }