// 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/data_dirs.h"

#include <algorithm>
#include <cerrno>
#include <cstddef>
#include <cstdint>
#include <memory>
#include <mutex>
#include <ostream>
#include <random>
#include <set>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <utility>
#include <vector>

#include <gflags/gflags.h>
#include <glog/logging.h>
#include <google/protobuf/stubs/common.h>

#include "kudu/fs/block_manager.h"
#include "kudu/fs/dir_util.h"
#include "kudu/fs/fs.pb.h"
#include "kudu/gutil/integral_types.h"
#include "kudu/gutil/macros.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/port.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/locks.h"
#include "kudu/util/metrics.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/random.h"
#include "kudu/util/status.h"
#include "kudu/util/test_util_prod.h"
#include "kudu/util/threadpool.h"

DEFINE_int32(fs_target_data_dirs_per_tablet, 3,
             "Indicates the target number of data dirs to spread each "
             "tablet's data across. If greater than the number of data dirs "
             "available, data will be striped across those available. A "
             "value of 0 indicates striping should occur across all healthy "
             "data dirs. Using fewer data dirs per tablet means a single "
             "drive failure will be less likely to affect a given tablet.");
DEFINE_validator(fs_target_data_dirs_per_tablet,
    [](const char* /*n*/, int32_t v) { return v >= 0; });
TAG_FLAG(fs_target_data_dirs_per_tablet, advanced);
TAG_FLAG(fs_target_data_dirs_per_tablet, evolving);

DEFINE_int64(fs_data_dirs_reserved_bytes, -1,
             "Number of bytes to reserve on each data directory filesystem for "
             "non-Kudu usage. The default, which is represented by -1, is that "
             "1% of the disk space on each disk will be reserved. Any other "
             "value specified represents the number of bytes reserved and must "
             "be greater than or equal to 0. Explicit percentages to reserve "
             "are not currently supported");
DEFINE_validator(fs_data_dirs_reserved_bytes, [](const char* /*n*/, int64_t v) { return v >= -1; });
TAG_FLAG(fs_data_dirs_reserved_bytes, runtime);
TAG_FLAG(fs_data_dirs_reserved_bytes, evolving);

DEFINE_int32(fs_data_dirs_available_space_cache_seconds, 10,
             "Number of seconds we cache the available disk space in the block manager.");
DEFINE_validator(fs_data_dirs_available_space_cache_seconds,
                 [](const char* /*n*/, int32_t v) { return v >= 0; });
TAG_FLAG(fs_data_dirs_available_space_cache_seconds, advanced);
TAG_FLAG(fs_data_dirs_available_space_cache_seconds, evolving);

DEFINE_int32(fs_wal_dir_available_space_cache_seconds, 10,
             "Number of seconds we cache the available disk space the WAL directory.");
DEFINE_validator(fs_wal_dir_available_space_cache_seconds,
                 [](const char* /*n*/, int32_t v) { return v >= 0; });
TAG_FLAG(fs_wal_dir_available_space_cache_seconds, advanced);
TAG_FLAG(fs_wal_dir_available_space_cache_seconds, evolving);

DEFINE_bool(fs_lock_data_dirs, true,
            "Lock the data directories to prevent concurrent usage. "
            "Note that read-only concurrent usage is still allowed.");
TAG_FLAG(fs_lock_data_dirs, unsafe);
TAG_FLAG(fs_lock_data_dirs, evolving);

DEFINE_bool(fs_data_dirs_consider_available_space, true,
            "Whether to consider available space when selecting a data "
            "directory during tablet or data block creation.");
TAG_FLAG(fs_data_dirs_consider_available_space, runtime);
TAG_FLAG(fs_data_dirs_consider_available_space, evolving);

DEFINE_uint64(fs_max_thread_count_per_data_dir, 8,
              "Maximum work thread per data directory.");
TAG_FLAG(fs_max_thread_count_per_data_dir, advanced);

METRIC_DEFINE_gauge_uint64(server, data_dirs_failed,
                           "Data Directories Failed",
                           kudu::MetricUnit::kDataDirectories,
                           "Number of data directories whose disks are currently "
                           "in a failed state",
                           kudu::MetricLevel::kWarn);
METRIC_DEFINE_gauge_uint64(server, data_dirs_full,
                           "Data Directories Full",
                           kudu::MetricUnit::kDataDirectories,
                           "Number of data directories whose disks are currently full",
                           kudu::MetricLevel::kWarn);

DECLARE_bool(enable_data_block_fsync);
DECLARE_string(block_manager);

namespace kudu {

namespace fs {

using internal::DataDirGroup;
using std::default_random_engine;
using std::pair;
using std::set;
using std::shuffle;
using std::string;
using std::unique_ptr;
using std::unordered_map;
using std::unordered_set;
using std::vector;
using strings::Substitute;
using strings::SubstituteAndAppend;


////////////////////////////////////////////////////////////
// DataDirMetrics
////////////////////////////////////////////////////////////

#define GINIT(member, x) member = METRIC_##x.Instantiate(metric_entity, 0)
DataDirMetrics::DataDirMetrics(const scoped_refptr<MetricEntity>& metric_entity) {
  GINIT(dirs_failed, data_dirs_failed);
  GINIT(dirs_full, data_dirs_full);
}
#undef GINIT

////////////////////////////////////////////////////////////
// DataDirGroup
////////////////////////////////////////////////////////////

DataDirGroup::DataDirGroup() {}

DataDirGroup::DataDirGroup(vector<int> uuid_indices)
    : uuid_indices_(std::move(uuid_indices)) {}

Status DataDirGroup::LoadFromPB(const UuidIndexByUuidMap& uuid_idx_by_uuid,
                                const DataDirGroupPB& pb) {
  vector<int> uuid_indices;
  for (const auto& uuid : pb.uuids()) {
    int uuid_idx;
    if (!FindCopy(uuid_idx_by_uuid, uuid, &uuid_idx)) {
      return Status::NotFound(Substitute(
          "could not find data dir with uuid $0", uuid));
    }
    uuid_indices.emplace_back(uuid_idx);
  }

  uuid_indices_ = std::move(uuid_indices);
  return Status::OK();
}

Status DataDirGroup::CopyToPB(const UuidByUuidIndexMap& uuid_by_uuid_idx,
                              DataDirGroupPB* pb) const {
  DCHECK(pb);
  DataDirGroupPB group;
  for (auto uuid_idx : uuid_indices_) {
    string uuid;
    if (!FindCopy(uuid_by_uuid_idx, uuid_idx, &uuid)) {
      return Status::NotFound(Substitute(
          "could not find data dir with uuid index $0", uuid_idx));
    }
    group.mutable_uuids()->Add(std::move(uuid));
  }

  *pb = std::move(group);
  return Status::OK();
}

////////////////////////////////////////////////////////////
// DataDir
////////////////////////////////////////////////////////////

DataDir::DataDir(Env* env, DirMetrics* metrics, FsType fs_type, std::string dir,
                 std::unique_ptr<DirInstanceMetadataFile> metadata_file,
                 std::unique_ptr<ThreadPool> pool)
    : Dir(env, metrics, fs_type, std::move(dir), std::move(metadata_file), std::move(pool)) {}

std::unique_ptr<Dir> DataDirManager::CreateNewDir(
    Env* env, DirMetrics* metrics, FsType fs_type,
    std::string dir, std::unique_ptr<DirInstanceMetadataFile> metadata_file,
    std::unique_ptr<ThreadPool> pool) {
  return unique_ptr<Dir>(new DataDir(env, metrics, fs_type, std::move(dir),
                                     std::move(metadata_file), std::move(pool)));
}

int DataDir::available_space_cache_secs() const {
  return FLAGS_fs_data_dirs_available_space_cache_seconds;
}

int DataDir::reserved_bytes() const {
  return FLAGS_fs_data_dirs_reserved_bytes;
}

////////////////////////////////////////////////////////////
// DataDirManager
////////////////////////////////////////////////////////////

DataDirManagerOptions::DataDirManagerOptions()
    : DirManagerOptions(FLAGS_block_manager) {}

DataDirManager::DataDirManager(Env* env,
                               const DataDirManagerOptions& opts,
                               CanonicalizedRootsList canonicalized_data_roots)
    : DirManager(env, opts.metric_entity ?
                          unique_ptr<DirMetrics>(new DataDirMetrics(opts.metric_entity)) : nullptr,
                 FLAGS_fs_max_thread_count_per_data_dir,
                 opts, std::move(canonicalized_data_roots)) {}

Status DataDirManager::OpenExistingForTests(Env* env,
                                            vector<string> data_fs_roots,
                                            const DataDirManagerOptions& opts,
                                            unique_ptr<DataDirManager>* dd_manager) {
  CanonicalizedRootsList roots;
  for (const auto& r : data_fs_roots) {
    roots.push_back({ r, Status::OK() });
  }
  return DataDirManager::OpenExisting(env, std::move(roots), opts, dd_manager);
}

Status DataDirManager::OpenExisting(Env* env, CanonicalizedRootsList data_fs_roots,
                                    const DataDirManagerOptions& opts,
                                    unique_ptr<DataDirManager>* dd_manager) {
  unique_ptr<DataDirManager> dm;
  dm.reset(new DataDirManager(env, opts, std::move(data_fs_roots)));
  RETURN_NOT_OK(dm->Open());
  dd_manager->swap(dm);
  return Status::OK();
}

Status DataDirManager::CreateNewForTests(Env* env, vector<string> data_fs_roots,
                                         const DataDirManagerOptions& opts,
                                         unique_ptr<DataDirManager>* dd_manager) {
  CanonicalizedRootsList roots;
  for (const auto& r : data_fs_roots) {
    roots.push_back({ r, Status::OK() });
  }
  return DataDirManager::CreateNew(env, std::move(roots), opts, dd_manager);
}

Status DataDirManager::CreateNew(Env* env, CanonicalizedRootsList data_fs_roots,
                                 const DataDirManagerOptions& opts,
                                 unique_ptr<DataDirManager>* dd_manager) {
  unique_ptr<DataDirManager> dm;
  dm.reset(new DataDirManager(env, opts, std::move(data_fs_roots)));
  RETURN_NOT_OK(dm->Create());
  RETURN_NOT_OK(dm->Open());
  dd_manager->swap(dm);
  return Status::OK();
}

bool DataDirManager::sync_dirs() const {
  return FLAGS_enable_data_block_fsync;
}

bool DataDirManager::lock_dirs() const {
  return FLAGS_fs_lock_data_dirs;
}

int DataDirManager::max_dirs() const {
  return opts_.dir_type == "file" ? (1 << 16) - 1 : kint32max;
}

Status DataDirManager::PopulateDirectoryMaps(const vector<unique_ptr<Dir>>& dirs) {
  if (opts_.dir_type == "log") {
    return DirManager::PopulateDirectoryMaps(dirs);
  }
  DCHECK_EQ("file", opts_.dir_type);
  // When assigning directories for the file block manager, the UUID indexes
  // must match what exists in the instance files' list of UUIDs.
  unordered_map<string, int> uuid_to_idx;
  for (const auto& dd : dirs) {
    // Find a healthy instance file and use its set of UUIDs.
    if (dd->instance()->healthy()) {
      const auto& dir_set = dd->instance()->metadata()->dir_set();
      VLOG(1) << Substitute("using dir set $0 as reference: $1",
          dd->instance()->path(), pb_util::SecureDebugString(dir_set));
      for (int idx = 0; idx < dir_set.all_uuids_size(); idx++) {
        const string& uuid = dir_set.all_uuids(idx);
        InsertIfNotPresent(&uuid_to_idx, uuid, idx);
      }
      break;
    }
  }
  // We should have the same number of UUID assignments as directories.
  if (dirs.size() != uuid_to_idx.size()) {
    return Status::Corruption(
        Substitute("instance files contain duplicate UUIDs: $0 unique UUIDs expected, got $1",
                    dirs.size(), uuid_to_idx.size()));
  }
  // Keep track of any dirs that were not referenced in the dir set. These
  // are presumably from instance files we failed to read. We'll assign them
  // indexes of those that remain.
  vector<Dir*> unassigned_dirs;
  for (const auto& dd : dirs) {
    const auto& uuid = dd->instance()->uuid();
    int* idx = FindOrNull(uuid_to_idx, uuid);
    if (idx) {
      InsertToMaps(uuid, *idx, dd.get());
      uuid_to_idx.erase(uuid);
    } else {
      LOG(WARNING) << Substitute("instance $0 has unknown UUID $1",
                                  dd->instance()->path(), uuid);
      unassigned_dirs.emplace_back(dd.get());
    }
  }
  DCHECK_EQ(unassigned_dirs.size(), uuid_to_idx.size());
  int unassigned_dir_idx = 0;
  for (const auto& failed_uuid_and_idx : uuid_to_idx) {
    InsertToMaps(failed_uuid_and_idx.first, failed_uuid_and_idx.second,
                    unassigned_dirs[unassigned_dir_idx++]);
  }
  return Status::OK();
}

Status DataDirManager::LoadDataDirGroupFromPB(const std::string& tablet_id,
                                              const DataDirGroupPB& pb) {
  std::lock_guard<percpu_rwlock> lock(dir_group_lock_);
  DataDirGroup group_from_pb;
  RETURN_NOT_OK_PREPEND(group_from_pb.LoadFromPB(idx_by_uuid_, pb), Substitute(
      "could not load data dir group for tablet $0", tablet_id));
  DataDirGroup* other = InsertOrReturnExisting(&group_by_tablet_map_,
                                               tablet_id,
                                               group_from_pb);
  if (other != nullptr) {
    return Status::AlreadyPresent(Substitute(
        "tried to load directory group for tablet $0 but one is already registered",
        tablet_id));
  }
  for (int uuid_idx : group_from_pb.uuid_indices()) {
    InsertOrDie(&FindOrDie(tablets_by_uuid_idx_map_, uuid_idx), tablet_id);
  }
  return Status::OK();
}

Status DataDirManager::CreateDataDirGroup(const string& tablet_id,
                                          DirDistributionMode mode) {
  std::lock_guard<percpu_rwlock> write_lock(dir_group_lock_);
  if (ContainsKey(group_by_tablet_map_, tablet_id)) {
    return Status::AlreadyPresent("Tried to create directory group for tablet but one is already "
                                  "registered", tablet_id);
  }
  // Adjust the disk group size to fit within the total number of data dirs.
  int group_target_size;
  if (FLAGS_fs_target_data_dirs_per_tablet == 0) {
    group_target_size = dirs_.size();
  } else {
    group_target_size = std::min(FLAGS_fs_target_data_dirs_per_tablet,
                                 static_cast<int>(dirs_.size()));
  }
  vector<int> group_indices;
  if (mode == DirDistributionMode::ACROSS_ALL_DIRS) {
    // If using all dirs, add all regardless of directory state.
    AppendKeysFromMap(dir_by_uuid_idx_, &group_indices);
  } else {
    // Randomly select directories, giving preference to those with fewer tablets.
    if (PREDICT_FALSE(!failed_dirs_.empty())) {
      group_target_size = std::min(group_target_size,
          static_cast<int>(dirs_.size()) - static_cast<int>(failed_dirs_.size()));

      // A size of 0 would indicate no healthy disks, which should crash the server.
      DCHECK_GE(group_target_size, 0);
      if (group_target_size == 0) {
        return Status::IOError("No healthy data directories available", "", ENODEV);
      }
    }
    GetDirsForGroupUnlocked(group_target_size, &group_indices);
    if (PREDICT_FALSE(group_indices.empty())) {
      return Status::IOError("All healthy data directories are full", "", ENOSPC);
    }
    if (PREDICT_FALSE(group_indices.size() < FLAGS_fs_target_data_dirs_per_tablet)) {
      string msg = Substitute("Could only allocate $0 dirs of requested $1 for tablet "
                              "$2. $3 dirs total", group_indices.size(),
                              FLAGS_fs_target_data_dirs_per_tablet, tablet_id, dirs_.size());
      if (metrics_) {
        SubstituteAndAppend(&msg, ", $0 dirs full, $1 dirs failed",
                            metrics_->dirs_full->value(), metrics_->dirs_failed->value());
      }
      LOG(INFO) << msg;
    }
  }
  InsertOrDie(&group_by_tablet_map_, tablet_id, DataDirGroup(group_indices));
  for (int uuid_idx : group_indices) {
    InsertOrDie(&FindOrDie(tablets_by_uuid_idx_map_, uuid_idx), tablet_id);
  }
  return Status::OK();
}

Status DataDirManager::GetDirForBlock(const CreateBlockOptions& opts, Dir** dir,
                                      int* new_target_group_size) const {
  shared_lock<rw_spinlock> lock(dir_group_lock_.get_lock());
  vector<int> healthy_uuid_indices;
  const DataDirGroup* group = nullptr;
  DataDirGroup group_for_tests;
  if (PREDICT_TRUE(!opts.tablet_id.empty())) {
    // Get the data dir group for the tablet.
    group = FindOrNull(group_by_tablet_map_, opts.tablet_id);
    if (group == nullptr) {
      return Status::NotFound("Tried to get directory but no directory group "
                              "registered for tablet", opts.tablet_id);
    }
  } else {
    // This should only be reached by some tests; in cases where there is no
    // natural tablet_id, select a data dir from any of the directories.
    CHECK(IsGTest());
    vector<int> all_uuid_indices;
    AppendKeysFromMap(dir_by_uuid_idx_, &all_uuid_indices);
    group_for_tests = DataDirGroup(std::move(all_uuid_indices));
    group = &group_for_tests;
  }
  // Within a given directory group, filter out the ones that are failed.
  if (PREDICT_TRUE(failed_dirs_.empty())) {
    healthy_uuid_indices = group->uuid_indices();
  } else {
    RemoveUnhealthyDataDirsUnlocked(group->uuid_indices(), &healthy_uuid_indices);
    if (healthy_uuid_indices.empty()) {
      return Status::IOError("No healthy directories exist in tablet's "
                             "directory group", opts.tablet_id, ENODEV);
    }
  }
  // Within a given directory group, filter out the ones that are full.
  vector<Dir*> candidate_dirs;
  for (auto uuid_idx : healthy_uuid_indices) {
    Dir* candidate = FindOrDie(dir_by_uuid_idx_, uuid_idx);
    Status s = candidate->RefreshAvailableSpace(Dir::RefreshMode::EXPIRED_ONLY);
    WARN_NOT_OK(s, Substitute("failed to refresh fullness of $0", candidate->dir()));
    if (s.ok() && !candidate->is_full()) {
      candidate_dirs.emplace_back(candidate);
    }
  }

  // If all the directories in the group are full, return an ENOSPC error.
  if (PREDICT_FALSE(candidate_dirs.empty())) {
    DCHECK(group);
    size_t num_total = group->uuid_indices().size();
    size_t num_full = 0;
    for (const auto& idx : group->uuid_indices()) {
      if (FindOrDie(dir_by_uuid_idx_, idx)->is_full()) {
        num_full++;
      }
    }
    size_t num_failed = num_total - num_full;
    *new_target_group_size = num_total + 1;
    return Status::IOError(
        Substitute("No directories available in $0's directory group ($1 dirs "
                   "total, $2 failed, $3 full)",
                   opts.tablet_id, num_total, num_failed, num_full),
        "", ENOSPC);
  }
  if (candidate_dirs.size() == 1) {
    *dir = candidate_dirs[0];
    return Status::OK();
  }
  // Pick two randomly and select the one with more space.
  shuffle(candidate_dirs.begin(), candidate_dirs.end(),
          default_random_engine(rng_.Next()));
  *dir = PREDICT_TRUE(FLAGS_fs_data_dirs_consider_available_space) &&
         candidate_dirs[0]->available_bytes() > candidate_dirs[1]->available_bytes() ?
           candidate_dirs[0] : candidate_dirs[1];
  return Status::OK();
}

void DataDirManager::DeleteDataDirGroup(const std::string& tablet_id) {
  std::lock_guard<percpu_rwlock> lock(dir_group_lock_);
  DataDirGroup* group = FindOrNull(group_by_tablet_map_, tablet_id);
  if (group == nullptr) {
    return;
  }
  // Remove the tablet_id from every data dir in its group.
  for (int uuid_idx : group->uuid_indices()) {
    FindOrDie(tablets_by_uuid_idx_map_, uuid_idx).erase(tablet_id);
  }
  group_by_tablet_map_.erase(tablet_id);
}

Status DataDirManager::GetDataDirGroupPB(const string& tablet_id,
                                         DataDirGroupPB* pb) const {
  shared_lock<rw_spinlock> lock(dir_group_lock_.get_lock());
  const DataDirGroup* group = FindOrNull(group_by_tablet_map_, tablet_id);
  if (group == nullptr) {
    return Status::NotFound(Substitute(
        "could not find data dir group for tablet $0", tablet_id));
  }
  RETURN_NOT_OK(group->CopyToPB(uuid_by_idx_, pb));
  return Status::OK();
}

Status DataDirManager::GetDirAddIfNecessary(const CreateBlockOptions& opts, Dir** dir) {
  int new_target_group_size = 0;
  Status s = GetDirForBlock(opts, dir, &new_target_group_size);
  if (PREDICT_TRUE(s.ok())) {
    return Status::OK();
  }
  const string& tablet_id = opts.tablet_id;
  if (tablet_id.empty()) {
    // This should only be reached by some tests; in cases where there is no
    // natural tablet_id. Just return whatever error we got.
    CHECK(IsGTest());
    return s;
  }
  // If we failed for a reason other than lack of space in the data dir gruop,
  // return the error.
  if (!s.IsIOError() || s.posix_code() != ENOSPC) {
    return s;
  }
  // If we couldn't get a directory because the group is out of space, try
  // adding a new directory to the group.
  DCHECK_GT(new_target_group_size, 0);
  std::lock_guard<percpu_rwlock> l(dir_group_lock_);
  const DataDirGroup& group = FindOrDie(group_by_tablet_map_, tablet_id);
  // If we're already at the new target group size (e.g. because another
  // thread has added a directory), just return the newly added directory.
  if (new_target_group_size <= group.uuid_indices().size()) {
    *dir = FindOrDie(dir_by_uuid_idx_, group.uuid_indices().back());
    return Status::OK();
  }
  vector<int> group_uuid_indices = group.uuid_indices();
  GetDirsForGroupUnlocked(new_target_group_size, &group_uuid_indices);
  if (PREDICT_FALSE(group_uuid_indices.size() < new_target_group_size)) {
    // If we couldn't add to the group, return an error.
    int num_total = dirs_.size();
    int num_failed = failed_dirs_.size();
    int num_full = 0;
    for (const auto& dd : dirs_) {
      if (dd->is_full()) num_full++;
    }
    return Status::IOError(
        Substitute("No directories could be added ($0 dirs total, $1 failed, $2 full): $3",
        num_total, num_failed, num_full, s.ToString()), "", ENODEV);
  }
  // Update the groups. The tablet ID should not be associated with the data
  // dir already, and we should update the existing data dir group.
  // NOTE: it is not the responsibility of the DataDirManager to persist groups
  // to disk. On-disk representations of data dir groups (e.g. those in the
  // TabletSuperBlockPB) should be re-written upon modifying them.
  int new_uuid_idx = group_uuid_indices.back();
  InsertOrDie(&FindOrDie(tablets_by_uuid_idx_map_, new_uuid_idx), tablet_id);
  CHECK(!EmplaceOrUpdate(&group_by_tablet_map_, tablet_id, DataDirGroup(group_uuid_indices)));
  *dir = FindOrDie(dir_by_uuid_idx_, new_uuid_idx);
  LOG(INFO) << Substitute("Added $0 to $1's directory group: $2",
                          (*dir)->dir(), tablet_id, s.ToString());
  return Status::OK();
}

void DataDirManager::GetDirsForGroupUnlocked(int target_size,
                                             vector<int>* group_indices) {
  DCHECK(dir_group_lock_.is_locked());
  vector<int> candidate_indices;
  unordered_set<int> existing_group_indices(group_indices->begin(), group_indices->end());
  for (auto& e : dir_by_uuid_idx_) {
    int uuid_idx = e.first;
    DCHECK_LT(uuid_idx, dirs_.size());
    if (ContainsKey(existing_group_indices, uuid_idx) ||
        ContainsKey(failed_dirs_, uuid_idx)) {
      continue;
    }
    Dir* dd = e.second;
    Status s = dd->RefreshAvailableSpace(Dir::RefreshMode::ALWAYS);
    WARN_NOT_OK(s, Substitute("failed to refresh fullness of $0", dd->dir()));
    if (s.ok() && !dd->is_full()) {
      // TODO(awong): If a disk is unhealthy at the time of group creation, the
      // resulting group may be below targeted size. Add functionality to
      // resize groups. See KUDU-2040 for more details.
      candidate_indices.push_back(uuid_idx);
    }
  }
  while (group_indices->size() < target_size && !candidate_indices.empty()) {
    shuffle(candidate_indices.begin(), candidate_indices.end(), default_random_engine(rng_.Next()));
    if (candidate_indices.size() == 1) {
      group_indices->push_back(candidate_indices[0]);
      candidate_indices.erase(candidate_indices.begin());
    } else {
      int tablets_in_first = FindOrDie(tablets_by_uuid_idx_map_, candidate_indices[0]).size();
      int tablets_in_second = FindOrDie(tablets_by_uuid_idx_map_, candidate_indices[1]).size();
      int selected_index = 0;
      if (tablets_in_first == tablets_in_second &&
          PREDICT_TRUE(FLAGS_fs_data_dirs_consider_available_space)) {
        int64_t space_in_first = FindOrDie(dir_by_uuid_idx_,
                                           candidate_indices[0])->available_bytes();
        int64_t space_in_second = FindOrDie(dir_by_uuid_idx_,
                                            candidate_indices[1])->available_bytes();
        selected_index = space_in_first > space_in_second ? 0 : 1;
      } else {
        selected_index = tablets_in_first < tablets_in_second ? 0 : 1;
      }
      group_indices->push_back(candidate_indices[selected_index]);
      candidate_indices.erase(candidate_indices.begin() + selected_index);
    }
  }
}

Status DataDirManager::FindDataDirsByTabletId(const string& tablet_id,
                                              vector<string>* data_dirs) const {
  CHECK(data_dirs);
  DataDirGroupPB group;
  RETURN_NOT_OK(GetDataDirGroupPB(tablet_id, &group));
  vector<string> data_dirs_tmp;
  data_dirs_tmp.reserve(group.uuids_size());
  for (const auto& uuid : group.uuids()) {
    int uuid_idx;
    if (!FindUuidIndexByUuid(uuid, &uuid_idx)) {
      return Status::NotFound("unable to find index for UUID", uuid);
    }
    const auto* data_dir = FindDirByUuidIndex(uuid_idx);
    if (!data_dir) {
      return Status::NotFound(
          Substitute("unable to find data dir for UUID $0 with index $1",
                     uuid, uuid_idx));
    }
    data_dirs_tmp.emplace_back(data_dir->dir());
  }
  std::sort(data_dirs_tmp.begin(), data_dirs_tmp.end());
  *data_dirs = std::move(data_dirs_tmp);
  return Status::OK();
}

void DataDirManager::RemoveUnhealthyDataDirsUnlocked(const vector<int>& uuid_indices,
                                                     vector<int>* healthy_indices) const {
  if (PREDICT_TRUE(failed_dirs_.empty())) {
    return;
  }
  healthy_indices->clear();
  for (int uuid_idx : uuid_indices) {
    if (!ContainsKey(failed_dirs_, uuid_idx)) {
      healthy_indices->emplace_back(uuid_idx);
    }
  }
}

} // namespace fs
} // namespace kudu