Status FsManager::Init()

in src/kudu/fs/fs_manager.cc [179:299]
92 lines of code
19 McCabe index (conditional complexity)

Status FsManager::Init() {
  if (initted_) {
    return Status::OK();
  }

  // The wal root must be set.
  if (opts_.wal_root.empty()) {
    return Status::IOError("Write-ahead log directory (fs_wal_dir) not provided");
  }

  // Deduplicate all of the roots.
  unordered_set<string> all_roots = { opts_.wal_root };
  all_roots.insert(opts_.data_roots.begin(), opts_.data_roots.end());

  // If the metadata root not set, Kudu will either use the wal root or the
  // first data root, in which case we needn't canonicalize additional roots.
  if (!opts_.metadata_root.empty()) {
    all_roots.insert(opts_.metadata_root);
  }

  // Build a map of original root --> canonicalized root, sanitizing each
  // root as we go and storing the canonicalization status.
  typedef unordered_map<string, CanonicalizedRootAndStatus> RootMap;
  RootMap canonicalized_roots;
  for (const string& root : all_roots) {
    if (root.empty()) {
      return Status::IOError("Empty string provided for path");
    }
    if (root[0] != '/') {
      return Status::IOError(
          Substitute("Relative path $0 provided", root));
    }
    string root_copy = root;
    StripWhiteSpace(&root_copy);
    if (root != root_copy) {
      return Status::IOError(
          Substitute("Path $0 contains illegal whitespace", root));
    }

    // Strip the basename when canonicalizing, as it may not exist. The
    // dirname, however, must exist.
    string canonicalized;
    Status s = env_->Canonicalize(DirName(root), &canonicalized);
    if (PREDICT_FALSE(!s.ok())) {
      if (s.IsNotFound() || s.IsDiskFailure()) {
        // If the directory fails to canonicalize due to disk failure, store
        // the non-canonicalized form and the returned error.
        canonicalized = DirName(root);
      } else {
        return s.CloneAndPrepend(Substitute("Failed to canonicalize $0", root));
      }
    }
    canonicalized = JoinPathSegments(canonicalized, BaseName(root));
    InsertOrDie(&canonicalized_roots, root, { canonicalized, s });
  }

  // All done, use the map to set the canonicalized state.

  canonicalized_wal_fs_root_ = FindOrDie(canonicalized_roots, opts_.wal_root);
  unordered_set<string> unique_roots;
  if (!opts_.data_roots.empty()) {
    for (const string& data_fs_root : opts_.data_roots) {
      const auto& root = FindOrDie(canonicalized_roots, data_fs_root);
      if (InsertIfNotPresent(&unique_roots, root.path)) {
        canonicalized_data_fs_roots_.emplace_back(root);
        canonicalized_all_fs_roots_.emplace_back(root);
      }
    }
  } else {
    LOG(INFO) << "Data directories (fs_data_dirs) not provided";
    LOG(INFO) << "Using write-ahead log directory (fs_wal_dir) as data directory";
    canonicalized_data_fs_roots_.emplace_back(canonicalized_wal_fs_root_);
  }
  if (InsertIfNotPresent(&unique_roots, canonicalized_wal_fs_root_.path)) {
    canonicalized_all_fs_roots_.emplace_back(canonicalized_wal_fs_root_);
  }

  // Decide on a metadata root to use.
  if (opts_.metadata_root.empty()) {
    // Check the first data root for metadata.
    const string meta_dir_in_data_root = JoinPathSegments(canonicalized_data_fs_roots_[0].path,
                                                          kTabletMetadataDirName);
    // If there is already metadata in the first data root, use it. Otherwise,
    // use the WAL root.
    LOG(INFO) << "Metadata directory not provided";
    if (env_->FileExists(meta_dir_in_data_root)) {
      canonicalized_metadata_fs_root_ = canonicalized_data_fs_roots_[0];
      LOG(INFO) << "Using existing metadata directory in first data directory";
    } else {
      canonicalized_metadata_fs_root_ = canonicalized_wal_fs_root_;
      LOG(INFO) << "Using write-ahead log directory (fs_wal_dir) as metadata directory";
    }
  } else {
    // Keep track of the explicitly-defined metadata root.
    canonicalized_metadata_fs_root_ = FindOrDie(canonicalized_roots, opts_.metadata_root);
    if (InsertIfNotPresent(&unique_roots, canonicalized_metadata_fs_root_.path)) {
      canonicalized_all_fs_roots_.emplace_back(canonicalized_metadata_fs_root_);
    }
  }

  // The server cannot start if the WAL root or metadata root failed to
  // canonicalize.
  const string& wal_root = canonicalized_wal_fs_root_.path;
  RETURN_NOT_OK_PREPEND(canonicalized_wal_fs_root_.status,
      Substitute("Write-ahead log directory $0 failed to canonicalize", wal_root));
  const string& meta_root = canonicalized_metadata_fs_root_.path;
  RETURN_NOT_OK_PREPEND(canonicalized_metadata_fs_root_.status,
      Substitute("Metadata directory $0 failed to canonicalize", meta_root));

  if (VLOG_IS_ON(1)) {
    VLOG(1) << "WAL root: " << canonicalized_wal_fs_root_.path;
    VLOG(1) << "Metadata root: " << canonicalized_metadata_fs_root_.path;
    VLOG(1) << "Data roots: " <<
      JoinStrings(DataDirManager::GetRootNames(canonicalized_data_fs_roots_), ",");
    VLOG(1) << "All roots: " <<
      JoinStrings(DataDirManager::GetRootNames(canonicalized_all_fs_roots_), ",");
  }

  initted_ = true;
  return Status::OK();
}