[9/9] mesos git commit: Added a test for nested container resource limitations.

[jp...@apache.org]

Added a test for nested container resource limitations.

Added a test to verify that resource limitations incurred by a nested
container are propagated all the way back to the scheduler through a
`TASK_FAILED` status update.

Review: https://reviews.apache.org/r/62648/


Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/cc29c27e
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/cc29c27e
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/cc29c27e

Branch: refs/heads/master
Commit: cc29c27ee597fbe79206d890d695c0a2e64c46aa
Parents: 08d9468
Author: James Peach <jp...@apache.org>
Authored: Sun Oct 15 15:21:59 2017 -0700
Committer: James Peach <jp...@apache.org>
Committed: Sun Oct 15 15:21:59 2017 -0700

----------------------------------------------------------------------
 src/tests/default_executor_tests.cpp | 116 ++++++++++++++++++++++++++++++
 1 file changed, 116 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/mesos/blob/cc29c27e/src/tests/default_executor_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/default_executor_tests.cpp b/src/tests/default_executor_tests.cpp
index 2b4c643..6831201 100644
--- a/src/tests/default_executor_tests.cpp
+++ b/src/tests/default_executor_tests.cpp
@@ -77,6 +77,7 @@ using testing::WithParamInterface;
 using mesos::internal::slave::Containerizer;
 using mesos::internal::slave::Fetcher;
 using mesos::internal::slave::MesosContainerizer;
+using mesos::internal::slave::Slave;
 
 using mesos::slave::ContainerTermination;
 
@@ -1362,6 +1363,121 @@ TEST_P(DefaultExecutorTest, ROOT_MultiTaskgroupSharePidNamespace)
 #endif // __linux__
 
 
+// This test verifies that a resource limitation incurred on a nested
+// container is propagated all the way up to the scheduler.
+TEST_P_TEMP_DISABLED_ON_WINDOWS(
+    DefaultExecutorTest, ResourceLimitation)
+{
+  Try<Owned<cluster::Master>> master = StartMaster();
+  ASSERT_SOME(master);
+
+  slave::Flags flags = CreateSlaveFlags();
+  flags.containerizers = GetParam();
+  flags.enforce_container_disk_quota = true;
+  flags.container_disk_watch_interval = Milliseconds(1);
+  flags.isolation = "disk/du";
+
+  Owned<MasterDetector> detector = master.get()->createDetector();
+  Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), flags);
+  ASSERT_SOME(slave);
+
+  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());
+
+  v1::Resources resources =
+    v1::Resources::parse("cpus:0.1;mem:32;disk:10").get();
+
+  v1::ExecutorInfo executorInfo = v1::createExecutorInfo(
+      v1::DEFAULT_EXECUTOR_ID,
+      None(),
+      resources,
+      v1::ExecutorInfo::DEFAULT,
+      frameworkId);
+
+  AWAIT_READY(offers);
+  ASSERT_FALSE(offers->offers().empty());
+
+  const v1::Offer& offer = offers->offers(0);
+  const v1::AgentID& agentId = offer.agent_id();
+
+  Future<v1::scheduler::Event::Update> running;
+  Future<v1::scheduler::Event::Update> failed;
+  EXPECT_CALL(*scheduler, update(_, _))
+    .WillOnce(
+        DoAll(
+            FutureArg<1>(&running),
+            v1::scheduler::SendAcknowledge(frameworkId, agentId)))
+    .WillOnce(
+        DoAll(
+            FutureArg<1>(&failed),
+            v1::scheduler::SendAcknowledge(frameworkId, agentId)));
+
+  // Since we requested 10MB each for the task and the executor,
+  // writing 30MB will violate our disk resource limit.
+  v1::TaskInfo taskInfo = v1::createTask(
+      agentId,
+      resources,
+      "dd if=/dev/zero of=dd.out bs=1048576 count=30; sleep 1000");
+
+  mesos.send(
+      v1::createCallAccept(
+          frameworkId,
+          offer,
+          {v1::LAUNCH_GROUP(
+              executorInfo, v1::createTaskGroupInfo({taskInfo}))}));
+
+  Future<Nothing> ack =
+    FUTURE_DISPATCH(_, &Slave::_statusUpdateAcknowledgement);
+
+  AWAIT_READY(running);
+  AWAIT_READY(ack);
+
+  EXPECT_EQ(TASK_RUNNING, running->status().state());
+  EXPECT_EQ(taskInfo.task_id(), running->status().task_id());
+
+  AWAIT_READY(failed);
+
+  // We expect the failure to be a disk limitation that tells us something
+  // about the disk resources.
+  EXPECT_EQ(TASK_FAILED, failed->status().state());
+  EXPECT_EQ(
+      TaskStatus::REASON_CONTAINER_LIMITATION_DISK,
+      failed->status().reason());
+
+  EXPECT_EQ(taskInfo.task_id(), failed->status().task_id());
+  ASSERT_TRUE(failed->status().has_limitation());
+  EXPECT_GT(failed->status().limitation().resources().size(), 0);
+
+  foreach (const v1::Resource& resource,
+           failed->status().limitation().resources()) {
+    EXPECT_EQ("disk", resource.name());
+    EXPECT_EQ(mesos::v1::Value::SCALAR, resource.type());
+  }
+}
+
+
 struct LauncherAndIsolationParam
 {
   LauncherAndIsolationParam(const string& _launcher, const string& _isolation)