#include <gflags/gflags.h>
#include <glog/logging.h>
#include <gtest/gtest.h>
#include <unistd.h>
#include <memory>
#include <string>
#include "client_wrapper.h"
#include "e2e_utils.h"
#include "process_handler.h"
#include "types.h"
#include "utils.h"
USE_engine_flags;
FLAG_etcd_endpoints;
FLAG_master_path;
FLAG_client_path;
FLAG_out_dir;
constexpr int master_port_base = 50051;
constexpr int client_port_base = 12888;
namespace mooncake {
namespace testing {
class ChaosTest : public ::testing::Test {
protected:
static void SetUpTestSuite() {
google::InitGoogleLogging("ClientIntegrationTest");
FLAGS_logtostderr = 1;
LOG(INFO) << "Protocol: " << FLAGS_protocol
<< ", Device name: " << FLAGS_device_name
<< ", Metadata URL: " << FLAGS_engine_meta_url;
master_view_helper_ = std::make_shared<MasterViewHelper>();
EXPECT_EQ(master_view_helper_->ConnectToEtcd(FLAGS_etcd_endpoints),
ErrorCode::OK)
<< "Failed to connect to etcd";
}
static void TearDownTestSuite() { google::ShutdownGoogleLogging(); }
void SetUp() override {
const int master_num = 3;
for (int i = 0; i < master_num; ++i) {
masters_.emplace_back(
std::make_unique<mooncake::testing::MasterProcessHandler>(
FLAGS_master_path, FLAGS_etcd_endpoints,
master_port_base + i, i, FLAGS_out_dir));
ASSERT_TRUE(masters_.back()->start());
}
WaitMasterViewChange();
GetLeaderIndex(leader_index_);
ASSERT_TRUE(leader_index_ >= 0 && leader_index_ < master_num);
}
void TearDown() override {
masters_.clear();
}
static std::shared_ptr<ClientTestWrapper> CreateClientWrapper(
const std::string& host_name) {
auto client_opt = ClientTestWrapper::CreateClientWrapper(
host_name,
FLAGS_engine_meta_url,
FLAGS_protocol, FLAGS_device_name,
"etcd://" + FLAGS_etcd_endpoints);
EXPECT_TRUE(client_opt.has_value()) << "Failed to create client";
if (!client_opt.has_value()) {
return nullptr;
}
return *client_opt;
}
static void GetLeaderIndex(int& leader_index) {
std::string master_address;
ViewVersionId version;
ASSERT_EQ(master_view_helper_->GetMasterView(master_address, version),
ErrorCode::OK);
size_t colon_pos = master_address.find(':');
ASSERT_NE(colon_pos, std::string::npos)
<< "Master address should contain port part separated by ':', got: "
<< master_address;
std::string port_str = master_address.substr(colon_pos + 1);
ASSERT_FALSE(port_str.empty())
<< "Port part should not be empty in master address: "
<< master_address;
int port;
try {
port = std::stoi(port_str);
} catch (const std::exception& e) {
FAIL() << "Port part should be a valid integer, got: '" << port_str
<< "' in master address: " << master_address;
}
leader_index = port - master_port_base;
}
static void WaitMasterViewChange() {
sleep(ETCD_MASTER_VIEW_LEASE_TTL * 3);
}
static void WaitClientCrashDetection() {
sleep(DEFAULT_CLIENT_LIVE_TTL_SEC + 5);
}
static std::shared_ptr<MasterViewHelper> master_view_helper_;
std::vector<std::unique_ptr<mooncake::testing::MasterProcessHandler>>
masters_;
int leader_index_;
};
std::shared_ptr<MasterViewHelper> ChaosTest::master_view_helper_ = nullptr;
TEST_F(ChaosTest, LeaderKilledFailover) {
std::shared_ptr<ClientTestWrapper> client =
CreateClientWrapper("0.0.0.0:" + std::to_string(client_port_base));
ASSERT_TRUE(client != nullptr);
void* buffer;
ASSERT_EQ(client->Mount(1024 * 1024 * 16, buffer), ErrorCode::OK);
std::string key = "key";
std::string value = "value";
ASSERT_EQ(client->Put(key, value), ErrorCode::OK);
ASSERT_TRUE(masters_[leader_index_]->kill());
WaitMasterViewChange();
std::string key2 = "key2";
std::string value2 = "value2";
ASSERT_EQ(client->Put(key2, value2), ErrorCode::OK);
std::string get_value2;
ASSERT_EQ(client->Get(key2, get_value2), ErrorCode::OK);
ASSERT_EQ(get_value2, value2);
}
TEST_F(ChaosTest, BackupMasterKilled) {
const int master_num = static_cast<int>(masters_.size());
std::shared_ptr<ClientTestWrapper> client =
CreateClientWrapper("0.0.0.0:" + std::to_string(client_port_base));
ASSERT_TRUE(client != nullptr);
void* buffer;
ASSERT_EQ(client->Mount(1024 * 1024 * 16, buffer), ErrorCode::OK);
std::string key = "key";
std::string value = "value";
ASSERT_EQ(client->Put(key, value), ErrorCode::OK);
for (int i = 0; i < master_num; ++i) {
if (i != leader_index_) {
ASSERT_TRUE(masters_[i]->kill());
}
}
std::string get_value;
ASSERT_EQ(client->Get(key, get_value), ErrorCode::OK);
ASSERT_EQ(get_value, value);
}
TEST_F(ChaosTest, AllMastersOtherThanOneBackedUpKilledFailover) {
const int master_num = static_cast<int>(masters_.size());
std::shared_ptr<ClientTestWrapper> client =
CreateClientWrapper("0.0.0.0:" + std::to_string(client_port_base));
ASSERT_TRUE(client != nullptr);
void* buffer;
ASSERT_EQ(client->Mount(1024 * 1024 * 16, buffer), ErrorCode::OK);
bool find_one_backup = false;
for (int i = 0; i < master_num; ++i) {
if (i != leader_index_ && !find_one_backup) {
find_one_backup = true;
} else {
ASSERT_TRUE(masters_[i]->kill());
}
}
WaitMasterViewChange();
std::string key = "key";
std::string value = "value";
ASSERT_EQ(client->Put(key, value), ErrorCode::OK);
std::string get_value;
ASSERT_EQ(client->Get(key, get_value), ErrorCode::OK);
ASSERT_EQ(get_value, value);
}
TEST_F(ChaosTest, AllMastersKilledThenRestartFailover) {
const int master_num = static_cast<int>(masters_.size());
std::shared_ptr<ClientTestWrapper> client =
CreateClientWrapper("0.0.0.0:" + std::to_string(client_port_base));
ASSERT_TRUE(client != nullptr);
void* buffer;
ASSERT_EQ(client->Mount(1024 * 1024 * 16, buffer), ErrorCode::OK);
for (int i = 0; i < master_num; ++i) {
ASSERT_TRUE(masters_[i]->kill());
}
for (int i = 0; i < master_num; ++i) {
ASSERT_TRUE(masters_[i]->start());
}
WaitMasterViewChange();
std::string key = "key";
std::string value = "value";
ASSERT_EQ(client->Put(key, value), ErrorCode::OK);
std::string get_value;
ASSERT_EQ(client->Get(key, get_value), ErrorCode::OK);
ASSERT_EQ(get_value, value);
}
TEST_F(ChaosTest, ClientGracefulClose) {
std::shared_ptr<ClientTestWrapper> to_close_client =
CreateClientWrapper("0.0.0.0:" + std::to_string(client_port_base));
ASSERT_TRUE(to_close_client != nullptr);
std::shared_ptr<ClientTestWrapper> other_client =
CreateClientWrapper("0.0.0.0:" + std::to_string(client_port_base + 1));
ASSERT_TRUE(other_client != nullptr);
void* buffer;
ASSERT_EQ(to_close_client->Mount(1024 * 1024 * 16, buffer), ErrorCode::OK);
std::string key = "key";
std::string value = "value";
ASSERT_EQ(other_client->Put(key, value), ErrorCode::OK);
to_close_client.reset();
std::string get_value;
ASSERT_EQ(other_client->Get(key, get_value), ErrorCode::OBJECT_NOT_FOUND);
ASSERT_EQ(other_client->Put(key, value), ErrorCode::NO_AVAILABLE_HANDLE);
}
TEST_F(ChaosTest, ClientKilledFailover) {
ClientRunnerConfig clt_runner_cfg = {
.put_prob = 0,
.get_prob = 0,
.mount_prob = 1000,
.unmount_prob = 0,
.port = client_port_base + 0,
.master_server_entry = "etcd://" + FLAGS_etcd_endpoints,
.engine_meta_url = FLAGS_engine_meta_url,
.protocol = FLAGS_protocol,
.device_name = FLAGS_device_name,
};
ClientProcessHandler to_close_client(FLAGS_client_path, 0, FLAGS_out_dir,
clt_runner_cfg);
ASSERT_TRUE(to_close_client.start());
std::shared_ptr<ClientTestWrapper> other_client =
CreateClientWrapper("0.0.0.0:" + std::to_string(client_port_base + 1));
ASSERT_TRUE(other_client != nullptr);
sleep(10);
std::string key = "key";
std::string value = "value";
ASSERT_EQ(other_client->Put(key, value), ErrorCode::OK);
std::string get_value;
ASSERT_EQ(other_client->Get(key, get_value), ErrorCode::OK);
ASSERT_EQ(get_value, value);
ASSERT_TRUE(to_close_client.kill());
WaitClientCrashDetection();
ASSERT_EQ(other_client->Get(key, get_value), ErrorCode::OBJECT_NOT_FOUND);
ASSERT_EQ(other_client->Put(key, value), ErrorCode::NO_AVAILABLE_HANDLE);
}
}
}
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
gflags::ParseCommandLineFlags(&argc, &argv, false);
return RUN_ALL_TESTS();
}