inline std::string createSlaveDirectory(
    const std::string& rootDir,
    const SlaveID& slaveId)
{
  std::string directory = getSlavePath(rootDir, slaveId);

  Try<Nothing> mkdir = os::mkdir(directory);

  CHECK_SOME(mkdir)
    << "Failed to create slave directory '" << directory << "'";

  // Remove the previous "latest" symlink.
  std::string latest = getLatestSlavePath(rootDir);

  if (os::exists(latest)) {
    CHECK_SOME(os::rm(latest))
      << "Failed to remove latest symlink '" << latest << "'";
  }

  // Symlink the new slave directory to "latest".
  Try<Nothing> symlink = fs::symlink(directory, latest);

  CHECK_SOME(symlink)
    << "Failed to symlink directory '" << directory
    << "' to '" << latest << "'";

  return directory;
}

  Flags()
  {
    // We log to stderr by default, but when running tests we'd prefer
    // less junk to fly by, so force one to specify the verbosity.
    add(&Flags::verbose,
        "verbose",
        "Log all severity levels to stderr",
        false);

    add(&Flags::benchmark,
        "benchmark",
        "Run the benchmark tests (and skip other tests)",
        false);

    // We determine the defaults for 'source_dir' and 'build_dir' from
    // preprocessor definitions (at the time this comment was written
    // these were set via '-DSOURCE_DIR=...' and '-DBUILD_DIR=...' in
    // src/Makefile.am).
    Result<std::string> path = os::realpath(SOURCE_DIR);
    CHECK_SOME(path);
    add(&Flags::source_dir,
        "source_dir",
        "Where to find the source directory",
        path.get());

    path = os::realpath(BUILD_DIR);
    CHECK_SOME(path);
    add(&Flags::build_dir,
        "build_dir",
        "Where to find the build directory",
        path.get());
  }

inline std::string createExecutorDirectory(
    const std::string& rootDir,
    const SlaveID& slaveId,
    const FrameworkID& frameworkId,
    const ExecutorID& executorId,
    const UUID& executorUUID)
{
  std::string directory =
    getExecutorRunPath(rootDir, slaveId, frameworkId, executorId, executorUUID);

  Try<Nothing> mkdir = os::mkdir(directory);

  CHECK_SOME(mkdir)
    << "Failed to create executor directory '" << directory << "'";

  // Remove the previous "latest" symlink.
  std::string latest =
    getExecutorLatestRunPath(rootDir, slaveId, frameworkId, executorId);

  if (os::exists(latest)) {
    CHECK_SOME(os::rm(latest))
      << "Failed to remove latest symlink '" << latest << "'";
  }

  // Symlink the new executor directory to "latest".
  Try<Nothing> symlink = fs::symlink(directory, latest);

  CHECK_SOME(symlink)
    << "Failed to symlink directory '" << directory
    << "' to '" << latest << "'";

  return directory;
}

string createSlaveDirectory(
    const string& rootDir,
    const SlaveID& slaveId)
{
  // `slaveId` should be valid because it's assigned by the master but
  // we do a sanity check here before using it to create a directory.
  CHECK_NONE(common::validation::validateSlaveID(slaveId));

  const string directory = getSlavePath(rootDir, slaveId);

  Try<Nothing> mkdir = os::mkdir(directory);

  CHECK_SOME(mkdir)
    << "Failed to create agent directory '" << directory << "'";

  // Remove the previous "latest" symlink.
  const string latest = getLatestSlavePath(rootDir);

  if (os::exists(latest)) {
    CHECK_SOME(os::rm(latest))
      << "Failed to remove latest symlink '" << latest << "'";
  }

  // Symlink the new slave directory to "latest".
  Try<Nothing> symlink = ::fs::symlink(directory, latest);

  CHECK_SOME(symlink)
    << "Failed to symlink directory '" << directory
    << "' to '" << latest << "'";

  return directory;
}

string createExecutorDirectory(
    const string& rootDir,
    const SlaveID& slaveId,
    const FrameworkID& frameworkId,
    const ExecutorID& executorId,
    const ContainerID& containerId,
    const Option<string>& user)
{
  const string directory =
    getExecutorRunPath(rootDir, slaveId, frameworkId, executorId, containerId);

  Try<Nothing> mkdir = os::mkdir(directory);

  CHECK_SOME(mkdir)
    << "Failed to create executor directory '" << directory << "'";

  // Remove the previous "latest" symlink.
  const string latest =
    getExecutorLatestRunPath(rootDir, slaveId, frameworkId, executorId);

  if (os::exists(latest)) {
    CHECK_SOME(os::rm(latest))
      << "Failed to remove latest symlink '" << latest << "'";
  }

  // Symlink the new executor directory to "latest".
  Try<Nothing> symlink = ::fs::symlink(directory, latest);

  CHECK_SOME(symlink)
    << "Failed to symlink directory '" << directory
    << "' to '" << latest << "'";

// `os::chown()` is not available on Windows.
#ifndef __WINDOWS__
  if (user.isSome()) {
    // Per MESOS-2592, we need to set the ownership of the executor
    // directory during its creation. We should not rely on subsequent
    // phases of the executor creation to ensure the ownership as
    // those may be conditional and in some cases leave the executor
    // directory owned by the slave user instead of the specified
    // framework or per-executor user.
    LOG(INFO) << "Trying to chown '" << directory << "' to user '"
              << user.get() << "'";
    Try<Nothing> chown = os::chown(user.get(), directory);
    if (chown.isError()) {
      // TODO(nnielsen): We currently have tests which depend on using
      // user names which may not be available on the test machines.
      // Therefore, we cannot make the chown validation a hard
      // CHECK().
      LOG(WARNING) << "Failed to chown executor directory '" << directory
                   << "'. This may be due to attempting to run the executor "
                   << "as a nonexistent user on the agent; see the description"
                   << " for the `--switch_user` flag for more information: "
                   << chown.error();
    }
  }
#endif // __WINDOWS__

  return directory;
}

// This test verifies that the pending future returned by
// 'Authenticator::authenticate()' is properly failed when the Authenticator is
// destructed in the middle of authentication.
TYPED_TEST(CRAMMD5Authentication, AuthenticatorDestructionRace)
{
  // Launch a dummy process (somebody to send the AuthenticateMessage).
  UPID pid = spawn(new ProcessBase(), true);

  Credential credential1;
  credential1.set_principal("benh");
  credential1.set_secret("secret");

  Credentials credentials;
  Credential* credential2 = credentials.add_credentials();
  credential2->set_principal(credential1.principal());
  credential2->set_secret(credential1.secret());

  secrets::load(credentials);

  Future<Message> message =
    FUTURE_MESSAGE(Eq(AuthenticateMessage().GetTypeName()), _, _);

  Try<Authenticatee*> authenticatee = TypeParam::TypeAuthenticatee::create();
  CHECK_SOME(authenticatee);

  Future<bool> client =
    authenticatee.get()->authenticate(pid, UPID(), credential1);

  AWAIT_READY(message);

  Try<Authenticator*> authenticator = TypeParam::TypeAuthenticator::create();
  CHECK_SOME(authenticator);

  authenticator.get()->initialize(message.get().from);

  // Drop the AuthenticationStepMessage from authenticator to keep
  // the authentication from getting completed.
  Future<AuthenticationStepMessage> authenticationStepMessage =
    DROP_PROTOBUF(AuthenticationStepMessage(), _, _);

  Future<Option<string>> principal = authenticator.get()->authenticate();

  AWAIT_READY(authenticationStepMessage);

  // At this point 'AuthenticatorProcess::authenticate()' has been
  // executed and its promise associated with the promise returned
  // by 'Authenticator::authenticate()'.
  // Authentication should be pending.
  ASSERT_TRUE(principal.isPending());

  // Now delete the authenticator.
  delete authenticator.get();

  // The future should be failed at this point.
  AWAIT_FAILED(principal);

  terminate(pid);
  delete authenticatee.get();
}

inline std::ostream& operator<<(
    std::ostream& stream,
    const ResourceProviderMessage& resourceProviderMessage)
{
  stream << stringify(resourceProviderMessage.type) << ": ";

  switch (resourceProviderMessage.type) {
    case ResourceProviderMessage::Type::UPDATE_STATE: {
      const Option<ResourceProviderMessage::UpdateState>&
        updateState = resourceProviderMessage.updateState;

      CHECK_SOME(updateState);

      return stream
          << updateState->info.id() << " "
          << updateState->totalResources;
    }

    case ResourceProviderMessage::Type::UPDATE_OPERATION_STATUS: {
      const Option<ResourceProviderMessage::UpdateOperationStatus>&
        updateOperationStatus =
          resourceProviderMessage.updateOperationStatus;

      CHECK_SOME(updateOperationStatus);

      return stream
          << "(uuid: "
          << updateOperationStatus->update.operation_uuid()
          << ") for framework "
          << updateOperationStatus->update.framework_id()
          << " (latest state: "
          << updateOperationStatus->update.latest_status().state()
          << ", status update state: "
          << updateOperationStatus->update.status().state() << ")";
    }

    case ResourceProviderMessage::Type::DISCONNECT: {
      const Option<ResourceProviderMessage::Disconnect>& disconnect =
        resourceProviderMessage.disconnect;

      CHECK_SOME(disconnect);

      return stream
          << "resource provider "
          << disconnect->resourceProviderId;
    }
  }

  UNREACHABLE();
}

// Using JSON base file for authentication without
// protobuf tools assistance.
TEST_F(CredentialsTest, AuthenticatedSlaveJSON)
{
  string path =  path::join(os::getcwd(), "credentials");

  Try<int> fd = os::open(
      path,
      O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC,
      S_IRUSR | S_IWUSR | S_IRGRP);

  CHECK_SOME(fd);

  // This unit tests our capacity to process JSON credentials without
  // using our protobuf tools.
  JSON::Object credential;
  credential.values["principal"] = DEFAULT_CREDENTIAL.principal();
  credential.values["secret"] = DEFAULT_CREDENTIAL.secret();

  JSON::Array array;
  array.values.push_back(credential);

  JSON::Object credentials;
  credentials.values["credentials"] = array;

  CHECK_SOME(os::write(fd.get(), stringify(credentials)))
      << "Failed to write credentials to '" << path << "'";

  CHECK_SOME(os::close(fd.get()));

  map<string, Option<string>> values{{"credentials", Some("file://" + path)}};

  master::Flags masterFlags = CreateMasterFlags();
  masterFlags.load(values, true);

  Try<PID<Master>> master = StartMaster(masterFlags);
  ASSERT_SOME(master);

  Future<SlaveRegisteredMessage> slaveRegisteredMessage =
    FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);

  slave::Flags slaveFlags = CreateSlaveFlags();
  slaveFlags.load(values, true);

  Try<PID<Slave>> slave = StartSlave(slaveFlags);
  ASSERT_SOME(slave);

  AWAIT_READY(slaveRegisteredMessage);
  ASSERT_NE("", slaveRegisteredMessage.get().slave_id().value());

  Shutdown();
}

  Try<Bytes> du(std::string path)
  {
    // Make sure 'path' starts with a '/'.
    path = path::join("", path);

    Try<std::string> command = strings::format(
        "%s fs -du -h '%s'", hadoop, path);

    CHECK_SOME(command);

    std::ostringstream output;

    Try<int> status = os::shell(&output, command.get() + " 2>&1");

    if (status.isError()) {
      return Error("HDFS du failed: " + status.error());
    }

    const std::vector<std::string>& s = strings::split(output.str(), " ");
    if (s.size() != 2) {
      return Error("HDFS du returned an unexpected number of results: '" +
                   output.str() + "'");
    }

    Result<size_t> size = numify<size_t>(s[0]);
    if (size.isError()) {
      return Error("HDFS du returned unexpected format: " + size.error());
    } else if (size.isNone()) {
      return Error("HDFS du returned unexpected format");
    }

    return Bytes(size.get());
  }

  Try<Bytes> du(std::string path)
  {
    path = absolutePath(path);

    Try<std::string> command = strings::format(
        "%s fs -du -h '%s'", hadoop, path);

    CHECK_SOME(command);

    // We are piping stderr to stdout so that we can see the error (if
    // any) in the logs emitted by `os::shell()` in case of failure.
    //
    // TODO(marco): this was the existing logic, but not sure it is
    // actually needed.
    Try<std::string> out = os::shell(command.get() + " 2>&1");

    if (out.isError()) {
      return Error("HDFS du failed: " + out.error());
    }

    const std::vector<std::string>& s = strings::split(out.get(), " ");
    if (s.size() != 2) {
      return Error("HDFS du returned an unexpected number of results: '" +
                   out.get() + "'");
    }

    Result<size_t> size = numify<size_t>(s[0]);
    if (size.isError()) {
      return Error("HDFS du returned unexpected format: " + size.error());
    } else if (size.isNone()) {
      return Error("HDFS du returned unexpected format");
    }

    return Bytes(size.get());
  }

 // Helper constructor for converting an `inet::Address`.
 Address(const inet::Address& address)
   : Address([](const Try<Address>& address) {
       // We expect our implementation of the cast operator to be
       // correct, hence `Address::create` should always succeed.
       CHECK_SOME(address);
       return address.get();
     }(Address::create((sockaddr_storage) address))) {}

void ZooKeeperTest::SetUpTestCase()
{
  if (!Jvm::created()) {
    std::string zkHome = flags.build_dir +
      "/3rdparty/zookeeper-" ZOOKEEPER_VERSION;

    std::string classpath = "-Djava.class.path=" +
      zkHome + "/zookeeper-" ZOOKEEPER_VERSION ".jar:" +
      zkHome + "/lib/log4j-1.2.15.jar";

    LOG(INFO) << "Using classpath setup: " << classpath << std::endl;

    std::vector<std::string> options;
    options.push_back(classpath);
    Try<Jvm*> jvm = Jvm::create(options);
    CHECK_SOME(jvm);

    if (!flags.verbose) {
      // Silence server logs.
      org::apache::log4j::Logger::getRootLogger()
        .setLevel(org::apache::log4j::Level::OFF);

      // Silence client logs.
      // TODO(jsirois): Create C++ ZooKeeper::setLevel.
      zoo_set_debug_level(ZOO_LOG_LEVEL_ERROR);
    }
  }
}

  Try<Nothing> copyFromLocal(
      const std::string& from,
      std::string to)
  {
    if (!os::exists(from)) {
      return Error("Failed to find " + from);
    }

    // Make sure 'to' starts with a '/'.
    to = path::join("", to);

    // Copy to HDFS.
    Try<std::string> command = strings::format(
        "%s fs -copyFromLocal '%s' '%s'", hadoop, from, to);

    CHECK_SOME(command);

    std::ostringstream output;

    Try<int> status = os::shell(&output, command.get() + " 2>&1");

    if (status.isError()) {
      return Error(status.error());
    } else if (status.get() != 0) {
      return Error(command.get() + "\n" + output.str());
    }

    return Nothing();
  }

 virtual ~Impl()
 {
   // Don't close if the socket was released.
   if (s >= 0) {
     CHECK_SOME(os::close(s)) << "Failed to close socket";
   }
 }

  Try<Nothing> copyFromLocal(
      const std::string& from,
      std::string to)
  {
    if (!os::exists(from)) {
      return Error("Failed to find " + from);
    }

    // Make sure 'to' starts with a '/'.
    to = path::join("", to);

    // Copy to HDFS.
    Try<std::string> command = strings::format(
        "%s fs -copyFromLocal '%s' '%s'", hadoop, from, to);

    CHECK_SOME(command);

    Try<std::string> out = os::shell(command.get());

    if (out.isError()) {
      return Error(out.error());
    }

    return Nothing();
  }

  void apply(const Offer::Operation& operation)
  {
    Try<Resources> resources = totalResources.apply(operation);
    CHECK_SOME(resources);

    totalResources = resources.get();
    checkpointedResources = totalResources.filter(needCheckpointing);
  }

mesos::allocator::Allocator* createAllocator()
{
  // T represents the allocator type. It can be a default built-in
  // allocator, or one provided by an allocator module.
  Try<mesos::allocator::Allocator*> instance = T::create();
  CHECK_SOME(instance);
  return CHECK_NOTNULL(instance.get());
}

TYPED_TEST(CRAMMD5Authentication, Success)
{
  // Launch a dummy process (somebody to send the AuthenticateMessage).
  UPID pid = spawn(new ProcessBase(), true);

  Credential credential1;
  credential1.set_principal("benh");
  credential1.set_secret("secret");

  Credentials credentials;
  Credential* credential2 = credentials.add_credentials();
  credential2->set_principal(credential1.principal());
  credential2->set_secret(credential1.secret());

  Future<Message> message =
    FUTURE_MESSAGE(Eq(AuthenticateMessage().GetTypeName()), _, _);

  Try<Authenticatee*> authenticatee = TypeParam::TypeAuthenticatee::create();
  CHECK_SOME(authenticatee);

  Future<bool> client =
    authenticatee.get()->authenticate(pid, UPID(), credential1);

  AWAIT_READY(message);

  Try<Authenticator*> authenticator = TypeParam::TypeAuthenticator::create();
  CHECK_SOME(authenticator);

  EXPECT_SOME(authenticator.get()->initialize(credentials));

  Future<Option<string>> principal =
    authenticator.get()->authenticate(message.get().from);

  AWAIT_EQ(true, client);
  AWAIT_READY(principal);
  EXPECT_SOME_EQ("benh", principal.get());

  terminate(pid);

  delete authenticator.get();
  delete authenticatee.get();
}

    static int on_body(http_parser* p, const char* data, size_t length)
    {
        StreamingResponseDecoder* decoder = (StreamingResponseDecoder*) p->data;

        CHECK_SOME(decoder->writer);

        http::Pipe::Writer writer = decoder->writer.get(); // Remove const.
        writer.write(std::string(data, length));

        return 0;
    }

ZooKeeperTestServer::ZooKeeperTestServer()
  : zooKeeperServer(NULL),
    connectionFactory(NULL),
    port(0),
    started(false)
{
  // Create temporary directories for the FileTxnSnapLog.
  Try<std::string> directory = os::mkdtemp();
  CHECK_SOME(directory);
  java::io::File dataDir(directory.get());
  dataDir.deleteOnExit();

  directory = os::mkdtemp();
  CHECK_SOME(directory);
  java::io::File snapDir(directory.get());
  snapDir.deleteOnExit();

  zooKeeperServer = new ZooKeeperServer(
      FileTxnSnapLog(dataDir, snapDir),
      ZooKeeperServer::BasicDataTreeBuilder());
}