TEST_F(SubprocessTest, Environment)
{
  Clock::pause();

  // Simple value.
  map<string, string> environment;
  environment["MESSAGE"] = "hello";

  Try<Subprocess> s = subprocess(
      "echo $MESSAGE",
      Subprocess::PIPE(),
      Subprocess::PIPE(),
      Subprocess::PIPE(),
      environment);

  ASSERT_SOME(s);
  ASSERT_SOME(s.get().out());
  AWAIT_EXPECT_EQ("hello\n", io::read(s.get().out().get()));

  // Advance time until the internal reaper reaps the subprocess.
  while (s.get().status().isPending()) {
    Clock::advance(Seconds(1));
    Clock::settle();
  }

  AWAIT_ASSERT_READY(s.get().status());
  ASSERT_SOME(s.get().status().get());

  int status = s.get().status().get().get();
  EXPECT_TRUE(WIFEXITED(status));
  EXPECT_EQ(0, WEXITSTATUS(status));

  // Multiple key-value pairs.
  environment.clear();
  environment["MESSAGE0"] = "hello";
  environment["MESSAGE1"] = "world";

  s = subprocess(
      "echo $MESSAGE0 $MESSAGE1",
      Subprocess::PIPE(),
      Subprocess::PIPE(),
      Subprocess::PIPE(),
      environment);

  ASSERT_SOME(s);
  ASSERT_SOME(s.get().out());
  AWAIT_EXPECT_EQ("hello world\n", io::read(s.get().out().get()));

  // Advance time until the internal reaper reaps the subprocess.
  while (s.get().status().isPending()) {
    Clock::advance(Seconds(1));
    Clock::settle();
  }

  AWAIT_ASSERT_READY(s.get().status());
  ASSERT_SOME(s.get().status().get());

  status = s.get().status().get().get();
  EXPECT_TRUE(WIFEXITED(status));
  EXPECT_EQ(0, WEXITSTATUS(status));

  Clock::resume();
}

int main(int argc, char* argv[])
{
  GOOGLE_PROTOBUF_VERIFY_VERSION;

  CommandInfo commandInfo;
  // Construct URIs from the encoded environment string.
  const std::string& uris = os::getenv("MESOS_EXECUTOR_URIS");
  foreach (const std::string& token, strings::tokenize(uris, " ")) {
    // Delimiter between URI, execute permission and extract options
    // Expected format: {URI}+[01][XN]
    //  {URI} - The actual URI for the asset to fetch
    //  [01]  - 1 if the execute permission should be set else 0
    //  [XN]  - X if we should extract the URI (if it's compressed) else N
    size_t pos = token.rfind("+");
    CHECK(pos != std::string::npos)
      << "Invalid executor uri token in env " << token;

    CommandInfo::URI uri;
    uri.set_value(token.substr(0, pos));
    uri.set_executable(token.substr(pos + 1, 1) == "1");
    uri.set_extract(token.substr(pos + 2, 1) == "X");

    commandInfo.add_uris()->MergeFrom(uri);
  }

  CHECK(os::hasenv("MESOS_WORK_DIRECTORY"))
    << "Missing MESOS_WORK_DIRECTORY environment variable";
  std::string directory = os::getenv("MESOS_WORK_DIRECTORY");

  // We cannot use Some in the ternary expression because the compiler needs to
  // be able to infer the type, thus the explicit Option<string>.
  // TODO(idownes): Add an os::hasenv that returns an Option<string>.
  Option<std::string> user = os::hasenv("MESOS_USER")
    ? Option<std::string>(os::getenv("MESOS_USER")) // Explicit so it compiles.
    : None();

  // Fetch each URI to a local file, chmod, then chown if a user is provided.
  foreach (const CommandInfo::URI& uri, commandInfo.uris()) {
    // Fetch the URI to a local file.
    Try<string> fetched = fetch(uri.value(), directory);
    if (fetched.isError()) {
      EXIT(1) << "Failed to fetch: " << uri.value();
    }

    // Chmod the fetched URI if it's executable, else assume it's an archive
    // that should be extracted.
    if (uri.executable()) {
      Try<Nothing> chmod = os::chmod(
          fetched.get(), S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
      if (chmod.isError()) {
        EXIT(1) << "Failed to chmod " << fetched.get() << ": " << chmod.error();
      }
    } else if (uri.extract()) {
      //TODO(idownes): Consider removing the archive once extracted.
      // Try to extract the file if it's recognized as an archive.
      Try<bool> extracted = extract(fetched.get(), directory);
      if (extracted.isError()) {
        EXIT(1) << "Failed to extract "
                << fetched.get() << ":" << extracted.error();
      }
    } else {
      LOG(INFO) << "Skipped extracting path '" << fetched.get() << "'";
    }

    // Recursively chown the directory if a user is provided.
    if (user.isSome()) {
      Try<Nothing> chowned = os::chown(user.get(), directory);
      if (chowned.isError()) {
        EXIT(1) << "Failed to chown " << directory << ": " << chowned.error();
      }
    }
  }

  return 0;
}

 Result(const Try<T>& _t)
   : data(_t.isSome() ?
          Try<Option<T>>(Some(_t.get())) :
          Try<Option<T>>(Error(_t.error()))) {}

// Checks that the DRF allocator implements the DRF algorithm
// correctly. The test accomplishes this by adding frameworks and
// slaves one at a time to the allocator, making sure that each time
// a new slave is added all of its resources are offered to whichever
// framework currently has the smallest share. Checking for proper DRF
// logic when resources are returned, frameworks exit, etc. is handled
// by SorterTest.DRFSorter.
TEST_F(DRFAllocatorTest, DRFAllocatorProcess)
{
  MockAllocatorProcess<HierarchicalDRFAllocatorProcess> allocator;

  EXPECT_CALL(allocator, initialize(_, _, _));

  master::Flags masterFlags = CreateMasterFlags();
  masterFlags.roles = Option<string>("role1,role2");
  Try<PID<Master> > master = StartMaster(&allocator, masterFlags);
  ASSERT_SOME(master);

  slave::Flags flags1 = CreateSlaveFlags();
  flags1.resources = Option<string>("cpus:2;mem:1024;disk:0");

  EXPECT_CALL(allocator, slaveAdded(_, _, _));

  Try<PID<Slave> > slave1 = StartSlave(flags1);
  ASSERT_SOME(slave1);
  // Total cluster resources now cpus=2, mem=1024.

  FrameworkInfo frameworkInfo1;
  frameworkInfo1.set_name("framework1");
  frameworkInfo1.set_user("user1");
  frameworkInfo1.set_role("role1");

  MockScheduler sched1;
  MesosSchedulerDriver driver1(&sched1, frameworkInfo1, master.get());

  EXPECT_CALL(allocator, frameworkAdded(_, _, _));

  EXPECT_CALL(sched1, registered(_, _, _));

  Future<vector<Offer> > offers1;
  EXPECT_CALL(sched1, resourceOffers(_, _))
    .WillOnce(FutureArg<1>(&offers1));

  driver1.start();

  AWAIT_READY(offers1);

  // framework1 will be offered all of slave1's resources since it is
  // the only framework running so far.
  EXPECT_THAT(offers1.get(), OfferEq(2, 1024));
  // user1 share = 1 (cpus=2, mem=1024)
  //   framework1 share = 1

  FrameworkInfo frameworkInfo2;
  frameworkInfo2.set_name("framework2");
  frameworkInfo2.set_user("user2");
  frameworkInfo2.set_role("role2");

  MockScheduler sched2;
  MesosSchedulerDriver driver2(&sched2, frameworkInfo2, master.get());

  Future<Nothing> frameworkAdded2;
  EXPECT_CALL(allocator, frameworkAdded(_, _, _))
    .WillOnce(DoAll(InvokeFrameworkAdded(&allocator),
		    FutureSatisfy(&frameworkAdded2)));

  EXPECT_CALL(sched2, registered(_, _, _));

  driver2.start();

  AWAIT_READY(frameworkAdded2);

  slave::Flags flags2 = CreateSlaveFlags();
  flags2.resources = Option<string>("cpus:1;mem:512;disk:0");

  EXPECT_CALL(allocator, slaveAdded(_, _, _));

  Future<vector<Offer> > offers2;
  EXPECT_CALL(sched2, resourceOffers(_, _))
    .WillOnce(FutureArg<1>(&offers2));

  Try<PID<Slave> > slave2 = StartSlave(flags2);
  ASSERT_SOME(slave2);
  // Total cluster resources now cpus=3, mem=1536.
  // user1 share = 0.66 (cpus=2, mem=1024)
  //   framework1 share = 1
  // user2 share = 0
  //   framework2 share = 0

  AWAIT_READY(offers2);

  // framework2 will be offered all of slave2's resources since user2
  // has the lowest user share, and framework2 is its only framework.
  EXPECT_THAT(offers2.get(), OfferEq(1, 512));
  // user1 share = 0.67 (cpus=2, mem=1024)
  //   framework1 share = 1
  // user2 share = 0.33 (cpus=1, mem=512)
  //   framework2 share = 1

  slave::Flags flags3 = CreateSlaveFlags();
//.........这里部分代码省略.........

// Checks that if a framework launches a task and then the slave the
// task was running on gets killed, the task's resources are properly
// recovered and, along with the rest of the resources from the killed
// slave, never offered again.
TYPED_TEST(AllocatorTest, SlaveLost)
{
  EXPECT_CALL(this->allocator, initialize(_, _, _));

  Try<PID<Master> > master = this->StartMaster(&this->allocator);
  ASSERT_SOME(master);

  MockExecutor exec(DEFAULT_EXECUTOR_ID);

  slave::Flags flags1 = this->CreateSlaveFlags();
  flags1.resources = Option<string>("cpus:2;mem:1024");

  EXPECT_CALL(this->allocator, slaveAdded(_, _, _));

  Try<PID<Slave> > slave1 = this->StartSlave(&exec, flags1);
  ASSERT_SOME(slave1);

  MockScheduler sched;
  MesosSchedulerDriver driver(&sched, DEFAULT_FRAMEWORK_INFO, master.get());

  EXPECT_CALL(this->allocator, frameworkAdded(_, _, _));

  EXPECT_CALL(sched, registered(_, _, _));

  // Initially, all of slave1's resources are available.
  EXPECT_CALL(sched, resourceOffers(_, OfferEq(2, 1024)))
    .WillOnce(LaunchTasks(1, 2, 512));

  EXPECT_CALL(this->allocator, resourcesUnused(_, _, _, _));

  EXPECT_CALL(exec, registered(_, _, _, _));

  Future<Nothing> launchTask;
  EXPECT_CALL(exec, launchTask(_, _))
    .WillOnce(DoAll(SendStatusUpdateFromTask(TASK_RUNNING),
                    FutureSatisfy(&launchTask)));

  EXPECT_CALL(sched, statusUpdate(_, _))
    .WillRepeatedly(DoDefault());

  driver.start();

  // Ensures the task is completely launched before we
  // kill the slave, to test that the task's resources
  // are recovered correctly (i.e. never reallocated
  // since the slave is killed)
  AWAIT_READY(launchTask);

  EXPECT_CALL(this->allocator, resourcesRecovered(_, _, _));

  Future<Nothing> slaveRemoved;
  EXPECT_CALL(this->allocator, slaveRemoved(_))
    .WillOnce(DoAll(InvokeSlaveRemoved(&this->allocator),
                    FutureSatisfy(&slaveRemoved)));

  EXPECT_CALL(exec, shutdown(_))
    .Times(AtMost(1));

  EXPECT_CALL(sched, slaveLost(_, _));

  this->ShutdownSlaves();

  AWAIT_READY(slaveRemoved);

  slave::Flags flags2 = this->CreateSlaveFlags();
  flags2.resources = Option<string>("cpus:3;mem:256");

  EXPECT_CALL(this->allocator, slaveAdded(_, _, _));

  // Eventually after slave2 is launched, we should get
  // an offer that contains all of slave2's resources
  // and none of slave1's resources.
  Future<Nothing> resourceOffers;
  EXPECT_CALL(sched, resourceOffers(_, OfferEq(3, 256)))
    .WillOnce(FutureSatisfy(&resourceOffers));

  Try<PID<Slave> > slave2 = this->StartSlave(flags2);
  ASSERT_SOME(slave2);

  AWAIT_READY(resourceOffers);

  // Shut everything down.
  EXPECT_CALL(this->allocator, resourcesRecovered(_, _, _))
    .WillRepeatedly(DoDefault());

  EXPECT_CALL(this->allocator, frameworkDeactivated(_))
    .Times(AtMost(1));

  EXPECT_CALL(this->allocator, frameworkRemoved(_))
    .Times(AtMost(1));

  driver.stop();
  driver.join();

  EXPECT_CALL(this->allocator, slaveRemoved(_))
    .Times(AtMost(1));
//.........这里部分代码省略.........

static Try<Nothing> increasePageCache(const vector<string>& tokens)
{
  const Bytes UNIT = Megabytes(1);

  if (tokens.size() < 2) {
    return Error("Expect at least one argument");
  }

  Try<Bytes> size = Bytes::parse(tokens[1]);
  if (size.isError()) {
    return Error("The first argument '" + tokens[1] + "' is not a byte size");
  }

  // TODO(chzhcn): Currently, we assume the current working directory
  // is a temporary directory and will be cleaned up when the test
  // finishes. Since the child process will inherit the current
  // working directory from the parent process, that means the test
  // that uses this helper probably needs to inherit from
  // TemporaryDirectoryTest. Consider relaxing this constraint.
  Try<string> path = os::mktemp(path::join(os::getcwd(), "XXXXXX"));
  if (path.isError()) {
    return Error("Failed to create a temporary file: " + path.error());
  }

  Try<int> fd = os::open(path.get(), O_WRONLY);
  if (fd.isError()) {
    return Error("Failed to open file: " + fd.error());
  }

  // NOTE: We are doing round-down here to calculate the number of
  // writes to do.
  for (uint64_t i = 0; i < size.get().bytes() / UNIT.bytes(); i++) {
    // Write UNIT size to disk at a time. The content isn't important.
    Try<Nothing> write = os::write(fd.get(), string(UNIT.bytes(), 'a'));
    if (write.isError()) {
      os::close(fd.get());
      return Error("Failed to write file: " + write.error());
    }

    // Use fsync to make sure data is written to disk.
    if (fsync(fd.get()) == -1) {
      // Save the error message because os::close below might
      // overwrite the errno.
      const string message = os::strerror(errno);

      os::close(fd.get());
      return Error("Failed to fsync: " + message);
    }
  }

  os::close(fd.get());
  return Nothing();
}

int main(int argc, char** argv)
{
  if (argc != 3) {
    std::cerr << "Usage: " << argv[0]
              << " <master> <balloon limit in MB>" << std::endl;
    return -1;
  }

  // Verify the balloon limit.
  Try<size_t> limit = numify<size_t>(argv[2]);
  if (limit.isError()) {
    std::cerr << "Balloon limit is not a valid number" << std::endl;
    return -1;
  }

  if (limit.get() < EXECUTOR_MEMORY_MB) {
    std::cerr << "Please use a balloon limit bigger than "
              << EXECUTOR_MEMORY_MB << " MB" << std::endl;
  }

  // Find this executable's directory to locate executor.
  string uri;
  Option<string> value = os::getenv("MESOS_BUILD_DIR");
  if (value.isSome()) {
    uri = path::join(value.get(), "src", "balloon-executor");
  } else {
    uri = path::join(
        os::realpath(Path(argv[0]).dirname()).get(),
        "balloon-executor");
  }

  ExecutorInfo executor;
  executor.mutable_executor_id()->set_value("default");
  executor.mutable_command()->set_value(uri);
  executor.set_name("Balloon Executor");
  executor.set_source("balloon_test");

  Resource* mem = executor.add_resources();
  mem->set_name("mem");
  mem->set_type(Value::SCALAR);
  mem->mutable_scalar()->set_value(EXECUTOR_MEMORY_MB);

  BalloonScheduler scheduler(executor, limit.get());

  FrameworkInfo framework;
  framework.set_user(""); // Have Mesos fill in the current user.
  framework.set_name("Balloon Framework (C++)");

  value = os::getenv("MESOS_CHECKPOINT");
  if (value.isSome()) {
    framework.set_checkpoint(
        numify<bool>(value.get()).get());
  }

  MesosSchedulerDriver* driver;
  value = os::getenv("MESOS_AUTHENTICATE");
  if (value.isSome()) {
    cout << "Enabling authentication for the framework" << endl;

    value = os::getenv("DEFAULT_PRINCIPAL");
    if (value.isNone()) {
      EXIT(EXIT_FAILURE)
        << "Expecting authentication principal in the environment";
    }

    Credential credential;
    credential.set_principal(value.get());

    framework.set_principal(value.get());

    value = os::getenv("DEFAULT_SECRET");
    if (value.isNone()) {
      EXIT(EXIT_FAILURE)
        << "Expecting authentication secret in the environment";
    }

    credential.set_secret(value.get());

    driver = new MesosSchedulerDriver(
        &scheduler, framework, argv[1], credential);
  } else {
    framework.set_principal("balloon-framework-cpp");

    driver = new MesosSchedulerDriver(
        &scheduler, framework, argv[1]);
  }

  int status = driver->run() == DRIVER_STOPPED ? 0 : 1;

  // Ensure that the driver process terminates.
  driver->stop();

  delete driver;
  return status;
}

  void launchTask(ExecutorDriver* driver, const TaskInfo& task)
  {
    if (run.isSome()) {
      // TODO(alexr): Use `protobuf::createTaskStatus()`
      // instead of manually setting fields.
      TaskStatus status;
      status.mutable_task_id()->CopyFrom(task.task_id());
      status.set_state(TASK_FAILED);
      status.set_message(
          "Attempted to run multiple tasks using a \"docker\" executor");

      driver->sendStatusUpdate(status);
      return;
    }

    // Capture the TaskID.
    taskId = task.task_id();

    // Capture the kill policy.
    if (task.has_kill_policy()) {
      killPolicy = task.kill_policy();
    }

    LOG(INFO) << "Starting task " << taskId.get();

    CHECK(task.has_container());
    CHECK(task.has_command());

    CHECK(task.container().type() == ContainerInfo::DOCKER);

    Try<Docker::RunOptions> runOptions = Docker::RunOptions::create(
        task.container(),
        task.command(),
        containerName,
        sandboxDirectory,
        mappedDirectory,
        task.resources() + task.executor().resources(),
        cgroupsEnableCfs,
        taskEnvironment,
        None(), // No extra devices.
        defaultContainerDNS
    );

    if (runOptions.isError()) {
      // TODO(alexr): Use `protobuf::createTaskStatus()`
      // instead of manually setting fields.
      TaskStatus status;
      status.mutable_task_id()->CopyFrom(task.task_id());
      status.set_state(TASK_FAILED);
      status.set_message(
        "Failed to create docker run options: " + runOptions.error());

      driver->sendStatusUpdate(status);

      _stop();
      return;
    }

    // We're adding task and executor resources to launch docker since
    // the DockerContainerizer updates the container cgroup limits
    // directly and it expects it to be the sum of both task and
    // executor resources. This does leave to a bit of unaccounted
    // resources for running this executor, but we are assuming
    // this is just a very small amount of overcommit.
    run = docker->run(
        runOptions.get(),
        Subprocess::FD(STDOUT_FILENO),
        Subprocess::FD(STDERR_FILENO));

    run->onAny(defer(self(), &Self::reaped, lambda::_1));

    // Delay sending TASK_RUNNING status update until we receive
    // inspect output. Note that we store a future that completes
    // after the sending of the running update. This allows us to
    // ensure that the terminal update is sent after the running
    // update (see `reaped()`).
    inspect = docker->inspect(containerName, DOCKER_INSPECT_DELAY)
      .then(defer(self(), [=](const Docker::Container& container) {
        if (!killed) {
          containerPid = container.pid;

          // TODO(alexr): Use `protobuf::createTaskStatus()`
          // instead of manually setting fields.
          TaskStatus status;
          status.mutable_task_id()->CopyFrom(taskId.get());
          status.set_state(TASK_RUNNING);
          status.set_data(container.output);
          if (container.ipAddress.isSome()) {
            // TODO(karya): Deprecated -- Remove after 0.25.0 has shipped.
            Label* label = status.mutable_labels()->add_labels();
            label->set_key("Docker.NetworkSettings.IPAddress");
            label->set_value(container.ipAddress.get());

            NetworkInfo* networkInfo =
              status.mutable_container_status()->add_network_infos();

            // Copy the NetworkInfo if it is specified in the
            // ContainerInfo. A Docker container has at most one
            // NetworkInfo, which is validated in containerizer.
            if (task.container().network_infos().size() > 0) {
//.........这里部分代码省略.........

Try<X509*> generate_x509(
    EVP_PKEY* subject_key,
    EVP_PKEY* sign_key,
    const Option<X509*>& parent_certificate,
    int serial,
    int days,
    Option<std::string> hostname)
{
  Option<X509_NAME*> issuer_name = None();
  if (parent_certificate.isNone()) {
    // If there is no parent certificate, then the subject and
    // signing key must be the same.
    if (subject_key != sign_key) {
      return Error("Subject vs signing key mismatch");
    }
  } else {
    // If there is a parent certificate, then set the issuer name to
    // be that of the parent.
    issuer_name = X509_get_subject_name(parent_certificate.get());

    if (issuer_name.get() == nullptr) {
      return Error("Failed to get subject name of parent certificate: "
        "X509_get_subject_name");
    }
  }

  // Allocate the in-memory structure for the certificate.
  X509* x509 = X509_new();
  if (x509 == nullptr) {
    return Error("Failed to allocate certification: X509_new");
  }

  // Set the version to V3.
  if (X509_set_version(x509, 2) != 1) {
    X509_free(x509);
    return Error("Failed to set version: X509_set_version");
  }

  // Set the serial number.
  if (ASN1_INTEGER_set(X509_get_serialNumber(x509), serial) != 1) {
    X509_free(x509);
    return Error("Failed to set serial number: ASN1_INTEGER_set");
  }

  // Make this certificate valid for 'days' number of days from now.
  if (X509_gmtime_adj(X509_get_notBefore(x509), 0) == nullptr ||
      X509_gmtime_adj(X509_get_notAfter(x509),
                      60L * 60L * 24L * days) == nullptr) {
    X509_free(x509);
    return Error("Failed to set valid days of certificate: X509_gmtime_adj");
  }

  // Set the public key for our certificate based on the subject key.
  if (X509_set_pubkey(x509, subject_key) != 1) {
    X509_free(x509);
    return Error("Failed to set public key: X509_set_pubkey");
  }

  // Figure out our hostname if one was not provided.
  if (hostname.isNone()) {
    const Try<std::string> _hostname = net::hostname();
    if (_hostname.isError()) {
      X509_free(x509);
      return Error("Failed to determine hostname");
    }

    hostname = _hostname.get();
  }

  // Grab the subject name of the new certificate.
  X509_NAME* name = X509_get_subject_name(x509);
  if (name == nullptr) {
    X509_free(x509);
    return Error("Failed to get subject name: X509_get_subject_name");
  }

  // Set the country code, organization, and common name.
  if (X509_NAME_add_entry_by_txt(
          name,
          "C",
          MBSTRING_ASC,
          reinterpret_cast<const unsigned char*>("US"),
          -1,
          -1,
          0) != 1) {
    X509_free(x509);
    return Error("Failed to set country code: X509_NAME_add_entry_by_txt");
  }

  if (X509_NAME_add_entry_by_txt(
          name,
          "O",
          MBSTRING_ASC,
          reinterpret_cast<const unsigned char*>("Test"),
          -1,
          -1,
          0) != 1) {
    X509_free(x509);
    return Error("Failed to set organization name: X509_NAME_add_entry_by_txt");
  }
//.........这里部分代码省略.........

// This test verifies that the slave run task label decorator can add
// and remove labels from a task during the launch sequence. A task
// with two labels ("foo":"bar" and "bar":"baz") is launched and will
// get modified by the slave hook to strip the "foo":"bar" pair and
// add a new "baz":"qux" pair.
TEST_F(HookTest, VerifySlaveRunTaskHook)
{
  Try<PID<Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockExecutor exec(DEFAULT_EXECUTOR_ID);

  TestContainerizer containerizer(&exec);

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

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, DEFAULT_FRAMEWORK_INFO, master.get(), DEFAULT_CREDENTIAL);

  EXPECT_CALL(sched, registered(&driver, _, _));

  Future<vector<Offer>> offers;
  EXPECT_CALL(sched, resourceOffers(&driver, _))
    .WillOnce(FutureArg<1>(&offers))
    .WillRepeatedly(Return()); // Ignore subsequent offers.

  driver.start();

  AWAIT_READY(offers);
  ASSERT_EQ(1u, offers.get().size());

  TaskInfo task;
  task.set_name("");
  task.mutable_task_id()->set_value("1");
  task.mutable_slave_id()->CopyFrom(offers.get()[0].slave_id());
  task.mutable_resources()->CopyFrom(offers.get()[0].resources());
  task.mutable_executor()->CopyFrom(DEFAULT_EXECUTOR_INFO);

  // Add two labels: (1) will be removed by the hook to ensure that
  // runTaskHook can remove labels (2) will be preserved to ensure
  // that the framework can add labels to the task and have those be
  // available by the end of the launch task sequence when hooks are
  // used (to protect against hooks removing labels completely).
  Labels* labels = task.mutable_labels();
  Label* label1 = labels->add_labels();
  label1->set_key("foo");
  label1->set_value("bar");

  Label* label2 = labels->add_labels();
  label2->set_key("bar");
  label2->set_value("baz");

  vector<TaskInfo> tasks;
  tasks.push_back(task);

  EXPECT_CALL(exec, registered(_, _, _, _));

  Future<TaskInfo> taskInfo;
  EXPECT_CALL(exec, launchTask(_, _))
    .WillOnce(DoAll(
        FutureArg<1>(&taskInfo),
        SendStatusUpdateFromTask(TASK_RUNNING)));

  driver.launchTasks(offers.get()[0].id(), tasks);

  AWAIT_READY(taskInfo);

  // The master hook will hang an extra label off.
  const Labels& labels_ = taskInfo.get().labels();

  ASSERT_EQ(3, labels_.labels_size());

  // The slave run task hook will prepend a new "baz":"qux" label.
  EXPECT_EQ(labels_.labels(0).key(), "baz");
  EXPECT_EQ(labels_.labels(0).value(), "qux");

  // Master launch task hook will still hang off test label.
  EXPECT_EQ(labels_.labels(1).key(), testLabelKey);
  EXPECT_EQ(labels_.labels(1).value(), testLabelValue);

  // And lastly, we only expect the "foo":"bar" pair to be stripped by
  // the module. The last pair should be the original "bar":"baz"
  // pair set by the test.
  EXPECT_EQ(labels_.labels(2).key(), "bar");
  EXPECT_EQ(labels_.labels(2).value(), "baz");

  driver.stop();
  driver.join();

  Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
}

// This test ensures we don't break the API when it comes to JSON
// representation of tasks.
TEST(HTTPTest, ModelTask)
{
  TaskID taskId;
  taskId.set_value("t");

  SlaveID slaveId;
  slaveId.set_value("s");

  ExecutorID executorId;
  executorId.set_value("t");

  FrameworkID frameworkId;
  frameworkId.set_value("f");

  TaskState state = TASK_RUNNING;

  vector<TaskStatus> statuses;

  TaskStatus status;
  status.mutable_task_id()->CopyFrom(taskId);
  status.set_state(state);
  status.mutable_slave_id()->CopyFrom(slaveId);
  status.mutable_executor_id()->CopyFrom(executorId);
  status.set_timestamp(0.0);

  statuses.push_back(status);

  Labels labels;
  labels.add_labels()->CopyFrom(createLabel("ACTION", "port:7987 DENY"));

  Ports ports;
  Port* port = ports.add_ports();
  port->set_number(80);
  port->mutable_labels()->CopyFrom(labels);

  DiscoveryInfo discovery;
  discovery.set_visibility(DiscoveryInfo::CLUSTER);
  discovery.set_name("discover");
  discovery.mutable_ports()->CopyFrom(ports);

  TaskInfo taskInfo;
  taskInfo.set_name("task");
  taskInfo.mutable_task_id()->CopyFrom(taskId);
  taskInfo.mutable_slave_id()->CopyFrom(slaveId);
  taskInfo.mutable_command()->set_value("echo hello");
  taskInfo.mutable_discovery()->CopyFrom(discovery);

  Task task = createTask(taskInfo, state, frameworkId);
  task.add_statuses()->CopyFrom(statuses[0]);

  JSON::Value object = model(task);

  Try<JSON::Value> expected = JSON::parse(
      "{"
      "  \"executor_id\":\"\","
      "  \"framework_id\":\"f\","
      "  \"id\":\"t\","
      "  \"name\":\"task\","
      "  \"resources\":"
      "  {"
      "    \"cpus\":0,"
      "    \"disk\":0,"
      "    \"gpus\":0,"
      "    \"mem\":0"
      "  },"
      "  \"slave_id\":\"s\","
      "  \"state\":\"TASK_RUNNING\","
      "  \"statuses\":"
      "  ["
      "    {"
      "      \"state\":\"TASK_RUNNING\","
      "      \"timestamp\":0"
      "    }"
      "  ],"
      " \"discovery\":"
      " {"
      "   \"name\":\"discover\","
      "   \"ports\":"
      "   {"
      "     \"ports\":"
      "     ["
      "       {"
      "         \"number\":80,"
      "         \"labels\":"
      "         {"
      "           \"labels\":"
      "           ["
      "             {"
      "              \"key\":\"ACTION\","
      "              \"value\":\"port:7987 DENY\""
      "             }"
      "           ]"
      "         }"
      "       }"
      "     ]"
      "   },"
      "   \"visibility\":\"CLUSTER\""
      " }"
//.........这里部分代码省略.........

// This tests the authorization of ACLs used for unreserve
// operations on dynamically reserved resources.
TYPED_TEST(AuthorizationTest, Unreserve)
{
  ACLs acls;

  // "foo" principal can unreserve its own resources.
  mesos::ACL::UnreserveResources* acl1 = acls.add_unreserve_resources();
  acl1->mutable_principals()->add_values("foo");
  acl1->mutable_reserver_principals()->add_values("foo");

  // "bar" principal cannot unreserve anyone's resources.
  mesos::ACL::UnreserveResources* acl2 = acls.add_unreserve_resources();
  acl2->mutable_principals()->add_values("bar");
  acl2->mutable_reserver_principals()->set_type(mesos::ACL::Entity::NONE);

  // "ops" principal can unreserve anyone's resources.
  mesos::ACL::UnreserveResources* acl3 = acls.add_unreserve_resources();
  acl3->mutable_principals()->add_values("ops");
  acl3->mutable_reserver_principals()->set_type(mesos::ACL::Entity::ANY);

  // No other principals can unreserve resources.
  mesos::ACL::UnreserveResources* acl4 = acls.add_unreserve_resources();
  acl4->mutable_principals()->set_type(mesos::ACL::Entity::ANY);
  acl4->mutable_reserver_principals()->set_type(mesos::ACL::Entity::NONE);

  // Create an Authorizer with the ACLs.
  Try<Authorizer*> create = TypeParam::create();
  ASSERT_SOME(create);
  Owned<Authorizer> authorizer(create.get());

  Try<Nothing> initialized = authorizer.get()->initialize(acls);
  ASSERT_SOME(initialized);

  // Principal "foo" can unreserve its own resources.
  mesos::ACL::UnreserveResources request1;
  request1.mutable_principals()->add_values("foo");
  request1.mutable_reserver_principals()->add_values("foo");
  AWAIT_EXPECT_TRUE(authorizer.get()->authorize(request1));

  // Principal "bar" cannot unreserve anyone's
  // resources, so requests 2 and 3 will fail.
  mesos::ACL::UnreserveResources request2;
  request2.mutable_principals()->add_values("bar");
  request2.mutable_reserver_principals()->add_values("foo");
  AWAIT_EXPECT_FALSE(authorizer.get()->authorize(request2));

  mesos::ACL::UnreserveResources request3;
  request3.mutable_principals()->add_values("bar");
  request3.mutable_reserver_principals()->add_values("bar");
  AWAIT_EXPECT_FALSE(authorizer.get()->authorize(request3));

  // Principal "ops" can unreserve anyone's resources,
  // so requests 4 and 5 will succeed.
  mesos::ACL::UnreserveResources request4;
  request4.mutable_principals()->add_values("ops");
  request4.mutable_reserver_principals()->add_values("foo");
  AWAIT_EXPECT_TRUE(authorizer.get()->authorize(request4));

  mesos::ACL::UnreserveResources request5;
  request5.mutable_principals()->add_values("ops");
  request5.mutable_reserver_principals()->add_values("foo");
  request5.mutable_reserver_principals()->add_values("bar");
  request5.mutable_reserver_principals()->add_values("ops");
  AWAIT_EXPECT_TRUE(authorizer.get()->authorize(request5));

  // Principal "zelda" is not mentioned in the ACLs of the Authorizer, so it
  // will be caught by the final ACL, which provides a default case that denies
  // access for all other principals. This case will fail.
  mesos::ACL::UnreserveResources request6;
  request6.mutable_principals()->add_values("zelda");
  request6.mutable_reserver_principals()->add_values("foo");
  AWAIT_EXPECT_FALSE(authorizer.get()->authorize(request6));
}

  // This method will be called when a container running as non-root user tries
  // to use a shared persistent volume or a PARENT type SANDBOX_PATH volume, the
  // parameter `path` will be the source path of the volume.
  Future<gid_t> allocate(const string& path, VolumeGidInfo::Type type)
  {
    gid_t gid;

    // If a gid has already been allocated for the specified path,
    // just return the gid.
    if (infos.contains(path)) {
      gid = infos[path].gid();

      LOG(INFO) << "Use the allocated gid " << gid << " of the volume path '"
                << path << "'";

      // If we are already setting ownership for the specified path, skip the
      // additional setting.
      if (setting.contains(path)) {
        return setting[path]->future();
      }
    } else {
      struct stat s;
      if (::stat(path.c_str(), &s) < 0) {
        return Failure("Failed to stat '" + path + "': " + os::strerror(errno));
      }

      // If the gid of the specified path is in the total gid range, just
      // return the gid. This could happen in the case that nested container
      // uses persistent volume, in which case we did a workaround in the
      // default executor to set up a volume mapping (i.e., map the persistent
      // volume to a PARENT type SANDBOX_PATH volume for the nested container)
      // so that the nested container can access the persistent volume.
      //
      // Please note that in the case of shared persistent volume, operator
      // should NOT restart agent with a different total gid range, otherwise
      // the gid of the shared persistent volume may be overwritten if a nested
      // container tries to use the shared persistent volume after the restart.
      if (totalGids.contains(s.st_gid)) {
        gid = s.st_gid;

        LOG(INFO) << "Use the gid " << gid << " for the volume path '" << path
                  << "' which should be the mount point of another volume "
                  << "which is actually allocated with the gid";
      } else {
        // Allocate a free gid to the specified path and then set the
        // ownership for it.
        if (freeGids.empty()) {
          return Failure(
              "Failed to allocate gid to the volume path '" + path +
              "' because the free gid range is exhausted");
        }

        gid = freeGids.begin()->lower();

        LOG(INFO) << "Allocating gid " << gid << " to the volume path '"
                  << path << "'";

        freeGids -= gid;
        --metrics.volume_gids_free;

        VolumeGidInfo info;
        info.set_type(type);
        info.set_path(path);
        info.set_gid(gid);

        infos.put(path, info);

        Try<Nothing> status = persist();
        if (status.isError()) {
          return Failure(
              "Failed to save state of volume gid infos: " + status.error());
        }

        Owned<Promise<gid_t>> promise(new Promise<gid_t>());

        Future<gid_t> future = async(&setVolumeOwnership, path, gid, true)
          .then([path, gid](const Try<Nothing>& result) -> Future<gid_t> {
            if (result.isError()) {
              return Failure(
                  "Failed to set the owner group of the volume path '" + path +
                  "' to " + stringify(gid) + ": " + result.error());
            }

            return gid;
          })
          .onAny(defer(self(), [=](const Future<gid_t>&) {
            setting.erase(path);
          }));

        promise->associate(future);
        setting[path] = promise;

        return promise->future();
      }
    }

    return gid;
  }

  // This method will be called in two cases:
  //   1. When a shared persistent volume is destroyed by agent, the parameter
  //      `path` will be the shared persistent volume's path.
  //   2. When a container is destroyed by containerizer, the parameter `path`
  //      will be the container's sandbox path.
  // We search if the given path is contained in `infos` (for the case 1) or is
  // the parent directory of any volume paths in `infos` (for the case 2, i.e.,
  // the PARENT type SANDBOX_PATH volume must be a subdirectory in the parent
  // container's sandbox) and then free the allocated gid for the found path(s).
  Future<Nothing> deallocate(const string& path)
  {
    vector<string> sandboxPathVolumes;

    bool changed = false;
    for (auto it = infos.begin(); it != infos.end(); ) {
      const VolumeGidInfo& info = it->second;
      const string& volumePath = info.path();

      if (strings::startsWith(volumePath, path)) {
        if (volumePath != path) {
          // This is the case of the PARENT type SANDBOX_PATH volume.
          sandboxPathVolumes.push_back(volumePath);
        }

        gid_t gid = info.gid();

        LOG(INFO) << "Deallocated gid " << gid << " for the volume path '"
                  << volumePath << "'";

        // Only return the gid to the free range if it is in the total
        // range. The gid may not be in the total range in the case that
        // Mesos agent is restarted with a different total range and we
        // deallocate gid for a previous volume path from the old range.
        if (totalGids.contains(gid)) {
          freeGids += gid;
          ++metrics.volume_gids_free;
        }

        it = infos.erase(it);
        changed = true;
      } else {
        ++it;
      }
    }

    // For the PARENT type SANDBOX_PATH volume, it will exist for a while
    // (depending on GC policy) after the container is destroyed. So to
    // avoid leaking it to other containers in the case that its gid is
    // allocated to another volume, we need to change its owner group back
    // to the original one (i.e., the primary group of its owner).
    vector<Future<Try<Nothing>>> futures;
    vector<pair<string, gid_t>> volumeGids;
    foreach (const string& volume, sandboxPathVolumes) {
      // Get the uid of the volume's owner.
      struct stat s;
      if (::stat(volume.c_str(), &s) < 0) {
        LOG(WARNING) << "Failed to stat '" << volume << "': "
                     << os::strerror(errno);

        continue;
      }

      Result<string> user = os::user(s.st_uid);
      if (!user.isSome()) {
        LOG(WARNING) << "Failed to get username for the uid " << s.st_uid
                     << ": " << (user.isError() ? user.error() : "not found");

        continue;
      }

      // Get the primary group ID of the user.
      Result<gid_t> gid = os::getgid(user.get());
      if (!gid.isSome()) {
        LOG(WARNING) << "Failed to get gid for the user '" << user.get()
                     << "': " << (gid.isError() ? gid.error() : "not found");

        continue;
      }

      futures.push_back(async(&setVolumeOwnership, volume, gid.get(), false));
      volumeGids.push_back({volume, gid.get()});
    }

    return await(futures)
      .then(defer(
          self(),
          [=](const vector<Future<Try<Nothing>>>& results) -> Future<Nothing> {
            for (size_t i = 0; i < results.size(); ++i) {
              const Future<Try<Nothing>>& result = results[i];
              const string& path = volumeGids[i].first;
              const gid_t gid = volumeGids[i].second;

              if (!result.isReady()) {
                LOG(WARNING) << "Failed to set the owner group of the volume "
                             << "path '" << path << "' back to " << gid << ": "
                             << (result.isFailed() ?
                                 result.failure() : "discarded");
              } else if (result->isError()) {
                LOG(WARNING) << "Failed to set the owner group of the volume "
                             << "path '" << path << "' back to " << gid << ": "
//.........这里部分代码省略.........

TEST_F(MesosContainerizerLaunchTest, ROOT_ChangeRootfs)
{
  Try<Owned<Rootfs>> rootfs =
    LinuxRootfs::create(path::join(os::getcwd(), "rootfs"));

  ASSERT_SOME(rootfs);

  // Add /usr/bin/stat into the rootfs.
  ASSERT_SOME(rootfs.get()->add("/usr/bin/stat"));

  Clock::pause();

  Try<Subprocess> s = run(
      "/usr/bin/stat -c %i / >" + path::join("/", &quo