Google Git
Sign in
apache / kudu / df40fff0dea06fde06c3cc4967047177524f0b09 / . / src / kudu / master / dynamic_multi_master-test.cc
blob: d605fe0d378b0b3c978e4b9e8aa98eb3a40f01f3 [file] [log] [blame]
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include <algorithm>
#include <atomic>
#include <cstdint>
#include <functional>
#include <iterator>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <ostream>
#include <string>
#include <thread>
#include <tuple>
#include <unordered_set>
#include <utility>
#include <vector>
#include <gflags/gflags.h>
#include <glog/logging.h>
#include <gtest/gtest.h>
#include <kudu/gutil/strings/util.h>
#include "kudu/client/client.h"
#include "kudu/client/schema.h"
#include "kudu/common/common.pb.h"
#include "kudu/common/wire_protocol.h"
#include "kudu/common/wire_protocol.pb.h"
#include "kudu/consensus/consensus.pb.h"
#include "kudu/consensus/consensus.proxy.h"
#include "kudu/consensus/metadata.pb.h"
#include "kudu/fs/dir_manager.h"
#include "kudu/fs/fs_manager.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/integration-tests/cluster_itest_util.h"
#include "kudu/integration-tests/cluster_verifier.h"
#include "kudu/integration-tests/test_workload.h"
#include "kudu/master/master.h"
#include "kudu/master/master.pb.h"
#include "kudu/master/master.proxy.h"
#include "kudu/master/sys_catalog.h"
#include "kudu/mini-cluster/external_mini_cluster.h"
#include "kudu/mini-cluster/mini_cluster.h"
#include "kudu/rpc/rpc_controller.h"
#include "kudu/tools/tool_test_util.h"
#include "kudu/tserver/tserver.pb.h"
#include "kudu/util/env.h"
#include "kudu/util/env_util.h"
#include "kudu/util/locks.h"
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/net/sockaddr.h"
#include "kudu/util/net/socket.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/random.h"
#include "kudu/util/scoped_cleanup.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
DECLARE_string(fs_wal_dir);
DECLARE_string(fs_data_dirs);
METRIC_DECLARE_histogram(log_gc_duration);
METRIC_DECLARE_entity(tablet);
using kudu::client::KuduClient;
using kudu::client::KuduColumnSchema;
using kudu::client::KuduSchema;
using kudu::client::KuduSchemaBuilder;
using kudu::client::KuduTableCreator;
using kudu::client::sp::shared_ptr;
using kudu::cluster::ExternalDaemonOptions;
using kudu::cluster::ExternalMaster;
using kudu::cluster::ExternalMiniCluster;
using kudu::cluster::ExternalMiniClusterOptions;
using kudu::cluster::MiniCluster;
using kudu::consensus::ConsensusServiceProxy;
using kudu::consensus::HealthReportPB;
using kudu::consensus::LeaderStepDownRequestPB;
using kudu::consensus::LeaderStepDownResponsePB;
using kudu::consensus::RaftConfigPB;
using kudu::consensus::RaftPeerPB;
using kudu::pb_util::SecureShortDebugString;
using kudu::rpc::RpcController;
using std::atomic;
using std::map;
using std::optional;
using std::string;
using std::thread;
using std::tuple;
using std::unique_ptr;
using std::unordered_set;
using std::vector;
using strings::Substitute;
namespace kudu {
namespace master {
namespace {
Status CreateTableWithClient(KuduClient* client, const std::string& table_name) {
KuduSchema schema;
KuduSchemaBuilder b;
b.AddColumn("key")->Type(KuduColumnSchema::INT32)->NotNull()->PrimaryKey();
RETURN_NOT_OK(b.Build(&schema));
unique_ptr<KuduTableCreator> table_creator(client->NewTableCreator());
return table_creator->table_name(table_name)
.schema(&schema)
.set_range_partition_columns({ "key" })
.num_replicas(1)
.Create();
}
Status CreateTable(ExternalMiniCluster* cluster,
const std::string& table_name) {
shared_ptr<KuduClient> client;
RETURN_NOT_OK(cluster->CreateClient(nullptr, &client));
return CreateTableWithClient(client.get(), table_name);
}
Status ReserveSocketForMaster(int master_idx, unique_ptr<Socket>* socket,
Sockaddr* addr, HostPort* hp) {
unique_ptr<Socket> s;
Sockaddr a;
RETURN_NOT_OK(MiniCluster::ReserveDaemonSocket(MiniCluster::MASTER, master_idx,
kDefaultBindMode, &s));
RETURN_NOT_OK(s->GetSocketAddress(&a));
*socket = std::move(s);
*addr = a;
*hp = HostPort(a);
return Status::OK();
}
// Functor that takes a leader_master_idx and runs the desired master RPC against
// the leader master returning the RPC status and the optional MasterErrorPB::Code.
typedef std::function<
std::pair<Status, optional<MasterErrorPB::Code>>(int leader_master_idx)> MasterRPC;
// Helper function that runs the master RPC against the leader master and retries the RPC
// if the expected leader master returns NOT_THE_LEADER error due to leadership change.
// Returns a single combined Status:
// - RPC return status if not OK.
// - IllegalState for a master response error other than NOT_THE_LEADER error.
// - TimedOut if all attempts to run the RPC against leader master are exhausted.
// - OK if the master RPC is successful.
Status RunLeaderMasterRPC(const MasterRPC& master_rpc, ExternalMiniCluster* cluster) {
int64_t time_left_to_sleep_msecs = 2000;
while (time_left_to_sleep_msecs > 0) {
int leader_master_idx;
RETURN_NOT_OK(cluster->GetLeaderMasterIndex(&leader_master_idx));
const auto& rpc_result = master_rpc(leader_master_idx);
RETURN_NOT_OK(rpc_result.first);
const auto& master_error = rpc_result.second;
if (!master_error) {
return Status::OK();
}
if (master_error != MasterErrorPB::NOT_THE_LEADER) {
// Some other master error.
return Status::IllegalState(Substitute("Master error: $0"),
MasterErrorPB_Code_Name(*master_error));
}
// NOT_THE_LEADER error, so retry after some duration.
static const MonoDelta kSleepDuration = MonoDelta::FromMilliseconds(100);
SleepFor(kSleepDuration);
time_left_to_sleep_msecs -= kSleepDuration.ToMilliseconds();
}
return Status::TimedOut("Failed contacting the right leader master after multiple attempts");
}
// Run ListMasters RPC, retrying on leadership change, returning the response
// in 'resp'.
Status RunListMasters(ListMastersResponsePB* resp, ExternalMiniCluster* cluster) {
auto list_masters = [&] (int leader_master_idx) {
ListMastersRequestPB req;
RpcController rpc;
Status s = cluster->master_proxy(leader_master_idx)->ListMasters(req, resp, &rpc);
optional<MasterErrorPB::Code> err_code;
if (resp->has_error()) {
err_code.emplace(resp->error().code());
}
return std::make_pair(s, err_code);
};
return RunLeaderMasterRPC(list_masters, cluster);
}
// Verify the 'cluster' contains 'num_masters' on all masters, and that they
// are all voters. Returns an error if the expected state is not present.
//
// This should be used instead of VerifyVoterMastersForCluster() if it is
// required that all masters have accepted the config change. E.g. tests that
// restart a cluster after adding a master should verify that all masters agree
// before restarting, in case lagging masters start up with stale configs.
//
// TODO(awong): we should be more robust to starting up with mismatched on-disk
// configs, if we can help it.
Status VerifyVotersOnAllMasters(int num_masters, ExternalMiniCluster* cluster) {
for (int i = 0; i < cluster->num_masters(); i++) {
ListMastersResponsePB resp;
ListMastersRequestPB req;
RpcController rpc;
RETURN_NOT_OK(cluster->master_proxy(i)->ListMasters(req, &resp, &rpc));
if (num_masters != resp.masters_size()) {
return Status::IllegalState(Substitute("expected $0 masters but got $1",
num_masters, resp.masters_size()));
}
for (const auto& master : resp.masters()) {
if ((master.role() != RaftPeerPB::LEADER && master.role() != RaftPeerPB::FOLLOWER) ||
master.member_type() != RaftPeerPB::VOTER ||
master.registration().rpc_addresses_size() != 1) {
return Status::IllegalState(Substitute("bad master: $0", SecureShortDebugString(master)));
}
}
}
return Status::OK();
}
// Verify the ExternalMiniCluster 'cluster' contains 'num_masters' overall and
// are all VOTERS. Populates the new master addresses in 'master_hps', if not
// nullptr. Returns an error if the expected state is not present.
//
// This should be used instead of VerifyVotersOnAllMasters() if the test does
// not need to wait for all masters to agree on the config.
Status VerifyVoterMastersForCluster(int num_masters, vector<HostPort>* master_hps,
ExternalMiniCluster* cluster) {
ListMastersResponsePB resp;
RETURN_NOT_OK(RunListMasters(&resp, cluster));
if (num_masters != resp.masters_size()) {
return Status::IllegalState(Substitute("expected $0 masters but got $1",
num_masters, resp.masters_size()));
}
vector<HostPort> hps;
for (const auto& master : resp.masters()) {
if ((master.role() != RaftPeerPB::LEADER && master.role() != RaftPeerPB::FOLLOWER) ||
master.member_type() != RaftPeerPB::VOTER ||
master.registration().rpc_addresses_size() != 1) {
return Status::IllegalState(Substitute("bad master: $0", SecureShortDebugString(master)));
}
hps.emplace_back(HostPortFromPB(master.registration().rpc_addresses(0)));
}
if (master_hps) {
*master_hps = std::move(hps);
}
return Status::OK();
}
// Initiates leadership transfer to the specified master returning status of
// the request. The request is performed synchronously, though the transfer of
// leadership is asynchronous -- callers need to wait to ensure leadership is
// actually transferred.
Status TransferMasterLeadershipAsync(ExternalMiniCluster* cluster, const string& master_uuid) {
int leader_master_idx;
RETURN_NOT_OK(cluster->GetLeaderMasterIndex(&leader_master_idx));
auto leader_master_addr = cluster->master(leader_master_idx)->bound_rpc_addr();
ConsensusServiceProxy consensus_proxy(cluster->messenger(), leader_master_addr,
leader_master_addr.host());
LeaderStepDownRequestPB req;
req.set_dest_uuid(cluster->master(leader_master_idx)->uuid());
req.set_tablet_id(master::SysCatalogTable::kSysCatalogTabletId);
req.set_new_leader_uuid(master_uuid);
req.set_mode(consensus::GRACEFUL);
LeaderStepDownResponsePB resp;
RpcController rpc;
RETURN_NOT_OK(consensus_proxy.LeaderStepDown(req, &resp, &rpc));
if (resp.has_error()) {
return StatusFromPB(resp.error().status());
}
return Status::OK();
}
// Transfers leadership among masters in the 'cluster' to the specified 'new_master_uuid'
// verifies the transfer is successful.
void TransferMasterLeadership(ExternalMiniCluster* cluster, const string& new_master_uuid) {
ASSERT_OK(TransferMasterLeadershipAsync(cluster, new_master_uuid));
// It takes some time for the leadership transfer to complete, hence the
// ASSERT_EVENTUALLY.
ASSERT_EVENTUALLY([&] {
int leader_master_idx = -1;
ASSERT_OK(cluster->GetLeaderMasterIndex(&leader_master_idx));
ASSERT_EQ(new_master_uuid, cluster->master(leader_master_idx)->uuid());
});
}
// Fetch uuid of the specified 'fs_wal_dir' and 'fs_data_dirs'.
Status GetFsUuid(const string& fs_wal_dir, const vector<string>& fs_data_dirs, string* uuid) {
google::FlagSaver saver;
FLAGS_fs_wal_dir = fs_wal_dir;
FLAGS_fs_data_dirs = JoinStrings(fs_data_dirs, ",");
FsManagerOpts fs_opts;
fs_opts.read_only = true;
fs_opts.update_instances = fs::UpdateInstanceBehavior::DONT_UPDATE;
FsManager fs_manager(Env::Default(), std::move(fs_opts));
RETURN_NOT_OK(fs_manager.PartialOpen());
*uuid = fs_manager.uuid();
return Status::OK();
}
} // anonymous namespace
// Test class for testing addition/removal of masters to a Kudu cluster.
class DynamicMultiMasterTest : public KuduTest {
protected:
typedef map<string, string> EnvVars;
void SetUpWithNumMasters(int num_masters) {
// Initial number of masters in the cluster before adding a master.
orig_num_masters_ = num_masters;
// Reserving a port upfront for the new master that'll be added to the cluster.
ASSERT_OK(ReserveSocketForMaster(/*index*/orig_num_masters_, &reserved_socket_,
&reserved_addr_, &reserved_hp_));
}
void StartCluster(const vector<string>& extra_master_flags = {},
bool supply_single_master_addr = true) {
opts_.num_masters = orig_num_masters_;
opts_.supply_single_master_addr = supply_single_master_addr;
opts_.extra_master_flags = extra_master_flags;
opts_.extra_master_flags.emplace_back("--master_auto_join_cluster=false");
cluster_.reset(new ExternalMiniCluster(opts_));
ASSERT_OK(cluster_->Start());
ASSERT_OK(cluster_->WaitForTabletServerCount(cluster_->num_tablet_servers(),
MonoDelta::FromSeconds(5)));
}
// Bring up a cluster with bunch of tables and ensure the system catalog WAL
// has been GC'ed.
// Out parameter 'master_hps' returns the HostPort of the masters in the original
// cluster.
void StartClusterWithSysCatalogGCed(vector<HostPort>* master_hps,
const vector<string>& extra_flags = {}) {
// Using low values of log flush threshold and segment size to trigger GC of the
// sys catalog WAL
vector<string> flags = {"--flush_threshold_secs=0", "--log_segment_size_mb=1"};
flags.insert(flags.end(), extra_flags.begin(), extra_flags.end());
NO_FATALS(StartCluster(flags));
// Verify that masters are running as VOTERs and collect their addresses to be used
// for starting the new master.
NO_FATALS(VerifyVoterMasters(orig_num_masters_, master_hps));
// Function to fetch the GC count of the system catalog WAL.
auto get_sys_catalog_wal_gc_count = [&] (int master_idx) {
int64_t sys_catalog_wal_gc_count = 0;
CHECK_OK(itest::GetInt64Metric(cluster_->master(master_idx)->bound_http_hostport(),
&METRIC_ENTITY_tablet,
master::SysCatalogTable::kSysCatalogTabletId,
&METRIC_log_gc_duration,
"total_count",
&sys_catalog_wal_gc_count));
return sys_catalog_wal_gc_count;
};
vector<int64_t> orig_gc_count(orig_num_masters_);
for (int master_idx = 0; master_idx < orig_num_masters_; master_idx++) {
orig_gc_count[master_idx] = get_sys_catalog_wal_gc_count(master_idx);
}
// Function to compute whether all masters have updated the system catalog WAL GC count.
// Ideally we could just check against the leader master but the leader master could
// potentially change hence checking across all masters.
auto all_masters_updated_wal_gc_count = [&] {
int num_masters_gc_updated = 0;
for (int master_idx = 0; master_idx < orig_num_masters_; master_idx++) {
if (get_sys_catalog_wal_gc_count(master_idx) > orig_gc_count[master_idx]) {
num_masters_gc_updated++;
}
}
return num_masters_gc_updated == orig_num_masters_;
};
// Create a bunch of tables to ensure sys catalog WAL gets GC'ed.
// Need to create around 1k tables even with lowest flush threshold and log segment size.
int i;
bool wal_gc_counts_updated = false;
for (i = 1; i < 1000; i++) {
if (all_masters_updated_wal_gc_count()) {
wal_gc_counts_updated = true;
break;
}
string table_name = Substitute("Table.$0.$1", CURRENT_TEST_NAME(), std::to_string(i));
ASSERT_OK(CreateTable(cluster_.get(), table_name));
}
LOG(INFO) << "Number of tables created: " << i - 1;
if (wal_gc_counts_updated) {
// We are done here and no need to wait further.
return;
}
MonoTime deadline = MonoTime::Now() + MonoDelta::FromSeconds(2);
while (MonoTime::Now() < deadline) {
if (all_masters_updated_wal_gc_count()) {
wal_gc_counts_updated = true;
break;
}
SleepFor(MonoDelta::FromMilliseconds(100));
}
ASSERT_TRUE(wal_gc_counts_updated) << "Timed out waiting for system catalog WAL to be GC'ed";
}
// Brings up a new master 'new_master_hp' where 'master_hps' contains master addresses including
// the new master to be added at the index 'new_master_idx' in the ExternalMiniCluster.
void StartNewMaster(const vector<HostPort>& master_hps,
const HostPort& new_master_hp,
int new_master_idx,
scoped_refptr<ExternalMaster>* new_master_out) {
vector<string> master_addresses;
master_addresses.reserve(master_hps.size());
for (const auto& hp : master_hps) {
master_addresses.emplace_back(hp.ToString());
}
ExternalDaemonOptions new_master_opts;
ASSERT_OK(BuildMasterOpts(new_master_idx, new_master_hp, &new_master_opts));
auto& flags = new_master_opts.extra_flags;
flags.insert(flags.end(), {
"--master_addresses=" + JoinStrings(master_addresses, ","),
"--master_address_add_new_master=" + new_master_hp.ToString(),
"--master_auto_join_cluster=false",
});
LOG(INFO) << "Bringing up the new master at: " << new_master_hp.ToString();
scoped_refptr<ExternalMaster> master = new ExternalMaster(new_master_opts);
ASSERT_OK(master->Start());
ASSERT_OK(master->WaitForCatalogManager());
Sockaddr new_master_addr;
ASSERT_OK(SockaddrFromHostPort(new_master_hp, &new_master_addr));
MasterServiceProxy new_master_proxy(new_master_opts.messenger, new_master_addr,
new_master_hp.host());
GetMasterRegistrationRequestPB req;
GetMasterRegistrationResponsePB resp;
RpcController rpc;
ASSERT_OK(new_master_proxy.GetMasterRegistration(req, &resp, &rpc));
ASSERT_FALSE(resp.has_error());
ASSERT_EQ(RaftPeerPB::NON_VOTER, resp.member_type());
ASSERT_EQ(RaftPeerPB::LEARNER, resp.role());
*new_master_out = std::move(master);
}
void VerifyVoterMasters(int num_masters, vector<HostPort>* master_hps = nullptr,
ExternalMiniCluster* cluster = nullptr) {
if (cluster == nullptr) {
cluster = cluster_.get();
}
NO_FATALS(VerifyVoterMastersForCluster(num_masters, master_hps, cluster));
}
// Fetch a follower (non-leader) master index from the cluster.
Status GetFollowerMasterIndex(int* follower_master_idx) {
int leader_master_idx;
RETURN_NOT_OK(cluster_->GetLeaderMasterIndex(&leader_master_idx));
int follower = -1;
for (int i = 0; i < cluster_->num_masters(); i++) {
if (i != leader_master_idx) {
follower = i;
break;
}
}
if (follower == -1) {
return Status::NotFound("No follower master found");
}
*follower_master_idx = follower;
return Status::OK();
}
// Builds and returns ExternalDaemonOptions used to add master with address 'rpc_addr' and
// 'master_idx' which helps with naming master's directories.
// Output 'kerberos_env_vars' parameter must be supplied for a secure cluster i.e. when Kerberos
// is enabled and returns Kerberos related environment variables.
Status BuildMasterOpts(int master_idx, const HostPort& rpc_addr,
ExternalDaemonOptions* master_opts,
EnvVars* kerberos_env_vars = nullptr) {
// Start with an existing master daemon's options, but modify them for use in a new master.
const string new_master_id = Substitute("master-$0", master_idx);
ExternalDaemonOptions opts = cluster_->master(0)->opts();
opts.rpc_bind_address = rpc_addr;
opts.wal_dir = cluster_->GetWalPath(new_master_id);
opts.data_dirs = cluster_->GetDataPaths(new_master_id);
opts.log_dir = cluster_->GetLogPath(new_master_id);
if (opts_.enable_kerberos) {
CHECK(kerberos_env_vars);
vector<string> flags;
RETURN_NOT_OK(cluster::ExternalDaemon::CreateKerberosConfig(
cluster_->kdc(), opts_.principal, rpc_addr.host(), &flags, kerberos_env_vars));
// Inserting the Kerberos related flags at the end will override flags from the master
// which was used as a basis for the new master.
opts.extra_flags.insert(opts.extra_flags.end(), std::make_move_iterator(flags.begin()),
std::make_move_iterator(flags.end()));
}
*master_opts = std::move(opts);
return Status::OK();
}
// Adds the specified master to the cluster using the CLI tool.
// Unset 'rpc_bind_address' in 'opts' can be used to indicate to not supply master address.
// Optional 'env_vars' can be used to set environment variables while running kudu tool.
// Optional 'wait_secs' can be used to supply wait timeout to the master add CLI tool.
// Optional 'kudu_binary' can be used to supply the path to the kudu binary.
// Returns generic RuntimeError() on failure with the actual error in the optional 'err'
// output parameter.
Status AddMasterToClusterUsingCLITool(const ExternalDaemonOptions& opts, string* err = nullptr,
EnvVars env_vars = {}, int wait_secs = 10,
const string& kudu_binary = "") {
auto hps = cluster_->master_rpc_addrs();
vector<string> addresses;
addresses.reserve(hps.size());
for (const auto& hp : hps) {
addresses.emplace_back(hp.ToString());
}
vector<string> cmd = {"master", "add", JoinStrings(addresses, ",")};
if (opts.rpc_bind_address.Initialized()) {
cmd.emplace_back(opts.rpc_bind_address.ToString());
}
vector<string> new_master_flags = ExternalMaster::GetMasterFlags(opts);
cmd.insert(cmd.end(), std::make_move_iterator(new_master_flags.begin()),
std::make_move_iterator(new_master_flags.end()));
cmd.insert(cmd.end(), opts.extra_flags.begin(), opts.extra_flags.end());
if (wait_secs != 0) {
cmd.emplace_back("-wait_secs=" + std::to_string(wait_secs));
}
if (!kudu_binary.empty()) {
cmd.emplace_back("-kudu_abs_path=" + kudu_binary);
}
RETURN_NOT_OK(env_util::CreateDirsRecursively(Env::Default(), opts.log_dir));
Status s = tools::RunKuduTool(cmd, nullptr, err, "", std::move(env_vars));
if (!s.ok() && err != nullptr) {
LOG(INFO) << "Add master stderr: " << *err;
}
RETURN_NOT_OK(s);
return Status::OK();
}
// Removes the specified master from the cluster using the CLI tool.
// Unset 'master_to_remove' can be used to indicate to not supply master address.
// Returns generic RuntimeError() on failure with the actual error in the optional 'err'
// output parameter.
Status RemoveMasterFromClusterUsingCLITool(const HostPort& master_to_remove,
string* err = nullptr,
const string& master_uuid = "") {
auto hps = cluster_->master_rpc_addrs();
vector<string> addresses;
addresses.reserve(hps.size());
for (const auto& hp : hps) {
addresses.emplace_back(hp.ToString());
}
vector<string> args = {"master", "remove", JoinStrings(addresses, ",")};
if (master_to_remove.Initialized()) {
args.push_back(master_to_remove.ToString());
}
if (!master_uuid.empty()) {
args.push_back("--master_uuid=" + master_uuid);
}
RETURN_NOT_OK(tools::RunKuduTool(args, nullptr, err));
return cluster_->RemoveMaster(master_to_remove);
}
// Fetch consensus state of the leader master.
void GetLeaderMasterConsensusState(RaftConfigPB* consensus_config) {
int leader_master_idx;
ASSERT_OK(cluster_->GetLeaderMasterIndex(&leader_master_idx));
auto leader_master_addr = cluster_->master(leader_master_idx)->bound_rpc_addr();
ConsensusServiceProxy consensus_proxy(cluster_->messenger(), leader_master_addr,
leader_master_addr.host());
consensus::GetConsensusStateRequestPB req;
consensus::GetConsensusStateResponsePB resp;
RpcController rpc;
req.set_dest_uuid(cluster_->master(leader_master_idx)->uuid());
req.set_report_health(consensus::INCLUDE_HEALTH_REPORT);
ASSERT_OK(consensus_proxy.GetConsensusState(req, &resp, &rpc));
ASSERT_FALSE(resp.has_error());
ASSERT_EQ(1, resp.tablets_size());
// Lookup the new_master from the consensus state of the system catalog.
const auto& sys_catalog = resp.tablets(0);
ASSERT_EQ(master::SysCatalogTable::kSysCatalogTabletId, sys_catalog.tablet_id());
const auto& cstate = sys_catalog.cstate();
*consensus_config = cstate.has_pending_config() ?
cstate.pending_config() : cstate.committed_config();
}
void VerifyDeadMasterInSpecifiedState(const string& dead_master_uuid,
HealthReportPB::HealthStatus expected_state) {
RaftConfigPB config;
NO_FATALS(GetLeaderMasterConsensusState(&config));
ASSERT_EQ(orig_num_masters_, config.peers_size());
bool dead_master_found = false;
for (const auto& peer : config.peers()) {
if (peer.permanent_uuid() == dead_master_uuid) {
dead_master_found = true;
ASSERT_EQ(expected_state, peer.health_report().overall_health());
break;
}
}
ASSERT_TRUE(dead_master_found);
}
// Verification steps after the new master has been added successfully and it's promoted
// as VOTER. The supplied 'master_hps' includes the new_master as well.
void VerifyClusterAfterMasterAddition(const vector<HostPort>& master_hps,
int expected_num_masters) {
// Collect information about the cluster for verification later before shutting
// it down.
UnorderedHostPortSet master_hps_set(master_hps.begin(), master_hps.end());
ASSERT_EQ(master_hps.size(), master_hps_set.size()) << "Duplicates found in master_hps";
unordered_set<string> master_uuids;
for (int i = 0; i < cluster_->num_masters(); i++) {
master_uuids.emplace(cluster_->master(i)->uuid());
}
string new_master_uuid;
ASSERT_OK(GetFsUuid(new_master_opts_.wal_dir, new_master_opts_.data_dirs, &new_master_uuid));
master_uuids.emplace(new_master_uuid);
// Shutdown the cluster and the new master daemon process.
// This allows ExternalMiniCluster to manage the newly added master and allows
// client to connect to the new master if it's elected the leader.
LOG(INFO) << "Shutting down the old cluster";
cluster_.reset();
LOG(INFO) << "Bringing up the migrated cluster";
opts_.num_masters = expected_num_masters;
opts_.master_rpc_addresses = master_hps;
ExternalMiniCluster migrated_cluster(opts_);
ASSERT_OK(migrated_cluster.Start());
for (int i = 0; i < migrated_cluster.num_masters(); i++) {
ASSERT_OK(migrated_cluster.master(i)->WaitForCatalogManager());
}
// Verify the cluster still has the same masters.
{
ListMastersResponsePB resp;
ASSERT_OK(RunListMasters(&resp, &migrated_cluster));
ASSERT_EQ(expected_num_masters, resp.masters_size());
UnorderedHostPortSet hps_found;
unordered_set<string> uuids_found;
for (const auto& master : resp.masters()) {
ASSERT_EQ(RaftPeerPB::VOTER, master.member_type());
ASSERT_TRUE(master.role() == RaftPeerPB::LEADER || master.role() == RaftPeerPB::FOLLOWER);
ASSERT_EQ(1, master.registration().rpc_addresses_size());
HostPort actual_hp = HostPortFromPB(master.registration().rpc_addresses(0));
ASSERT_TRUE(ContainsKey(master_hps_set, actual_hp));
hps_found.insert(actual_hp);
ASSERT_TRUE(ContainsKey(master_uuids, master.instance_id().permanent_uuid()));
uuids_found.emplace(master.instance_id().permanent_uuid());
}
ASSERT_EQ(master_hps_set, hps_found);
ASSERT_EQ(master_uuids, uuids_found);
}
// Transfer leadership to the new master.
LOG(INFO) << "Transferring leadership to master: " << new_master_uuid;
NO_FATALS(TransferMasterLeadership(&migrated_cluster, new_master_uuid));
shared_ptr<KuduClient> client;
ASSERT_OK(migrated_cluster.CreateClient(nullptr, &client));
ClusterVerifier cv(&migrated_cluster);
NO_FATALS(cv.CheckCluster());
LOG(INFO) << "Verifying the first table";
NO_FATALS(cv.CheckRowCount(kTableName, ClusterVerifier::EXACTLY, 0));
LOG(INFO) << "Creating and verifying the second table";
// Perform an operation that requires replication to masters.
ASSERT_OK(CreateTable(&migrated_cluster, "second_table"));
NO_FATALS(cv.CheckRowCount("second_table", ClusterVerifier::EXACTLY, 0));
// Pause one master at a time and create table at the same time which will allow
// new leader to be elected if the paused master is a leader.
// Need at least 3 masters to form consensus and elect a new leader.
if (expected_num_masters >= 3) {
LOG(INFO) << "Pausing and resuming individual masters";
string table_name = kTableName;
for (int i = 0; i < expected_num_masters; i++) {
auto* master = migrated_cluster.master(i);
LOG(INFO) << Substitute("Pausing master $0, $1", master->uuid(),
master->bound_rpc_hostport().ToString());
ASSERT_OK(master->Pause());
cluster::ScopedResumeExternalDaemon resume_daemon(master);
// We can run into table not found error in cases where the
// previously paused master that's leader of prior term resumes
// and the up to date follower doesn't become leader and the resumed
// master from previous term isn't up to date. See KUDU-3266 for details.
ASSERT_EVENTUALLY([&] {
NO_FATALS(cv.CheckRowCount(table_name, ClusterVerifier::EXACTLY, 0));
});
// See MasterFailoverTest.TestCreateTableSync to understand why we must
// check for IsAlreadyPresent as well.
table_name = Substitute("table-$0", i);
Status s = CreateTable(&migrated_cluster, table_name);
ASSERT_TRUE(s.ok() || s.IsAlreadyPresent());
}
}
}
// Function to prevent leadership changes among masters for quick tests.
void DisableMasterLeadershipTransfer() {
for (int i = 0 ; i < cluster_->num_masters(); i++) {
// Starting the cluster with following flag leads to a case sometimes
// wherein no leader gets elected leading to failure in ConnectToMaster() RPC.
// So instead set the flag after the cluster is running.
ASSERT_OK(cluster_->SetFlag(cluster_->master(i),
"leader_failure_max_missed_heartbeat_periods", "10.0"));
}
}
// Helper function for recovery tests that shuts down and deletes state of a follower master
// and removes it from the Raft configuration.
// Output parameters:
// dead_master_idx: Index of the follower master selected for recovery in the
// ExternalMiniCluster
// dead_master_hp: HostPort of the follower master selected for recovery
// src_master_hp: HostPort of an existing master that can be used as a source for recovery
void FailAndRemoveFollowerMaster(const vector<HostPort>& master_hps, int* dead_master_idx,
HostPort* dead_master_hp, HostPort* src_master_hp) {
// We'll be selecting a follower master to be shutdown to simulate a dead master
// and to prevent the test from being flaky we disable master leadership change.
NO_FATALS(DisableMasterLeadershipTransfer());
int master_to_recover_idx = -1;
ASSERT_OK(GetFollowerMasterIndex(&master_to_recover_idx));
ASSERT_NE(master_to_recover_idx, -1);
scoped_refptr<ExternalMaster> master_to_recover(cluster_->master(master_to_recover_idx));
const auto master_to_recover_hp = master_to_recover->bound_rpc_hostport();
LOG(INFO) << Substitute("Shutting down and deleting the state of the master to be recovered "
"HostPort: $0, UUID: $1, index : $2", master_to_recover_hp.ToString(),
master_to_recover->uuid(), master_to_recover_idx);
NO_FATALS(master_to_recover->Shutdown());
ASSERT_OK(master_to_recover->DeleteFromDisk());
LOG(INFO) << "Detecting transition to terminal FAILED state";
ASSERT_EVENTUALLY([&] {
VerifyDeadMasterInSpecifiedState(master_to_recover->uuid(), HealthReportPB::FAILED);
});
// Verify the master to be removed is part of the list of masters.
ASSERT_NE(std::find(master_hps.begin(), master_hps.end(), master_to_recover_hp),
master_hps.end());
// Source master to copy system catalog out of at least 2 masters in the cluster.
// Recovery tests start with at least 2 masters.
// Need to capture the source master before removing master as once the master
// is removed it won't be tracked by ExternalMiniCluster and the indices would change.
ASSERT_GE(cluster_->num_masters(), 2);
int src_master_idx = master_to_recover_idx == 0 ? 1 : 0;
*src_master_hp = cluster_->master(src_master_idx)->bound_rpc_hostport();
ASSERT_OK(RemoveMasterFromClusterUsingCLITool(master_to_recover_hp));
// Verify we have one master less and the desired master was removed.
vector<HostPort> updated_master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_ - 1, &updated_master_hps));
UnorderedHostPortSet expected_master_hps(master_hps.begin(), master_hps.end());
expected_master_hps.erase(master_to_recover_hp);
UnorderedHostPortSet actual_master_hps(updated_master_hps.begin(), updated_master_hps.end());
ASSERT_EQ(expected_master_hps, actual_master_hps);
ClusterVerifier cv(cluster_.get());
NO_FATALS(cv.CheckCluster());
NO_FATALS(cv.CheckRowCount(kTableName, ClusterVerifier::EXACTLY, 0));
// Set the remaining output parameters.
*dead_master_idx = master_to_recover_idx;
*dead_master_hp = master_to_recover_hp;
}
// Tracks the current number of masters in the cluster
int orig_num_masters_;
ExternalMiniClusterOptions opts_;
unique_ptr<ExternalMiniCluster> cluster_;
// Socket, HostPort, proxy etc. for the new master to be added
unique_ptr<Socket> reserved_socket_;
Sockaddr reserved_addr_;
HostPort reserved_hp_;
ExternalDaemonOptions new_master_opts_;
static const char* const kTableName;
};
const char* const DynamicMultiMasterTest::kTableName = "first_table";
// Parameterized DynamicMultiMasterTest class that works with different initial number of masters
// and secure/unsecure clusters.
class ParameterizedAddMasterTest : public DynamicMultiMasterTest,
public ::testing::WithParamInterface<tuple<int, bool>> {
public:
void SetUp() override {
NO_FATALS(SetUpWithNumMasters(std::get<0>(GetParam())));
opts_.enable_kerberos = std::get<1>(GetParam());
}
};
INSTANTIATE_TEST_SUITE_P(, ParameterizedAddMasterTest,
::testing::Combine(
// Initial number of masters in the cluster before adding a new
// master
::testing::Values(1, 2),
// Whether Kerberos is enabled
::testing::Bool()));
// This test starts a cluster, creates a table and then adds a new master.
// For a system catalog with little data, the new master can be caught up from WAL and
// promoted to a VOTER without requiring tablet copy.
TEST_P(ParameterizedAddMasterTest, TestAddMasterCatchupFromWAL) {
SKIP_IF_SLOW_NOT_ALLOWED();
NO_FATALS(StartCluster());
// Verify that masters are running as VOTERs and collect their addresses to be used
// for starting the new master.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
ASSERT_OK(CreateTable(cluster_.get(), kTableName));
// Add new master to the cluster.
{
string err;
EnvVars env_vars;
ASSERT_OK(BuildMasterOpts(orig_num_masters_, reserved_hp_, &new_master_opts_, &env_vars));
ASSERT_OK(AddMasterToClusterUsingCLITool(new_master_opts_, &err, std::move(env_vars),
4 /* wait_secs */, tools::GetKuduToolAbsolutePath()));
ASSERT_STR_CONTAINS(err, Substitute("Master $0 successfully caught up from WAL.",
new_master_opts_.rpc_bind_address.ToString()));
}
// Adding the same master again should print a message but not throw an error.
{
string err;
ExternalDaemonOptions opts;
EnvVars env_vars;
ASSERT_OK(BuildMasterOpts(orig_num_masters_, reserved_hp_, &opts, &env_vars));
ASSERT_OK(AddMasterToClusterUsingCLITool(opts, &err, std::move(env_vars)));
ASSERT_STR_CONTAINS(err, "Master already present");
}
// Adding one of the running former masters will throw an error.
{
string err;
const auto& hp = master_hps[0];
ExternalDaemonOptions opts;
EnvVars env_vars;
ASSERT_OK(BuildMasterOpts(0, hp, &opts, &env_vars));
Status s = AddMasterToClusterUsingCLITool(opts, &err, std::move(env_vars));
ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
ASSERT_STR_CONTAINS(err, Substitute("Master $0 already running", hp.ToString()));
}
master_hps.emplace_back(reserved_hp_);
NO_FATALS(VerifyClusterAfterMasterAddition(master_hps, orig_num_masters_ + 1));
}
// This test goes through the workflow required to copy system catalog to the newly added master.
TEST_P(ParameterizedAddMasterTest, TestAddMasterSysCatalogCopy) {
SKIP_IF_SLOW_NOT_ALLOWED();
vector<HostPort> master_hps;
NO_FATALS(StartClusterWithSysCatalogGCed(&master_hps));
ASSERT_OK(CreateTable(cluster_.get(), kTableName));
// Add new master to the cluster.
{
string err;
EnvVars env_vars;
ASSERT_OK(BuildMasterOpts(orig_num_masters_, reserved_hp_, &new_master_opts_, &env_vars));
ASSERT_OK(AddMasterToClusterUsingCLITool(new_master_opts_, &err, std::move(env_vars),
4 /* wait_secs */));
ASSERT_STR_CONTAINS(err, Substitute("Master $0 could not be caught up from WAL.",
reserved_hp_.ToString()));
ASSERT_STR_CONTAINS(err, "Successfully copied system catalog and new master is healthy");
}
// Adding the same master again should print a message but not throw an error.
{
string err;
ExternalDaemonOptions opts;
EnvVars env_vars;
ASSERT_OK(BuildMasterOpts(orig_num_masters_, reserved_hp_, &opts, &env_vars));
ASSERT_OK(AddMasterToClusterUsingCLITool(opts, &err, std::move(env_vars)));
ASSERT_STR_CONTAINS(err, "Master already present");
}
// Adding one of the running former masters will throw an error.
{
string err;
const auto& hp = master_hps[0];
ExternalDaemonOptions opts;
EnvVars env_vars;
ASSERT_OK(BuildMasterOpts(0, hp, &opts, &env_vars));
Status s = AddMasterToClusterUsingCLITool(opts, &err, std::move(env_vars));
ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
ASSERT_STR_CONTAINS(err, Substitute("Master $0 already running", hp.ToString()));
}
master_hps.emplace_back(reserved_hp_);
NO_FATALS(VerifyClusterAfterMasterAddition(master_hps, orig_num_masters_ + 1));
}
class ParameterizedRemoveMasterTest : public DynamicMultiMasterTest,
public ::testing::WithParamInterface<tuple<int, bool>> {
public:
void SetUp() override {
NO_FATALS(SetUpWithNumMasters(std::get<0>(GetParam())));
}
};
INSTANTIATE_TEST_SUITE_P(, ParameterizedRemoveMasterTest,
::testing::Combine(
// Initial number of masters in the cluster before removing a master
::testing::Values(2, 3),
// Whether the master to be removed is dead/shutdown
::testing::Bool()));
// Tests removing a non-leader master from the cluster.
TEST_P(ParameterizedRemoveMasterTest, TestRemoveMaster) {
NO_FATALS(StartCluster({// Keeping RPC timeouts short to quickly detect downed servers.
// This will put the health status into an UNKNOWN state until the point
// where they are considered FAILED.
"--consensus_rpc_timeout_ms=2000",
"--follower_unavailable_considered_failed_sec=4"}));
// Verify that existing masters are running as VOTERs.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
// When an ExternalMiniCluster is restarted after removal of a master then one of the
// remaining masters can get reassigned to the same wal dir which was previously assigned
// to the removed master. This causes problems during verification, so we always try to
// remove the last master in terms of index for test purposes.
int leader_master_idx = -1;
ASSERT_OK(cluster_->GetLeaderMasterIndex(&leader_master_idx));
ASSERT_NE(leader_master_idx, -1);
const int non_leader_master_idx = orig_num_masters_ - 1;
if (leader_master_idx == non_leader_master_idx) {
// Move the leader to the first master index
auto first_master_uuid = cluster_->master(0)->uuid();
LOG(INFO) << "Transferring leadership to master: " << first_master_uuid;
NO_FATALS(TransferMasterLeadership(cluster_.get(), first_master_uuid));
}
ASSERT_OK(cluster_->GetLeaderMasterIndex(&leader_master_idx));
ASSERT_NE(leader_master_idx, non_leader_master_idx);
const auto master_to_remove = cluster_->master(non_leader_master_idx)->bound_rpc_hostport();
const auto master_to_remove_uuid = cluster_->master(non_leader_master_idx)->uuid();
// A NO_OP operation is issued after assuming leadership so that ChangeConfig operation
// can be issued against the new leader in the current term.
// Don't know of a good way to wait/verify that the NO_OP operation has completed. Table
// creation helps with a new operation in the current term and is used later for verification.
// Hence creating a table after possible master leadership transfer and before initiating remove
// master ChangeConfig request.
ASSERT_OK(CreateTable(cluster_.get(), kTableName));
bool shutdown = std::get<1>(GetParam());
if (shutdown) {
LOG(INFO) << "Shutting down the master to be removed";
cluster_->master(non_leader_master_idx)->Shutdown();
LOG(INFO) << "Detecting transition to terminal FAILED state";
ASSERT_EVENTUALLY([&] {
VerifyDeadMasterInSpecifiedState(master_to_remove_uuid, HealthReportPB::FAILED);
});
}
// Verify the master to be removed is part of the list of masters.
ASSERT_NE(std::find(master_hps.begin(), master_hps.end(), master_to_remove), master_hps.end());
ASSERT_OK(RemoveMasterFromClusterUsingCLITool(master_to_remove, nullptr, master_to_remove_uuid));
// Verify we have one master less and the desired master was removed.
vector<HostPort> updated_master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_ - 1, &updated_master_hps));
UnorderedHostPortSet expected_master_hps(master_hps.begin(), master_hps.end());
expected_master_hps.erase(master_to_remove);
UnorderedHostPortSet actual_master_hps(updated_master_hps.begin(), updated_master_hps.end());
ASSERT_EQ(expected_master_hps, actual_master_hps);
ClusterVerifier cv(cluster_.get());
NO_FATALS(cv.CheckCluster());
NO_FATALS(cv.CheckRowCount(kTableName, ClusterVerifier::EXACTLY, 0));
// Removing the same master again should result in an error
string err;
Status s = RemoveMasterFromClusterUsingCLITool(master_to_remove, &err, master_to_remove_uuid);
ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
ASSERT_STR_CONTAINS(err, Substitute("Master $0 not found", master_to_remove.ToString()));
// Attempt transferring leadership to the removed master
LOG(INFO) << "Transferring leadership to master: " << master_to_remove_uuid;
s = TransferMasterLeadershipAsync(cluster_.get(), master_to_remove_uuid);
ASSERT_TRUE(s.IsInvalidArgument()) << s.ToString();
ASSERT_STR_CONTAINS(s.ToString(),
Substitute("tablet server $0 is not a voter in the active config",
master_to_remove_uuid));
LOG(INFO) << "Shutting down the old cluster";
cluster_.reset();
LOG(INFO) << "Bringing up the migrated cluster";
opts_.num_masters = orig_num_masters_ - 1;
opts_.master_rpc_addresses = updated_master_hps;
ExternalMiniCluster migrated_cluster(opts_);
ASSERT_OK(migrated_cluster.Start());
for (int i = 0; i < migrated_cluster.num_masters(); i++) {
ASSERT_OK(migrated_cluster.master(i)->WaitForCatalogManager());
}
vector<HostPort> migrated_master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_ - 1, &migrated_master_hps,
&migrated_cluster));
actual_master_hps.clear();
actual_master_hps.insert(migrated_master_hps.begin(), migrated_master_hps.end());
ASSERT_EQ(expected_master_hps, actual_master_hps);
ClusterVerifier mcv(&migrated_cluster);
NO_FATALS(mcv.CheckCluster());
NO_FATALS(mcv.CheckRowCount(kTableName, ClusterVerifier::EXACTLY, 0));
}
class ParameterizedRecoverMasterTest : public DynamicMultiMasterTest,
public ::testing::WithParamInterface<int> {
public:
void SetUp() override {
NO_FATALS(SetUpWithNumMasters(GetParam()));
}
};
INSTANTIATE_TEST_SUITE_P(, ParameterizedRecoverMasterTest,
// Number of masters in a cluster
::testing::Values(2, 3));
// Tests recovering a dead master at the same HostPort without explicit system catalog copy
TEST_P(ParameterizedRecoverMasterTest, TestRecoverDeadMasterCatchupfromWAL) {
SKIP_IF_SLOW_NOT_ALLOWED();
NO_FATALS(StartCluster({// Keeping RPC timeouts short to quickly detect downed servers.
// This will put the health status into an UNKNOWN state until the point
// where they are considered FAILED.
"--consensus_rpc_timeout_ms=2000",
"--follower_unavailable_considered_failed_sec=4"}));
// Verify that existing masters are running as VOTERs.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
ASSERT_OK(CreateTable(cluster_.get(), kTableName));
int master_to_recover_idx = -1;
HostPort master_to_recover_hp;
HostPort src_master_hp;
NO_FATALS(FailAndRemoveFollowerMaster(master_hps, &master_to_recover_idx, &master_to_recover_hp,
&src_master_hp));
NO_FATALS(VerifyVoterMasters(orig_num_masters_ - 1));
// Add new master at the same HostPort
{
string err;
ASSERT_OK(BuildMasterOpts(master_to_recover_idx, master_to_recover_hp, &new_master_opts_));
ASSERT_OK(AddMasterToClusterUsingCLITool(new_master_opts_, &err));
ASSERT_STR_CONTAINS(err, Substitute("Master $0 successfully caught up from WAL.",
new_master_opts_.rpc_bind_address.ToString()));
}
NO_FATALS(VerifyClusterAfterMasterAddition(master_hps, orig_num_masters_));
}
// Tests recovering a dead master at the same HostPort with explicit system catalog copy
TEST_P(ParameterizedRecoverMasterTest, TestRecoverDeadMasterSysCatalogCopy) {
SKIP_IF_SLOW_NOT_ALLOWED();
vector<HostPort> master_hps;
NO_FATALS(StartClusterWithSysCatalogGCed(
&master_hps,
// Keeping RPC timeouts short to quickly detect downed servers.
// This will put the health status into an UNKNOWN state until the point
// where they are considered FAILED.
{"--consensus_rpc_timeout_ms=2000",
"--follower_unavailable_considered_failed_sec=4"}));
// Verify that existing masters are running as VOTERs.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
ASSERT_OK(CreateTable(cluster_.get(), kTableName));
int master_to_recover_idx = -1;
HostPort master_to_recover_hp;
HostPort src_master_hp;
NO_FATALS(FailAndRemoveFollowerMaster(master_hps, &master_to_recover_idx, &master_to_recover_hp,
&src_master_hp));
NO_FATALS(VerifyVoterMasters(orig_num_masters_ - 1));
// Add new master at the same HostPort
string err;
ASSERT_OK(BuildMasterOpts(master_to_recover_idx, master_to_recover_hp, &new_master_opts_));
ASSERT_OK(AddMasterToClusterUsingCLITool(new_master_opts_, &err));
ASSERT_STR_CONTAINS(err, Substitute("Master $0 could not be caught up from WAL.",
master_to_recover_hp.ToString()));
ASSERT_STR_CONTAINS(err, "Successfully copied system catalog and new master is healthy");
NO_FATALS(VerifyClusterAfterMasterAddition(master_hps, orig_num_masters_));
}
// Test that brings up a single master cluster with 'last_known_addr' not populated by
// not specifying '--master_addresses' and then attempts to add a new master which is
// expected to fail due to invalid Raft config.
TEST_F(DynamicMultiMasterTest, TestAddMasterWithNoLastKnownAddr) {
NO_FATALS(SetUpWithNumMasters(1));
NO_FATALS(StartCluster({}, false/* supply_single_master_addr */));
// Verify that existing masters are running as VOTERs.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
// Add master to the cluster.
string err;
ExternalDaemonOptions opts;
ASSERT_OK(BuildMasterOpts(orig_num_masters_, reserved_hp_, &opts));
Status actual = AddMasterToClusterUsingCLITool(opts, &err);
ASSERT_TRUE(actual.IsRuntimeError()) << actual.ToString();
ASSERT_STR_MATCHES(err, "'last_known_addr' field in single master Raft configuration not set. "
"Please restart master with --master_addresses flag");
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
}
// Test that attempts to add a new master without enabling the feature flag for master Raft
// change config.
TEST_F(DynamicMultiMasterTest, TestAddMasterFeatureFlagNotSpecified) {
NO_FATALS(SetUpWithNumMasters(1));
NO_FATALS(StartCluster({ "--master_support_change_config=false" }));
// Verify that existing masters are running as VOTERs.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
// Add master to the cluster.
string err;
ExternalDaemonOptions opts;
ASSERT_OK(BuildMasterOpts(orig_num_masters_, reserved_hp_, &opts));
Status actual = AddMasterToClusterUsingCLITool(opts, &err);
ASSERT_TRUE(actual.IsRuntimeError()) << actual.ToString();
ASSERT_STR_CONTAINS(err, "cluster does not support AddMaster "
"with feature(s) DYNAMIC_MULTI_MASTER");
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
}
// Test that attempts to remove an existing master without enabling the feature flag for master
// Raft change config.
TEST_F(DynamicMultiMasterTest, TestRemoveMasterFeatureFlagNotSpecified) {
NO_FATALS(SetUpWithNumMasters(2));
NO_FATALS(StartCluster({"--master_support_change_config=false"}));
// Verify that existing masters are running as VOTERs.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
// Try removing non-leader master.
{
int non_leader_master_idx = -1;
ASSERT_OK(GetFollowerMasterIndex(&non_leader_master_idx));
auto master_to_remove = cluster_->master(non_leader_master_idx)->bound_rpc_hostport();
string err;
Status s = RemoveMasterFromClusterUsingCLITool(master_to_remove, &err);
ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
ASSERT_STR_CONTAINS(err, "cluster does not support RemoveMaster "
"with feature(s) DYNAMIC_MULTI_MASTER");
}
// Try removing leader master
{
int leader_master_idx = -1;
ASSERT_OK(cluster_->GetLeaderMasterIndex(&leader_master_idx));
auto master_to_remove = cluster_->master(leader_master_idx)->bound_rpc_hostport();
string err;
Status s = RemoveMasterFromClusterUsingCLITool(master_to_remove, &err);
ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
ASSERT_STR_CONTAINS(err, "cluster does not support RemoveMaster "
"with feature(s) DYNAMIC_MULTI_MASTER");
}
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
}
// Test that attempts to request a non-leader master to add a new master.
TEST_F(DynamicMultiMasterTest, TestAddMasterToNonLeader) {
NO_FATALS(SetUpWithNumMasters(2));
NO_FATALS(StartCluster());
// Verify that existing masters are running as VOTERs and collect their addresses to be used
// for starting the new master.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
// Bring up the new master and add to the cluster.
master_hps.emplace_back(reserved_hp_);
scoped_refptr<ExternalMaster> master;
NO_FATALS(StartNewMaster(master_hps, reserved_hp_, orig_num_masters_, &master));
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
// Verify sending add master request to a non-leader master returns NOT_THE_LEADER error.
// It's possible there is a leadership change between querying for leader master and
// sending the add master request to non-leader master and hence using ASSERT_EVENTUALLY.
ASSERT_EVENTUALLY([&] {
AddMasterRequestPB req;
AddMasterResponsePB resp;
RpcController rpc;
*req.mutable_rpc_addr() = HostPortToPB(reserved_hp_);
rpc.RequireServerFeature(MasterFeatures::DYNAMIC_MULTI_MASTER);
int leader_master_idx;
ASSERT_OK(cluster_->GetLeaderMasterIndex(&leader_master_idx));
ASSERT_TRUE(leader_master_idx == 0 || leader_master_idx == 1);
int non_leader_master_idx = !leader_master_idx;
ASSERT_OK(cluster_->master_proxy(non_leader_master_idx)->AddMaster(req, &resp, &rpc));
ASSERT_TRUE(resp.has_error());
ASSERT_EQ(MasterErrorPB::NOT_THE_LEADER, resp.error().code());
});
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
}
// Test that attempts to request a non-leader master to remove a master.
TEST_F(DynamicMultiMasterTest, TestRemoveMasterToNonLeader) {
NO_FATALS(SetUpWithNumMasters(2));
NO_FATALS(StartCluster());
// Verify that existing masters are running as VOTERs and collect their addresses to be used
// for starting the new master.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
// In test below we use the master RPC directly to the non-leader master and a retry
// will have unintended consequences hence disabling master leadership transfer.
NO_FATALS(DisableMasterLeadershipTransfer());
// Verify sending remove master request to a non-leader master returns NOT_THE_LEADER error.
RemoveMasterRequestPB req;
RemoveMasterResponsePB resp;
RpcController rpc;
int leader_master_idx;
ASSERT_OK(cluster_->GetLeaderMasterIndex(&leader_master_idx));
ASSERT_TRUE(leader_master_idx == 0 || leader_master_idx == 1);
int non_leader_master_idx = !leader_master_idx;
*req.mutable_rpc_addr() = HostPortToPB(cluster_->master(leader_master_idx)->bound_rpc_hostport());
rpc.RequireServerFeature(MasterFeatures::DYNAMIC_MULTI_MASTER);
// Using the master proxy directly instead of using CLI as this test wants to test
// invoking RemoveMaster RPC to non-leader master.
ASSERT_OK(cluster_->master_proxy(non_leader_master_idx)->RemoveMaster(req, &resp, &rpc));
ASSERT_TRUE(resp.has_error());
ASSERT_EQ(MasterErrorPB::NOT_THE_LEADER, resp.error().code());
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
}
// Test that attempts to add a master with missing master address and a non-routable incorrect
// address.
TEST_F(DynamicMultiMasterTest, TestAddMasterMissingAndIncorrectAddress) {
NO_FATALS(SetUpWithNumMasters(1));
NO_FATALS(StartCluster());
// Verify that existing masters are running as VOTERs.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
// Empty HostPort
{
string err;
ExternalDaemonOptions opts;
ASSERT_OK(BuildMasterOpts(orig_num_masters_, HostPort(), &opts));
Status actual = AddMasterToClusterUsingCLITool(opts, &err);
ASSERT_TRUE(actual.IsRuntimeError()) << actual.ToString();
ASSERT_STR_CONTAINS(err, "must provide positional argument master_address");
}
// Non-routable incorrect hostname.
{
string err;
ExternalDaemonOptions opts;
ASSERT_OK(BuildMasterOpts(orig_num_masters_,
HostPort("non-existent-path.local", Master::kDefaultPort), &opts));
Status actual = AddMasterToClusterUsingCLITool(opts, &err);
ASSERT_TRUE(actual.IsRuntimeError()) << actual.ToString();
ASSERT_STR_CONTAINS(err, "unable to resolve address for non-existent-path.local");
}
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
}
// Test that attempts to remove a master with missing master address and a non-existent
// hostname.
TEST_F(DynamicMultiMasterTest, TestRemoveMasterMissingAndIncorrectHostname) {
NO_FATALS(SetUpWithNumMasters(2));
NO_FATALS(StartCluster());
// Verify that existing masters are running as VOTERs.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
// Empty HostPort.
{
string err;
Status actual = RemoveMasterFromClusterUsingCLITool(HostPort(), &err);
ASSERT_TRUE(actual.IsRuntimeError()) << actual.ToString();
ASSERT_STR_CONTAINS(err, "must provide positional argument master_address");
}
// Non-existent hostname.
{
HostPort dummy_hp = HostPort("non-existent-path.local", Master::kDefaultPort);
string err;
Status actual = RemoveMasterFromClusterUsingCLITool(dummy_hp, &err);
ASSERT_TRUE(actual.IsRuntimeError()) << actual.ToString();
ASSERT_STR_CONTAINS(err, Substitute("Master $0 not found", dummy_hp.ToString()));
}
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
}
// Test that attempts to remove a master with mismatching hostname and uuid.
TEST_F(DynamicMultiMasterTest, TestRemoveMasterMismatchHostnameAndUuid) {
NO_FATALS(SetUpWithNumMasters(2));
NO_FATALS(StartCluster());
// Verify that existing masters are running as VOTERs.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
// Master leadership transfer could result in a different error and hence disabling it.
NO_FATALS(DisableMasterLeadershipTransfer());
// Random uuid
Random rng(SeedRandom());
auto random_uuid = std::to_string(rng.Next64());
int non_leader_idx = -1;
ASSERT_OK(GetFollowerMasterIndex(&non_leader_idx));
auto master_to_remove = cluster_->master(non_leader_idx)->bound_rpc_hostport();
ASSERT_NE(random_uuid, cluster_->master(non_leader_idx)->uuid());
string err;
Status actual = RemoveMasterFromClusterUsingCLITool(master_to_remove, &err, random_uuid);
ASSERT_TRUE(actual.IsRuntimeError()) << actual.ToString();
ASSERT_STR_CONTAINS(err,
Substitute("Mismatch in UUID of the master $0 to be removed. "
"Expected: $1, supplied: $2.", master_to_remove.ToString(),
cluster_->master(non_leader_idx)->uuid(), random_uuid));
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
}
// Test that attempts to add a master with non-existent kudu executable path.
TEST_F(DynamicMultiMasterTest, TestAddMasterIncorrectKuduBinary) {
NO_FATALS(SetUpWithNumMasters(1));
NO_FATALS(StartCluster());
// Verify that existing masters are running as VOTERs.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
// Add master to the cluster.
string err;
string kudu_abs_path = "/tmp/path/to/nowhere";
ExternalDaemonOptions opts;
ASSERT_OK(BuildMasterOpts(orig_num_masters_, reserved_hp_, &opts));
Status actual = AddMasterToClusterUsingCLITool(opts, &err, {} /* env_vars */, 4, kudu_abs_path);
ASSERT_TRUE(actual.IsRuntimeError()) << actual.ToString();
ASSERT_STR_CONTAINS(err, Substitute("kudu binary not found at $0", kudu_abs_path));
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_));
}
// Test that attempts removing a leader master itself from a cluster with
// 1 or 2 masters.
class ParameterizedRemoveLeaderMasterTest : public DynamicMultiMasterTest,
public ::testing::WithParamInterface<int> {
public:
void SetUp() override {
NO_FATALS(SetUpWithNumMasters(GetParam()));
}
};
INSTANTIATE_TEST_SUITE_P(, ParameterizedRemoveLeaderMasterTest,
::testing::Values(1, 2));
TEST_P(ParameterizedRemoveLeaderMasterTest, TestRemoveLeaderMaster) {
NO_FATALS(StartCluster());
// Verify that existing masters are running as VOTERs and collect their addresses to be used
// for starting the new master.
vector<HostPort> master_hps;
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
// In test below a retry in case of master leadership transfer will have unintended
// consequences and hence disabling master leadership transfer.
NO_FATALS(DisableMasterLeadershipTransfer());
int leader_master_idx;
ASSERT_OK(cluster_->GetLeaderMasterIndex(&leader_master_idx));
const auto master_to_remove = cluster_->master(leader_master_idx)->bound_rpc_hostport();
string err;
Status s = RemoveMasterFromClusterUsingCLITool(master_to_remove, &err);
ASSERT_TRUE(s.IsRuntimeError()) << s.ToString();
if (orig_num_masters_ == 1) {
ASSERT_STR_CONTAINS(err, Substitute("Can't remove master $0 in a single master "
"configuration", master_to_remove.ToString()));
} else {
ASSERT_GT(orig_num_masters_, 1);
ASSERT_STR_CONTAINS(err, Substitute("Can't remove the leader master $0",
master_to_remove.ToString()));
}
// Verify no change in number of masters.
NO_FATALS(VerifyVoterMasters(orig_num_masters_, &master_hps));
}
struct MultiMasterClusterArgs {
int orig_num_masters;
bool is_secure;
};
class AutoAddMasterTest : public KuduTest {
public:
Status SetUpWithTestArgs(const MultiMasterClusterArgs& args) {
opts_.num_masters = args.orig_num_masters;
opts_.enable_kerberos = args.is_secure;
args_ = args;
cluster_.reset(new ExternalMiniCluster(opts_));
RETURN_NOT_OK(cluster_->Start());
return cluster_->WaitForTabletServerCount(cluster_->num_tablet_servers(),
MonoDelta::FromSeconds(10));
}
void SetUp() override {
ASSERT_OK(SetUpWithTestArgs({ /*orig_num_masters*/2, /*is_secure*/false }));
TestWorkload w(cluster_.get());
w.set_num_replicas(1);
w.Setup();
}
protected:
MultiMasterClusterArgs args_;
ExternalMiniClusterOptions opts_;
unique_ptr<ExternalMiniCluster> cluster_;
};
constexpr const int64_t kShortRetryIntervalSecs = 1;
// Test that nothing goes wrong when starting up masters but the entire cluster
// isn't fully healthy. The auto-add checks should still run, but should be
// inconsequential if they fail because the entire cluster isn't healthy.
TEST_F(AutoAddMasterTest, TestRestartMastersWhileSomeDown) {
SKIP_IF_SLOW_NOT_ALLOWED();
// We'll start with three masters, and then restart two, leaving one down.
ASSERT_OK(cluster_->AddMaster());
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVoterMastersForCluster(cluster_->num_masters(), nullptr, cluster_.get()));
});
// Emulate one of the masters going down by only restarting two.
cluster_->Shutdown();
for (int i = 1; i < cluster_->num_masters(); i++) {
ASSERT_OK(cluster_->master(i)->Restart());
}
int table_idx = 0;
constexpr const char* kTablePrefix = "default.table";
const auto& deadline = MonoTime::Now() + MonoDelta::FromSeconds(30);
while (MonoTime::Now() > deadline) {
SleepFor(MonoDelta::FromSeconds(1));
// Nothing sinister should happen despite one master being down. The
// remaining masters should be operable and alive.
ASSERT_OK(CreateTable(cluster_.get(), Substitute("$0-$1", kTablePrefix, ++table_idx)));
for (int i = 1; i < cluster_->num_masters(); i++) {
ASSERT_TRUE(cluster_->master(i)->IsProcessAlive());
}
}
ASSERT_OK(cluster_->master(0)->Restart());
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVoterMastersForCluster(cluster_->num_masters(), nullptr, cluster_.get()));
});
}
// Test the procedure when some masters aren't reachable.
TEST_F(AutoAddMasterTest, TestSomeMastersUnreachable) {
SKIP_IF_SLOW_NOT_ALLOWED();
auto* stopped_master = cluster_->master(0);
ASSERT_OK(stopped_master->Pause());
// Adding a master to a cluster wherein a master is already down will fail.
// This is similar behavior to starting a new master while some are down
// since the new master can't resolve all peers' UUIDs. Shorten the time
// masters will wait to communicate to all peers to speed up this test.
ASSERT_OK(cluster_->AddMaster({ "--raft_get_node_instance_timeout_ms=3000" }));
auto* new_master = cluster_->master(args_.orig_num_masters);
ASSERT_EVENTUALLY([&] {
ASSERT_FALSE(new_master->IsProcessAlive());
});
ASSERT_OK(stopped_master->Resume());
// Even after restarting, we still won't be quite able to start healthily
// because our previous crashes will have left an unusable set of metadata
// (i.e. no consensus metadata).
new_master->Shutdown();
ASSERT_OK(new_master->Restart());
ASSERT_EVENTUALLY([&] {
ASSERT_FALSE(new_master->IsProcessAlive());
});
// If we blow away our new master and start anew, we should be able to
// proceed.
new_master->Shutdown();
ASSERT_OK(new_master->DeleteFromDisk());
ASSERT_OK(new_master->Restart());
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVoterMastersForCluster(cluster_->num_masters(), nullptr, cluster_.get()));
});
// Ensure that even after waiting a bit, our cluster is stable.
SleepFor(MonoDelta::FromSeconds(3));
ASSERT_OK(VerifyVoterMastersForCluster(cluster_->num_masters(), nullptr, cluster_.get()));
}
// Test if we fail to replicate the AddMaster request.
TEST_F(AutoAddMasterTest, TestFailWithoutReplicatingAddMaster) {
// Make master followers unable to accept updates, including config changes.
// We'll set this for all masters including leaders for simplicity.
for (int i = 0; i < cluster_->num_masters(); i++) {
ASSERT_OK(cluster_->SetFlag(cluster_->master(i),
"follower_reject_update_consensus_requests", "true"));
}
// Upon starting, the master will attempt to add itself, but fail to do so.
// Even after several attempts.
ASSERT_OK(cluster_->AddMaster({ Substitute("--master_auto_join_retry_interval_secs=$0",
kShortRetryIntervalSecs) }));
auto* new_master = cluster_->master(args_.orig_num_masters);
SleepFor(MonoDelta::FromSeconds(5 * kShortRetryIntervalSecs));
// The new master should still be around, but not be added as a part of the
// Raft group.
ASSERT_TRUE(new_master->IsProcessAlive());
Status s = VerifyVoterMastersForCluster(cluster_->num_masters(), nullptr, cluster_.get());
ASSERT_FALSE(s.ok());
ASSERT_STR_CONTAINS(s.ToString(), "expected 3 masters but got 2");
// Since nothing was successfully replicated, it shouldn't be a problem to
// start up again and re-add.
new_master->Shutdown();
for (int i = 0; i < cluster_->num_masters() - 1; i++) {
ASSERT_OK(cluster_->SetFlag(cluster_->master(i),
"follower_reject_update_consensus_requests", "false"));
}
ASSERT_OK(new_master->Restart());
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVoterMastersForCluster(cluster_->num_masters(), nullptr, cluster_.get()));
});
}
// Test when the new master fails to copy.
TEST_F(AutoAddMasterTest, TestFailTabletCopy) {
SKIP_IF_SLOW_NOT_ALLOWED();
ASSERT_OK(cluster_->AddMaster({ "--tablet_copy_fault_crash_during_download_wal=1" }));
auto* new_master = cluster_->master(args_.orig_num_masters);
ASSERT_EVENTUALLY([&] {
ASSERT_FALSE(new_master->IsProcessAlive());
});
// We should have been able to add the master to the Raft quorum, but been
// able to copy. Upon doing so, the new master should fail to come up.
new_master->Shutdown();
new_master->mutable_flags()->emplace_back("--tablet_copy_fault_crash_during_download_wal=0");
ASSERT_OK(new_master->Restart());
ASSERT_EVENTUALLY([&] {
ASSERT_FALSE(new_master->IsProcessAlive());
});
// Even blowing the new master away entirely will result in a new master
// being unable to join. The cluster already believes there to be a new
// master, but no live majority, so we're unable to add _another_ master.
ASSERT_OK(new_master->DeleteFromDisk());
new_master->Shutdown();
ASSERT_OK(new_master->Restart());
ASSERT_EVENTUALLY([&] {
ASSERT_FALSE(new_master->IsProcessAlive());
});
new_master->Shutdown();
// So, we first need to remove the master from the quorum, and then restart,
// at which point the new master should be able to join the cluster.
vector<string> addresses;
for (const auto& hp : cluster_->master_rpc_addrs()) {
addresses.emplace_back(hp.ToString());
}
// TODO(awong): we should really consider automating this step from the
// leader master.
ASSERT_OK(tools::RunKuduTool({ "master", "remove", JoinStrings(addresses, ","),
new_master->bound_rpc_hostport().ToString() }));
ASSERT_OK(new_master->Restart());
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVoterMastersForCluster(cluster_->num_masters(), nullptr, cluster_.get()));
});
SleepFor(MonoDelta::FromSeconds(3));
ASSERT_OK(VerifyVoterMastersForCluster(cluster_->num_masters(), nullptr, cluster_.get()));
}
TEST_F(AutoAddMasterTest, TestAddWithOnGoingDdl) {
simple_spinlock master_addrs_lock;
vector<string> master_addrs_unlocked;
for (const auto& hp : cluster_->master_rpc_addrs()) {
master_addrs_unlocked.emplace_back(hp.ToString());
}
// Start a thread that creates a client and tries to create tables.
const auto generate_client = [&] (shared_ptr<KuduClient>* c) {
vector<string> master_addrs;
{
std::lock_guard<simple_spinlock> l(master_addrs_lock);
master_addrs = master_addrs_unlocked;
}
shared_ptr<KuduClient> client;
RETURN_NOT_OK(client::KuduClientBuilder()
.master_server_addrs(master_addrs)
.Build(&client));
*c = std::move(client);
return Status::OK();
};
atomic<bool> proceed = true;
constexpr const int kNumThreads = 2;
vector<thread> threads;
threads.reserve(kNumThreads);
vector<Status> errors(kNumThreads);
for (int i = 0; i < kNumThreads; i++) {
threads.emplace_back([&, i] {
int idx = 0;
while (proceed) {
client::sp::shared_ptr<KuduClient> c;
Status s = generate_client(&c).AndThen([&] {
return CreateTableWithClient(c.get(), Substitute("default.$0_$1", i, ++idx));
});
if (!s.ok()) {
errors[i] = s;
}
SleepFor(MonoDelta::FromSeconds(1));
}
});
}
auto thread_joiner = MakeScopedCleanup([&] {
proceed = false;
for (auto& t : threads) {
t.join();
}
});
int num_masters = args_.orig_num_masters;
for (int i = 0; i < 3; i++) {
ASSERT_OK(cluster_->AddMaster());
auto* new_master = cluster_->master(args_.orig_num_masters);
ASSERT_OK(new_master->WaitForCatalogManager());
num_masters++;
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVotersOnAllMasters(num_masters, cluster_.get()));
});
{
std::lock_guard<simple_spinlock> l(master_addrs_lock);
master_addrs_unlocked.emplace_back(new_master->bound_rpc_hostport().ToString());
}
cluster_->Shutdown();
ASSERT_OK(cluster_->Restart());
ASSERT_OK(cluster_->WaitForTabletServerCount(cluster_->num_tablet_servers(),
MonoDelta::FromSeconds(5)));
}
proceed = false;
thread_joiner.cancel();
for (auto& t : threads) {
t.join();
}
for (const auto& e : errors) {
// NOTE: the table may exist if the CreateTable call is retried.
if (e.ok() || e.IsTimedOut() || e.IsAlreadyPresent()) {
continue;
}
// TODO(awong): we should relax the need for clients to have the precise
// list of masters.
if (e.IsConfigurationError()) {
ASSERT_STR_CONTAINS(e.ToString(), "cluster indicates it expects");
continue;
}
// TODO(KUDU-1358): we should probably allow clients to retry if the RF is
// within some normal-looking range.
ASSERT_TRUE(e.IsInvalidArgument()) << e.ToString();
ASSERT_STR_CONTAINS(e.ToString(), "not enough live tablet servers");
}
}
TEST_F(AutoAddMasterTest, TestAddNewMaster) {
SKIP_IF_SLOW_NOT_ALLOWED();
// Let's get the current master addresses and add a new one to them
vector<HostPort> master_addrs = cluster_->master_rpc_addrs();
unique_ptr<Socket> reserved_socket;
ASSERT_OK(MiniCluster::ReserveDaemonSocket(
MiniCluster::DaemonType::MASTER,
master_addrs.size(),
opts_.bind_mode,
&reserved_socket));
Sockaddr addr;
ASSERT_OK(reserved_socket->GetSocketAddress(&addr));
master_addrs.emplace_back(addr.host(), addr.port());
const auto& new_master_addrs_list =
HostPort::ToCommaSeparatedString(master_addrs);
cluster_->Shutdown();
// We have to remove the previous '--master_addresses' flag to ensure there
// are no duplicates, then we set the new addresses on the existing masters.
for (int i = 0; i < cluster_->num_masters(); i++) {
auto* flags = cluster_->master(i)->mutable_flags();
flags->erase(
remove_if(flags->begin(),
flags->end(),
[](const string &s) {
return HasPrefixString(s, "--master_addresses=");
}), flags->end());
}
for (int i = 0; i < cluster_->num_masters(); i++) {
cluster_->master(i)->mutable_flags()->emplace_back(
Substitute("--master_addresses=$0", new_master_addrs_list));
}
// Starting up the two existing masters with three addresses should cause no
// issues.
ASSERT_OK(cluster_->Restart());
ASSERT_OK(cluster_->master(0)->WaitForCatalogManager());
// Let's create the new master and start it to ensure it starts up okay.
scoped_refptr<ExternalMaster> peer;
auto idx = cluster_->master_rpc_addrs().size();
ASSERT_OK(cluster_->CreateMaster(master_addrs, idx, &peer));
ASSERT_OK(peer->Start());
ASSERT_OK(peer->WaitForCatalogManager());
auto expected_num_masters = ++idx;
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVotersOnAllMasters(expected_num_masters, cluster_.get()));
});
NO_FATALS(cluster_->AssertNoCrashes());
}
class ParameterizedAutoAddMasterTest : public AutoAddMasterTest,
public ::testing::WithParamInterface<tuple<int, bool>> {
public:
void SetUp() override {
ASSERT_OK(SetUpWithTestArgs({ /*orig_num_masters*/std::get<0>(GetParam()),
/*is_secure*/std::get<1>(GetParam()) }));
}
};
TEST_P(ParameterizedAutoAddMasterTest, TestBasicAddition) {
TestWorkload w(cluster_.get());
w.set_num_replicas(1);
w.Setup();
w.Start();
int num_masters = args_.orig_num_masters;
for (int i = 0; i < 3; i++) {
ASSERT_OK(cluster_->AddMaster());
auto* new_master = cluster_->master(args_.orig_num_masters);
ASSERT_OK(new_master->WaitForCatalogManager());
num_masters++;
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVoterMastersForCluster(num_masters, nullptr, cluster_.get()));
});
}
w.StopAndJoin();
ClusterVerifier cv(cluster_.get());
NO_FATALS(cv.CheckCluster());
NO_FATALS(cv.CheckRowCount(w.kDefaultTableName, ClusterVerifier::EXACTLY, w.rows_inserted()));
cluster_->Shutdown();
ASSERT_OK(cluster_->Restart());
ASSERT_EVENTUALLY([&] {
ASSERT_OK(VerifyVoterMastersForCluster(num_masters, nullptr, cluster_.get()));
});
NO_FATALS(cv.CheckCluster());
NO_FATALS(cv.CheckRowCount(w.kDefaultTableName, ClusterVerifier::EXACTLY, w.rows_inserted()));
}
INSTANTIATE_TEST_SUITE_P(,
ParameterizedAutoAddMasterTest, ::testing::Combine(
::testing::Values(1, 2),
::testing::Bool()),
[] (const ::testing::TestParamInfo<ParameterizedAutoAddMasterTest::ParamType>& info) {
return Substitute("$0_orig_masters_$1secure", std::get<0>(info.param),
std::get<1>(info.param) ? "" : "not_");
});
} // namespace master
} // namespace kudu