// 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 <gmock/gmock.h>

#include <gtest/gtest.h>

#include <mesos/slave/container_logger.hpp>

#include <mesos/v1/mesos.hpp>

#include <process/collect.hpp>
#include <process/future.hpp>
#include <process/gmock.hpp>
#include <process/io.hpp>
#include <process/owned.hpp>
#include <process/subprocess.hpp>

#include <stout/duration.hpp>
#include <stout/uuid.hpp>

#ifdef __linux__
#include "linux/cgroups.hpp"
#include "linux/fs.hpp"
#endif // __linux__

#include "messages/messages.hpp"

#include "slave/containerizer/docker.hpp"
#include "slave/containerizer/fetcher.hpp"

#include "slave/containerizer/mesos/isolators/cgroups/constants.hpp"

#include "slave/paths.hpp"
#include "slave/slave.hpp"
#include "slave/state.hpp"

#include "tests/flags.hpp"
#include "tests/mesos.hpp"
#include "tests/mock_docker.hpp"

#include "tests/containerizer/docker_common.hpp"

using namespace mesos::internal::slave::paths;
using namespace mesos::internal::slave::state;

using namespace process;

using mesos::internal::master::Master;

using mesos::internal::slave::DockerContainerizer;
using mesos::internal::slave::DockerContainerizerProcess;
using mesos::internal::slave::Fetcher;
using mesos::internal::slave::Slave;

using mesos::master::detector::MasterDetector;

using mesos::slave::ContainerConfig;
using mesos::slave::ContainerLogger;
using mesos::slave::ContainerTermination;

using std::list;
using std::string;
using std::vector;

using testing::_;
using testing::DoAll;
using testing::DoDefault;
using testing::Eq;
using testing::Invoke;
using testing::Return;

namespace process {

// We need to reinitialize libprocess in order to test against
// different configurations, such as when libprocess is initialized
// with SSL or IPv6 enabled.
void reinitialize(
    const Option<string>& delegate,
    const Option<string>& readonlyAuthenticationRealm,
    const Option<string>& readwriteAuthenticationRealm);

} // namespace process {


namespace mesos {
namespace internal {
namespace tests {

#ifdef __WINDOWS__
static constexpr char DOCKER_INKY_IMAGE[] = "akagup/inky";
#else
static constexpr char DOCKER_INKY_IMAGE[] = "mesosphere/inky";
#endif // __WINDOWS__


static
ContainerInfo createDockerInfo(const string& imageName)
{
  ContainerInfo containerInfo;

  containerInfo.set_type(ContainerInfo::DOCKER);
  containerInfo.mutable_docker()->set_image(imageName);

  return containerInfo;
}


class DockerContainerizerTest : public MesosTest
{
public:
  static string containerName(const ContainerID& containerId)
  {
    return slave::DOCKER_NAME_PREFIX + containerId.value();
  }

  enum ContainerState
  {
    EXISTS,
    RUNNING
  };

  static bool exists(
      const process::Shared<Docker>& docker,
      const ContainerID& containerId,
      ContainerState state = ContainerState::EXISTS,
      bool retry = true)
  {
    Duration waited = Duration::zero();
    string expectedName = containerName(containerId);

#ifdef __WINDOWS__
    constexpr Duration waitInspect = Seconds(10);
    constexpr Duration waitInterval = Milliseconds(500);
    constexpr Duration waitMax = Seconds(15);
#else
    constexpr Duration waitInspect = Seconds(3);
    constexpr Duration waitInterval = Milliseconds(200);
    constexpr Duration waitMax = Seconds(5);
#endif // __WINDOWS__

    do {
      Future<Docker::Container> inspect = docker->inspect(expectedName);

      if (!inspect.await(waitInspect)) {
        return false;
      }

      if (inspect.isReady()) {
        switch (state) {
          case ContainerState::RUNNING:
            if (inspect->pid.isSome()) {
              return true;
            }
            // Retry looking for running pid until timeout.
            break;
          case ContainerState::EXISTS:
            return true;
        }
      }

      os::sleep(waitInterval);
      waited += waitInterval;
    } while (retry && waited < waitMax);

    return false;
  }

  static bool containsLine(
    const vector<string>& lines,
    const string& expectedLine)
  {
    foreach (const string& line, lines) {
      if (line == expectedLine) {
        return true;
      }
    }

    return false;
  }

  void SetUp() override
  {
    MesosTest::SetUp();

    Future<std::tuple<Nothing, Nothing>> pulls = process::collect(
        pullDockerImage(DOCKER_TEST_IMAGE),
        pullDockerImage(DOCKER_INKY_IMAGE));

    // The pull should only need to happen once since we don't delete the
    // image. So, we only log the warning once.
    LOG_FIRST_N(WARNING, 1) << "Pulling " << string(DOCKER_TEST_IMAGE)
                            << " and " << string(DOCKER_INKY_IMAGE) << ". "
                            << "This might take a while...";

    // The Windows images are ~200 MB, while the Linux images are ~2MB, so
    // hopefully this is enough time for the Windows images. There should
    // be some parallelism too, since we're pulling them simultaneously and
    // they share the same base Windows layer.
    AWAIT_READY_FOR(pulls, Minutes(10));
  }

  void TearDown() override
  {
    Try<Owned<Docker>> docker = Docker::create(
        tests::flags.docker,
        tests::flags.docker_socket,
        false);

    ASSERT_SOME(docker);

    Future<vector<Docker::Container>> containers =
      docker.get()->ps(true, slave::DOCKER_NAME_PREFIX);

    AWAIT_READY(containers);

    // Cleanup all mesos launched containers.
    foreach (const Docker::Container& container, containers.get()) {
      AWAIT_READY_FOR(docker.get()->rm(container.id, true), Seconds(30));
    }

    MesosTest::TearDown();
  }
};


// Only enable executor launch on linux as other platforms
// requires running linux VM and need special port forwarding
// to get host networking to work.
#ifdef __linux__
TEST_F(DockerContainerizerTest, ROOT_DOCKER_Launch_Executor)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

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

  CommandInfo command;
  command.set_value("/bin/test-executor");

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command,
      executorId);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_executor()->mutable_container()->CopyFrom(
      createDockerInfo("tnachen/test-executor"));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  AWAIT_READY_FOR(statusFinished, Seconds(60));
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

  ASSERT_TRUE(exists(docker, containerId.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));
}


// This test verifies that a custom executor can be launched and
// registered with the slave with docker bridge network enabled.
// We're assuming that the custom executor is registering its public
// ip instead of 0.0.0.0 or equivalent to the slave as that's the
// default behavior for libprocess.
//
// Currently this test fails on ubuntu and centos since the slave is
// binding and advertising 127.0.x.x address and unreachable by executor
// in bridge network.
// TODO(tnachen): Re-enable this test when we are able to fix MESOS-3123.
TEST_F(DockerContainerizerTest, DISABLED_ROOT_DOCKER_Launch_Executor_Bridged)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

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

  CommandInfo command;
  command.set_value("/bin/test-executor");

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command,
      executorId);

  // TODO(tnachen): Use local image to test if possible.
  ContainerInfo containerInfo = createDockerInfo("alpine");

  containerInfo.mutable_docker()->set_network(
      ContainerInfo::DockerInfo::BRIDGE);

  task.mutable_executor()->mutable_container()->CopyFrom(containerInfo);

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  AWAIT_READY_FOR(statusFinished, Seconds(60));
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

  ASSERT_TRUE(exists(docker, containerId.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));
}
#endif // __linux__


TEST_F(DockerContainerizerTest, ROOT_DOCKER_Launch)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  ASSERT_TRUE(statusRunning->has_data());

  Try<JSON::Array> array = JSON::parse<JSON::Array>(statusRunning->data());
  ASSERT_SOME(array);

  // Check if container information is exposed through master's state endpoint.
  Future<http::Response> response = http::get(
      master.get()->pid,
      "state",
      None(),
      createBasicAuthHeaders(DEFAULT_CREDENTIAL));

  AWAIT_EXPECT_RESPONSE_STATUS_EQ(process::http::OK().status, response);

  Try<JSON::Object> parse = JSON::parse<JSON::Object>(response->body);
  ASSERT_SOME(parse);

  Result<JSON::Value> find = parse->find<JSON::Value>(
      "frameworks[0].tasks[0].container.docker.privileged");

  EXPECT_SOME_FALSE(find);

  // Check if container information is exposed through slave's state endpoint.
  response = http::get(
      slave.get()->pid,
      "state",
      None(),
      createBasicAuthHeaders(DEFAULT_CREDENTIAL));

  AWAIT_EXPECT_RESPONSE_STATUS_EQ(process::http::OK().status, response);

  parse = JSON::parse<JSON::Object>(response->body);
  ASSERT_SOME(parse);

  find = parse->find<JSON::Value>(
      "frameworks[0].executors[0].tasks[0].container.docker.privileged");

  EXPECT_SOME_FALSE(find);

  // Now verify the ContainerStatus fields in the TaskStatus.
  ASSERT_TRUE(statusRunning->has_container_status());
  EXPECT_TRUE(statusRunning->container_status().has_container_id());
  ASSERT_EQ(1, statusRunning->container_status().network_infos().size());
  EXPECT_EQ(1, statusRunning->container_status().network_infos(0).ip_addresses().size()); // NOLINT(whitespace/line_length)

  ASSERT_TRUE(exists(docker, containerId.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));
}


// This test verifies that docker executor will terminate a task after it
// reaches `max_completion_time`.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_MaxCompletionTime)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  const Offer& offer = offers.get()[0];

  TaskInfo task =
    createTask(offer.slave_id(), offer.resources(), SLEEP_COMMAND(1000));

  // Set a `max_completion_time` for 10 seconds on Windows and 2 seconds on
  // other platforms. Hopefully this should not block test too long and still
  // keep it reliable.
#ifdef __WINDOWS__
  task.mutable_max_completion_time()->set_nanoseconds(Seconds(10).ns());
#else
  task.mutable_max_completion_time()->set_nanoseconds(Seconds(2).ns());
#endif // __WINDOWS__

  ContainerInfo containerInfo;
  containerInfo.set_type(ContainerInfo::DOCKER);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFailed;

  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFailed));

  Future<Nothing> executorTerminated =
    FUTURE_DISPATCH(_, &Slave::executorTerminated);

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  AWAIT_READY(statusFailed);
  EXPECT_EQ(TASK_FAILED, statusFailed->state());

  EXPECT_EQ(
      TaskStatus::REASON_MAX_COMPLETION_TIME_REACHED, statusFailed->reason());

  AWAIT_READY(executorTerminated);

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


TEST_F(DockerContainerizerTest, ROOT_DOCKER_Kill)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  ASSERT_TRUE(
    exists(docker, containerId.get(), ContainerState::RUNNING));

  Future<TaskStatus> statusKilled;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusKilled));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

  driver.killTask(task.task_id());

  AWAIT_READY(statusKilled);
  EXPECT_EQ(TASK_KILLED, statusKilled->state());

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  // Even though the task is killed, the executor should exit gracefully.
  ASSERT_TRUE(termination.get()->has_status());
  EXPECT_EQ(0, termination.get()->status());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));

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


// Ensures that the framework will receive a TASK_KILLING update
// before TASK_KILLED, if the capability is supported.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_TaskKillingCapability)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

  // Start the framework with the task killing capability.
  FrameworkInfo::Capability capability;
  capability.set_type(FrameworkInfo::Capability::TASK_KILLING_STATE);

  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.add_capabilities()->CopyFrom(capability);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  ASSERT_TRUE(
    exists(docker, containerId.get(), ContainerState::RUNNING));

  Future<TaskStatus> statusKilling, statusKilled;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusKilling))
    .WillOnce(FutureArg<1>(&statusKilled));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

  driver.killTask(task.task_id());

  AWAIT_READY(statusKilling);
  EXPECT_EQ(TASK_KILLING, statusKilling->state());

  AWAIT_READY(statusKilled);
  EXPECT_EQ(TASK_KILLED, statusKilled->state());

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));

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


// This test tests DockerContainerizer::usage().
TEST_F(DockerContainerizerTest, ROOT_DOCKER_Usage)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  slave::Flags flags = CreateSlaveFlags();
  flags.resources = Option<string>("cpus:1;mem:1024");

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  CommandInfo command;
  // Run a CPU intensive command, so we can measure utime and stime later.
#ifdef __WINDOWS__
  command.set_value("for /L %n in (1, 0, 2) do rem");
#else
  command.set_value("dd if=/dev/zero of=/dev/null");
#endif // __WINDOWS__

  Value::Scalar cpuLimit, memLimit;
  cpuLimit.set_value(2);
  memLimit.set_value(2048);

  google::protobuf::Map<string, Value::Scalar> resourceLimits;
  resourceLimits.insert({"cpus", cpuLimit});
  resourceLimits.insert({"mem", memLimit});

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command,
      None(),
      "test-task",
      id::UUID::random().toString(),
      resourceLimits);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  // We ignore all update calls to prevent resizing cgroup limits.
  EXPECT_CALL(dockerContainerizer, update(_, _, _))
    .WillRepeatedly(Return(Nothing()));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  // Verify the usage.
  ResourceStatistics statistics;
  Duration waited = Duration::zero();
  do {
    Future<ResourceStatistics> usage =
      dockerContainerizer.usage(containerId.get());
    // TODO(tnachen): Replace await with AWAIT_COMPLETED once
    // implemented.
    ASSERT_TRUE(usage.await(Seconds(3)));

    if (usage.isReady()) {
      statistics = usage.get();

      if (statistics.cpus_user_time_secs() > 0 &&
          statistics.cpus_system_time_secs() > 0) {
        break;
      }
    }

    os::sleep(Milliseconds(200));
    waited += Milliseconds(200);
  } while (waited < Seconds(3));

  EXPECT_EQ(1, statistics.cpus_soft_limit());
  EXPECT_EQ(2, statistics.cpus_limit());
  EXPECT_EQ(Gigabytes(1).bytes(), statistics.mem_soft_limit_bytes());
  EXPECT_EQ(Gigabytes(2).bytes(), statistics.mem_limit_bytes());

#ifndef __WINDOWS__
  // These aren't provided by the Windows Container APIs, so skip them.
  EXPECT_LT(0, statistics.cpus_user_time_secs());
  EXPECT_LT(0, statistics.cpus_system_time_secs());
  EXPECT_GT(statistics.mem_rss_bytes(), 0u);
#endif // __WINDOWS__

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

  dockerContainerizer.destroy(containerId.get());

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  // Usage() should fail again since the container is destroyed.
  Future<ResourceStatistics> usage =
    dockerContainerizer.usage(containerId.get());

  AWAIT_FAILED(usage);

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


TEST_F(DockerContainerizerTest, ROOT_DOCKER_Recover)
{
  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  SlaveID slaveId;
  slaveId.set_value("s1");
  ContainerID containerId;
  containerId.set_value(id::UUID::random().toString());
  ContainerID reapedContainerId;
  reapedContainerId.set_value(id::UUID::random().toString());

  string container1 = containerName(containerId);
  string container2 = containerName(reapedContainerId);

  // Clean up artifacts if containers still exists.
  ASSERT_TRUE(docker->rm(container1, true).await(Seconds(30)));
  ASSERT_TRUE(docker->rm(container2, true).await(Seconds(30)));

  Resources resources = Resources::parse("cpus:1;mem:512").get();

  // TODO(tnachen): Use local image to test if possible.
  ContainerInfo containerInfo = createDockerInfo(DOCKER_TEST_IMAGE);

  CommandInfo commandInfo;
  commandInfo.set_value(SLEEP_COMMAND(1000));

  Try<Docker::RunOptions> runOptions = Docker::RunOptions::create(
      containerInfo,
      commandInfo,
      container1,
      flags.work_dir,
      flags.sandbox_directory,
      resources);

  ASSERT_SOME(runOptions);

  docker->run(runOptions.get());

  Try<Docker::RunOptions> orphanOptions = Docker::RunOptions::create(
      containerInfo,
      commandInfo,
      container2,
      flags.work_dir,
      flags.sandbox_directory,
      resources);

  ASSERT_SOME(orphanOptions);

  Future<Option<int>> orphanRun = docker->run(orphanOptions.get());

  ASSERT_TRUE(
    exists(docker, containerId, ContainerState::RUNNING));
  ASSERT_TRUE(
    exists(docker, reapedContainerId, ContainerState::RUNNING));

  Future<Docker::Container> inspect = docker->inspect(container2);
  AWAIT_READY(inspect);

  SlaveState slaveState;
  slaveState.id = slaveId;
  FrameworkState frameworkState;

  ExecutorID execId;
  execId.set_value("e1");

  ExecutorState execState;
  ExecutorInfo execInfo;
  execState.info = execInfo;
  execState.latest = containerId;

  Try<process::Subprocess> wait =
    process::subprocess(tests::flags.docker + " wait " + container1);

  ASSERT_SOME(wait);

  FrameworkID frameworkId;

  RunState runState;
  runState.id = containerId;
  runState.forkedPid = wait->pid();

  execState.runs.put(containerId, runState);
  frameworkState.executors.put(execId, execState);

  slaveState.frameworks.put(frameworkId, frameworkState);

  Future<Nothing> recover = dockerContainerizer.recover(slaveState);

  AWAIT_READY(recover);

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId);

  ASSERT_FALSE(termination.isFailed());

  // The reaped container should be cleaned up and unknown at this point.
  Future<Option<ContainerTermination>> termination2 =
    dockerContainerizer.wait(reapedContainerId);

  AWAIT_READY(termination2);
  EXPECT_NONE(termination2.get());

  // Expect the orphan to be stopped!
  assertDockerKillStatus(orphanRun);
}


// This test ensures that orphaned docker containers unknown to the current
// agent instance are properly cleaned up.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_KillOrphanContainers)
{
  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

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

  ContainerID containerId;
  containerId.set_value(id::UUID::random().toString());

  ContainerID orphanContainerId;
  orphanContainerId.set_value(id::UUID::random().toString());

  string container1 = containerName(containerId);

  // Start the orphan container with the old slave id.
  string container2 = containerName(orphanContainerId);

  // Clean up artifacts if containers still exists.
  ASSERT_TRUE(docker->rm(container1, true).await(Seconds(30)));
  ASSERT_TRUE(docker->rm(container2, true).await(Seconds(30)));

  Resources resources = Resources::parse("cpus:1;mem:512").get();

  // TODO(tnachen): Use local image to test if possible.
  ContainerInfo containerInfo = createDockerInfo(DOCKER_TEST_IMAGE);

  CommandInfo commandInfo;
  commandInfo.set_value(SLEEP_COMMAND(1000));

  Try<Docker::RunOptions> runOptions = Docker::RunOptions::create(
      containerInfo,
      commandInfo,
      container1,
      flags.work_dir,
      flags.sandbox_directory,
      resources);

  ASSERT_SOME(runOptions);

  docker->run(runOptions.get());

  Try<Docker::RunOptions> orphanOptions = Docker::RunOptions::create(
      containerInfo,
      commandInfo,
      container2,
      flags.work_dir,
      flags.sandbox_directory,
      resources);

  ASSERT_SOME(orphanOptions);

  Future<Option<int>> orphanRun = docker->run(orphanOptions.get());

  ASSERT_TRUE(
    exists(docker, containerId, ContainerState::RUNNING));

  ASSERT_TRUE(
    exists(docker, orphanContainerId, ContainerState::RUNNING));

  SlaveState slaveState;
  slaveState.id = slaveId;

  FrameworkState frameworkState;

  ExecutorID execId;
  execId.set_value("e1");

  ExecutorState execState;
  ExecutorInfo execInfo;

  execState.info = execInfo;
  execState.latest = containerId;

  Try<process::Subprocess> wait =
    process::subprocess(tests::flags.docker + " wait " + container1);

  ASSERT_SOME(wait);

  FrameworkID frameworkId;

  RunState runState;
  runState.id = containerId;
  runState.forkedPid = wait->pid();

  execState.runs.put(containerId, runState);
  frameworkState.executors.put(execId, execState);

  slaveState.frameworks.put(frameworkId, frameworkState);

  Future<Nothing> recover = dockerContainerizer.recover(slaveState);

  AWAIT_READY(recover);

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId);

  ASSERT_FALSE(termination.isFailed());

  // The orphaned container should be correctly cleaned up.
  Future<Option<ContainerTermination>> termination2 =
    dockerContainerizer.wait(orphanContainerId);

  AWAIT_READY(termination2);
  EXPECT_NONE(termination2.get());
  ASSERT_FALSE(
      exists(docker, orphanContainerId, ContainerState::EXISTS, false));

  assertDockerKillStatus(orphanRun);
}


// This test checks the docker containerizer doesn't recover executors
// that were started by another containerizer (e.g: mesos).
TEST_F(DockerContainerizerTest, ROOT_DOCKER_SkipRecoverNonDocker)
{
  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  ContainerID containerId;
  containerId.set_value(id::UUID::random().toString());

  ExecutorID executorId;
  executorId.set_value(id::UUID::random().toString());

  ExecutorInfo executorInfo;
  executorInfo.mutable_container()->set_type(ContainerInfo::MESOS);

  ExecutorState executorState;
  executorState.info = executorInfo;
  executorState.latest = containerId;

  RunState runState;
  runState.id = containerId;
  executorState.runs.put(containerId, runState);

  FrameworkState frameworkState;
  frameworkState.executors.put(executorId, executorState);

  SlaveState slaveState;
  FrameworkID frameworkId;
  frameworkId.set_value(id::UUID::random().toString());
  slaveState.frameworks.put(frameworkId, frameworkState);

  Future<Nothing> recover = dockerContainerizer.recover(slaveState);
  AWAIT_READY(recover);

  Future<hashset<ContainerID>> containers = dockerContainerizer.containers();
  AWAIT_READY(containers);

  // A MesosContainerizer task shouldn't be recovered by
  // DockerContainerizer.
  EXPECT_TRUE(containers->empty());
}


// This test checks the docker containerizer doesn't recover containers
// with malformed uuid.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_SkipRecoverMalformedUUID)
{
  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();
  flags.docker_kill_orphans = true;

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  SlaveID slaveId;
  slaveId.set_value("s1");
  ContainerID containerId;
  containerId.set_value("malformedUUID");

  string container = containerName(containerId);

  // Clean up container if it still exists.
  ASSERT_TRUE(docker->rm(container, true).await(Seconds(30)));

  Resources resources = Resources::parse("cpus:1;mem:512").get();

  // TODO(tnachen): Use local image to test if possible.

  CommandInfo commandInfo;
  commandInfo.set_value(SLEEP_COMMAND(1000));

  Try<Docker::RunOptions> runOptions = Docker::RunOptions::create(
      createDockerInfo(DOCKER_TEST_IMAGE),
      commandInfo,
      container,
      flags.work_dir,
      flags.sandbox_directory,
      resources);

  Future<Option<int>> run = docker->run(runOptions.get());

  ASSERT_TRUE(
    exists(docker, containerId, ContainerState::RUNNING));

  SlaveState slaveState;
  slaveState.id = slaveId;
  FrameworkState frameworkState;

  ExecutorID execId;
  execId.set_value("e1");

  ExecutorState execState;
  ExecutorInfo execInfo;
  execState.info = execInfo;

  FrameworkID frameworkId;
  frameworkState.executors.put(execId, execState);
  slaveState.frameworks.put(frameworkId, frameworkState);

  Future<Nothing> recover = dockerContainerizer.recover(slaveState);
  AWAIT_READY(recover);

  // The container should still exist and should not get killed
  // by containerizer recovery.
  ASSERT_TRUE(exists(docker, containerId));
}


// TOOD(akagup): Persistent volumes aren't implemented on Windows, but these
// tests should be enabled once we implement them. See MESOS-5461.
#ifdef __linux__
// This test verifies that we can launch a docker container with
// persistent volume.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_LaunchWithPersistentVolumes)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();
  flags.resources = "cpu:2;mem:2048;disk(role1):2048";

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.set_roles(0, "role1");

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  Resource volume = createPersistentVolume(
    Megabytes(64),
    "role1",
    "id1",
    "path1",
    None(),
    None(),
    frameworkInfo.principal());

  CommandInfo command;
  command.set_value("echo abc > " +
                    path::join(flags.sandbox_directory, "path1", "file"));

  TaskInfo task = createTask(
      offers->front().slave_id(),
      Resources::parse("cpus:1;mem:64").get() + volume,
      command);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  // We use the filter explicitly here so that the resources will not
  // be filtered for 5 seconds (the default).
  Filters filters;
  filters.set_refuse_seconds(0);

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished))
    .WillRepeatedly(DoDefault());

  driver.acceptOffers(
      {offers->front().id()},
      {CREATE(volume), LAUNCH({task})},
      filters);

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);

  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());

  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  AWAIT_READY(statusFinished);
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));

  const string& volumePath = getPersistentVolumePath(
      flags.work_dir,
      volume);

  EXPECT_SOME_EQ("abc\n", os::read(path::join(volumePath, "file")));

  Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
  EXPECT_SOME(table);

  // Verify that the persistent volume is unmounted.
  foreach (const fs::MountInfoTable::Entry& entry, table->entries) {
    EXPECT_FALSE(strings::contains(
        entry.target,
        path::join(containerConfig->directory(), "path1")));
  }
}


// This test checks the docker containerizer is able to recover containers
// with persistent volumes and destroy it properly.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_RecoverPersistentVolumes)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();
  flags.resources = "cpu:2;mem:2048;disk(role1):2048";

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  Owned<MockDockerContainerizer> dockerContainerizer(
      new MockDockerContainerizer(
          flags,
          &fetcher,
          Owned<ContainerLogger>(logger.get()),
          docker));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), dockerContainerizer.get(), flags);
  ASSERT_SOME(slave);

  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.set_roles(0, "role1");
  frameworkInfo.set_checkpoint(true);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

  Filters filters;
  filters.set_refuse_seconds(0);

  // NOTE: We set filter explicitly here so that the resources will
  // not be filtered for 5 seconds (the default).
  Future<vector<Offer>> offers;
  EXPECT_CALL(sched, resourceOffers(&driver, _))
    .WillOnce(FutureArg<1>(&offers))
    .WillRepeatedly(DeclineOffers(filters));      // Ignore subsequent offers.

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  Resource volume = createPersistentVolume(
    Megabytes(64),
    "role1",
    "id1",
    "path1",
    None(),
    None(),
    frameworkInfo.principal());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      Resources::parse("cpus:1;mem:64").get() + volume,
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(*dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(dockerContainerizer.get(),
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.acceptOffers(
      {offers->front().id()},
      {CREATE(volume), LAUNCH({task})},
      filters);

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  // Recreate containerizer and start slave again.
  slave.get()->terminate();
  slave->reset();

  logger = ContainerLogger::create(flags.container_logger);
  ASSERT_SOME(logger);

  dockerContainerizer.reset(new MockDockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker));

  slave = StartSlave(detector.get(), dockerContainerizer.get(), flags);
  ASSERT_SOME(slave);

  Future<Nothing> _recover = FUTURE_DISPATCH(_, &Slave::_recover);

  // Wait until containerizer recover is complete.
  AWAIT_READY(_recover);

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer->destroy(containerId.get());

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
  EXPECT_SOME(table);

  // Verify that the recovered container's persistent volume is
  // unmounted.
  foreach (const fs::MountInfoTable::Entry& entry, table->entries) {
    EXPECT_FALSE(strings::contains(
        entry.target,
        path::join(containerConfig->directory(), "path1")));
  }

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


// This test checks the docker containerizer is able to clean up
// orphaned containers with persistent volumes.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_RecoverOrphanedPersistentVolumes)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();
  flags.resources = "cpu:2;mem:2048;disk(role1):2048";

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  Owned<MockDockerContainerizer> dockerContainerizer(
      new MockDockerContainerizer(
          flags,
          &fetcher,
          Owned<ContainerLogger>(logger.get()),
          docker));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), dockerContainerizer.get(), flags);
  ASSERT_SOME(slave);

  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.set_roles(0, "role1");
  frameworkInfo.set_checkpoint(true);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

  Filters filters;
  filters.set_refuse_seconds(0);

  // NOTE: We set filter explicitly here so that the resources will
  // not be filtered for 5 seconds (the default).
  Future<vector<Offer>> offers;
  EXPECT_CALL(sched, resourceOffers(&driver, _))
    .WillOnce(FutureArg<1>(&offers))
    .WillRepeatedly(DeclineOffers(filters)); // Ignore subsequent offers.

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  Resource volume = createPersistentVolume(
    Megabytes(64),
    "role1",
    "id1",
    "path1",
    None(),
    None(),
    frameworkInfo.principal());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      Resources::parse("cpus:1;mem:64").get() + volume,
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(*dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(dockerContainerizer.get(),
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.acceptOffers(
      {offers->front().id()},
      {CREATE(volume), LAUNCH({task})},
      filters);

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  // Recreate containerizer and start slave again.
  slave.get()->terminate();
  slave->reset();

  // Wipe the framework directory so that the slave will treat the
  // above running task as an orphan. We don't want to wipe the whole
  // meta directory since Docker Containerizer will skip recover if
  // state is not found.
  ASSERT_SOME(
      os::rmdir(getFrameworkPath(
          getMetaRootDir(flags.work_dir),
          offers->front().slave_id(),
          frameworkId.get())));

  logger = ContainerLogger::create(flags.container_logger);
  ASSERT_SOME(logger);

  dockerContainerizer.reset(new MockDockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker));

  slave = StartSlave(detector.get(), dockerContainerizer.get(), flags);
  ASSERT_SOME(slave);

  Future<Nothing> _recover = FUTURE_DISPATCH(_, &Slave::_recover);

  // Wait until containerizer recover is complete.
  AWAIT_READY(_recover);

  Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
  EXPECT_SOME(table);

  // Verify that the orphaned container's persistent volume is
  // unmounted.
  foreach (const fs::MountInfoTable::Entry& entry, table->entries) {
    EXPECT_FALSE(strings::contains(
        entry.target,
        path::join(containerConfig->directory(), "path1")));
  }

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

  slave->reset();

  EXPECT_FALSE(
      exists(docker, containerId.get(), ContainerState::EXISTS, false));
}
#endif // __linux__


TEST_F(DockerContainerizerTest, ROOT_DOCKER_Logs)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  // We skip stopping the docker container because stopping a container
  // even when it terminated might not flush the logs and we end up
  // not getting stdout/stderr in our tests.
  EXPECT_CALL(*mockDocker, stop(_, _, _))
    .WillRepeatedly(Return(Nothing()));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  string uuid = id::UUID::random().toString();

  CommandInfo command;
#ifdef __WINDOWS__
  // We avoid spaces in `echo` since `echo` in `cmd.exe` treats spaces
  // in the argument as literal spaces so `echo X<SPACE>` outputs X<SPACE>.
  // We don't use powershell here since `Write-Error` is verbose and causes
  // the script to return a failure.
  command.set_value(
      "echo out" + uuid + "&"
      "(echo err" + uuid + ")1>&2");
#else
  // NOTE: We prefix `echo` with `unbuffer` so that we can immediately
  // flush the output of `echo`.  This mitigates a race in Docker where
  // it mangles reads from stdout/stderr and commits suicide.
  // See MESOS-4676 for more information.
  command.set_value(
      "unbuffer echo out" + uuid + " ; "
      "unbuffer echo err" + uuid + " 1>&2");
#endif // __WINDOWS__

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command);

  // TODO(tnachen): Use local image to test if possible.
#ifdef __WINDOWS__
  const ContainerInfo containerInfo =
    createDockerInfo(DOCKER_TEST_IMAGE);
#else
  // NOTE: This is an image that is exactly
  // `docker run -t -i alpine /bin/sh -c "apk add --update expect"`.
  const ContainerInfo containerInfo =
    createDockerInfo("mesosphere/alpine-expect");
#endif // __WINDOWS__

  task.mutable_container()->CopyFrom(containerInfo);

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  AWAIT_READY_FOR(statusFinished, Seconds(60));
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

  // Now check that the proper output is in stderr and stdout (which
  // might also contain other things, hence the use of a UUID).
  Try<string> read =
    os::read(path::join(containerConfig->directory(), "stderr"));

  ASSERT_SOME(read);

  vector<string> lines = strings::split(read.get(), "\n");

  EXPECT_TRUE(containsLine(lines, "err" + uuid));
  EXPECT_FALSE(containsLine(lines, "out" + uuid));

  read = os::read(path::join(containerConfig->directory(), "stdout"));
  ASSERT_SOME(read);

  lines = strings::split(read.get(), "\n");

  EXPECT_TRUE(containsLine(lines, "out" + uuid));
  EXPECT_FALSE(containsLine(lines, "err" + uuid));

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


// The following test uses a Docker image (mesosphere/inky) that has
// an entrypoint "echo" and a default command "inky".
TEST_F(DockerContainerizerTest, ROOT_DOCKER_Default_CMD)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  // We skip stopping the docker container because stopping a container
  // even when it terminated might not flush the logs and we end up
  // not getting stdout/stderr in our tests.
  EXPECT_CALL(*mockDocker, stop(_, _, _))
    .WillRepeatedly(Return(Nothing()));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  CommandInfo command;
  command.set_shell(false);

  // NOTE: By not setting CommandInfo::value we're testing that we
  // will still be able to run the container because it has a default
  // entrypoint!

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_INKY_IMAGE));

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  AWAIT_READY_FOR(statusFinished, Seconds(60));
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

  Try<string> read =
    os::read(path::join(containerConfig->directory(), "stdout"));

  ASSERT_SOME(read);

  vector<string> lines = strings::split(read.get(), "\n");

  // Since we're not passing any command value, we're expecting the
  // default entry point to be run which is 'echo' with the default
  // command from the image which is 'inky'.
  EXPECT_TRUE(containsLine(lines, "inky"));

  read = os::read(path::join(containerConfig->directory(), "stderr"));
  ASSERT_SOME(read);

  lines = strings::split(read.get(), "\n");

  EXPECT_FALSE(containsLine(lines, "inky"));

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


// The following test uses a Docker image (mesosphere/inky) that has
// an entrypoint "echo" and a default command "inky".
TEST_F(DockerContainerizerTest, ROOT_DOCKER_Default_CMD_Override)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  // We skip stopping the docker container because stopping  a container
  // even when it terminated might not flush the logs and we end up
  // not getting stdout/stderr in our tests.
  EXPECT_CALL(*mockDocker, stop(_, _, _))
    .WillRepeatedly(Return(Nothing()));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  string uuid = id::UUID::random().toString();

  CommandInfo command;
  command.set_shell(false);

  // We can set the value to just the 'uuid' since it should get
  // passed as an argument to the entrypoint, i.e., 'echo uuid'.
  command.set_value(uuid);

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_INKY_IMAGE));

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  AWAIT_READY_FOR(statusFinished, Seconds(60));
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

  // Now check that the proper output is in stderr and stdout.
  Try<string> read =
    os::read(path::join(containerConfig->directory(), "stdout"));

  ASSERT_SOME(read);

  vector<string> lines = strings::split(read.get(), "\n");

  // We expect the passed in command value to override the image's
  // default command, thus we should see the value of 'uuid' in the
  // output instead of the default command which is 'inky'.
  EXPECT_TRUE(containsLine(lines, uuid));
  EXPECT_FALSE(containsLine(lines, "inky"));

  read = os::read(path::join(containerConfig->directory(), "stderr"));
  ASSERT_SOME(read);

  lines = strings::split(read.get(), "\n");

  EXPECT_FALSE(containsLine(lines, "inky"));
  EXPECT_FALSE(containsLine(lines, uuid));

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


// The following test uses a Docker image (mesosphere/inky) that has
// an entrypoint "echo" and a default command "inky".
TEST_F(DockerContainerizerTest, ROOT_DOCKER_Default_CMD_Args)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  // We skip stopping the docker container because stopping a container
  // even when it terminated might not flush the logs and we end up
  // not getting stdout/stderr in our tests.
  EXPECT_CALL(*mockDocker, stop(_, _, _))
    .WillRepeatedly(Return(Nothing()));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  string uuid = id::UUID::random().toString();

  CommandInfo command;
  command.set_shell(false);

  // We should also be able to skip setting the command value and just
  // set the arguments and those should also get passed through to the
  // entrypoint!
  command.add_arguments(uuid);

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_INKY_IMAGE));

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  AWAIT_READY_FOR(statusFinished, Seconds(60));
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

  // Now check that the proper output is in stderr and stdout.
  Try<string> read =
    os::read(path::join(containerConfig->directory(), "stdout"));

  ASSERT_SOME(read);

  vector<string> lines = strings::split(read.get(), "\n");

  // We expect the passed in command arguments to override the image's
  // default command, thus we should see the value of 'uuid' in the
  // output instead of the default command which is 'inky'.
  EXPECT_TRUE(containsLine(lines, uuid));
  EXPECT_FALSE(containsLine(lines, "inky"));

  read = os::read(path::join(containerConfig->directory(), "stderr"));
  ASSERT_SOME(read);

  lines = strings::split(read.get(), "\n");

  EXPECT_FALSE(containsLine(lines, "inky"));
  EXPECT_FALSE(containsLine(lines, uuid));

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


// The slave is stopped before the first update for a task is received
// from the executor. When it comes back up we make sure the executor
// reregisters and the slave properly sends the update.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_SlaveRecoveryTaskContainer)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  // This is owned by the containerizer, so we'll need one per containerizer.
  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  Owned<MockDockerContainerizer> dockerContainerizer(
      new MockDockerContainerizer(
          flags,
          &fetcher,
          Owned<ContainerLogger>(logger.get()),
          docker));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), dockerContainerizer.get(), flags);
  ASSERT_SOME(slave);

  // Enable checkpointing for the framework.
  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.set_checkpoint(true);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<ContainerID> containerId;
  EXPECT_CALL(*dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(dockerContainerizer.get(),
                           &MockDockerContainerizer::_launch)));

  // Drop the status updates from the executor. We actually wait until we can
  // drop the `TASK_RUNNING` update here because the window between the two is
  // small enough that we could still successfully receive `TASK_RUNNING` after
  // we have dropped `TASK_STARTING`.
  Future<StatusUpdateMessage> runningUpdate =
    DROP_PROTOBUF(StatusUpdateMessage(), _, _);

  Future<StatusUpdateMessage> startingUpdate =
    DROP_PROTOBUF(StatusUpdateMessage(), _, _);

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY(containerId);

  // Stop the slave before the status updates are received.
  AWAIT_READY(startingUpdate);
  AWAIT_READY(runningUpdate);

  slave.get()->terminate();

  Future<Message> reregisterExecutorMessage =
    FUTURE_MESSAGE(Eq(ReregisterExecutorMessage().GetTypeName()), _, _);

  Future<TaskStatus> status;
  EXPECT_CALL(sched, statusUpdate(_, _))
    .WillOnce(FutureArg<1>(&status))
    .WillRepeatedly(Return());       // Ignore subsequent updates.

  // This is owned by the containerizer, so we'll need one per containerizer.
  logger = ContainerLogger::create(flags.container_logger);
  ASSERT_SOME(logger);

  dockerContainerizer.reset(new MockDockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker));

  slave = StartSlave(detector.get(), dockerContainerizer.get(), flags);
  ASSERT_SOME(slave);

  // Ensure the executor reregisters.
  AWAIT_READY(reregisterExecutorMessage);

  ReregisterExecutorMessage reregister;
  reregister.ParseFromString(reregisterExecutorMessage->body);

  // Executor should inform about the unacknowledged updates.
  ASSERT_EQ(2, reregister.updates_size());

  const StatusUpdate& updateStarting = reregister.updates(0);
  ASSERT_EQ(task.task_id(), updateStarting.status().task_id());
  ASSERT_EQ(TASK_STARTING, updateStarting.status().state());

  const StatusUpdate& updateRunning = reregister.updates(1);
  ASSERT_EQ(task.task_id(), updateRunning.status().task_id());
  ASSERT_EQ(TASK_RUNNING, updateRunning.status().state());

  // Scheduler should receive the recovered update.
  AWAIT_READY(status);
  ASSERT_EQ(TASK_STARTING, status->state());

  ASSERT_TRUE(exists(docker, containerId.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer->wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}

#ifndef __WINDOWS__
// The slave is stopped before the first update for a task is received
// from the executor. When it comes back up we make sure the executor
// reregisters and the slave properly sends the update.
//
// The test is removed on Windows, because the `mesosphere/test-executor`
// image doesn't work on Windows and probably won't ever be ported.
//
// TODO(benh): This test is currently disabled because the executor
// inside the image mesosphere/test-executor does not properly set the
// executor PID that is uses during registration, so when the new
// slave recovers it can't reconnect and instead destroys that
// container. In particular, it uses '0' for its IP which we properly
// parse and can even properly use for sending other messages, but the
// current implementation of 'UPID::operator bool()' fails if the IP
// component of a PID is '0'.
//
// TODO(alexr): Enable after MESOS-8708 is resolved.
TEST_F(DockerContainerizerTest,
       DISABLED_ROOT_DOCKER_SlaveRecoveryExecutorContainer)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  // This is owned by the containerizer, so we'll need one per containerizer.
  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  Owned<MockDockerContainerizer> dockerContainerizer(
      new MockDockerContainerizer(
          flags,
          &fetcher,
          Owned<ContainerLogger>(logger.get()),
          docker));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), dockerContainerizer.get(), flags);
  ASSERT_SOME(slave);

  // Enable checkpointing for the framework.
  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.set_checkpoint(true);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

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

  CommandInfo command;
  command.set_value("test-executor");

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command,
      executorId);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_executor()->mutable_container()->CopyFrom(
      createDockerInfo("mesosphere/test-executor"));

  Future<ContainerID> containerId;
  EXPECT_CALL(*dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(dockerContainerizer.get(),
                           &MockDockerContainerizer::_launch)));

  // We need to wait until the container's pid has been been
  // checkpointed so that when the next slave recovers it won't treat
  // the executor as having gone lost! We know this has completed
  // after Containerizer::launch returns and the
  // Slave::executorLaunched gets dispatched.
  Future<Nothing> executorLaunched =
    FUTURE_DISPATCH(_, &Slave::executorLaunched);

  // The test-executor in the image immediately sends a TASK_RUNNING
  // followed by TASK_FINISHED (no sleep/delay in between) so we need
  // to drop the first TWO updates that come from the executor rather
  // than only the first update like above where we can control how
  // the length of the task.
  Future<StatusUpdateMessage> statusUpdateMessage1 =
    DROP_PROTOBUF(StatusUpdateMessage(), _, _);

  // Drop the first update from the executor.
  Future<StatusUpdateMessage> statusUpdateMessage2 =
    DROP_PROTOBUF(StatusUpdateMessage(), _, _);

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY(containerId);

  AWAIT_READY(executorLaunched);
  AWAIT_READY(statusUpdateMessage1);
  AWAIT_READY(statusUpdateMessage2);

  slave.get()->terminate();

  Future<Message> reregisterExecutorMessage =
    FUTURE_MESSAGE(Eq(ReregisterExecutorMessage().GetTypeName()), _, _);

  Future<TaskStatus> status;
  EXPECT_CALL(sched, statusUpdate(_, _))
    .WillOnce(FutureArg<1>(&status))
    .WillRepeatedly(Return());       // Ignore subsequent updates.

  // This is owned by the containerizer, so we'll need one per containerizer.
  logger = ContainerLogger::create(flags.container_logger);
  ASSERT_SOME(logger);

  dockerContainerizer.reset(new MockDockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker));

  slave = StartSlave(detector.get(), dockerContainerizer.get(), flags);
  ASSERT_SOME(slave);

  // Ensure the executor reregisters.
  AWAIT_READY(reregisterExecutorMessage);

  ReregisterExecutorMessage reregister;
  reregister.ParseFromString(reregisterExecutorMessage->body);

  // Executor should inform about the unacknowledged update.
  ASSERT_EQ(1, reregister.updates_size());
  const StatusUpdate& update = reregister.updates(0);
  ASSERT_EQ(task.task_id(), update.status().task_id());
  ASSERT_EQ(TASK_STARTING, update.status().state());

  // Scheduler should receive the recovered update.
  AWAIT_READY(status);
  ASSERT_EQ(TASK_STARTING, status->state());

  ASSERT_TRUE(exists(docker, containerId.get()));

  driver.stop();
  driver.join();
}
#endif // __WINDOWS__


// This test verifies that port mapping with bridge network is
// exposing the host port to the container port, by sending data
// to the host port and receiving it in the container by listening
// to the mapped container port.
//
// TODO(akagup): This test requres netcat on the Windows host before
// it can be ported. We could provide a build of netcat or just replace
// it with powershell for this test.
TEST_F_TEMP_DISABLED_ON_WINDOWS(
    DockerContainerizerTest, ROOT_DOCKER_NC_PortMapping)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();
  flags.resources = "cpus:1;mem:1024;ports:[10000-10000]";

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  // We skip stopping the docker container because stopping a container
  // even when it terminated might not flush the logs and we end up
  // not getting stdout/stderr in our tests.
  EXPECT_CALL(*mockDocker, stop(_, _, _))
    .WillRepeatedly(Return(Nothing()));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  CommandInfo command;
  command.set_shell(false);
  command.set_value("nc");
  command.add_arguments("-l");
  command.add_arguments("-p");
  command.add_arguments("1000");

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command);

  // TODO(tnachen): Use local image to test if possible.
  ContainerInfo containerInfo = createDockerInfo("alpine");

  containerInfo.mutable_docker()->set_network(
      ContainerInfo::DockerInfo::BRIDGE);

  ContainerInfo::DockerInfo::PortMapping portMapping;
  portMapping.set_host_port(10000);
  portMapping.set_container_port(1000);

  containerInfo.mutable_docker()->add_port_mappings()->CopyFrom(portMapping);

  task.mutable_container()->CopyFrom(containerInfo);

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  ASSERT_TRUE(
    exists(docker,
           containerId.get(),
           ContainerState::RUNNING));

  string uuid = id::UUID::random().toString();

  // Write uuid to docker mapped host port.
  Try<process::Subprocess> s = process::subprocess(
      "echo " + uuid + " | nc localhost 10000");

  ASSERT_SOME(s);
  AWAIT_READY_FOR(s->status(), Seconds(60));

  AWAIT_READY_FOR(statusFinished, Seconds(60));
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

  // Now check that the proper output is in stdout.
  Try<string> read =
    os::read(path::join(containerConfig->directory(), "stdout"));

  ASSERT_SOME(read);

  const vector<string> lines = strings::split(read.get(), "\n");

  // We expect the uuid that is sent to host port to be written
  // to stdout by the docker container running nc -l.
  EXPECT_TRUE(containsLine(lines, uuid));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}

#ifndef __WINDOWS__
// This test verifies that sandbox with ':' in the path can still
// run successfully. This a limitation of the Docker CLI where
// the volume map parameter treats colons (:) as separators,
// and incorrectly separates the sandbox directory.
//
// On Windows, colons aren't a legal path character, so this test is skipped.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_LaunchSandboxWithColon)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  // Create a sleep task whose name is "test:colon".
  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000),
      None(),
      "test:colon");

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  ASSERT_TRUE(exists(docker, containerId.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}
#endif // __WINDOWS__


TEST_F(DockerContainerizerTest, ROOT_DOCKER_DestroyWhileFetching)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  // The docker containerizer will free the process, so we must
  // allocate on the heap.
  MockDockerContainerizerProcess* process =
    new MockDockerContainerizerProcess(
        flags,
        &fetcher,
        Owned<ContainerLogger>(logger.get()),
        docker);

  MockDockerContainerizer dockerContainerizer(
      (Owned<DockerContainerizerProcess>(process)));

  Promise<Nothing> promise;
  Future<Nothing> fetch;

  // We want to pause the fetch call to simulate a long fetch time.
  EXPECT_CALL(*process, fetch(_))
    .WillOnce(DoAll(FutureSatisfy(&fetch),
                    Return(promise.future())));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<TaskStatus> statusFailed;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusFailed));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));

  AWAIT_READY(fetch);

  dockerContainerizer.destroy(containerId.get());

  // Resume docker launch.
  promise.set(Nothing());

  AWAIT_READY(statusFailed);

  EXPECT_EQ(TASK_FAILED, statusFailed->state());

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


TEST_F(DockerContainerizerTest, ROOT_DOCKER_DestroyWhilePulling)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  // The docker containerizer will free the process, so we must
  // allocate on the heap.
  MockDockerContainerizerProcess* process =
    new MockDockerContainerizerProcess(
        flags,
        &fetcher,
        Owned<ContainerLogger>(logger.get()),
        docker);

  MockDockerContainerizer dockerContainerizer(
      (Owned<DockerContainerizerProcess>(process)));

  Future<Nothing> fetch;
  EXPECT_CALL(*process, fetch(_))
    .WillOnce(DoAll(FutureSatisfy(&fetch),
                    Return(Nothing())));

  Promise<Nothing> promise;

  // We want to pause the fetch call to simulate a long fetch time.
  EXPECT_CALL(*process, pull(_))
    .WillOnce(Return(promise.future()));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<TaskStatus> statusFailed;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusFailed));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));

  // Wait until fetch is finished.
  AWAIT_READY(fetch);

  dockerContainerizer.destroy(containerId.get());

  // Resume docker launch.
  promise.set(Nothing());

  AWAIT_READY(statusFailed);

  EXPECT_EQ(TASK_FAILED, statusFailed->state());

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


// Ensures the containerizer responds correctly (false Future) to
// a request to destroy an unknown container.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_DestroyUnknownContainer)
{
  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<DockerContainerizer*> create =
    DockerContainerizer::create(flags, &fetcher);

  ASSERT_SOME(create);

  DockerContainerizer* containerizer = create.get();

  ContainerID containerId;
  containerId.set_value(id::UUID::random().toString());

  Future<Option<ContainerTermination>> destroyed =
    containerizer->destroy(containerId);

  AWAIT_READY(destroyed);
  EXPECT_NONE(destroyed.get());
}


// This test checks that when a docker containerizer update failed
// and the container failed before the executor started, the executor
// is properly killed and cleaned up.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_ExecutorCleanupWhenLaunchFailed)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  // The docker containerizer will free the process, so we must
  // allocate on the heap.
  MockDockerContainerizerProcess* process =
    new MockDockerContainerizerProcess(
        flags,
        &fetcher,
        Owned<ContainerLogger>(logger.get()),
        docker);

  MockDockerContainerizer dockerContainerizer(
      (Owned<DockerContainerizerProcess>(process)));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.add_capabilities()->set_type(
      FrameworkInfo::Capability::PARTITION_AWARE);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      "exit 0");

  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<TaskStatus> statusGone;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusGone));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  // Fail the update so we don't proceed to send run task to the executor.
  EXPECT_CALL(dockerContainerizer, update(_, _, _))
    .WillRepeatedly(Return(Failure("Fail resource update")));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));

  AWAIT_READY(statusGone);
  EXPECT_EQ(TASK_GONE, statusGone->state());
  EXPECT_EQ(TaskStatus::REASON_CONTAINER_UPDATE_FAILED,
            statusGone->reason());

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


// When the fetch fails we should send the scheduler a status
// update with message the shows the actual error.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_FetchFailure)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  // The docker containerizer will free the process, so we must
  // allocate on the heap.
  MockDockerContainerizerProcess* process =
    new MockDockerContainerizerProcess(
        flags,
        &fetcher,
        Owned<ContainerLogger>(logger.get()),
        docker);

  MockDockerContainerizer dockerContainerizer(
      (Owned<DockerContainerizerProcess>(process)));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      "exit 0");

  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<TaskStatus> statusFailed;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusFailed));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  EXPECT_CALL(*process, fetch(_))
    .WillOnce(Return(Failure("some error from fetch")));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));

  AWAIT_READY(statusFailed);

  EXPECT_EQ(TASK_FAILED, statusFailed->state());
  EXPECT_EQ("Failed to launch container: some error from fetch",
             statusFailed->message());

  // TODO(jaybuff): When MESOS-2035 is addressed we should validate
  // that statusFailed->reason() is correctly set here.

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


// When the docker pull fails we should send the scheduler a status
// update with message the shows the actual error.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_DockerPullFailure)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  // The docker containerizer will free the process, so we must
  // allocate on the heap.
  MockDockerContainerizerProcess* process =
    new MockDockerContainerizerProcess(
        flags,
        &fetcher,
        Owned<ContainerLogger>(logger.get()),
        docker);

  MockDockerContainerizer dockerContainerizer(
      (Owned<DockerContainerizerProcess>(process)));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      "exit 0");

  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<TaskStatus> statusFailed;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusFailed));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  EXPECT_CALL(*mockDocker, pull(_, _, _))
    .WillOnce(Return(Failure("some error from docker pull")));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));

  AWAIT_READY(statusFailed);

  EXPECT_EQ(TASK_FAILED, statusFailed->state());
  EXPECT_EQ("Failed to launch container: some error from docker pull",
             statusFailed->message());

  // TODO(jaybuff): When MESOS-2035 is addressed we should validate
  // that statusFailed->reason() is correctly set here.

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


// When the docker executor container fails to launch, docker inspect
// future that is in a retry loop should be discarded.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_DockerInspectDiscard)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  // The docker containerizer will free the process, so we must
  // allocate on the heap.
  MockDockerContainerizerProcess* process =
    new MockDockerContainerizerProcess(
        flags,
        &fetcher,
        Owned<ContainerLogger>(logger.get()),
        docker);

  MockDockerContainerizer dockerContainerizer(
      (Owned<DockerContainerizerProcess>(process)));

  Future<Docker::Container> inspect;
  EXPECT_CALL(*mockDocker, inspect(_, _))
    .WillOnce(FutureResult(&inspect,
                           Invoke((MockDocker*) docker.get(),
                                  &MockDocker::_inspect)));

  EXPECT_CALL(*mockDocker, run(_, _, _))
    .WillOnce(Return(Failure("Run failed")));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

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

  CommandInfo command;
#ifdef __WINDOWS__
  command.set_value(SLEEP_COMMAND(1000));
#else
  command.set_value("/bin/test-executor");
#endif // __WINDOWS__

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command,
      executorId);

  // TODO(tnachen): Use local image to test if possible.
#ifdef __WINDOWS__
  const ContainerInfo containerInfo = createDockerInfo(DOCKER_TEST_IMAGE);
#else
  const ContainerInfo containerInfo = createDockerInfo("tnachen/test-executor");
#endif // __WINDOWS__

  task.mutable_executor()->mutable_container()->CopyFrom(containerInfo);

  Future<TaskStatus> statusFailed;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusFailed));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<Nothing> executorLost;
  EXPECT_CALL(sched, executorLost(&driver, executorId, _, _))
    .WillOnce(FutureSatisfy(&executorLost));

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));

  AWAIT_READY(statusFailed);
  EXPECT_EQ(TASK_FAILED, statusFailed->state());

  AWAIT_READY(executorLost);

  AWAIT_DISCARDED(inspect);

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


// Ensures the containerizer responds correctly (returns None)
// to a request to wait on an unknown container.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_WaitUnknownContainer)
{
  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<DockerContainerizer*> create =
    DockerContainerizer::create(flags, &fetcher);

  ASSERT_SOME(create);

  DockerContainerizer* containerizer = create.get();

  ContainerID containerId;
  containerId.set_value(id::UUID::random().toString());

  Future<Option<ContainerTermination>> wait = containerizer->wait(containerId);

  AWAIT_READY(wait);
  EXPECT_NONE(wait.get());
}


// This test ensures that a task will transition straight from `TASK_KILLING` to
// `TASK_KILLED`, even if the health check begins to fail during the kill policy
// grace period.
//
// TODO(gkleiman): this test takes about 7 seconds to run, consider using mock
// tasks and health checkers to speed it up.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_NoTransitionFromKillingToRunning)
{
  Shared<Docker> docker(new MockDocker(
      tests::flags.docker, tests::flags.docker_socket));

  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  slave::Flags agentFlags = CreateSlaveFlags();

  Fetcher fetcher(agentFlags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(agentFlags.container_logger);
  ASSERT_SOME(logger);

  MockDockerContainerizer containerizer(
      agentFlags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> agent =
    StartSlave(detector.get(), &containerizer, agentFlags);
  ASSERT_SOME(agent);

  // Start the framework with the task killing capability.
  FrameworkInfo::Capability capability;
  capability.set_type(FrameworkInfo::Capability::TASK_KILLING_STATE);

  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.add_capabilities()->CopyFrom(capability);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, 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);
  EXPECT_EQ(1u, offers->size());

  const uint16_t testPort = getFreePort().get();

#ifdef __WINDOWS__
  // On Windows, we will do a command health check instead of a TCP one,
  // so that this test will work on Windows 10 (Hyper-V isolation) containers.
  const string command = SLEEP_COMMAND(1000);
#else
  // Launch a HTTP server until SIGTERM is received, then sleep for
  // 15 seconds to let the health check fail.
  const string command = strings::format(
      "trap \"sleep 15\" SIGTERM && nc -lk -p %u -e echo",
      testPort).get();
#endif // __WINDOWS__

  TaskInfo task = createTask(offers->front(), command);

  // TODO(tnachen): Use local image to test if possible.
  ContainerInfo containerInfo = createDockerInfo(DOCKER_TEST_IMAGE);

  // On Linux, the docker container runs in host network mode.
#ifndef __WINDOWS__
  containerInfo.mutable_docker()->set_network(
      ContainerInfo::DockerInfo::HOST);
#endif // __WINDOWS__

  task.mutable_container()->CopyFrom(containerInfo);

  HealthCheck healthCheck;
#ifdef __WINDOWS__
  healthCheck.set_type(HealthCheck::COMMAND);

  // The first `mkdir` will succeed, but the later ones will fail, so we get
  // the same behavior as the Linux test.
  healthCheck.mutable_command()->set_value("mkdir C:\\healthcheck-test");
#else
  healthCheck.set_type(HealthCheck::TCP);
  healthCheck.mutable_tcp()->set_port(testPort);
#endif // __WINDOWS__

  // Set `grace_period_seconds` here because it takes some time to launch
  // Netcat to serve requests.
  healthCheck.set_delay_seconds(0);
  healthCheck.set_grace_period_seconds(15);
  healthCheck.set_interval_seconds(0);

  task.mutable_health_check()->CopyFrom(healthCheck);

  // Set the kill policy grace period to 5 seconds.
  KillPolicy killPolicy;
  killPolicy.mutable_grace_period()->set_nanoseconds(Seconds(5).ns());

  task.mutable_kill_policy()->CopyFrom(killPolicy);

  Future<ContainerID> containerId;
  EXPECT_CALL(containerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&containerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusHealthy;
  Future<TaskStatus> statusKilling;
  Future<TaskStatus> statusKilled;

  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusHealthy))
    .WillOnce(FutureArg<1>(&statusKilling))
    .WillOnce(FutureArg<1>(&statusKilled));

  driver.launchTasks(offers->front().id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  AWAIT_READY(statusHealthy);
  EXPECT_EQ(TASK_RUNNING, statusHealthy->state());
  EXPECT_TRUE(statusHealthy->has_healthy());
  EXPECT_TRUE(statusHealthy->healthy());

  driver.killTask(task.task_id());

  AWAIT_READY(statusKilling);
  EXPECT_EQ(TASK_KILLING, statusKilling->state());
  EXPECT_FALSE(statusKilling->has_healthy());

  AWAIT_READY(statusKilled);
  EXPECT_EQ(TASK_KILLED, statusKilled->state());
  EXPECT_FALSE(statusKilled->has_healthy());

  Future<Option<ContainerTermination>> termination =
    containerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}


#ifndef __WINDOWS__
// This test ensures that a task will transition from `TASK_KILLING`
// to `TASK_KILLED` rather than `TASK_FINISHED` when it is killed,
// even if it returns an "EXIT_STATUS" of 0 on receiving a SIGTERM.
//
// This test is ignored on Windows, since Windows containers seem to
// always return `STATUS_CONTROL_C_EXIT` and `STATUS_UNSUCCESSFUL` for
// graceful and forceful shutdown.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_NoTransitionFromKillingToFinished)
{
  Shared<Docker> docker(new MockDocker(
      tests::flags.docker, tests::flags.docker_socket));

  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  slave::Flags agentFlags = CreateSlaveFlags();

  Fetcher fetcher(agentFlags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(agentFlags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer containerizer(
      agentFlags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> agent =
    StartSlave(detector.get(), &containerizer, agentFlags);

  ASSERT_SOME(agent);

  // Start the framework with the task killing capability.
  FrameworkInfo::Capability capability;
  capability.set_type(FrameworkInfo::Capability::TASK_KILLING_STATE);

  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.add_capabilities()->CopyFrom(capability);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, 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);
  EXPECT_EQ(1u, offers->size());

  CommandInfo command;
  command.set_shell(false);

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command);

  // The "nginx:alpine" container returns an "EXIT_STATUS" of 0 on
  // receiving a SIGTERM.
  //
  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("nginx:alpine"));

  Future<ContainerID> containerId;
  EXPECT_CALL(containerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&containerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusKilling;
  Future<TaskStatus> statusKilled;

  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusKilling))
    .WillOnce(FutureArg<1>(&statusKilled));

  driver.launchTasks(offers->front().id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));

  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());

  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  // Docker executor will call "docker stop ..." to send SIGTERM
  // to kill the task.
  driver.killTask(task.task_id());

  AWAIT_READY(statusKilling);
  EXPECT_EQ(TASK_KILLING, statusKilling->state());

  AWAIT_READY(statusKilled);
  EXPECT_EQ(TASK_KILLED, statusKilled->state());

  Future<Option<ContainerTermination>> termination =
    containerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}
#endif // __WINDOWS__


#ifdef __linux__
// This test ensures that when `cgroups_enable_cfs` is set on agent,
// the docker container launched through docker containerizer has
// `cpuQuotas` limit.
// Cgroups cpu quota is only available on Linux.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_CGROUPS_CFS_CgroupsEnableCFS)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();
  flags.cgroups_enable_cfs = true;
  flags.resources = "cpus:1;mem:128";

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  ASSERT_TRUE(statusRunning->has_data());

  // Find cgroups cpu hierarchy of the container and verifies
  // quota is set.
  string name = containerName(containerId.get());
  Future<Docker::Container> inspect = docker->inspect(name);
  AWAIT_READY(inspect);

  Result<string> cpuHierarchy = cgroups::hierarchy("cpu");
  ASSERT_SOME(cpuHierarchy);

  Option<pid_t> pid = inspect->pid;
  ASSERT_SOME(pid);

  Result<string> cpuCgroup = cgroups::cpu::cgroup(pid.get());
  ASSERT_SOME(cpuCgroup);

  Try<Duration> cfsQuota =
    cgroups::cpu::cfs_quota_us(cpuHierarchy.get(), cpuCgroup.get());

  ASSERT_SOME(cfsQuota);

  const Duration expectedCpuQuota = mesos::internal::slave::CPU_CFS_PERIOD * 1;
  EXPECT_EQ(expectedCpuQuota, cfsQuota.get());

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}


// This test verifies that a task launched with resource limits specified
// will have its CPU and memory's soft & hard limits and OOM score adjustment
// set correctly.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_CGROUPS_CFS_CommandTaskLimits)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  // Start agent with 2 CPUs and total host memory.
  Try<os::Memory> memory = os::memory();
  ASSERT_SOME(memory);

  uint64_t totalMemInMB = memory->total.bytes() / 1024 / 1024;

  slave::Flags flags = CreateSlaveFlags();
  flags.resources =
    strings::format("cpus:2;mem:%d", totalMemInMB).get();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);

  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  // Launch a task with 1 cpu request, 2 cpu limit, half of host total
  // memory as memory request and host total memory as memory limit.
  string resourceRequests = strings::format(
      "cpus:1;mem:%d;disk:1024",
      totalMemInMB/2).get();

  Value::Scalar cpuLimit, memLimit;
  cpuLimit.set_value(2);
  memLimit.set_value(totalMemInMB);

  google::protobuf::Map<string, Value::Scalar> resourceLimits;
  resourceLimits.insert({"cpus", cpuLimit});
  resourceLimits.insert({"mem", memLimit});

  TaskInfo task = createTask(
      offers.get()[0].slave_id(),
      Resources::parse(resourceRequests).get(),
      SLEEP_COMMAND(1000),
      None(),
      "test-task",
      id::UUID::random().toString(),
      resourceLimits);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  ASSERT_TRUE(statusRunning->has_data());

  string name = containerName(containerId.get());
  Future<Docker::Container> inspect = docker->inspect(name);
  AWAIT_READY(inspect);

  Result<string> cpuHierarchy = cgroups::hierarchy("cpu");
  Result<string> memoryHierarchy = cgroups::hierarchy("memory");

  ASSERT_SOME(cpuHierarchy);
  ASSERT_SOME(memoryHierarchy);

  Option<pid_t> pid = inspect->pid;
  ASSERT_SOME(pid);

  Result<string> cpuCgroup = cgroups::cpu::cgroup(pid.get());
  ASSERT_SOME(cpuCgroup);

  EXPECT_SOME_EQ(
      mesos::internal::slave::CPU_SHARES_PER_CPU,
      cgroups::cpu::shares(cpuHierarchy.get(), cpuCgroup.get()));

  Try<Duration> cfsQuota =
    cgroups::cpu::cfs_quota_us(cpuHierarchy.get(), cpuCgroup.get());

  ASSERT_SOME(cfsQuota);

  const Duration expectedCpuQuota =
    mesos::internal::slave::CPU_CFS_PERIOD * cpuLimit.value();

  EXPECT_EQ(expectedCpuQuota, cfsQuota.get());

  Result<string> memoryCgroup = cgroups::memory::cgroup(pid.get());
  ASSERT_SOME(memoryCgroup);

  EXPECT_SOME_EQ(
      Megabytes(totalMemInMB/2),
      cgroups::memory::soft_limit_in_bytes(
          memoryHierarchy.get(), memoryCgroup.get()));

  EXPECT_SOME_EQ(
      Megabytes(memLimit.value()),
      cgroups::memory::limit_in_bytes(
          memoryHierarchy.get(), memoryCgroup.get()));

  // Ensure the OOM score adjustment is correctly set for the container.
  Try<string> read = os::read(
      strings::format("/proc/%d/oom_score_adj", pid.get()).get());

  ASSERT_SOME(read);

  // Since the memory request is half of host total memory so the OOM score
  // adjustment should be about 500.
  Try<int32_t> oomScoreAdj = numify<int32_t>(strings::trim(read.get()));
  ASSERT_SOME(oomScoreAdj);

  EXPECT_GT(502, oomScoreAdj.get());
  EXPECT_LT(498, oomScoreAdj.get());

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}


// This test verifies that a task launched with infinite resource
// limits specified will have its CPU and memory's hard limits set
// correctly to infinite values.
TEST_F(
    DockerContainerizerTest,
    ROOT_DOCKER_CGROUPS_CFS_CommandTaskInfiniteLimits)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);

  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  // Launch a task with infinite resource limits.
  Value::Scalar cpuLimit, memLimit;
  cpuLimit.set_value(std::numeric_limits<double>::infinity());
  memLimit.set_value(std::numeric_limits<double>::infinity());

  google::protobuf::Map<string, Value::Scalar> resourceLimits;
  resourceLimits.insert({"cpus", cpuLimit});
  resourceLimits.insert({"mem", memLimit});

  TaskInfo task = createTask(
      offers.get()[0].slave_id(),
      offers.get()[0].resources(),
      SLEEP_COMMAND(1000),
      None(),
      "test-task",
      id::UUID::random().toString(),
      resourceLimits);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  ASSERT_TRUE(statusRunning->has_data());

  string name = containerName(containerId.get());
  Future<Docker::Container> inspect = docker->inspect(name);
  AWAIT_READY(inspect);

  Result<string> cpuHierarchy = cgroups::hierarchy("cpu");
  Result<string> memoryHierarchy = cgroups::hierarchy("memory");

  ASSERT_SOME(cpuHierarchy);
  ASSERT_SOME(memoryHierarchy);

  Option<pid_t> pid = inspect->pid;
  ASSERT_SOME(pid);

  Result<string> cpuCgroup = cgroups::cpu::cgroup(pid.get());
  ASSERT_SOME(cpuCgroup);

  // The CFS quota should be -1 which means infinite quota.
  Try<string> quota =
    cgroups::read(cpuHierarchy.get(), cpuCgroup.get(), "cpu.cfs_quota_us");

  ASSERT_SOME(quota);
  EXPECT_EQ("-1", strings::trim(quota.get()));

  Result<string> memoryCgroup = cgroups::memory::cgroup(pid.get());
  ASSERT_SOME(memoryCgroup);

  // Root cgroup (e.g., `/sys/fs/cgroup/memory/`) cannot have any limits set, so
  // its hard limit must be infinity.
  Try<Bytes> rootCgrouplimit =
    cgroups::memory::limit_in_bytes(memoryHierarchy.get(), "");

  ASSERT_SOME(rootCgrouplimit);

  // The memory hard limit should be same as root cgroup's, i.e. infinity.
  EXPECT_SOME_EQ(
      rootCgrouplimit.get(),
      cgroups::memory::limit_in_bytes(
          memoryHierarchy.get(), memoryCgroup.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}
#endif // __linux__


#ifndef __WINDOWS__
// Run a task as non root while inheriting this ownership from the
// framework supplied default user. Tests if the sandbox "stdout"
// is correctly owned and writeable by the tasks user.
// This test isn't run on Windows, because the `switch_user` flag
// isn't supported.
TEST_F(DockerContainerizerTest,
       ROOT_DOCKER_UNPRIVILEGED_USER_NonRootSandbox)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

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

  Shared<Docker> docker(
      new MockDocker(tests::flags.docker, tests::flags.docker_socket));

  slave::Flags flags = CreateSlaveFlags();
  flags.switch_user = true;

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);
  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

  AWAIT_READY(slaveRegisteredMessage);
  SlaveID slaveId = slaveRegisteredMessage->slave_id();

  Option<string> user = os::getenv("SUDO_USER");
  ASSERT_SOME(user);

  FrameworkInfo framework;
  framework.set_name("default");
  framework.set_user(user.get());
  framework.set_principal(DEFAULT_CREDENTIAL.principal());
  framework.add_capabilities()->set_type(
      FrameworkInfo::Capability::RESERVATION_REFINEMENT);

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  // Start the task as a user without supplying an explicit command
  // user. This should inherit the framework user for the task
  // ownership.
  CommandInfo command;
  command.set_value("echo \"foo\" && sleep 1000");

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      command);

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());

  ASSERT_TRUE(exists(docker, containerId.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));

  // Check that the sandbox was written to.
  const string sandboxDirectory = slave::paths::getExecutorRunPath(
      flags.work_dir,
      slaveId,
      frameworkId.get(),
      statusRunning->executor_id(),
      containerId.get());

  ASSERT_TRUE(os::exists(sandboxDirectory));

  // Check the sandbox "stdout" was written to.
  const string stdoutPath = path::join(sandboxDirectory, "stdout");
  ASSERT_TRUE(os::exists(stdoutPath));

  // Check the sandbox "stdout" is owned by the framework default user.
  struct stat stdoutStat;
  ASSERT_GE(::stat(stdoutPath.c_str(), &stdoutStat), 0);

  Result<uid_t> uid = os::getuid(framework.user());
  ASSERT_SOME(uid);

  ASSERT_EQ(stdoutStat.st_uid, uid.get());

  // Validate that our task was able to log into the sandbox.
  Result<string> stdout = os::read(stdoutPath);
  ASSERT_SOME(stdout);

  EXPECT_TRUE(strings::contains(stdout.get(), "foo"));
}
#endif // __WINDOWS__


// This test verifies the DNS configuration of the Docker container
// can be successfully set with the agent flag `--default_container_dns`.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_DefaultDNS)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

#ifdef __WINDOWS__
  // --dns-option and --dns-search are not supported on Windows.
  // See https://docs.microsoft.com/en-us/virtualization/windowscontainers/manage-containers/container-networking // NOLINT(whitespace/line_length)
  Try<ContainerDNSInfo> parse = flags::parse<ContainerDNSInfo>(
      R"~(
      {
        "docker": [
          {
            "network_mode": "BRIDGE",
            "dns": {
              "nameservers": [ "8.8.8.8", "8.8.4.4" ]
            }
          }
        ]
      })~");
#else
  Try<ContainerDNSInfo> parse = flags::parse<ContainerDNSInfo>(
      R"~(
      {
        "docker": [
          {
            "network_mode": "BRIDGE",
            "dns": {
              "nameservers": [ "8.8.8.8", "8.8.4.4" ],
              "search": [ "example1.com", "example2.com" ],
              "options": [ "timeout:3", "attempts:2" ]
            }
          }
        ]
      })~");
#endif // __WINDOWS__

  ASSERT_SOME(parse);

  flags.default_container_dns = parse.get();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  ContainerInfo containerInfo = createDockerInfo(DOCKER_TEST_IMAGE);

  containerInfo.mutable_docker()->set_network(
      ContainerInfo::DockerInfo::BRIDGE);

  task.mutable_container()->CopyFrom(containerInfo);

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  ASSERT_TRUE(statusRunning->has_data());

  // Find the DNS configuration of the container and verify
  // if it is consistent with `flags.default_container_dns`.
  string name = containerName(containerId.get());
  Future<Docker::Container> inspect = docker->inspect(name);
  AWAIT_READY(inspect);

  vector<string> defaultDNS;
  std::copy(
      flags.default_container_dns->docker(0).dns().nameservers().begin(),
      flags.default_container_dns->docker(0).dns().nameservers().end(),
      std::back_inserter(defaultDNS));

  EXPECT_EQ(inspect->dns, defaultDNS);

#ifndef __WINDOWS__
  vector<string> defaultDNSSearch;
  std::copy(
      flags.default_container_dns->docker(0).dns().search().begin(),
      flags.default_container_dns->docker(0).dns().search().end(),
      std::back_inserter(defaultDNSSearch));

  EXPECT_EQ(inspect->dnsSearch, defaultDNSSearch);

  vector<string> defaultDNSOption;
  std::copy(
      flags.default_container_dns->docker(0).dns().options().begin(),
      flags.default_container_dns->docker(0).dns().options().end(),
      std::back_inserter(defaultDNSOption));

  EXPECT_EQ(inspect->dnsOptions, defaultDNSOption);
#endif // __WINDOWS__

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());
}


// This test verifies that the `MESOS_ALLOCATION_ROLE`
// environment variable is set properly.
TEST_F(DockerContainerizerTest, ROOT_DOCKER_AllocationRoleEnvironmentVariable)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  // We skip stopping the docker container because stopping a container
  // even when it terminated might not flush the logs and we end up
  // not getting stdout/stderr in our tests.
  EXPECT_CALL(*mockDocker, stop(_, _, _))
    .WillRepeatedly(Return(Nothing()));

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

  // Start the framework with a role specified.
  FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
  frameworkInfo.clear_roles();
  frameworkInfo.add_roles("role1");

  MockScheduler sched;
  MesosSchedulerDriver driver(
      &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL);

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  TaskInfo task = createTask(
      offers->front().slave_id(),
      offers->front().resources(),
#ifdef __WINDOWS__
      "if %MESOS_ALLOCATION_ROLE% == \"role1\" (exit 1)");
#else
      "if [ \"$MESOS_ALLOCATION_ROLE\" != \"role1\" ]; then exit 1; fi");
#endif // __WINDOWS__

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<ContainerID> containerId;
  Future<ContainerConfig> containerConfig;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    FutureArg<1>(&containerConfig),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  Future<TaskStatus> statusFinished;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillOnce(FutureArg<1>(&statusFinished))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY(containerConfig);
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  AWAIT_READY_FOR(statusFinished, Seconds(60));
  EXPECT_EQ(TASK_FINISHED, statusFinished->state());

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


// Fixture for testing IPv6 support for docker containers on host network.
//
// TODO(akagup): Windows containers do not support IPv6, but they should
// in the future, so enable these when IPv6 is supported. See MESOS-8566.
//
// TODO(asridharan): Currently in the `Setup` and `TearDown` methods
// of this class we re-initialize libprocess to take an IPv6 address.
// Ideally, we should be moving this into a more general test fixture
// in mesos.hpp to be used by any other tests for IPv6. This might
// need changes to `MesosTest` in to order to allow for multiple
// inheritance.
class DockerContainerizerIPv6Test : public DockerContainerizerTest
{
protected:
  void SetUp() override
  {
    os::setenv("LIBPROCESS_IP6", "::1234");
    process::reinitialize(
        None(),
        READWRITE_HTTP_AUTHENTICATION_REALM,
        READONLY_HTTP_AUTHENTICATION_REALM);

    DockerContainerizerTest::SetUp();
  }

  void TearDown() override
  {
    DockerContainerizerTest::TearDown();

    os::unsetenv("LIBPROCESS_IP6");
    process::reinitialize(
        None(),
        READWRITE_HTTP_AUTHENTICATION_REALM,
        READONLY_HTTP_AUTHENTICATION_REALM);
  }
};


// Launches a docker container on the host network. The host network
// is assumed to have an IPv4 address and an IPv6 address. The test
// passes if the Mesos state correctly exposes both the IPv4 and IPv6
// address.
TEST_F_TEMP_DISABLED_ON_WINDOWS(
    DockerContainerizerIPv6Test,
    ROOT_DOCKER_LaunchIPv6HostNetwork)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);

  ASSERT_SOME(slave);

  // Check if the slave has the IPv6 address stored in its PID.
  EXPECT_SOME(slave.get()->pid.addresses.v6);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  const Offer& offer = offers.get()[0];

  TaskInfo task = createTask(
      offer.slave_id(),
      offer.resources(),
      SLEEP_COMMAND(10000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo("alpine"));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  ASSERT_TRUE(statusRunning->has_data());

  Try<JSON::Array> array = JSON::parse<JSON::Array>(statusRunning->data());
  ASSERT_SOME(array);

  // Check if container information is exposed through master's state endpoint.
  Future<http::Response> response = http::get(
      master.get()->pid,
      "state",
      None(),
      createBasicAuthHeaders(DEFAULT_CREDENTIAL));

  AWAIT_EXPECT_RESPONSE_STATUS_EQ(process::http::OK().status, response);

  Try<JSON::Object> parse = JSON::parse<JSON::Object>(response->body);
  ASSERT_SOME(parse);

  Result<JSON::Value> find = parse->find<JSON::Value>(
      "frameworks[0].tasks[0].container.docker.privileged");

  EXPECT_SOME_FALSE(find);

  // Check if container information is exposed through slave's state endpoint.
  response = http::get(
      slave.get()->pid,
      "state",
      None(),
      createBasicAuthHeaders(DEFAULT_CREDENTIAL));

  AWAIT_EXPECT_RESPONSE_STATUS_EQ(process::http::OK().status, response);

  parse = JSON::parse<JSON::Object>(response->body);
  ASSERT_SOME(parse);

  find = parse->find<JSON::Value>(
      "frameworks[0].executors[0].tasks[0].container.docker.privileged");

  EXPECT_SOME_FALSE(find);

  // Now verify the ContainerStatus fields in the TaskStatus.
  ASSERT_TRUE(statusRunning->has_container_status());
  EXPECT_TRUE(statusRunning->container_status().has_container_id());
  ASSERT_EQ(1, statusRunning->container_status().network_infos().size());
  EXPECT_EQ(2, statusRunning->container_status().network_infos(0).ip_addresses().size()); // NOLINT(whitespace/line_length)

  Option<string> containerIPv4 = None();
  Option<string> containerIPv6 = None();

  foreach(const NetworkInfo::IPAddress& ipAddress,
          statusRunning->container_status().network_infos(0).ip_addresses()) {
    if (ipAddress.protocol() == NetworkInfo::IPv4) {
      containerIPv4 = ipAddress.ip_address();
    }

    if (ipAddress.protocol() == NetworkInfo::IPv6) {
      containerIPv6 = ipAddress.ip_address();
    }
  }

  EXPECT_SOME(containerIPv4);
  ASSERT_SOME(containerIPv6);
  EXPECT_EQ(containerIPv6.get(), "::1234");

  ASSERT_TRUE(exists(docker, containerId.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));
}


// Fixture for testing IPv6 support for docker containers on docker
// user network.
class DockerContainerizerIPv6UserNetworkTest : public DockerContainerizerTest
{
protected:
  void SetUp() override
  {
    createDockerIPv6UserNetwork();
    DockerContainerizerTest::SetUp();
  }

  void TearDown() override
  {
    DockerContainerizerTest::TearDown();
    removeDockerIPv6UserNetwork();
  }
};


// Launches a docker container on the docker user network. The docker network
// is assumed to have an IPv4 address and an IPv6 address. The test passes if
// the Mesos state correctly exposes both the IPv4 and IPv6 address.
TEST_F_TEMP_DISABLED_ON_WINDOWS(
    DockerContainerizerIPv6UserNetworkTest,
    ROOT_DOCKER_USERNETWORK_LaunchIPv6Container)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);

  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  const Offer& offer = offers.get()[0];

  TaskInfo task = createTask(
      offer.slave_id(),
      offer.resources(),
      SLEEP_COMMAND(10000));

  // TODO(tnachen): Use local image to test if possible.
  ContainerInfo containerInfo = createDockerInfo("alpine");

  containerInfo.mutable_docker()->set_network(
      ContainerInfo::DockerInfo::USER);

  // Setup the docker IPv6 network.
  NetworkInfo networkInfo;
  networkInfo.set_name(DOCKER_IPv6_NETWORK);
  containerInfo.add_network_infos()->CopyFrom(networkInfo);

  task.mutable_container()->CopyFrom(containerInfo);

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusStarting;
  Future<TaskStatus> statusRunning;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusStarting))
    .WillOnce(FutureArg<1>(&statusRunning))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(statusStarting, Seconds(60));
  EXPECT_EQ(TASK_STARTING, statusStarting->state());
  AWAIT_READY_FOR(statusRunning, Seconds(60));
  EXPECT_EQ(TASK_RUNNING, statusRunning->state());
  ASSERT_TRUE(statusRunning->has_data());

  // Now verify the ContainerStatus fields in the TaskStatus.
  ASSERT_TRUE(statusRunning->has_container_status());
  EXPECT_TRUE(statusRunning->container_status().has_container_id());
  ASSERT_EQ(1, statusRunning->container_status().network_infos().size());
  ASSERT_EQ(2, statusRunning->container_status().network_infos(0).ip_addresses().size()); // NOLINT(whitespace/line_length)

  Option<string> containerIPv4 = None();
  Option<string> containerIPv6 = None();

  foreach(const NetworkInfo::IPAddress& ipAddress,
          statusRunning->container_status().network_infos(0).ip_addresses()) {
    if (ipAddress.protocol() == NetworkInfo::IPv4) {
      containerIPv4 = ipAddress.ip_address();
    }

    if (ipAddress.protocol() == NetworkInfo::IPv6) {
      containerIPv6 = ipAddress.ip_address();
    }
  }

  ASSERT_SOME(containerIPv4);
  ASSERT_SOME(containerIPv6);

  // Check if container information is exposed through slave's state endpoint.
  Future<http::Response> response = http::get(
      slave.get()->pid,
      "state",
      None(),
      createBasicAuthHeaders(DEFAULT_CREDENTIAL));

  AWAIT_EXPECT_RESPONSE_STATUS_EQ(process::http::OK().status, response);

  Try<JSON::Object> parse = JSON::parse<JSON::Object>(response->body);
  ASSERT_SOME(parse);

  // Verify that the slave state information has the same container
  // status as received in the status update message.
  for (int i = 0; i < 2; i++) {
    Result<JSON::String> protocol = parse->find<JSON::String>(
        "frameworks[0].executors[0].tasks[0].statuses[1]"
        ".container_status.network_infos[0].ip_addresses[" +
        stringify(i) + "].protocol");

    ASSERT_SOME(protocol);

    Result<JSON::String> ip = parse->find<JSON::String>(
        "frameworks[0].executors[0].tasks[0].statuses[1]"
        ".container_status.network_infos[0].ip_addresses[" +
        stringify(i) + "].ip_address");

    ASSERT_SOME(ip);

    if (protocol->value == "IPv4") {
      EXPECT_EQ(ip->value, containerIPv4.get());
    } else {
      EXPECT_EQ(ip->value, containerIPv6.get());
    }

    LOG(INFO) << "IP: " << ip->value;
  }

  ASSERT_TRUE(exists(docker, containerId.get()));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

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

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  ASSERT_FALSE(
    exists(docker, containerId.get(), ContainerState::RUNNING, false));
}


class HungDockerTest : public DockerContainerizerTest
{
public:
  const string testDockerBinary = "docker";
#ifdef __WINDOWS__
  const string testDockerScript = "test-docker.bat";
  const string testDockerEnvFile = "test-docker-env.bat";
#else
  const string testDockerScript = "test-docker.sh";
  const string testDockerEnvFile = "test-docker.env";
#endif // __WINDOWS__

  string commandsEnv;
  string delayEnv;

  slave::Flags CreateSlaveFlags() override
  {
    slave::Flags flags = MesosTest::CreateSlaveFlags();

    flags.docker = path::join(os::getcwd(), testDockerScript);

    return flags;
  }

  void writeEnv()
  {
    // TODO(greggomann): This write operation is not atomic, which means an
    // ill-timed write may cause the shell script to be invoked when this
    // file is in an unintended state. We should make this atomic.

    Try<Nothing> write =
#ifdef __WINDOWS__
      os::write(testDockerEnvFile, commandsEnv + "\r\n" + delayEnv);
#else
      os::write(testDockerEnvFile, commandsEnv + "\n" + delayEnv);
#endif // __WINDOWS__

    ASSERT_SOME(write);
  }

  void setDelayedCommands(const std::vector<string>& commands)
  {
#ifdef __WINDOWS__
    commandsEnv = "set ";
#else
    commandsEnv = "";
#endif // __WINDOWS__

    commandsEnv += "DELAYED_COMMANDS=( ";
    foreach (const string& command, commands) {
      commandsEnv += (command + " ");
    }
    commandsEnv += ")";

    writeEnv();
  }

  void setDelay(const int seconds)
  {
#ifdef __WINDOWS__
    delayEnv = "set ";
#else
    delayEnv = "";
#endif // __WINDOWS__

    delayEnv += "DELAY_SECONDS=" + stringify(seconds);

    writeEnv();
  }

  void SetUp() override
  {
    DockerContainerizerTest::SetUp();

    // Write a wrapper script which allows us to delay Docker commands.
#ifdef __WINDOWS__
    const string dockerScriptText =
      "@echo off\r\n"
      "setlocal enabledelayedexpansion\r\n"
      "call \"" + path::join(os::getcwd(), testDockerEnvFile) + "\"\r\n"
      "set ACTIVE_COMMAND=%3\r\n"
      "if not defined DELAYED_COMMANDS set DELAYED_COMMANDS=()\r\n"
      "for %%G in %DELAYED_COMMANDS% do (\r\n"
      "  if %ACTIVE_COMMAND% == %%G (\r\n"
      "    ping -n %DELAY_SECONDS% 127.0.0.1 > NUL\r\n"
      "  )\r\n"
      ")\r\n" +
      testDockerBinary + " %*\r\n";
#else
    const string dockerScriptText =
      "#!/usr/bin/env bash\n"
      "source " + stringify(path::join(os::getcwd(), testDockerEnvFile)) + "\n"
      "ACTIVE_COMMAND=$3\n"
      "for DELAYED_COMMAND in \"${DELAYED_COMMANDS[@]}\"; do\n"
      "  if [ \"$ACTIVE_COMMAND\" == \"$DELAYED_COMMAND\" ]; then\n"
      "    sleep $DELAY_SECONDS\n"
      "  fi\n"
      "done\n" +
      testDockerBinary + " \"$@\"\n";
#endif // __WINDOWS__

    Try<Nothing> write = os::write(testDockerScript, dockerScriptText);
    ASSERT_SOME(write);

#ifndef __WINDOWS__
    Try<Nothing> chmod = os::chmod(
        testDockerScript, S_IRWXU | S_IRGRP | S_IXGRP | S_IROTH | S_IXOTH);
    ASSERT_SOME(chmod);
#endif // __WINDOWS__

    // Set a very long delay by default to simulate an indefinitely
    // hung Docker daemon.
    setDelay(999999);
  }
};


TEST_F(HungDockerTest, ROOT_DOCKER_InspectHungDuringPull)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  slave::Flags flags = CreateSlaveFlags();

  // When the 'executor_registration_timeout' elapses, the agent will destroy
  // the container whose 'docker pull' command is stuck. This should cause the
  // launch to fail and the terminal task status update to be sent.
  flags.executor_registration_timeout = Milliseconds(100);

  MockDocker* mockDocker =
    new MockDocker(flags.docker, tests::flags.docker_socket);
  Shared<Docker> docker(mockDocker);

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

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

  Future<FrameworkID> frameworkId;
  EXPECT_CALL(sched, registered(&driver, _, _))
    .WillOnce(FutureArg<1>(&frameworkId));

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

  driver.start();

  AWAIT_READY(frameworkId);

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->empty());

  // Causing the 'docker inspect' call preceding the container pull to hang
  // should result in a TASK_FAILED update.
  setDelayedCommands({"inspect"});

  const Offer& offer = offers.get()[0];

  TaskInfo task = createTask(
      offer.slave_id(),
      offer.resources(),
      SLEEP_COMMAND(1000));

  // TODO(tnachen): Use local image to test if possible.
  task.mutable_container()->CopyFrom(createDockerInfo(DOCKER_TEST_IMAGE));

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<TaskStatus> statusFailed;
  EXPECT_CALL(sched, statusUpdate(&driver, _))
    .WillOnce(FutureArg<1>(&statusFailed))
    .WillRepeatedly(DoDefault());

  driver.launchTasks(offers.get()[0].id(), {task});

  AWAIT_READY(statusFailed);
  EXPECT_EQ(TASK_FAILED, statusFailed->state());
  EXPECT_EQ(
      TaskStatus::REASON_CONTAINER_LAUNCH_FAILED,
      statusFailed->reason());

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


// This test is disabled on windows due to the bash-specific
// command used in the task below.
TEST_F_TEMP_DISABLED_ON_WINDOWS(
    DockerContainerizerTest, ROOT_DOCKER_OverrideKillPolicy)
{
  Try<Owned<cluster::Master>> master = StartMaster();
  ASSERT_SOME(master);

  MockDocker* mockDocker =
    new MockDocker(tests::flags.docker, tests::flags.docker_socket);

  Shared<Docker> docker(mockDocker);

  slave::Flags flags = CreateSlaveFlags();

  Fetcher fetcher(flags);

  Try<ContainerLogger*> logger =
    ContainerLogger::create(flags.container_logger);

  ASSERT_SOME(logger);

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

  MockDockerContainerizer dockerContainerizer(
      flags,
      &fetcher,
      Owned<ContainerLogger>(logger.get()),
      docker);

  Owned<MasterDetector> detector = master.get()->createDetector();

  Try<Owned<cluster::Slave>> slave =
    StartSlave(detector.get(), &dockerContainerizer, flags);
  ASSERT_SOME(slave);

  AWAIT_READY(slaveRegisteredMessage);

  auto scheduler = std::make_shared<v1::MockHTTPScheduler>();

  EXPECT_CALL(*scheduler, connected(_))
    .WillOnce(v1::scheduler::SendSubscribe(v1::DEFAULT_FRAMEWORK_INFO));

  Future<v1::scheduler::Event::Subscribed> subscribed;
  EXPECT_CALL(*scheduler, subscribed(_, _))
    .WillOnce(FutureArg<1>(&subscribed));

  Future<v1::scheduler::Event::Offers> offers;
  EXPECT_CALL(*scheduler, offers(_, _))
    .WillOnce(FutureArg<1>(&offers))
    .WillRepeatedly(Return());

  EXPECT_CALL(*scheduler, heartbeat(_))
    .WillRepeatedly(Return()); // Ignore heartbeats.

  v1::scheduler::TestMesos mesos(
      master.get()->pid,
      ContentType::PROTOBUF,
      scheduler);

  AWAIT_READY(subscribed);
  v1::FrameworkID frameworkId(subscribed->framework_id());

  AWAIT_READY(offers);
  ASSERT_FALSE(offers->offers().empty());

  const v1::Offer& offer = offers->offers(0);
  const v1::AgentID& agentId = offer.agent_id();

  Try<v1::Resources> parsed =
    v1::Resources::parse("cpus:0.1;mem:32;disk:32");

  ASSERT_SOME(parsed);

  v1::Resources resources = parsed.get();

  // Create a task which ignores SIGTERM so that we can detect
  // when the task receives SIGKILL.
  v1::TaskInfo taskInfo = v1::createTask(
      agentId,
      resources,
      "trap \"echo 'SIGTERM received'\" SIGTERM; sleep 999999");

  // TODO(tnachen): Use local image to test if possible.
  taskInfo.mutable_container()->CopyFrom(
      evolve(createDockerInfo(DOCKER_TEST_IMAGE)));

  {
    // Set a long grace period on the task's kill policy so that we
    // can detect if the override is effective.
    mesos::v1::DurationInfo gracePeriod;
    gracePeriod.set_nanoseconds(Minutes(10).ns());

    mesos::v1::KillPolicy killPolicy;
    killPolicy.mutable_grace_period()->CopyFrom(gracePeriod);

    taskInfo.mutable_kill_policy()->CopyFrom(killPolicy);
  }

  Future<ContainerID> containerId;
  EXPECT_CALL(dockerContainerizer, launch(_, _, _, _))
    .WillOnce(DoAll(FutureArg<0>(&containerId),
                    Invoke(&dockerContainerizer,
                           &MockDockerContainerizer::_launch)));

  Future<v1::scheduler::Event::Update> startingUpdate;
  Future<v1::scheduler::Event::Update> runningUpdate;
  EXPECT_CALL(*scheduler, update(_, _))
    .WillOnce(DoAll(
        FutureArg<1>(&startingUpdate),
        v1::scheduler::SendAcknowledge(frameworkId, agentId)))
    .WillOnce(DoAll(
        FutureArg<1>(&runningUpdate),
        v1::scheduler::SendAcknowledge(frameworkId, agentId)));

  mesos.send(
      v1::createCallAccept(
          frameworkId,
          offer,
          {v1::LAUNCH({taskInfo})}));

  AWAIT_READY_FOR(containerId, Seconds(60));
  AWAIT_READY_FOR(startingUpdate, Seconds(60));
  EXPECT_EQ(v1::TASK_STARTING, startingUpdate->status().state());
  AWAIT_READY_FOR(runningUpdate, Seconds(60));
  EXPECT_EQ(v1::TASK_RUNNING, runningUpdate->status().state());

  ASSERT_TRUE(
    exists(docker, containerId.get(), ContainerState::RUNNING));

  Future<v1::scheduler::Event::Update> killedUpdate;
  EXPECT_CALL(*scheduler, update(_, _))
    .WillOnce(FutureArg<1>(&killedUpdate));

  Future<Option<ContainerTermination>> termination =
    dockerContainerizer.wait(containerId.get());

  {
    // Set a short grace period on the kill call so that we
    // can detect if the override is effective.
    mesos::v1::DurationInfo gracePeriod;
    gracePeriod.set_nanoseconds(100);

    mesos::v1::KillPolicy killPolicy;
    killPolicy.mutable_grace_period()->CopyFrom(gracePeriod);

    mesos.send(
        v1::createCallKill(
            frameworkId,
            taskInfo.task_id(),
            agentId,
            killPolicy));
  }

  AWAIT_READY(killedUpdate);
  EXPECT_EQ(v1::TASK_KILLED, killedUpdate->status().state());

  AWAIT_READY(termination);
  EXPECT_SOME(termination.get());

  // Even though the task is killed, the executor should exit gracefully.
  ASSERT_TRUE(termination.get()->has_status());
  EXPECT_EQ(0, termination.get()->status());

  ASSERT_FALSE(
      exists(docker, containerId.get(), ContainerState::RUNNING, false));
}

} // namespace tests {
} // namespace internal {
} // namespace mesos {