blob: fd8a81f95a59c6d8fadd034408264d90c852f651 [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 "kudu/fs/dir_manager.h"
#include <errno.h>
#include <algorithm>
#include <functional>
#include <iterator>
#include <memory>
#include <ostream>
#include <set>
#include <string>
#include <unordered_set>
#include <utility>
#include <vector>
#include <glog/logging.h>
#include "kudu/fs/dir_util.h"
#include "kudu/fs/fs.pb.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/port.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/util/env.h"
#include "kudu/util/env_util.h"
#include "kudu/util/oid_generator.h"
#include "kudu/util/path_util.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/random_util.h"
#include "kudu/util/scoped_cleanup.h"
#include "kudu/util/stopwatch.h"
#include "kudu/util/threadpool.h"
using std::set;
using std::string;
using std::unique_ptr;
using std::unordered_map;
using std::unordered_set;
using std::vector;
using strings::Substitute;
namespace kudu {
namespace {
// Wrapper for env_util::DeleteTmpFilesRecursively that is suitable for parallel
// execution on a data directory's thread pool (which requires the return value
// be void).
void DeleteTmpFilesRecursively(Env* env, const string& path) {
WARN_NOT_OK(env_util::DeleteTmpFilesRecursively(env, path),
"Error while deleting temp files");
}
} // anonymous namespace
namespace fs {
Dir::Dir(Env* env,
DirMetrics* metrics,
FsType fs_type,
string dir,
unique_ptr<DirInstanceMetadataFile> metadata_file,
unique_ptr<ThreadPool> pool)
: env_(env),
metrics_(metrics),
fs_type_(fs_type),
dir_(std::move(dir)),
metadata_file_(std::move(metadata_file)),
pool_(std::move(pool)),
is_shutdown_(false),
is_full_(false),
available_bytes_(0) {
}
Dir::~Dir() {
Shutdown();
}
void Dir::Shutdown() {
if (is_shutdown_) {
return;
}
WaitOnClosures();
pool_->Shutdown();
is_shutdown_ = true;
}
void Dir::ExecClosure(const std::function<void()>& task) {
Status s = pool_->Submit(task);
if (!s.ok()) {
WARN_NOT_OK(
s, "Could not submit task to thread pool, running it synchronously");
task();
}
}
void Dir::WaitOnClosures() {
pool_->Wait();
}
Status Dir::RefreshAvailableSpace(RefreshMode mode) {
switch (mode) {
case RefreshMode::EXPIRED_ONLY: {
std::lock_guard<simple_spinlock> l(lock_);
DCHECK(last_space_check_.Initialized());
MonoTime expiry = last_space_check_ + MonoDelta::FromSeconds(
available_space_cache_secs());
if (MonoTime::Now() < expiry) {
break;
}
FALLTHROUGH_INTENDED; // Root was previously full, check again.
}
case RefreshMode::ALWAYS: {
int64_t available_bytes_new;
Status s = env_util::VerifySufficientDiskSpace(
env_, dir_, 0, reserved_bytes(), &available_bytes_new);
bool is_full_new;
if (PREDICT_FALSE(s.IsIOError() && s.posix_code() == ENOSPC)) {
LOG(WARNING) << Substitute(
"Insufficient disk space under path $0: will retry after $1 seconds: $2",
dir_, available_space_cache_secs(), s.ToString());
s = Status::OK();
is_full_new = true;
} else {
is_full_new = false;
}
RETURN_NOT_OK_PREPEND(s, "Could not refresh fullness"); // Catch other types of IOErrors, etc.
{
std::lock_guard<simple_spinlock> l(lock_);
if (metrics_ && is_full_ != is_full_new) {
metrics_->dirs_full->IncrementBy(is_full_new ? 1 : -1);
}
is_full_ = is_full_new;
last_space_check_ = MonoTime::Now();
available_bytes_ = available_bytes_new;
}
break;
}
default:
LOG(FATAL) << "Unknown check mode";
}
return Status::OK();
}
DirManagerOptions::DirManagerOptions(const string& dir_type)
: dir_type(dir_type), read_only(false),
update_instances(UpdateInstanceBehavior::UPDATE_AND_IGNORE_FAILURES) {}
vector<string> DirManager::GetRootNames(const CanonicalizedRootsList& root_list) {
vector<string> roots;
std::transform(root_list.begin(), root_list.end(), std::back_inserter(roots),
[&] (const CanonicalizedRootAndStatus& r) { return r.path; });
return roots;
}
vector<string> DirManager::GetRoots() const {
return GetRootNames(canonicalized_fs_roots_);
}
vector<string> DirManager::GetDirs() const {
return JoinPathSegmentsV(GetRoots(), dir_name());
}
DirManager::DirManager(Env* env,
unique_ptr<DirMetrics> dir_metrics,
int num_threads_per_dir,
const DirManagerOptions& opts,
CanonicalizedRootsList canonicalized_data_roots)
: env_(env),
num_threads_per_dir_(num_threads_per_dir),
opts_(opts),
canonicalized_fs_roots_(std::move(canonicalized_data_roots)),
metrics_(std::move(dir_metrics)),
rng_(GetRandomSeed32()) {
DCHECK_GT(canonicalized_fs_roots_.size(), 0);
DCHECK(opts_.update_instances == UpdateInstanceBehavior::DONT_UPDATE || !opts_.read_only);
DCHECK(!opts_.dir_type.empty());
}
DirManager::~DirManager() {
Shutdown();
}
void DirManager::WaitOnClosures() {
for (const auto& dir : dirs_) {
dir->WaitOnClosures();
}
}
void DirManager::Shutdown() {
// We may be waiting here for a while on outstanding closures.
LOG_SLOW_EXECUTION(INFO, 1000,
Substitute("waiting on $0 block manager thread pools",
dirs_.size())) {
for (const auto& dir : dirs_) {
dir->Shutdown();
}
}
}
Status DirManager::Create() {
CHECK(!opts_.read_only);
vector<string> all_uuids;
for (const auto& r : canonicalized_fs_roots_) {
RETURN_NOT_OK_PREPEND(r.status, "Could not create directory manager with disks failed");
}
vector<unique_ptr<DirInstanceMetadataFile>> loaded_instances;
bool has_existing_instances;
RETURN_NOT_OK(LoadInstances(&loaded_instances, &has_existing_instances));
if (has_existing_instances) {
return Status::AlreadyPresent("instance files already exist");
}
// If none of the instances exist, we can assume this is a new deployment and
// we should try creating some a new set of instance files.
RETURN_NOT_OK_PREPEND(CreateNewDirectoriesAndUpdateInstances(std::move(loaded_instances)),
"could not create new data directories");
return Status::OK();
}
Status DirManager::CreateNewDirectoriesAndUpdateInstances(
vector<unique_ptr<DirInstanceMetadataFile>> instances) {
CHECK(!opts_.read_only);
CHECK_NE(UpdateInstanceBehavior::DONT_UPDATE, opts_.update_instances);
vector<string> created_dirs;
vector<string> created_files;
auto deleter = MakeScopedCleanup([&]() {
// Delete files first so that the directories will be empty when deleted.
for (const auto& f : created_files) {
WARN_NOT_OK(env_->DeleteFile(f), "Could not delete file " + f);
}
// Delete directories in reverse order since parent directories will have
// been added before child directories.
for (auto it = created_dirs.rbegin(); it != created_dirs.rend(); it++) {
WARN_NOT_OK(env_->DeleteDir(*it), "Could not delete dir " + *it);
}
});
// First, de-duplicate the instance UUIDs. If we have duplicates, something's
// wrong. Maybe an operator manually duplicated some instance files.
set<string> all_uuids;
for (const auto& instance : instances) {
InsertIfNotPresent(&all_uuids, instance->uuid());
}
if (all_uuids.size() != instances.size()) {
return Status::InvalidArgument(
Substitute("instance files contain duplicate UUIDs: $0 directories provided, "
"$1 unique UUIDs found ($2)", instances.size(), all_uuids.size(),
JoinStrings(all_uuids, ", ")));
}
// Determine which instance files are healthy (and can thus be updated), and
// which don't exist. Create any that don't exist.
//
// Note: we don't bother trying to create/update the instance if the file is
// otherwise unhealthy.
vector<unique_ptr<DirInstanceMetadataFile>> healthy_instances;
for (auto& instance : instances) {
if (instance->healthy()) {
healthy_instances.emplace_back(std::move(instance));
continue;
}
if (instance->health_status().IsNotFound()) {
bool created_dir = false;
RETURN_NOT_OK(instance->Create(all_uuids, &created_dir));
if (created_dir) {
created_dirs.emplace_back(instance->dir());
}
created_files.emplace_back(instance->path());
}
}
// Go through the healthy instances and look for instances that don't have
// the full complete set of instance UUIDs.
vector<unique_ptr<DirInstanceMetadataFile>> instances_to_update;
for (auto& instance : healthy_instances) {
DCHECK(instance->healthy());
const auto& dir_set = instance->metadata()->dir_set();
set<string> instance_uuids;
for (int i = 0; i < dir_set.all_uuids_size(); i++) {
InsertIfNotPresent(&instance_uuids, dir_set.all_uuids(i));
}
// If an instance file disagrees with the expected UUIDs, rewrite it.
if (all_uuids != instance_uuids) {
instances_to_update.emplace_back(std::move(instance));
}
}
// If any of the instance files need to be updated because they didn't match
// the expected set of UUIDs, update them now.
// Note: Having a consistent set of instance files isn't a correctness
// requirement, but it can be useful for degbugging.
if (!instances_to_update.empty()) {
RETURN_NOT_OK(UpdateHealthyInstances(instances_to_update, all_uuids));
}
// Ensure newly created directories are synchronized to disk.
if (sync_dirs()) {
WARN_NOT_OK(env_util::SyncAllParentDirs(env_, created_dirs, created_files),
"could not sync newly created data directories");
}
// Success: don't delete any files.
deleter.cancel();
return Status::OK();
}
Status DirManager::UpdateHealthyInstances(
const vector<unique_ptr<DirInstanceMetadataFile>>& instances_to_update,
const set<string>& new_all_uuids) {
unordered_map<string, string> copies_to_restore;
unordered_set<string> copies_to_delete;
auto cleanup = MakeScopedCleanup([&] {
for (const auto& f : copies_to_delete) {
WARN_NOT_OK(env_->DeleteFile(f), Substitute("Could not delete file $0", f));
}
for (const auto& copy_and_original : copies_to_restore) {
const auto& copy_filename = copy_and_original.first;
const auto& original_filename = copy_and_original.second;
WARN_NOT_OK(env_->RenameFile(copy_filename, original_filename),
Substitute("Could not restore file $0 from $1", original_filename, copy_filename));
}
});
// Make a copy of every existing instance metadata file. This is done before
// performing any updates, so that if there's a failure while copying,
// there's no metadata to restore.
//
// We'll keep track of the copies so we can delete them on success, or use
// them to restore on failure.
WritableFileOptions opts;
opts.sync_on_close = true;
for (const auto& instance : instances_to_update) {
if (!instance->healthy()) {
continue;
}
const string& instance_filename = instance->path();
string copy_filename = instance_filename + kTmpInfix;
Status s = env_util::CopyFile(env_, instance_filename, copy_filename, opts);
if (PREDICT_FALSE(!s.ok())) {
s = s.CloneAndPrepend("unable to backup existing instance file");
instance->SetInstanceFailed(s);
LOG(WARNING) << s.ToString();
continue;
}
InsertOrDie(&copies_to_delete, copy_filename);
}
// Update the instance metadata files with the new set of UUIDs.
for (const auto& instance : instances_to_update) {
if (!instance->healthy()) {
continue;
}
const string& instance_filename = instance->path();
string copy_filename = instance_filename + kTmpInfix;
// Put together the PB and perform the update.
DirInstanceMetadataPB new_pb = *instance->metadata();
new_pb.mutable_dir_set()->mutable_all_uuids()->Clear();
for (const auto& uuid : new_all_uuids) {
new_pb.mutable_dir_set()->add_all_uuids(uuid);
}
// We're about to update the file; if we fail midway, we should try to
// restore them from our backups if we can.
InsertOrDie(&copies_to_restore, copy_filename, instance_filename);
CHECK_EQ(1, copies_to_delete.erase(copy_filename));
Status s = pb_util::WritePBContainerToPath(
env_, instance_filename, new_pb, pb_util::OVERWRITE,
sync_dirs() ? pb_util::SYNC : pb_util::NO_SYNC);
// We've failed to update for some reason, so restore our original file.
// Since we're renaming our copy, we don't have to delete it.
if (PREDICT_FALSE(!s.ok())) {
s = s.CloneAndPrepend("unable to update instance file");
instance->SetInstanceFailed(s);
LOG(WARNING) << Substitute("unable to overwrite existing instance file $0: $1",
instance_filename, s.ToString());
}
}
// If we are not tolerating errors (e.g. we're running the update_dirs tool)
// and we've hit an error, return now and clean up what we've changed.
if (opts_.update_instances == UpdateInstanceBehavior::UPDATE_AND_ERROR_ON_FAILURE) {
for (const auto& instance : instances_to_update) {
RETURN_NOT_OK_PREPEND(instance->health_status(),
"at least one instance file failed to update");
}
}
// Success; we only need to delete our copies.
InsertKeysFromMap(copies_to_restore, &copies_to_delete);
copies_to_restore.clear();
return Status::OK();
}
Status DirManager::LoadInstances(
vector<unique_ptr<DirInstanceMetadataFile>>* instance_files,
bool* has_existing_instances) {
LockMode lock_mode;
if (!lock_dirs()) {
lock_mode = LockMode::NONE;
} else if (opts_.read_only) {
lock_mode = LockMode::OPTIONAL;
} else {
lock_mode = LockMode::MANDATORY;
}
vector<string> missing_roots_tmp;
vector<unique_ptr<DirInstanceMetadataFile>> loaded_instances;
ObjectIdGenerator gen;
for (int i = 0; i < canonicalized_fs_roots_.size(); i++) {
const auto& root = canonicalized_fs_roots_[i];
string dir = JoinPathSegments(root.path, dir_name());
string instance_filename = JoinPathSegments(dir, instance_metadata_filename());
// Initialize the instance with a backup UUID. In case the load fails, this
// will be the UUID for our instnace.
string backup_uuid = gen.Next();
unique_ptr<DirInstanceMetadataFile> instance(
new DirInstanceMetadataFile(env_, std::move(backup_uuid), opts_.dir_type,
instance_filename));
if (PREDICT_FALSE(!root.status.ok())) {
instance->SetInstanceFailed(root.status);
} else {
// This may return OK and mark 'instance' as unhealthy if the file could
// not be loaded (e.g. not found, disk errors).
RETURN_NOT_OK_PREPEND(instance->LoadFromDisk(),
Substitute("could not load $0", instance_filename));
}
// Try locking the instance.
if (instance->healthy() && lock_mode != LockMode::NONE) {
// This may return OK and mark 'instance' as unhealthy if the file could
// not be locked due to non-locking issues (e.g. disk errors).
Status s = instance->Lock();
if (!s.ok()) {
if (lock_mode == LockMode::OPTIONAL) {
LOG(WARNING) << s.ToString();
LOG(WARNING) << "Proceeding without lock";
} else {
DCHECK(LockMode::MANDATORY == lock_mode);
return s;
}
}
}
loaded_instances.emplace_back(std::move(instance));
}
int num_healthy_instances = 0;
for (const auto& instance : loaded_instances) {
if (instance->healthy()) {
num_healthy_instances++;
}
}
if (has_existing_instances) {
*has_existing_instances = num_healthy_instances > 0;
}
instance_files->swap(loaded_instances);
return Status::OK();
}
Status DirManager::PopulateDirectoryMaps(const vector<unique_ptr<Dir>>& dirs) {
// Go through our instances and assign them each a UUID index.
for (int idx = 0; idx < dirs.size(); idx++) {
Dir* dir = dirs[idx].get();
InsertToMaps(dir->instance()->uuid(), idx, dir);
}
return Status::OK();
}
void DirManager::InsertToMaps(const string& uuid, int idx, Dir* dir) {
if (!dir->instance()->healthy()) {
if (metrics_) {
metrics_->dirs_failed->IncrementBy(1);
}
InsertOrDie(&failed_dirs_, idx);
}
InsertOrDie(&uuid_by_root_, DirName(dir->dir()), uuid);
InsertOrDie(&uuid_by_idx_, idx, uuid);
InsertOrDie(&idx_by_uuid_, uuid, idx);
InsertOrDie(&dir_by_uuid_idx_, idx, dir);
InsertOrDie(&uuid_idx_by_dir_, dir, idx);
InsertOrDie(&tablets_by_uuid_idx_map_, idx, {});
}
Status DirManager::Open() {
if (canonicalized_fs_roots_.size() > max_dirs()) {
return Status::InvalidArgument(Substitute("too many directories provided $0, max is $1",
canonicalized_fs_roots_.size(), max_dirs()));
}
vector<unique_ptr<DirInstanceMetadataFile>> loaded_instances;
// Load the instance files from disk.
bool has_existing_instances;
RETURN_NOT_OK_PREPEND(LoadInstances(&loaded_instances, &has_existing_instances),
"failed to load instance files");
if (!has_existing_instances) {
return Status::NotFound(
"could not open directory manager, no healthy directories found");
}
// Note: the file block manager should not be updated because its block
// indexing algorithm depends on a fixed set of directories.
if (!opts_.read_only && opts_.dir_type != "file" &&
opts_.update_instances != UpdateInstanceBehavior::DONT_UPDATE) {
RETURN_NOT_OK_PREPEND(
CreateNewDirectoriesAndUpdateInstances(
std::move(loaded_instances)),
"could not add new directories");
RETURN_NOT_OK_PREPEND(LoadInstances(&loaded_instances, &has_existing_instances),
"failed to load instance files after updating");
if (!has_existing_instances) {
return Status::IOError(
"could not open directory manager, no healthy directories found");
}
}
// All instances are present and accounted for. Time to create the in-memory
// directory structures.
vector<unique_ptr<Dir>> dirs;
for (int i = 0; i < loaded_instances.size(); i++) {
auto& instance = loaded_instances[i];
const string dir = instance->dir();
// Figure out what filesystem the directory is on.
FsType fs_type = FsType::OTHER;
if (instance->healthy()) {
bool result = false;
Status fs_check = env_->IsOnExtFilesystem(dir, &result);
if (fs_check.ok()) {
if (result) {
fs_type = FsType::EXT;
} else {
fs_check = env_->IsOnXfsFilesystem(dir, &result);
if (fs_check.ok() && result) {
fs_type = FsType::XFS;
}
}
}
// If we hit a disk error, consider the directory failed.
if (PREDICT_FALSE(fs_check.IsDiskFailure())) {
instance->SetInstanceFailed(fs_check.CloneAndPrepend("failed to check FS type"));
} else {
RETURN_NOT_OK(fs_check);
}
}
// Create a per-dir thread pool.
unique_ptr<ThreadPool> pool;
RETURN_NOT_OK(ThreadPoolBuilder(Substitute("dir $0", i))
.set_max_threads(num_threads_per_dir_)
.set_trace_metric_prefix("dirs")
.Build(&pool));
unique_ptr<Dir> new_dir = CreateNewDir(env_, metrics_.get(), fs_type, dir, std::move(instance),
std::move(pool));
dirs.emplace_back(std::move(new_dir));
}
// Use the per-dir thread pools to delete temporary files in parallel.
for (const auto& dir : dirs) {
if (dir->instance()->healthy()) {
auto* d = dir.get();
dir->ExecClosure([this, d]() { DeleteTmpFilesRecursively(this->env_, d->dir()); });
}
}
for (const auto& dir : dirs) {
dir->WaitOnClosures();
}
RETURN_NOT_OK(PopulateDirectoryMaps(dirs));
dirs_ = std::move(dirs);
// From this point onwards, the in-memory maps are the source of truth about
// the state of each dir.
// Initialize the 'fullness' status of the directories.
for (const auto& dd : dirs_) {
int uuid_idx;
CHECK(FindUuidIndexByDir(dd.get(), &uuid_idx));
if (ContainsKey(failed_dirs_, uuid_idx)) {
continue;
}
Status refresh_status = dd->RefreshAvailableSpace(Dir::RefreshMode::ALWAYS);
if (PREDICT_FALSE(!refresh_status.ok())) {
if (refresh_status.IsDiskFailure()) {
RETURN_NOT_OK(MarkDirFailed(uuid_idx, refresh_status.ToString()));
continue;
}
return refresh_status;
}
}
return Status::OK();
}
Dir* DirManager::FindDirByUuidIndex(int uuid_idx) const {
DCHECK_LT(uuid_idx, dirs_.size());
return FindPtrOrNull(dir_by_uuid_idx_, uuid_idx);
}
bool DirManager::FindUuidIndexByDir(Dir* dir, int* uuid_idx) const {
return FindCopy(uuid_idx_by_dir_, dir, uuid_idx);
}
bool DirManager::FindUuidIndexByRoot(const string& root, int* uuid_idx) const {
string uuid;
if (FindUuidByRoot(root, &uuid)) {
return FindUuidIndexByUuid(uuid, uuid_idx);
}
return false;
}
bool DirManager::FindUuidIndexByUuid(const string& uuid, int* uuid_idx) const {
return FindCopy(idx_by_uuid_, uuid, uuid_idx);
}
bool DirManager::FindUuidByRoot(const string& root, string* uuid) const {
return FindCopy(uuid_by_root_, root, uuid);
}
set<string> DirManager::FindTabletsByDirUuidIdx(int uuid_idx) const {
DCHECK_LT(uuid_idx, dirs_.size());
shared_lock<rw_spinlock> lock(dir_group_lock_.get_lock());
const set<string>* tablet_set_ptr = FindOrNull(tablets_by_uuid_idx_map_, uuid_idx);
if (tablet_set_ptr) {
return *tablet_set_ptr;
}
return {};
}
void DirManager::MarkDirFailedByUuid(const std::string& uuid) {
int uuid_idx;
CHECK(FindUuidIndexByUuid(uuid, &uuid_idx));
WARN_NOT_OK(MarkDirFailed(uuid_idx), "Failed to handle disk failure");
}
Status DirManager::MarkDirFailed(int uuid_idx, const string& error_message) {
DCHECK_LT(uuid_idx, dirs_.size());
std::lock_guard<percpu_rwlock> lock(dir_group_lock_);
Dir* dir = FindDirByUuidIndex(uuid_idx);
DCHECK(dir);
if (InsertIfNotPresent(&failed_dirs_, uuid_idx)) {
if (failed_dirs_.size() == dirs_.size()) {
// TODO(awong): pass 'error_message' as a Status instead of an string so
// we can avoid returning this artificial status.
return Status::IOError(Substitute("All dirs have failed: ", error_message));
}
if (metrics_) {
metrics_->dirs_failed->IncrementBy(1);
}
string error_prefix = "";
if (!error_message.empty()) {
error_prefix = Substitute("$0: ", error_message);
}
LOG(ERROR) << error_prefix << Substitute("Directory $0 marked as failed", dir->dir());
}
return Status::OK();
}
bool DirManager::IsDirFailed(int uuid_idx) const {
DCHECK_LT(uuid_idx, dirs_.size());
shared_lock<rw_spinlock> lock(dir_group_lock_.get_lock());
return ContainsKey(failed_dirs_, uuid_idx);
}
bool DirManager::IsTabletInFailedDir(const string& tablet_id) const {
const set<int> failed_dirs = GetFailedDirs();
for (int failed_dir : failed_dirs) {
if (ContainsKey(FindTabletsByDirUuidIdx(failed_dir), tablet_id)) {
return true;
}
}
return false;
}
} // namespace fs
} // namespace kudu