[2/3] mesos git commit: Added task validation for revocable resources.

[vi...@apache.org]

Added task validation for revocable resources.

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


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

Branch: refs/heads/master
Commit: c6a9fff39e6ab625560894cf727bce54cd785113
Parents: 7840d32
Author: Vinod Kone <vi...@gmail.com>
Authored: Mon Jun 1 10:14:17 2015 -0700
Committer: Vinod Kone <vi...@gmail.com>
Committed: Wed Jun 10 16:55:54 2015 -0700

----------------------------------------------------------------------
 src/master/validation.cpp             | 85 ++++++++++++++++-----------
 src/master/validation.hpp             |  9 +++
 src/slave/slave.cpp                   |  8 ++-
 src/tests/master_validation_tests.cpp | 92 ++++++++++++++++++++++++++++++
 4 files changed, 157 insertions(+), 37 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/mesos/blob/c6a9fff3/src/master/validation.cpp
----------------------------------------------------------------------
diff --git a/src/master/validation.cpp b/src/master/validation.cpp
index 6c9dc04..889387d 100644
--- a/src/master/validation.cpp
+++ b/src/master/validation.cpp
@@ -187,6 +187,8 @@ Option<Error> validate(const RepeatedPtrField<Resource>& resources)
 
 namespace task {
 
+namespace internal {
+
 // Validates that a task id is valid, i.e., contains only valid
 // characters.
 Option<Error> validateTaskID(const TaskInfo& task)
@@ -295,33 +297,6 @@ Option<Error> validateCheckpoint(Framework* framework, Slave* slave)
   return None();
 }
 
-// Validates that the resources specified by the framework are valid.
-Option<Error> validateResources(const TaskInfo& task)
-{
-  Option<Error> error = resource::validate(task.resources());
-  if (error.isSome()) {
-    return Error("Task uses invalid resources: " + error.get().message);
-  }
-
-  Resources total = task.resources();
-
-  if (task.has_executor()) {
-    error = resource::validate(task.executor().resources());
-    if (error.isSome()) {
-      return Error("Executor uses invalid resources: " + error.get().message);
-    }
-
-    total += task.executor().resources();
-  }
-
-  error = resource::validateUniquePersistenceID(total);
-  if (error.isSome()) {
-    return error;
-  }
-
-  return None();
-}
-
 
 // Validates that the task and the executor are using proper amount of
 // resources. For instance, the used resources by a task on each slave
@@ -390,6 +365,47 @@ Option<Error> validateResourceUsage(
 }
 
 
+// Validates that the resources specified by the task are valid.
+Option<Error> validateResources(const TaskInfo& task)
+{
+  Option<Error> error = resource::validate(task.resources());
+  if (error.isSome()) {
+    return Error("Task uses invalid resources: " + error.get().message);
+  }
+
+  Resources total = task.resources();
+
+  if (task.has_executor()) {
+    error = resource::validate(task.executor().resources());
+    if (error.isSome()) {
+      return Error("Executor uses invalid resources: " + error.get().message);
+    }
+
+    total += task.executor().resources();
+  }
+
+  // A task and its executor can either use non-revocable resources
+  // or revocable resources of a given name but not both.
+  foreach (const string& name, total.names()) {
+    Resources resources = total.get(name);
+    if (!resources.revocable().empty() &&
+        resources != resources.revocable()) {
+      return Error("Task (and its executor, if exists) uses both revocable and"
+                   " non-revocable " + name);
+    }
+  }
+
+  error = resource::validateUniquePersistenceID(total);
+  if (error.isSome()) {
+    return error;
+  }
+
+  return None();
+}
+
+} // namespace internal {
+
+
 Option<Error> validate(
     const TaskInfo& task,
     Framework* framework,
@@ -404,13 +420,14 @@ Option<Error> validate(
   // assumes that ExecutorInfo is valid which is verified by
   // 'validateExecutorInfo'.
   vector<lambda::function<Option<Error>(void)>> validators = {
-    lambda::bind(validateTaskID, task),
-    lambda::bind(validateUniqueTaskID, task, framework),
-    lambda::bind(validateSlaveID, task, slave),
-    lambda::bind(validateExecutorInfo, task, framework, slave),
-    lambda::bind(validateCheckpoint, framework, slave),
-    lambda::bind(validateResources, task),
-    lambda::bind(validateResourceUsage, task, framework, slave, offered)
+    lambda::bind(internal::validateTaskID, task),
+    lambda::bind(internal::validateUniqueTaskID, task, framework),
+    lambda::bind(internal::validateSlaveID, task, slave),
+    lambda::bind(internal::validateExecutorInfo, task, framework, slave),
+    lambda::bind(internal::validateCheckpoint, framework, slave),
+    lambda::bind(internal::validateResources, task),
+    lambda::bind(
+        internal::validateResourceUsage, task, framework, slave, offered)
   };
 
   // TODO(benh): Add a validateHealthCheck function.

http://git-wip-us.apache.org/repos/asf/mesos/blob/c6a9fff3/src/master/validation.hpp
----------------------------------------------------------------------
diff --git a/src/master/validation.hpp b/src/master/validation.hpp
index a74e844..469d6f5 100644
--- a/src/master/validation.hpp
+++ b/src/master/validation.hpp
@@ -59,6 +59,15 @@ Option<Error> validate(
     Slave* slave,
     const Resources& offered);
 
+
+// Functions in this namespace are only exposed for testing.
+namespace internal {
+
+// Validates resources of the task and executor (if present).
+Option<Error> validateResources(const TaskInfo& task);
+
+} // namespace internal {
+
 } // namespace task {
 
 

http://git-wip-us.apache.org/repos/asf/mesos/blob/c6a9fff3/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index b9fb929..2eef391 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -3098,9 +3098,11 @@ ExecutorInfo Slave::getExecutorInfo(
 
     // Add an allowance for the command executor. This does lead to a
     // small overcommit of resources.
-    //
-    // TODO(bmahler): If revocable resources are used, this leads to
-    // mixing of resources.
+    // TODO(vinod): If a task is using revocable resources, mark the
+    // corresponding executor resource (e.g., cpus) to be also
+    // revocable. Currently, it is OK because the containerizer is
+    // given task + executor resources on task launch resulting in
+    // the container being correctly marked as revocable.
     executor.mutable_resources()->MergeFrom(
         Resources::parse(
           "cpus:" + stringify(DEFAULT_EXECUTOR_CPUS) + ";" +

http://git-wip-us.apache.org/repos/asf/mesos/blob/c6a9fff3/src/tests/master_validation_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_validation_tests.cpp b/src/tests/master_validation_tests.cpp
index dc9e91e..dbd5118 100644
--- a/src/tests/master_validation_tests.cpp
+++ b/src/tests/master_validation_tests.cpp
@@ -1100,6 +1100,98 @@ TEST_F(TaskValidationTest, ExecutorInfoDiffersOnDifferentSlaves)
   Shutdown();
 }
 
+
+// This test verifies that a task is not allowed to mix revocable and
+// non-revocable resources.
+TEST_F(TaskValidationTest, TaskUsesRevocableResources)
+{
+  TaskInfo task;
+  task.set_name("");
+  task.mutable_task_id()->set_value("task");
+  task.mutable_slave_id()->set_value("slave");
+
+  // Non-revocable cpus.
+  Resource cpus;
+  cpus.set_name("cpus");
+  cpus.set_type(Value::SCALAR);
+  cpus.mutable_scalar()->set_value(2);
+
+  // A task with only non-revocable cpus is valid.
+  task.add_resources()->CopyFrom(cpus);
+  EXPECT_NONE(task::internal::validateResources(task));
+
+  // Revocable cpus.
+  Resource revocableCpus = cpus;
+  revocableCpus.mutable_revocable();
+
+  // A task with only revocable cpus is valid.
+  task.clear_resources();
+  task.add_resources()->CopyFrom(revocableCpus);
+  EXPECT_NONE(task::internal::validateResources(task));
+
+  // A task with both revocable and non-revocable cpus is invalid.
+  task.clear_resources();
+  task.add_resources()->CopyFrom(cpus);
+  task.add_resources()->CopyFrom(revocableCpus);
+  EXPECT_SOME(task::internal::validateResources(task));
+}
+
+
+// This test verifies that a task and its executor are not allowed to
+// mix revocable and non-revocable resources.
+TEST_F(TaskValidationTest, TaskAndExecutorUseRevocableResources)
+{
+  TaskInfo task;
+  task.set_name("");
+  task.mutable_task_id()->set_value("task");
+  task.mutable_slave_id()->set_value("slave");
+
+  ExecutorInfo executor = DEFAULT_EXECUTOR_INFO;
+
+  // Non-revocable cpus.
+  Resource cpus;
+  cpus.set_name("cpus");
+  cpus.set_type(Value::SCALAR);
+  cpus.mutable_scalar()->set_value(2);
+
+  // A task and executor with only non-revocable cpus is valid.
+  task.add_resources()->CopyFrom(cpus);
+  executor.add_resources()->CopyFrom(cpus);
+  task.mutable_executor()->CopyFrom(executor);
+  EXPECT_NONE(task::internal::validateResources(task));
+
+  // Revocable cpus.
+  Resource revocableCpus = cpus;
+  revocableCpus.mutable_revocable();
+
+  // A task and executor with only revocable cpus is valid.
+  task.clear_resources();
+  task.add_resources()->CopyFrom(revocableCpus);
+  executor.clear_resources();
+  executor.add_resources()->CopyFrom(revocableCpus);
+  task.mutable_executor()->CopyFrom(executor);
+  EXPECT_NONE(task::internal::validateResources(task));
+
+  // A task with revocable cpus and its executor with non-revocable
+  // cpus is invalid.
+  task.clear_resources();
+  task.add_resources()->CopyFrom(revocableCpus);
+  executor.clear_resources();
+  executor.add_resources()->CopyFrom(cpus);
+  task.mutable_executor()->CopyFrom(executor);
+  EXPECT_SOME(task::internal::validateResources(task));
+
+  // A task with non-revocable cpus and its executor with
+  // non-revocable cpus is invalid.
+  task.clear_resources();
+  task.add_resources()->CopyFrom(cpus);
+  executor.clear_resources();
+  executor.add_resources()->CopyFrom(revocableCpus);
+  task.mutable_executor()->CopyFrom(executor);
+  EXPECT_SOME(task::internal::validateResources(task));
+}
+
+
 // TODO(jieyu): Add tests for checking duplicated persistence ID
 // against offered resources.