Status FsManager::Open()

in src/kudu/fs/fs_manager.cc [315:452]
103 lines of code
20 McCabe index (conditional complexity)

Status FsManager::Open(FsReport* report) {
  RETURN_NOT_OK(Init());

  // Load and verify the instance metadata files.
  //
  // Done first to minimize side effects in the case that the configured roots
  // are not yet initialized on disk.
  CanonicalizedRootsList missing_roots;
  for (auto& root : canonicalized_all_fs_roots_) {
    if (!root.status.ok()) {
      continue;
    }
    gscoped_ptr<InstanceMetadataPB> pb(new InstanceMetadataPB);
    Status s = pb_util::ReadPBContainerFromPath(env_, GetInstanceMetadataPath(root.path),
                                                pb.get());
    if (PREDICT_FALSE(!s.ok())) {
      if (s.IsNotFound()) {
        missing_roots.emplace_back(root);
        continue;
      }
      if (s.IsDiskFailure()) {
        root.status = s.CloneAndPrepend("Failed to open instance file");
        continue;
      }
      return s;
    }

    if (!metadata_) {
      metadata_.reset(pb.release());
    } else if (pb->uuid() != metadata_->uuid()) {
      return Status::Corruption(Substitute(
          "Mismatched UUIDs across filesystem roots: $0 vs. $1; configuring "
          "multiple Kudu processes with the same directory is not supported",
          metadata_->uuid(), pb->uuid()));
    }
  }

  if (!metadata_) {
    return Status::NotFound("could not find a healthy instance file");
  }

  // Ensure all of the ancillary directories exist.
  vector<string> ancillary_dirs = { GetWalsRootDir(),
                                    GetTabletMetadataDir(),
                                    GetConsensusMetadataDir() };
  for (const auto& d : ancillary_dirs) {
    bool is_dir;
    RETURN_NOT_OK_PREPEND(env_->IsDirectory(d, &is_dir),
                          Substitute("could not verify required directory $0", d));
    if (!is_dir) {
      return Status::Corruption(
          Substitute("Required directory $0 exists but is not a directory", d));
    }
  }

  // In the event of failure, delete everything we created.
  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);
    }
  });

  // Create any missing roots, if desired.
  if (opts_.consistency_check == ConsistencyCheckBehavior::UPDATE_ON_DISK) {
    RETURN_NOT_OK_PREPEND(CreateFileSystemRoots(
        missing_roots, *metadata_, &created_dirs, &created_files),
                          "unable to create missing filesystem roots");
  }

  // Open the directory manager if it has not been opened already.
  if (!dd_manager_) {
    DataDirManagerOptions dm_opts;
    dm_opts.metric_entity = opts_.metric_entity;
    dm_opts.block_manager_type = opts_.block_manager_type;
    dm_opts.read_only = opts_.read_only;
    dm_opts.consistency_check = opts_.consistency_check;
    LOG_TIMING(INFO, "opening directory manager") {
      RETURN_NOT_OK(DataDirManager::OpenExisting(env_,
          canonicalized_data_fs_roots_, std::move(dm_opts), &dd_manager_));
    }
  }

  // Only clean temporary files after the data dir manager successfully opened.
  // This ensures that we were able to obtain the exclusive directory locks
  // on the data directories before we start deleting files.
  if (!opts_.read_only) {
    CleanTmpFiles();
    CheckAndFixPermissions();
  }

  // Set an initial error handler to mark data directories as failed.
  error_manager_->SetErrorNotificationCb(ErrorHandlerType::DISK_ERROR,
      Bind(&DataDirManager::MarkDataDirFailedByUuid, Unretained(dd_manager_.get())));

  // Finally, initialize and open the block manager.
  InitBlockManager();
  LOG_TIMING(INFO, "opening block manager") {
    RETURN_NOT_OK(block_manager_->Open(report));
  }

  // Report wal and metadata directories.
  if (report) {
    report->wal_dir = canonicalized_wal_fs_root_.path;
    report->metadata_dir = canonicalized_metadata_fs_root_.path;
  }

  if (FLAGS_enable_data_block_fsync) {
    // Files/directories created by the directory manager in the fs roots have
    // been synchronized, so now is a good time to sync the roots themselves.
    WARN_NOT_OK(env_util::SyncAllParentDirs(env_, created_dirs, created_dirs),
                "could not sync newly created fs roots");
  }

  LOG(INFO) << "Opened local filesystem: " <<
      JoinStrings(DataDirManager::GetRootNames(canonicalized_all_fs_roots_), ",")
            << std::endl << SecureDebugString(*metadata_);

  if (!created_dirs.empty()) {
    LOG(INFO) << "New directories created while opening local filesystem: " <<
        JoinStrings(created_dirs, ", ");
  }
  if (!created_files.empty()) {
    LOG(INFO) << "New files created while opening local filesystem: " <<
        JoinStrings(created_files, ", ");
  }

  // Success: do not delete any missing roots created.
  deleter.cancel();
  return Status::OK();
}