| // 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 <csignal> |
| #include <cstdint> |
| #include <fstream> // IWYU pragma: keep |
| #include <functional> |
| #include <iostream> |
| #include <iterator> |
| #include <map> |
| #include <memory> |
| #include <optional> |
| #include <set> |
| #include <string> |
| #include <tuple> |
| #include <unordered_map> |
| #include <utility> |
| #include <vector> |
| |
| #include <gflags/gflags.h> |
| #include <glog/logging.h> |
| |
| #include "kudu/common/common.pb.h" |
| #include "kudu/common/wire_protocol.h" |
| #include "kudu/common/wire_protocol.pb.h" |
| #include "kudu/consensus/metadata.pb.h" |
| #include "kudu/gutil/map-util.h" |
| #include "kudu/gutil/strings/join.h" |
| #include "kudu/gutil/strings/split.h" |
| #include "kudu/gutil/strings/stringpiece.h" |
| #include "kudu/gutil/strings/substitute.h" |
| #include "kudu/master/master.h" |
| #include "kudu/master/master.pb.h" |
| #include "kudu/master/master.proxy.h" |
| #include "kudu/master/master_runner.h" |
| #include "kudu/master/sys_catalog.h" |
| #include "kudu/rpc/response_callback.h" |
| #include "kudu/rpc/rpc_controller.h" |
| #include "kudu/tools/ksck.h" |
| #include "kudu/tools/ksck_remote.h" |
| #include "kudu/tools/master_rebuilder.h" |
| #include "kudu/tools/tool_action.h" |
| #include "kudu/tools/tool_action_common.h" |
| #include "kudu/util/env.h" |
| #include "kudu/util/flags.h" |
| #include "kudu/util/init.h" |
| #include "kudu/util/monotime.h" |
| #include "kudu/util/net/net_util.h" |
| #include "kudu/util/scoped_cleanup.h" |
| #include "kudu/util/status.h" |
| #include "kudu/util/string_case.h" |
| #include "kudu/util/subprocess.h" |
| |
| DECLARE_bool(force); |
| DECLARE_int64(negotiation_timeout_ms); |
| DECLARE_int64(timeout_ms); |
| DECLARE_string(columns); |
| DECLARE_string(fs_wal_dir); |
| DECLARE_string(fs_data_dirs); |
| DECLARE_string(tables); |
| |
| DEFINE_string(master_uuid, "", "Permanent UUID of the master. Only needed to disambiguate in case " |
| "of multiple masters with same RPC address"); |
| |
| // For catching up system catalog of a new master from WAL, we expect 8-10 secs. |
| // However bringing up a new master can take longer due to --ntp_initial_sync_wait_secs (60 secs) |
| // hence the wait secs is set higher. |
| DEFINE_int64(wait_secs, 64, |
| "Timeout in seconds to wait while retrying operations like bringing up new master, " |
| "running ksck, waiting for the new master to be promoted as VOTER, etc. This flag " |
| "is not passed to the new master."); |
| DEFINE_string(kudu_abs_path, "", "Absolute file path of the 'kudu' executable used to bring up " |
| "new master and other workflow steps. This flag is not passed " |
| "to the new master."); |
| |
| using kudu::master::AddMasterRequestPB; |
| using kudu::master::AddMasterResponsePB; |
| using kudu::master::ListMastersRequestPB; |
| using kudu::master::ListMastersResponsePB; |
| using kudu::master::Master; |
| using kudu::master::MasterServiceProxy; |
| using kudu::master::RefreshAuthzCacheRequestPB; |
| using kudu::master::RefreshAuthzCacheResponsePB; |
| using kudu::master::RemoveMasterRequestPB; |
| using kudu::master::RemoveMasterResponsePB; |
| using kudu::consensus::RaftPeerPB; |
| using kudu::rpc::RpcController; |
| using std::cout; |
| using std::endl; |
| using std::map; |
| using std::set; |
| using std::string; |
| using std::unique_ptr; |
| using std::vector; |
| using strings::Substitute; |
| |
| namespace kudu { |
| namespace tools { |
| namespace { |
| |
| const char* const kTabletServerAddressArg = "tserver_addresses"; |
| const char* const kTabletServerAddressDesc = "Address list of Kudu tablet servers" |
| " of form 'hostname-0:port-0 hostname-1:port-1 ... hostname-n:port-n'. Port may" |
| " be omitted if the tablet server is bound to the default port."; |
| const char* const kFlagArg = "flag"; |
| const char* const kValueArg = "value"; |
| |
| Status MasterGetFlags(const RunnerContext& context) { |
| const string& address = FindOrDie(context.required_args, kMasterAddressArg); |
| return PrintServerFlags(address, Master::kDefaultPort); |
| } |
| |
| Status MasterRun(const RunnerContext& context) { |
| RETURN_NOT_OK(InitKudu()); |
| |
| // Enable redaction by default. Unlike most tools, we don't want user data |
| // printed to the console/log to be shown by default. |
| CHECK_NE("", google::SetCommandLineOptionWithMode("redact", |
| "all", google::FlagSettingMode::SET_FLAGS_DEFAULT)); |
| |
| master::SetMasterFlagDefaults(); |
| return master::RunMasterServer(); |
| } |
| |
| Status MasterSetFlag(const RunnerContext& context) { |
| const string& address = FindOrDie(context.required_args, kMasterAddressArg); |
| const string& flag = FindOrDie(context.required_args, kFlagArg); |
| const string& value = FindOrDie(context.required_args, kValueArg); |
| return SetServerFlag(address, Master::kDefaultPort, flag, value); |
| } |
| |
| Status MasterSetAllMasterFlag(const RunnerContext& context) { |
| LeaderMasterProxy proxy; |
| RETURN_NOT_OK(proxy.Init(context)); |
| |
| ListMastersRequestPB req; |
| ListMastersResponsePB resp; |
| |
| RETURN_NOT_OK((proxy.SyncRpc<ListMastersRequestPB, ListMastersResponsePB>( |
| req, &resp, "ListMasters", &MasterServiceProxy::ListMastersAsync))); |
| |
| if (resp.has_error()) { |
| return StatusFromPB(resp.error().status()); |
| } |
| |
| const auto hostport_to_string = [] (const HostPortPB& hostport) { |
| return Substitute("$0:$1", hostport.host(), hostport.port()); |
| }; |
| |
| vector<ServerEntryPB> masters; |
| std::copy_if(resp.masters().begin(), resp.masters().end(), std::back_inserter(masters), |
| [](const ServerEntryPB& master) { |
| if (master.has_error()) { |
| LOG(WARNING) << "Failed to retrieve info for master: " |
| << StatusFromPB(master.error()).ToString(); |
| return false; |
| } |
| return true; |
| }); |
| vector<string> master_addresses; |
| for (const auto& master : masters) { |
| master_addresses.push_back(JoinMapped(master.registration().rpc_addresses(), |
| hostport_to_string, ",")); |
| } |
| const string& flag = FindOrDie(context.required_args, kFlagArg); |
| const string& value = FindOrDie(context.required_args, kValueArg); |
| bool set_failed_flag = false; |
| for (const auto& addr : master_addresses) { |
| Status s = SetServerFlag(addr, Master::kDefaultPort, flag, value); |
| if (!s.ok()) { |
| set_failed_flag = true; |
| LOG(WARNING) << Substitute("Set config {$0:$1} for $2 failed, error message: $3", |
| flag, value, addr, s.ToString()); |
| } |
| } |
| if (set_failed_flag) { |
| return Status::RuntimeError("Some Masters set flag failed!"); |
| } |
| return Status::OK(); |
| } |
| |
| Status MasterStatus(const RunnerContext& context) { |
| const string& address = FindOrDie(context.required_args, kMasterAddressArg); |
| return PrintServerStatus(address, Master::kDefaultPort); |
| } |
| |
| Status MasterTimestamp(const RunnerContext& context) { |
| const string& address = FindOrDie(context.required_args, kMasterAddressArg); |
| return PrintServerTimestamp(address, Master::kDefaultPort); |
| } |
| |
| // Bring up new master at the specified HostPort 'hp' with user-supplied 'flags' using kudu |
| // executable 'kudu_abs_path'. |
| // 'master_addresses' is list of masters in existing cluster including the new |
| // master. |
| Status BringUpNewMaster(const string& kudu_abs_path, |
| vector<string> flags, |
| const HostPort& hp, |
| const vector<string>& master_addresses, |
| unique_ptr<Subprocess>* new_master_out) { |
| // Ensure the new master is not already running at the specified RPC address. |
| unique_ptr<MasterServiceProxy> proxy; |
| RETURN_NOT_OK_PREPEND(BuildProxy(hp.ToString(), Master::kDefaultPort, &proxy), |
| "Failed building proxy for new master"); |
| |
| auto is_catalog_mngr_running = [](MasterServiceProxy* master_proxy) { |
| master::GetMasterRegistrationRequestPB req; |
| master::GetMasterRegistrationResponsePB resp; |
| RpcController rpc; |
| Status s = master_proxy->GetMasterRegistration(req, &resp, &rpc); |
| return s.ok() && !resp.has_error(); |
| }; |
| |
| if (is_catalog_mngr_running(proxy.get())) { |
| return Status::IllegalState(Substitute("Master $0 already running", hp.ToString())); |
| } |
| |
| flags.emplace_back("--master_addresses=" + JoinStrings(master_addresses, ",")); |
| flags.emplace_back("--master_address_add_new_master=" + hp.ToString()); |
| // In case of an error in bringing up the master we want to ensure the last log lines |
| // are emitted to help debug the issue. Hence don't use async logging. |
| flags.emplace_back("--log_async=false"); |
| vector<string> argv = { kudu_abs_path, "master", "run" }; |
| argv.insert(argv.end(), std::make_move_iterator(flags.begin()), |
| std::make_move_iterator(flags.end())); |
| auto new_master = std::make_unique<Subprocess>(argv); |
| RETURN_NOT_OK_PREPEND(new_master->Start(), "Failed starting new master"); |
| auto stop_master = MakeScopedCleanup([&] { |
| WARN_NOT_OK(new_master->KillAndWait(SIGKILL), "Failed stopping new master"); |
| }); |
| MonoTime deadline = MonoTime::Now() + MonoDelta::FromSeconds(FLAGS_wait_secs); |
| do { |
| Status wait_status = new_master->WaitNoBlock(); |
| if (!wait_status.IsTimedOut()) { |
| return Status::RuntimeError("Failed to bring up new master"); |
| } |
| if (is_catalog_mngr_running(proxy.get())) { |
| stop_master.cancel(); |
| *new_master_out = std::move(new_master); |
| return Status::OK(); |
| } |
| SleepFor(MonoDelta::FromMilliseconds(100)); |
| } while (MonoTime::Now() < deadline); |
| |
| return Status::TimedOut("Timed out waiting for the new master to come up"); |
| } |
| |
| // Check health and consensus status of masters in ksck. |
| Status CheckMastersHealthy(const vector<string>& master_addresses) { |
| std::shared_ptr<KsckCluster> cluster; |
| RETURN_NOT_OK(RemoteKsckCluster::Build(master_addresses, &cluster)); |
| |
| // Print to an unopened ofstream to discard ksck output. |
| // See https://stackoverflow.com/questions/8243743. |
| std::ofstream null_stream; |
| Ksck ksck(cluster, &null_stream); |
| RETURN_NOT_OK(ksck.CheckMasterHealth()); |
| RETURN_NOT_OK(ksck.CheckMasterConsensus()); |
| return Status::OK(); |
| } |
| |
| // Check new master 'new_master_hp' is promoted to a VOTER and all masters are healthy. |
| // Returns the last Raft role and member_type in output parameters 'master_role' and 'master_type' |
| // respectively. |
| Status CheckMasterVoterAndHealthy(LeaderMasterProxy* proxy, |
| const vector<string>& master_addresses, |
| const HostPort& new_master_hp, |
| RaftPeerPB::Role* master_role, |
| RaftPeerPB::MemberType* master_type) { |
| *master_role = RaftPeerPB::UNKNOWN_ROLE; |
| *master_type = RaftPeerPB::UNKNOWN_MEMBER_TYPE; |
| MonoTime deadline = MonoTime::Now() + MonoDelta::FromSeconds(FLAGS_wait_secs); |
| do { |
| ListMastersRequestPB req; |
| ListMastersResponsePB resp; |
| RETURN_NOT_OK((proxy->SyncRpc<ListMastersRequestPB, ListMastersResponsePB>( |
| req, &resp, "ListMasters", &MasterServiceProxy::ListMastersAsync))); |
| if (resp.has_error()) { |
| return StatusFromPB(resp.error().status()); |
| } |
| |
| int i = 0; |
| bool new_master_found = false; |
| for (; i < resp.masters_size(); i++) { |
| const auto& master = resp.masters(i); |
| if (master.has_error()) { |
| LOG(WARNING) << "Failed to retrieve info for master: " |
| << StatusFromPB(master.error()).ToString(); |
| continue; |
| } |
| for (const auto& master_hp : master.registration().rpc_addresses()) { |
| if (new_master_hp == HostPortFromPB(master_hp)) { |
| // Found the newly added master |
| new_master_found = true; |
| break; |
| } |
| } |
| if (new_master_found) { |
| break; |
| } |
| } |
| if (!new_master_found) { |
| return Status::NotFound(Substitute("New master $0 not found. Retry adding the master", |
| new_master_hp.ToString())); |
| } |
| CHECK_LT(i, resp.masters_size()); |
| const auto& master = resp.masters(i); |
| *master_role = master.role(); |
| *master_type = master.member_type(); |
| if (*master_type == RaftPeerPB::VOTER && |
| (*master_role == RaftPeerPB::FOLLOWER || *master_role == RaftPeerPB::LEADER)) { |
| // Check the master ksck state as well. |
| if (CheckMastersHealthy(master_addresses).ok()) { |
| return Status::OK(); |
| } |
| } |
| SleepFor(MonoDelta::FromMilliseconds(100)); |
| } while (MonoTime::Now() < deadline); |
| |
| return Status::TimedOut("Timed out waiting for master to catch up from WAL"); |
| } |
| |
| // Deletes local system catalog on 'dst_master' and copies system catalog from one of the masters |
| // in 'master_addresses' which includes the dst_master using the kudu executable 'kudu_abs_path'. |
| Status CopyRemoteSystemCatalog(const string& kudu_abs_path, |
| const HostPort& dst_master, |
| const vector<string>& master_addresses) { |
| // Find source master to copy system catalog from. |
| string src_master; |
| const auto& dst_master_str = dst_master.ToString(); |
| for (const auto& addr : master_addresses) { |
| if (addr != dst_master_str) { |
| src_master = addr; |
| break; |
| } |
| } |
| if (src_master.empty()) { |
| return Status::RuntimeError("Failed to find source master to copy system catalog"); |
| } |
| |
| vector<string> delete_args = |
| { kudu_abs_path, "local_replica", "delete", master::SysCatalogTable::kSysCatalogTabletId, |
| "--fs_wal_dir=" + FLAGS_fs_wal_dir, "--fs_data_dirs=" + FLAGS_fs_data_dirs, |
| "-clean_unsafe" }; |
| vector<string> copy_args = |
| { kudu_abs_path, "local_replica", "copy_from_remote", |
| master::SysCatalogTable::kSysCatalogTabletId, src_master, |
| "--fs_wal_dir=" + FLAGS_fs_wal_dir, "--fs_data_dirs=" + FLAGS_fs_data_dirs }; |
| |
| if (Env::Default()->IsEncryptionEnabled()) { |
| delete_args.emplace_back("--encrypt_data_at_rest=true"); |
| copy_args.emplace_back("--encrypt_data_at_rest=true"); |
| } |
| |
| LOG(INFO) << Substitute("Deleting system catalog on $0", dst_master_str); |
| RETURN_NOT_OK_PREPEND( |
| Subprocess::Call(delete_args), |
| "Failed to delete system catalog"); |
| LOG(INFO) << Substitute("Copying system catalog from master $0", src_master); |
| RETURN_NOT_OK_PREPEND( |
| Subprocess::Call(copy_args), |
| "Failed to copy system catalog"); |
| |
| return Status::OK(); |
| } |
| |
| // Invoke the add master Raft change config RPC to add the supplied master 'hp'. |
| // If the master is already present then Status::OK is returned and a log message is printed. |
| Status AddMasterRaftConfig(const RunnerContext& context, |
| const HostPort& hp) { |
| LeaderMasterProxy proxy; |
| RETURN_NOT_OK(proxy.Init(context)); |
| AddMasterRequestPB req; |
| AddMasterResponsePB resp; |
| *req.mutable_rpc_addr() = HostPortToPB(hp); |
| |
| Status s = proxy.SyncRpc<AddMasterRequestPB, AddMasterResponsePB>( |
| req, &resp, "AddMaster", &MasterServiceProxy::AddMasterAsync, |
| {master::MasterFeatures::DYNAMIC_MULTI_MASTER}); |
| // It's possible this is a retry request in which case instead of returning |
| // the master is already present in the Raft config error we make further checks |
| // whether the master has been promoted to a VOTER. |
| bool master_already_present = |
| resp.has_error() && resp.error().code() == master::MasterErrorPB::MASTER_ALREADY_PRESENT; |
| if (!s.ok() && !master_already_present) { |
| return s; |
| } |
| if (master_already_present) { |
| LOG(INFO) << "Master already present."; |
| } |
| LOG(INFO) << Substitute("Successfully added master $0 to the Raft configuration.", |
| hp.ToString()); |
| return Status::OK(); |
| } |
| |
| Status AddMaster(const RunnerContext& context) { |
| static const char* const kPostSuccessMsg = |
| "Please follow the next steps which includes updating master addresses, updating client " |
| "configuration etc. from the Kudu administration documentation on " |
| "\"Migrating to Multiple Kudu Masters\"."; |
| static const char* const kFailedStopMsg = |
| "Failed to stop new master after successfully adding to the cluster. Stop the master before " |
| "proceeding further."; |
| |
| // Parse input arguments. |
| vector<string> master_addresses; |
| RETURN_NOT_OK(ParseMasterAddresses(context, &master_addresses)); |
| // "master_addresses" after being successfully parsed above don't contain duplicates. |
| // So "master_hps" won't contain any duplicates as well. |
| vector<HostPort> master_hps; |
| RETURN_NOT_OK(HostPort::ParseAddresses(master_addresses, Master::kDefaultPort, &master_hps)); |
| DCHECK_EQ(master_addresses.size(), master_hps.size()); |
| |
| const string& new_master_address = FindOrDie(context.required_args, kMasterAddressArg); |
| HostPort hp; |
| RETURN_NOT_OK(hp.ParseString(new_master_address, Master::kDefaultPort)); |
| |
| // Check for the 'kudu' executable if the user has supplied one. |
| if (!FLAGS_kudu_abs_path.empty() && !Env::Default()->FileExists(FLAGS_kudu_abs_path)) { |
| return Status::NotFound(Substitute("kudu binary not found at $0", FLAGS_kudu_abs_path)); |
| } |
| string kudu_abs_path = FLAGS_kudu_abs_path; |
| if (kudu_abs_path.empty()) { |
| RETURN_NOT_OK(GetKuduToolAbsolutePathSafe(&kudu_abs_path)); |
| } |
| |
| // Get the flags that'll be needed to bring up new master and for system catalog copy. |
| if (FLAGS_fs_wal_dir.empty()) { |
| return Status::InvalidArgument("Flag -fs_wal_dir not supplied"); |
| } |
| if (FLAGS_fs_data_dirs.empty()) { |
| return Status::InvalidArgument("Flag -fs_data_dirs not supplied"); |
| } |
| GFlagsMap flags_map = GetNonDefaultFlagsMap(); |
| // Remove the optional parameters for this command. |
| // Remaining optional flags need to be passed to the new master. |
| flags_map.erase("wait_secs"); |
| flags_map.erase("kudu_abs_path"); |
| |
| // Bring up the new master. |
| vector<string> new_master_flags; |
| new_master_flags.reserve(flags_map.size()); |
| for (const auto& name_flag_pair : flags_map) { |
| const auto& flag = name_flag_pair.second; |
| new_master_flags.emplace_back(Substitute("--$0=$1", flag.name, flag.current_value)); |
| } |
| new_master_flags.emplace_back("--master_auto_join_cluster=false"); |
| |
| // Bring up the new master that includes master addresses of the cluster and itself. |
| // It's possible this is a retry in which case the new master is already part of |
| // master_addresses. |
| // Using HostPort for comparing instead of strings as in case of latter, the default port |
| // may not be specified. |
| if (std::find(master_hps.begin(), master_hps.end(), hp) == master_hps.end()) { |
| master_addresses.emplace_back(hp.ToString()); |
| master_hps.push_back(hp); |
| } |
| |
| unique_ptr<Subprocess> new_master; |
| RETURN_NOT_OK_PREPEND(BringUpNewMaster(kudu_abs_path, new_master_flags, hp, master_addresses, |
| &new_master), |
| "Failed bringing up new master. See Kudu Master log for details."); |
| auto stop_master = MakeScopedCleanup([&] { |
| auto* master_ptr = new_master.get(); |
| if (master_ptr != nullptr) { |
| WARN_NOT_OK(master_ptr->KillAndWait(SIGKILL), "Failed stopping new master"); |
| } |
| }); |
| // Call the add master Raft change config RPC. |
| RETURN_NOT_OK(AddMasterRaftConfig(context, hp)); |
| |
| // If the system catalog of the new master can be caught up from the WAL then the new master will |
| // transition to FAILED_UNRECOVERABLE state. However this can take some time, so we'll |
| // try for a few seconds. It's perfectly normal for the new master to be not caught up from |
| // the WAL in which case subsequent steps of system catalog tablet copy need to be carried out. |
| RaftPeerPB::Role master_role; |
| RaftPeerPB::MemberType master_type; |
| LeaderMasterProxy proxy; |
| // Since new master is now part of the Raft configuration, use updated 'master_addresses'. |
| RETURN_NOT_OK(proxy.Init( |
| master_addresses, MonoDelta::FromMilliseconds(FLAGS_timeout_ms), |
| MonoDelta::FromMilliseconds(FLAGS_negotiation_timeout_ms))); |
| |
| LOG(INFO) << "Checking for master consensus and health status..."; |
| Status wal_catchup_status = CheckMasterVoterAndHealthy(&proxy, master_addresses, hp, |
| &master_role, &master_type); |
| if (wal_catchup_status.ok()) { |
| stop_master.cancel(); |
| RETURN_NOT_OK_PREPEND(new_master->KillAndWait(SIGTERM), kFailedStopMsg); |
| LOG(INFO) << Substitute("Master $0 successfully caught up from WAL.", hp.ToString()); |
| LOG(INFO) << kPostSuccessMsg; |
| return Status::OK(); |
| } |
| if (!wal_catchup_status.IsTimedOut()) { |
| RETURN_NOT_OK_PREPEND(wal_catchup_status, |
| "Unexpected error waiting for master to catchup from WAL."); |
| return wal_catchup_status; |
| } |
| |
| // New master could not be caught up from WAL, so next we'll attempt copying |
| // system catalog. |
| LOG(INFO) << Substitute("Master $0 status; role: $1, member_type: $2", |
| hp.ToString(), RaftPeerPB::Role_Name(master_role), |
| RaftPeerPB::MemberType_Name(master_type)); |
| LOG(INFO) << Substitute("Master $0 could not be caught up from WAL.", hp.ToString()); |
| |
| RETURN_NOT_OK_PREPEND(new_master->KillAndWait(SIGTERM), |
| "Unable to stop master to proceed with system catalog copy"); |
| new_master.reset(); |
| |
| RETURN_NOT_OK(CopyRemoteSystemCatalog(kudu_abs_path, hp, master_addresses)); |
| |
| RETURN_NOT_OK_PREPEND(BringUpNewMaster(kudu_abs_path, new_master_flags, hp, master_addresses, |
| &new_master), |
| "Failed bringing up new master after system catalog copy. " |
| "See Kudu Master log for details."); |
| |
| RETURN_NOT_OK_PREPEND(CheckMasterVoterAndHealthy(&proxy, master_addresses, hp, &master_role, |
| &master_type), |
| "Failed waiting for new master to be healthy after system catalog copy"); |
| |
| stop_master.cancel(); |
| RETURN_NOT_OK_PREPEND(new_master->KillAndWait(SIGTERM), kFailedStopMsg); |
| LOG(INFO) << "Successfully copied system catalog and new master is healthy."; |
| LOG(INFO) << kPostSuccessMsg; |
| return Status::OK(); |
| } |
| |
| Status RemoveMasterChangeConfig(const RunnerContext& context) { |
| const string& master_hp_str = FindOrDie(context.required_args, kMasterAddressArg); |
| HostPort hp; |
| RETURN_NOT_OK(hp.ParseString(master_hp_str, Master::kDefaultPort)); |
| |
| LeaderMasterProxy proxy; |
| RETURN_NOT_OK(proxy.Init(context)); |
| |
| RemoveMasterRequestPB req; |
| RemoveMasterResponsePB resp; |
| *req.mutable_rpc_addr() = HostPortToPB(hp); |
| if (!FLAGS_master_uuid.empty()) { |
| *req.mutable_master_uuid() = FLAGS_master_uuid; |
| } |
| |
| RETURN_NOT_OK((proxy.SyncRpc<RemoveMasterRequestPB, RemoveMasterResponsePB>( |
| req, &resp, "RemoveMaster", &MasterServiceProxy::RemoveMasterAsync, |
| { master::MasterFeatures::DYNAMIC_MULTI_MASTER }))); |
| |
| if (resp.has_error()) { |
| return StatusFromPB(resp.error().status()); |
| } |
| |
| LOG(INFO) << Substitute("Successfully removed master $0 from the cluster. Please follow the next " |
| "steps from the Kudu administration documentation on " |
| "\"Removing Kudu Masters from a Multi-Master Deployment\" or " |
| "\"Recovering from a dead Kudu Master in a Multi-Master Deployment\" " |
| "as appropriate.", |
| hp.ToString()); |
| return Status::OK(); |
| } |
| |
| Status ListMasters(const RunnerContext& context) { |
| LeaderMasterProxy proxy; |
| RETURN_NOT_OK(proxy.Init(context)); |
| |
| ListMastersRequestPB req; |
| ListMastersResponsePB resp; |
| |
| RETURN_NOT_OK((proxy.SyncRpc<ListMastersRequestPB, ListMastersResponsePB>( |
| req, &resp, "ListMasters", &MasterServiceProxy::ListMastersAsync))); |
| |
| if (resp.has_error()) { |
| return StatusFromPB(resp.error().status()); |
| } |
| |
| DataTable table({}); |
| |
| vector<ServerEntryPB> masters; |
| std::copy_if(resp.masters().begin(), resp.masters().end(), std::back_inserter(masters), |
| [](const ServerEntryPB& master) { |
| if (master.has_error()) { |
| LOG(WARNING) << "Failed to retrieve info for master: " |
| << StatusFromPB(master.error()).ToString(); |
| return false; |
| } |
| return true; |
| }); |
| |
| auto hostport_to_string = [] (const HostPortPB& hostport) { |
| return Substitute("$0:$1", hostport.host(), hostport.port()); |
| }; |
| |
| for (const auto& column : strings::Split(FLAGS_columns, ",", strings::SkipEmpty())) { |
| vector<string> values; |
| if (iequals(column.ToString(), "uuid")) { |
| for (const auto& master : masters) { |
| values.push_back(master.instance_id().permanent_uuid()); |
| } |
| } else if (iequals(column.ToString(), "cluster_id")) { |
| for (const auto& master : masters) { |
| values.emplace_back(master.has_cluster_id() ? master.cluster_id() : ""); |
| } |
| } else if (iequals(column.ToString(), "seqno")) { |
| for (const auto& master : masters) { |
| values.push_back(std::to_string(master.instance_id().instance_seqno())); |
| } |
| } else if (iequals(column.ToString(), "rpc-addresses") || |
| iequals(column.ToString(), "rpc_addresses")) { |
| for (const auto& master : masters) { |
| values.push_back(JoinMapped(master.registration().rpc_addresses(), |
| hostport_to_string, ",")); |
| } |
| } else if (iequals(column.ToString(), "http-addresses") || |
| iequals(column.ToString(), "http_addresses")) { |
| for (const auto& master : masters) { |
| values.push_back(JoinMapped(master.registration().http_addresses(), |
| hostport_to_string, ",")); |
| } |
| } else if (iequals(column.ToString(), "version")) { |
| for (const auto& master : masters) { |
| values.push_back(master.registration().software_version()); |
| } |
| } else if (iequals(column.ToString(), "start_time")) { |
| for (const auto& master : masters) { |
| values.emplace_back(StartTimeToString(master.registration())); |
| } |
| } else if (iequals(column.ToString(), "role")) { |
| for (const auto& master : masters) { |
| values.emplace_back(RaftPeerPB::Role_Name(master.role())); |
| } |
| } else if (iequals(column.ToString(), "member_type")) { |
| for (const auto& master : masters) { |
| values.emplace_back(RaftPeerPB::MemberType_Name(master.member_type())); |
| } |
| } else { |
| return Status::InvalidArgument("unknown column (--columns)", column); |
| } |
| table.AddColumn(column.ToString(), std::move(values)); |
| } |
| |
| RETURN_NOT_OK(table.PrintTo(cout)); |
| return Status::OK(); |
| } |
| |
| Status MasterDumpMemTrackers(const RunnerContext& context) { |
| const auto& address = FindOrDie(context.required_args, kMasterAddressArg); |
| return DumpMemTrackers(address, Master::kDefaultPort); |
| } |
| |
| Status PrintRebuildReport(const RebuildReport& rebuild_report) { |
| cout << "Rebuild Report" << endl; |
| cout << "Tablet Servers" << endl; |
| DataTable tserver_table({ "address", "status" }); |
| int bad_tservers = 0; |
| for (const auto& pair : rebuild_report.tservers) { |
| const Status& s = pair.second; |
| tserver_table.AddRow({ pair.first, s.ToString() }); |
| if (!s.ok()) bad_tservers++; |
| } |
| RETURN_NOT_OK(tserver_table.PrintTo(cout)); |
| cout << Substitute("Rebuilt from $0 tablet servers, of which $1 had errors", |
| rebuild_report.tservers.size(), bad_tservers) |
| << endl << endl; |
| |
| cout << "Replicas" << endl; |
| DataTable replica_table({ "table", "tablet", "tablet server", "status" }); |
| int bad_replicas = 0; |
| for (const auto& entry : rebuild_report.replicas) { |
| const auto& key = entry.first; |
| const Status& s = entry.second; |
| replica_table.AddRow({ std::get<0>(key), std::get<1>(key), std::get<2>(key), s.ToString() }); |
| if (!s.ok()) bad_replicas++; |
| } |
| RETURN_NOT_OK(replica_table.PrintTo(cout)); |
| cout << Substitute("Rebuilt from $0 replicas, of which $1 had errors", |
| rebuild_report.replicas.size(), bad_replicas) |
| << endl; |
| |
| return Status::OK(); |
| } |
| |
| Status RefreshAuthzCacheAtMaster(const string& master_address) { |
| unique_ptr<MasterServiceProxy> proxy; |
| RETURN_NOT_OK(BuildProxy(master_address, Master::kDefaultPort, &proxy)); |
| |
| RpcController ctl; |
| ctl.set_timeout(MonoDelta::FromMilliseconds(FLAGS_timeout_ms)); |
| |
| RefreshAuthzCacheRequestPB req; |
| RefreshAuthzCacheResponsePB resp; |
| RETURN_NOT_OK(proxy->RefreshAuthzCache(req, &resp, &ctl)); |
| if (resp.has_error()) { |
| return StatusFromPB(resp.error().status()); |
| } |
| return Status::OK(); |
| } |
| |
| Status RefreshAuthzCache(const RunnerContext& context) { |
| vector<string> master_addresses; |
| RETURN_NOT_OK(ParseMasterAddresses(context, &master_addresses)); |
| |
| if (!FLAGS_force) { |
| // Make sure the list of master addresses specified for the command |
| // matches the actual list of masters in the cluster. |
| RETURN_NOT_OK(VerifyMasterAddressList(master_addresses)); |
| } |
| |
| // It makes sense to refresh privileges cache at every master in the cluster. |
| // Otherwise, the authorization provider might return inconsistent results for |
| // authz requests upon master leadership change. |
| vector<Status> statuses; |
| statuses.reserve(master_addresses.size()); |
| for (const auto& address : master_addresses) { |
| auto status = RefreshAuthzCacheAtMaster(address); |
| statuses.emplace_back(std::move(status)); |
| } |
| DCHECK_EQ(master_addresses.size(), statuses.size()); |
| string err_str; |
| for (auto i = 0; i < statuses.size(); ++i) { |
| const auto& s = statuses[i]; |
| if (s.ok()) { |
| continue; |
| } |
| err_str += Substitute(" error from master at $0: $1", |
| master_addresses[i], s.ToString()); |
| } |
| if (err_str.empty()) { |
| return Status::OK(); |
| } |
| return Status::Incomplete(err_str); |
| } |
| |
| Status RebuildMaster(const RunnerContext& context) { |
| MasterRebuilder master_rebuilder(context.variadic_args); |
| RETURN_NOT_OK(master_rebuilder.RebuildMaster()); |
| PrintRebuildReport(master_rebuilder.GetRebuildReport()); |
| return Status::OK(); |
| } |
| |
| } // anonymous namespace |
| |
| unique_ptr<Mode> BuildMasterMode() { |
| ModeBuilder builder("master"); |
| builder.Description("Operate on a Kudu Master"); |
| |
| { |
| unique_ptr<Action> action_refresh = |
| ClusterActionBuilder("refresh", &RefreshAuthzCache) |
| .Description("Refresh the authorization policies") |
| .AddOptionalParameter( |
| "force", std::nullopt, |
| string( |
| "Ignore mismatches of the specified and the actual lists " |
| "of master addresses in the cluster")) |
| .Build(); |
| |
| unique_ptr<Mode> mode_authz_cache = ModeBuilder("authz_cache") |
| .Description("Operate on the authz caches of the Kudu Masters") |
| .AddAction(std::move(action_refresh)) |
| .Build(); |
| builder.AddMode(std::move(mode_authz_cache)); |
| } |
| { |
| unique_ptr<Action> dump_memtrackers = |
| MasterActionBuilder("dump_memtrackers", &MasterDumpMemTrackers) |
| .Description("Dump the memtrackers from a Kudu Master") |
| .AddOptionalParameter("format") |
| .AddOptionalParameter("memtracker_output") |
| .Build(); |
| builder.AddAction(std::move(dump_memtrackers)); |
| } |
| { |
| unique_ptr<Action> get_flags = |
| MasterActionBuilder("get_flags", &MasterGetFlags) |
| .Description("Get the gflags for a Kudu Master") |
| .AddOptionalParameter("all_flags") |
| .AddOptionalParameter("flags") |
| .AddOptionalParameter("flag_tags") |
| .Build(); |
| builder.AddAction(std::move(get_flags)); |
| } |
| { |
| unique_ptr<Action> run = |
| ActionBuilder("run", &MasterRun) |
| .ProgramName("kudu-master") |
| .Description("Run a Kudu Master") |
| .ExtraDescription("Note: The master server is started in this process and " |
| "runs until interrupted.\n\n" |
| "The most common configuration flags are described below. " |
| "For all the configuration options pass --helpfull or see " |
| "https://kudu.apache.org/docs/configuration_reference.html" |
| "#kudu-master_supported") |
| .AddOptionalParameter("master_addresses") |
| // Even though fs_wal_dir is required, we don't want it to be positional argument. |
| // This allows it to be passed as a standard flag. |
| .AddOptionalParameter("fs_wal_dir") |
| .AddOptionalParameter("fs_data_dirs") |
| .AddOptionalParameter("fs_metadata_dir") |
| .AddOptionalParameter("log_dir") |
| // Unlike most tools we don't log to stderr by default to match the |
| // kudu-master binary as closely as possible. |
| .AddOptionalParameter("logtostderr", string("false")) |
| .Build(); |
| builder.AddAction(std::move(run)); |
| } |
| { |
| unique_ptr<Action> set_flag = |
| MasterActionBuilder("set_flag", &MasterSetFlag) |
| .Description("Change a gflag value on a Kudu Master") |
| .AddRequiredParameter({ kFlagArg, "Name of the gflag" }) |
| .AddRequiredParameter({ kValueArg, "New value for the gflag" }) |
| .AddOptionalParameter("force") |
| .Build(); |
| builder.AddAction(std::move(set_flag)); |
| } |
| { |
| unique_ptr<Action> set_flag_for_all = |
| ClusterActionBuilder("set_flag_for_all", &MasterSetAllMasterFlag) |
| .Description("Change a gflag value for all Kudu Masters in the cluster") |
| .AddRequiredParameter({ kFlagArg, "Name of the gflag" }) |
| .AddRequiredParameter({ kValueArg, "New value for the gflag" }) |
| .AddOptionalParameter("force") |
| .Build(); |
| builder.AddAction(std::move(set_flag_for_all)); |
| } |
| { |
| unique_ptr<Action> status = |
| MasterActionBuilder("status", &MasterStatus) |
| .Description("Get the status of a Kudu Master") |
| .Build(); |
| builder.AddAction(std::move(status)); |
| } |
| { |
| unique_ptr<Action> timestamp = |
| MasterActionBuilder("timestamp", &MasterTimestamp) |
| .Description("Get the current timestamp of a Kudu Master") |
| .Build(); |
| builder.AddAction(std::move(timestamp)); |
| } |
| { |
| unique_ptr<Action> list_masters = |
| ClusterActionBuilder("list", &ListMasters) |
| .Description("List masters in a Kudu cluster") |
| .AddOptionalParameter( |
| "columns", |
| string("uuid,rpc-addresses,role"), |
| string("Comma-separated list of master info fields to " |
| "include in output.\nPossible values: uuid, cluster_id, " |
| "rpc-addresses, http-addresses, version, seqno, " |
| "start_time, role and member_type")) |
| .AddOptionalParameter("format") |
| .Build(); |
| builder.AddAction(std::move(list_masters)); |
| } |
| { |
| unique_ptr<Action> add_master = |
| ActionBuilder("add", &AddMaster) |
| .Description("Add a master to the Kudu cluster") |
| .ExtraDescription( |
| "This is an advanced command that orchestrates the workflow to bring up and add a " |
| "new master to the Kudu cluster. It must be run locally on the new master being added " |
| "and not on the leader master. This tool shuts down the new master after completion " |
| "of the command regardless of whether the new master addition is successful. " |
| "After the command completes successfully, user is expected to explicitly " |
| "start the new master using the same flags as supplied to this tool.\n\n" |
| "Supply all the necessary flags to bring up the new master. " |
| "The most common configuration flags used to bring up a master are described " |
| "below. See \"Kudu Configuration Reference\" for all the configuration options that " |
| "apply to a Kudu master.\n\n" |
| "Please refer to the Kudu administration documentation on " |
| "\"Migrating to Multiple Kudu Masters\" for the complete steps.") |
| .AddRequiredParameter({ kMasterAddressesArg, kMasterAddressesArgDesc }) |
| .AddRequiredParameter({ kMasterAddressArg, kMasterAddressDesc }) |
| .AddOptionalParameter("wait_secs") |
| .AddOptionalParameter("kudu_abs_path") |
| .AddOptionalParameter("fs_wal_dir") |
| .AddOptionalParameter("fs_data_dirs") |
| .AddOptionalParameter("fs_metadata_dir") |
| .AddOptionalParameter("log_dir") |
| // Unlike most tools we don't log to stderr by default to match the |
| // kudu-master binary as closely as possible. |
| .AddOptionalParameter("logtostderr", string("false")) |
| .Build(); |
| builder.AddAction(std::move(add_master)); |
| } |
| { |
| unique_ptr<Action> remove_master = |
| ActionBuilder("remove", &RemoveMasterChangeConfig) |
| .Description("Remove a master from the Kudu cluster") |
| .ExtraDescription( |
| "Removes a master from the Raft configuration of the Kudu cluster.\n\n" |
| "Please refer to the Kudu administration documentation on " |
| "\"Removing Kudu Masters from a Multi-Master Deployment\" or " |
| "\"Recovering from a dead Kudu Master in a Multi-Master Deployment\" as appropriate.") |
| .AddRequiredParameter({ kMasterAddressesArg, kMasterAddressesArgDesc }) |
| .AddRequiredParameter({ kMasterAddressArg, kMasterAddressDesc }) |
| .AddOptionalParameter("master_uuid") |
| .Build(); |
| builder.AddAction(std::move(remove_master)); |
| } |
| |
| { |
| const char* rebuild_extra_description = "Attempts to create or update on-disk metadata " |
| "that can be used by a non-replicated master to recover a Kudu cluster " |
| "that has permanently lost its masters. It has a number of limitations:\n\n" |
| " - Security metadata like cryptographic keys are not rebuilt. Tablet servers " |
| "and clients must be restarted before starting the new master in order to " |
| "communicate with the new master.\n\n" |
| " - Table IDs are known only by the masters. Reconstructed tables will have " |
| "new IDs.\n\n" |
| " - If a create, delete, or alter table was in progress when the masters were lost, " |
| "it may not be possible to restore the table.\n\n" |
| " - If all replicas of a tablet are missing, it may not be able to recover the " |
| "table fully. Moreover, the rebuild tool cannot detect that a tablet is " |
| "missing.\n\n" |
| " - It's not possible to determine the replication factor of a table from tablet " |
| "server metadata. The rebuild tool sets the replication factor of each " |
| "table to --default_num_replicas instead.\n\n" |
| " - It's not possible to determine the next column id for a table from tablet " |
| "server metadata. Instead, the rebuilt tool sets the next column id to " |
| "a very large number.\n\n" |
| " - Table metadata like comments, owners, and configurations are not stored on " |
| "tablet servers and are thus not restored.\n\n" |
| " - Without '--tables', the tool will build a brand new syscatalog table based on " |
| "tablet data on tablet server metadata.\n\n" |
| " - With '--tables', the tool will update specific tables in current syscatalog.\n\n" |
| "WARNING: This tool is potentially unsafe. Only use it when there is no " |
| "possibility of recovering the original masters, and you know what you " |
| "are doing.\n"; |
| unique_ptr<Action> unsafe_rebuild = |
| ActionBuilder("unsafe_rebuild", &RebuildMaster) |
| .Description("Rebuild a Kudu master from tablet server metadata") |
| .ExtraDescription(rebuild_extra_description) |
| .AddRequiredVariadicParameter({ kTabletServerAddressArg, kTabletServerAddressDesc }) |
| .AddOptionalParameter("default_num_replicas") |
| .AddOptionalParameter("default_schema_version") |
| .AddOptionalParameter("fs_data_dirs") |
| .AddOptionalParameter("fs_metadata_dir") |
| .AddOptionalParameter("fs_wal_dir") |
| .AddOptionalParameter("tables") |
| .Build(); |
| builder.AddAction(std::move(unsafe_rebuild)); |
| } |
| |
| return builder.Build(); |
| } |
| |
| } // namespace tools |
| } // namespace kudu |
| |
