You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by vi...@apache.org on 2016/09/19 20:31:59 UTC
[1/4] mesos git commit: Refactored commandScheduler in
cli/execute.cpp to take TaskInfo.
Repository: mesos
Updated Branches:
refs/heads/master be5b76bbd -> f162ade0f
Refactored commandScheduler in cli/execute.cpp to take TaskInfo.
This refactoring was needed for better accomodating pod support
for mesos-execute.
Review: https://reviews.apache.org/r/51978/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/4e8ac999
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/4e8ac999
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/4e8ac999
Branch: refs/heads/master
Commit: 4e8ac999fe871fd8c085c43a9a9f940c56e83e26
Parents: be5b76b
Author: Abhishek Dasgupta <a1...@linux.vnet.ibm.com>
Authored: Mon Sep 19 13:31:29 2016 -0700
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Sep 19 13:31:29 2016 -0700
----------------------------------------------------------------------
src/cli/execute.cpp | 350 ++++++++++++++++++++++-------------------------
1 file changed, 167 insertions(+), 183 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/4e8ac999/src/cli/execute.cpp
----------------------------------------------------------------------
diff --git a/src/cli/execute.cpp b/src/cli/execute.cpp
index 525c898..f2e28e4 100644
--- a/src/cli/execute.cpp
+++ b/src/cli/execute.cpp
@@ -201,8 +201,8 @@ public:
"Example:\n"
"[\n"
" {\n"
- " \"container_path\":\"/path/to/container\"\n"
- " \"mode\":\"RW\"\n"
+ " \"container_path\":\"/path/to/container\",\n"
+ " \"mode\":\"RW\",\n"
" \"source\":\n"
" {\n"
" \"docker_volume\":\n"
@@ -253,35 +253,15 @@ public:
CommandScheduler(
const FrameworkInfo& _frameworkInfo,
const string& _master,
- const string& _name,
- const bool _shell,
- const Option<string>& _command,
- const Option<hashmap<string, string>>& _environment,
- const string& _resources,
- const Option<string>& _uri,
- const Option<string>& _appcImage,
- const Option<string>& _dockerImage,
- const vector<Volume>& _volumes,
- const string& _containerizer,
const Option<Duration>& _killAfter,
- const Option<string>& _networks,
- const Option<Credential> _credential)
+ const Option<Credential>& _credential,
+ const TaskInfo& _task)
: state(DISCONNECTED),
frameworkInfo(_frameworkInfo),
master(_master),
- name(_name),
- shell(_shell),
- command(_command),
- environment(_environment),
- resources(_resources),
- uri(_uri),
- appcImage(_appcImage),
- dockerImage(_dockerImage),
- volumes(_volumes),
- containerizer(_containerizer),
killAfter(_killAfter),
- networks(_networks),
credential(_credential),
+ task(_task),
launched(false) {}
virtual ~CommandScheduler() {}
@@ -355,26 +335,17 @@ protected:
{
CHECK_EQ(SUBSCRIBED, state);
- static const Try<Resources> TASK_RESOURCES = Resources::parse(resources);
-
- if (TASK_RESOURCES.isError()) {
- EXIT(EXIT_FAILURE)
- << "Failed to parse resources '" << resources << "': "
- << TASK_RESOURCES.error();
- }
-
foreach (const Offer& offer, offers) {
Resources offered = offer.resources();
- if (!launched && offered.flatten().contains(TASK_RESOURCES.get())) {
- TaskInfo task;
- task.set_name(name);
- task.mutable_task_id()->set_value(name);
- task.mutable_agent_id()->MergeFrom(offer.agent_id());
+ TaskInfo _task = task;
+
+ if (!launched && offered.flatten().contains(_task.resources())) {
+ _task.mutable_agent_id()->MergeFrom(offer.agent_id());
// Takes resources first from the specified role, then from '*'.
Try<Resources> flattened =
- TASK_RESOURCES.get().flatten(frameworkInfo.role());
+ Resources(_task.resources()).flatten(frameworkInfo.role());
// `frameworkInfo.role()` must be valid as it's allowed to register.
CHECK_SOME(flattened);
@@ -382,46 +353,7 @@ protected:
CHECK_SOME(resources);
- task.mutable_resources()->CopyFrom(resources.get());
-
- CommandInfo* commandInfo = task.mutable_command();
-
- if (shell) {
- CHECK_SOME(command);
-
- commandInfo->set_shell(true);
- commandInfo->set_value(command.get());
- } else {
- // TODO(gilbert): Treat 'command' as executable value and arguments.
- commandInfo->set_shell(false);
- }
-
- if (environment.isSome()) {
- Environment* environment_ = commandInfo->mutable_environment();
- foreachpair (
- const string& name, const string& value, environment.get()) {
- Environment::Variable* environmentVariable =
- environment_->add_variables();
-
- environmentVariable->set_name(name);
- environmentVariable->set_value(value);
- }
- }
-
- if (uri.isSome()) {
- task.mutable_command()->add_uris()->set_value(uri.get());
- }
-
- Result<ContainerInfo> containerInfo = getContainerInfo();
-
- if (containerInfo.isError()){
- EXIT(EXIT_FAILURE) << containerInfo.error();
- return;
- }
-
- if (containerInfo.isSome()) {
- task.mutable_container()->CopyFrom(containerInfo.get());
- }
+ _task.mutable_resources()->CopyFrom(resources.get());
Call call;
call.set_type(Call::ACCEPT);
@@ -435,11 +367,11 @@ protected:
Offer::Operation* operation = accept->add_operations();
operation->set_type(Offer::Operation::LAUNCH);
- operation->mutable_launch()->add_task_infos()->CopyFrom(task);
+ operation->mutable_launch()->add_task_infos()->CopyFrom(_task);
mesos->send(call);
- cout << "Submitted task '" << name << "' to agent '"
+ cout << "Submitted task '" << _task.name() << "' to agent '"
<< offer.agent_id() << "'" << endl;
launched = true;
@@ -512,7 +444,8 @@ protected:
void update(const TaskStatus& status)
{
CHECK_EQ(SUBSCRIBED, state);
- CHECK_EQ(name, status.task_id().value());
+
+ CHECK_EQ(task.name(), status.task_id().value());
cout << "Received status update " << status.state()
<< " for task '" << status.task_id() << "'" << endl;
@@ -566,112 +499,110 @@ private:
SUBSCRIBED
} state;
- // TODO(jojy): Consider breaking down the method for each 'containerizer'.
- Result<ContainerInfo> getContainerInfo() const
- {
- if (containerizer.empty()) {
- return None();
- }
+ FrameworkInfo frameworkInfo;
+ const string master;
+ const Option<Duration> killAfter;
+ const Option<Credential> credential;
+ const TaskInfo task;
+ bool launched;
+ Owned<Mesos> mesos;
+};
+
- ContainerInfo containerInfo;
+// TODO(jojy): Consider breaking down the method for each 'containerizer'.
+static Result<ContainerInfo> getContainerInfo(
+ const string& containerizer,
+ const Option<vector<Volume>>& volumes,
+ const Option<string>& networks,
+ const Option<string>& appcImage,
+ const Option<string>& dockerImage)
+{
+ if (containerizer.empty()) {
+ return None();
+ }
+
+ ContainerInfo containerInfo;
- foreach (const Volume& volume, volumes) {
+ if (volumes.isSome()) {
+ foreach (const Volume& volume, volumes.get()) {
containerInfo.add_volumes()->CopyFrom(volume);
}
+ }
- // Mesos containerizer supports 'appc' and 'docker' images.
- if (containerizer == "mesos") {
- if (dockerImage.isNone() && appcImage.isNone() &&
- (networks.isNone() || networks->empty()) &&
- volumes.empty()) {
- return None();
- }
+ // Mesos containerizer supports 'appc' and 'docker' images.
+ if (containerizer == "mesos") {
+ if (dockerImage.isNone() && appcImage.isNone() &&
+ (networks.isNone() || networks->empty()) &&
+ (volumes.isNone() || volumes->empty())) {
+ return None();
+ }
- containerInfo.set_type(ContainerInfo::MESOS);
+ containerInfo.set_type(ContainerInfo::MESOS);
- if (dockerImage.isSome()) {
- Image* image = containerInfo.mutable_mesos()->mutable_image();
- image->set_type(Image::DOCKER);
- image->mutable_docker()->set_name(dockerImage.get());
- } else if (appcImage.isSome()) {
- Image::Appc appc;
+ if (dockerImage.isSome()) {
+ Image* image = containerInfo.mutable_mesos()->mutable_image();
+ image->set_type(Image::DOCKER);
+ image->mutable_docker()->set_name(dockerImage.get());
+ } else if (appcImage.isSome()) {
+ Image::Appc appc;
- appc.set_name(appcImage.get());
+ appc.set_name(appcImage.get());
- // TODO(jojy): Labels are hard coded right now. Consider
- // adding label flags for customization.
- Label arch;
- arch.set_key("arch");
- arch.set_value("amd64");
+ // TODO(jojy): Labels are hard coded right now. Consider
+ // adding label flags for customization.
+ Label arch;
+ arch.set_key("arch");
+ arch.set_value("amd64");
- Label os;
- os.set_key("os");
- os.set_value("linux");
+ Label os;
+ os.set_key("os");
+ os.set_value("linux");
- Labels labels;
- labels.add_labels()->CopyFrom(os);
- labels.add_labels()->CopyFrom(arch);
+ Labels labels;
+ labels.add_labels()->CopyFrom(os);
+ labels.add_labels()->CopyFrom(arch);
- appc.mutable_labels()->CopyFrom(labels);
+ appc.mutable_labels()->CopyFrom(labels);
- Image* image = containerInfo.mutable_mesos()->mutable_image();
- image->set_type(Image::APPC);
- image->mutable_appc()->CopyFrom(appc);
- }
+ Image* image = containerInfo.mutable_mesos()->mutable_image();
+ image->set_type(Image::APPC);
+ image->mutable_appc()->CopyFrom(appc);
+ }
- if (networks.isSome() && !networks->empty()) {
- foreach (const string& network,
- strings::tokenize(networks.get(), ",")) {
- containerInfo.add_network_infos()->set_name(network);
- }
+ if (networks.isSome() && !networks->empty()) {
+ foreach (const string& network,
+ strings::tokenize(networks.get(), ",")) {
+ containerInfo.add_network_infos()->set_name(network);
}
+ }
- return containerInfo;
- } else if (containerizer == "docker") {
- // 'docker' containerizer only supports 'docker' images.
- if (dockerImage.isNone()) {
- return Error("'Docker' containerizer requires docker image name");
- }
+ return containerInfo;
+ } else if (containerizer == "docker") {
+ // 'docker' containerizer only supports 'docker' images.
+ if (dockerImage.isNone()) {
+ return Error("'Docker' containerizer requires docker image name");
+ }
- containerInfo.set_type(ContainerInfo::DOCKER);
- containerInfo.mutable_docker()->set_image(dockerImage.get());
+ containerInfo.set_type(ContainerInfo::DOCKER);
+ containerInfo.mutable_docker()->set_image(dockerImage.get());
- if (networks.isSome() && !networks->empty()) {
- vector<string> tokens = strings::tokenize(networks.get(), ",");
- if (tokens.size() > 1) {
- EXIT(EXIT_FAILURE)
- << "'Docker' containerizer can only support a single network";
- } else {
- containerInfo.mutable_docker()->set_network(
- ContainerInfo::DockerInfo::USER);
- containerInfo.add_network_infos()->set_name(tokens.front());
- }
+ if (networks.isSome() && !networks->empty()) {
+ vector<string> tokens = strings::tokenize(networks.get(), ",");
+ if (tokens.size() > 1) {
+ EXIT(EXIT_FAILURE)
+ << "'Docker' containerizer can only support a single network";
+ } else {
+ containerInfo.mutable_docker()->set_network(
+ ContainerInfo::DockerInfo::USER);
+ containerInfo.add_network_infos()->set_name(tokens.front());
}
-
- return containerInfo;
}
- return Error("Unsupported containerizer: " + containerizer);
+ return containerInfo;
}
- FrameworkInfo frameworkInfo;
- const string master;
- const string name;
- bool shell;
- const Option<string> command;
- const Option<hashmap<string, string>> environment;
- const string resources;
- const Option<string> uri;
- const Option<string> appcImage;
- const Option<string> dockerImage;
- const vector<Volume> volumes;
- const string containerizer;
- const Option<Duration> killAfter;
- const Option<string> networks;
- const Option<Credential> credential;
- bool launched;
- Owned<Mesos> mesos;
-};
+ return Error("Unsupported containerizer: " + containerizer);
+}
int main(int argc, char** argv)
@@ -835,20 +766,25 @@ int main(int argc, char** argv)
}
}
- vector<Volume> volumes;
+ Option<vector<Volume>> volumes = None();
+
if (flags.volumes.isSome()) {
- Try<RepeatedPtrField<Volume>> _volumes =
+ Try<RepeatedPtrField<Volume>> parse =
::protobuf::parse<RepeatedPtrField<Volume>>(flags.volumes.get());
- if (_volumes.isError()) {
+ if (parse.isError()) {
cerr << "Failed to convert '--volumes' to protobuf: "
- << _volumes.error() << endl;
+ << parse.error() << endl;
return EXIT_FAILURE;
}
- foreach (const Volume& volume, _volumes.get()) {
- volumes.push_back(volume);
+ vector<Volume> _volumes;
+
+ foreach (const Volume& volume, parse.get()) {
+ _volumes.push_back(volume);
}
+
+ volumes = _volumes;
}
FrameworkInfo frameworkInfo;
@@ -873,23 +809,71 @@ int main(int argc, char** argv)
}
}
+ TaskInfo task;
+ task.set_name(flags.name.get());
+ task.mutable_task_id()->set_value(flags.name.get());
+
+ static const Try<Resources> resources = Resources::parse(flags.resources);
+
+ if (resources.isError()) {
+ EXIT(EXIT_FAILURE)
+ << "Failed to parse resources '" << flags.resources << "': "
+ << resources.error();
+ }
+
+ task.mutable_resources()->CopyFrom(resources.get());
+
+ CommandInfo* commandInfo = task.mutable_command();
+
+ if (flags.shell) {
+ CHECK_SOME(flags.command);
+
+ commandInfo->set_shell(true);
+ commandInfo->set_value(flags.command.get());
+ } else {
+ // TODO(gilbert): Treat 'command' as executable value and arguments.
+ commandInfo->set_shell(false);
+ }
+
+ if (flags.environment.isSome()) {
+ Environment* environment_ = commandInfo->mutable_environment();
+ foreachpair (
+ const string& name, const string& value, environment.get()) {
+ Environment::Variable* environmentVariable =
+ environment_->add_variables();
+
+ environmentVariable->set_name(name);
+ environmentVariable->set_value(value);
+ }
+ }
+
+ if (uri.isSome()) {
+ task.mutable_command()->add_uris()->set_value(uri.get());
+ }
+
+ Result<ContainerInfo> containerInfo =
+ getContainerInfo(
+ flags.containerizer,
+ volumes,
+ flags.networks,
+ appcImage,
+ dockerImage);
+
+ if (containerInfo.isError()){
+ EXIT(EXIT_FAILURE) << containerInfo.error();
+ }
+
+ if (containerInfo.isSome()) {
+ task.mutable_container()->CopyFrom(containerInfo.get());
+ }
+
Owned<CommandScheduler> scheduler(
new CommandScheduler(
frameworkInfo,
flags.master.get(),
- flags.name.get(),
- flags.shell,
- flags.command,
- environment,
- flags.resources,
- uri,
- appcImage,
- dockerImage,
- volumes,
- flags.containerizer,
flags.kill_after,
- flags.networks,
- credential));
+ credential,
+ task));
process::spawn(scheduler.get());
process::wait(scheduler.get());
[3/4] mesos git commit: Added << operator to stringify
v1::TaskGroupInfo protobuf.
Posted by vi...@apache.org.
Added << operator to stringify v1::TaskGroupInfo protobuf.
This patch is needed as Flags add() function needs to stringify
v1::TaskGroupInfo before it is parsed to a variable of that
type.
Review: https://reviews.apache.org/r/51990/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/e84c2c45
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/e84c2c45
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/e84c2c45
Branch: refs/heads/master
Commit: e84c2c451dbb971ec91a70ef6c077b4a666ae311
Parents: 9a9610d
Author: Abhishek Dasgupta <a1...@linux.vnet.ibm.com>
Authored: Mon Sep 19 13:31:38 2016 -0700
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Sep 19 13:31:38 2016 -0700
----------------------------------------------------------------------
include/mesos/v1/mesos.hpp | 5 +++++
src/v1/mesos.cpp | 6 ++++++
2 files changed, 11 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/e84c2c45/include/mesos/v1/mesos.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/v1/mesos.hpp b/include/mesos/v1/mesos.hpp
index 936af77..f171458 100644
--- a/include/mesos/v1/mesos.hpp
+++ b/include/mesos/v1/mesos.hpp
@@ -285,6 +285,11 @@ std::ostream& operator<<(std::ostream& stream, const MachineID& machineId);
std::ostream& operator<<(std::ostream& stream, const TaskInfo& task);
+std::ostream& operator<<(
+ std::ostream& stream,
+ const TaskGroupInfo& taskGroupInfo);
+
+
std::ostream& operator<<(std::ostream& stream, const TaskState& state);
http://git-wip-us.apache.org/repos/asf/mesos/blob/e84c2c45/src/v1/mesos.cpp
----------------------------------------------------------------------
diff --git a/src/v1/mesos.cpp b/src/v1/mesos.cpp
index c6a8799..0f88aba 100644
--- a/src/v1/mesos.cpp
+++ b/src/v1/mesos.cpp
@@ -468,6 +468,12 @@ ostream& operator<<(ostream& stream, const TaskInfo& task)
}
+ostream& operator<<(ostream& stream, const TaskGroupInfo& taskGroupInfo)
+{
+ return stream << taskGroupInfo.DebugString();
+}
+
+
ostream& operator<<(ostream& stream, const TaskState& state)
{
return stream << TaskState_Name(state);
[2/4] mesos git commit: Added parse function in flags namespace for
v1::TaskGroupInfo protobuf.
Posted by vi...@apache.org.
Added parse function in flags namespace for v1::TaskGroupInfo protobuf.
Review: https://reviews.apache.org/r/51991/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/9a9610d0
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/9a9610d0
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/9a9610d0
Branch: refs/heads/master
Commit: 9a9610d0444d97b7226012ec82f9a0eb59a48959
Parents: 4e8ac99
Author: Abhishek Dasgupta <a1...@linux.vnet.ibm.com>
Authored: Mon Sep 19 13:31:34 2016 -0700
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Sep 19 13:31:34 2016 -0700
----------------------------------------------------------------------
src/common/parse.hpp | 16 ++++++++++++++++
1 file changed, 16 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/9a9610d0/src/common/parse.hpp
----------------------------------------------------------------------
diff --git a/src/common/parse.hpp b/src/common/parse.hpp
index 51582a4..62a333c 100644
--- a/src/common/parse.hpp
+++ b/src/common/parse.hpp
@@ -15,6 +15,8 @@
#include <mesos/mesos.hpp>
+#include <mesos/v1/mesos.hpp>
+
#include <mesos/authorizer/acls.hpp>
#include <mesos/module/module.hpp>
@@ -88,6 +90,20 @@ inline Try<mesos::ContainerInfo> parse(const std::string& value)
}
+template <>
+inline Try<mesos::v1::TaskGroupInfo> parse(const std::string& value)
+{
+ // Convert from string or file to JSON.
+ Try<JSON::Object> json = parse<JSON::Object>(value);
+ if (json.isError()) {
+ return Error(json.error());
+ }
+
+ // Convert from JSON to Protobuf.
+ return protobuf::parse<mesos::v1::TaskGroupInfo>(json.get());
+}
+
+
// When the same variable is listed multiple times,
// uses only the last value.
template <>
[4/4] mesos git commit: Updated mesos-execute to support task groups.
Posted by vi...@apache.org.
Updated mesos-execute to support task groups.
Through this patch, mesos-execute supports running
task groups. Single task provided through command
line option is supported as well for backward
compatibility.
Review: https://reviews.apache.org/r/51623/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/f162ade0
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/f162ade0
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/f162ade0
Branch: refs/heads/master
Commit: f162ade0f9d4c93129f935c54d100cc173579361
Parents: e84c2c4
Author: Abhishek Dasgupta <a1...@linux.vnet.ibm.com>
Authored: Mon Sep 19 13:31:43 2016 -0700
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Sep 19 13:31:43 2016 -0700
----------------------------------------------------------------------
src/cli/execute.cpp | 299 ++++++++++++++++++++++++++++++++++-------------
1 file changed, 221 insertions(+), 78 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/f162ade0/src/cli/execute.cpp
----------------------------------------------------------------------
diff --git a/src/cli/execute.cpp b/src/cli/execute.cpp
index f2e28e4..f180672 100644
--- a/src/cli/execute.cpp
+++ b/src/cli/execute.cpp
@@ -62,6 +62,7 @@ using mesos::v1::CommandInfo;
using mesos::v1::ContainerInfo;
using mesos::v1::Credential;
using mesos::v1::Environment;
+using mesos::v1::ExecutorInfo;
using mesos::v1::FrameworkID;
using mesos::v1::FrameworkInfo;
using mesos::v1::Image;
@@ -69,6 +70,7 @@ using mesos::v1::Label;
using mesos::v1::Labels;
using mesos::v1::Offer;
using mesos::v1::Resources;
+using mesos::v1::TaskGroupInfo;
using mesos::v1::TaskID;
using mesos::v1::TaskInfo;
using mesos::v1::TaskState;
@@ -93,6 +95,46 @@ public:
"master",
"Mesos master (e.g., IP:PORT).");
+ add(&task_group,
+ "task_group",
+ "The value could be a JSON-formatted string of `TaskGroupInfo` or a\n"
+ "file path containing the JSON-formatted `TaskGroupInfo`. Path must\n"
+ "be of the form `file:///path/to/file` or `/path/to/file`."
+ "\n"
+ "See the `TaskGroupInfo` message in `mesos.proto` for the expected\n"
+ "format. NOTE: `agent_id` need not to be set.\n"
+ "\n"
+ "Example:\n"
+ "{\n"
+ " \"tasks\":\n"
+ " [\n"
+ " {\n"
+ " \"name\": \"Name of the task\",\n"
+ " \"task_id\": {\"value\" : \"Id of the task\"},\n"
+ " \"agent_id\": {\"value\" : \"\"},\n"
+ " \"resources\": [{\n"
+ " \"name\": \"cpus\",\n"
+ " \"type\": \"SCALAR\",\n"
+ " \"scalar\": {\n"
+ " \"value\": 0.1\n"
+ " },\n"
+ " \"role\": \"*\"\n"
+ " },\n"
+ " {\n"
+ " \"name\": \"mem\",\n"
+ " \"type\": \"SCALAR\",\n"
+ " \"scalar\": {\n"
+ " \"value\": 32\n"
+ " },\n"
+ " \"role\": \"*\"\n"
+ " }],\n"
+ " \"command\": {\n"
+ " \"value\": \"sleep 1000\"\n"
+ " }\n"
+ " }\n"
+ " ]\n"
+ "}");
+
add(&name,
"name",
"Name for the command.");
@@ -225,6 +267,7 @@ public:
Option<string> master;
Option<string> name;
+ Option<TaskGroupInfo> task_group;
bool shell;
Option<string> command;
Option<hashmap<string, string>> environment;
@@ -255,14 +298,17 @@ public:
const string& _master,
const Option<Duration>& _killAfter,
const Option<Credential>& _credential,
- const TaskInfo& _task)
+ const Option<TaskInfo>& _task,
+ const Option<TaskGroupInfo> _taskGroup)
: state(DISCONNECTED),
frameworkInfo(_frameworkInfo),
master(_master),
killAfter(_killAfter),
credential(_credential),
task(_task),
- launched(false) {}
+ taskGroup(_taskGroup),
+ launched(false),
+ terminatedTaskCount(0) {}
virtual ~CommandScheduler() {}
@@ -338,43 +384,106 @@ protected:
foreach (const Offer& offer, offers) {
Resources offered = offer.resources();
- TaskInfo _task = task;
+ Resources requiredResources;
- if (!launched && offered.flatten().contains(_task.resources())) {
- _task.mutable_agent_id()->MergeFrom(offer.agent_id());
+ CHECK_NE(task.isSome(), taskGroup.isSome())
+ << "Either task or task group should be set but not both";
- // Takes resources first from the specified role, then from '*'.
- Try<Resources> flattened =
- Resources(_task.resources()).flatten(frameworkInfo.role());
+ if (task.isSome()) {
+ requiredResources = Resources(task.get().resources());
+ } else {
+ foreach (const TaskInfo& _task, taskGroup->tasks()) {
+ requiredResources += Resources(_task.resources());
+ }
+ }
- // `frameworkInfo.role()` must be valid as it's allowed to register.
- CHECK_SOME(flattened);
- Option<Resources> resources = offered.find(flattened.get());
+ if (!launched && offered.flatten().contains(requiredResources)) {
+ TaskInfo _task;
+ TaskGroupInfo _taskGroup;
- CHECK_SOME(resources);
+ if (task.isSome()) {
+ _task = task.get();
+ _task.mutable_agent_id()->MergeFrom(offer.agent_id());
- _task.mutable_resources()->CopyFrom(resources.get());
+ // Takes resources first from the specified role, then from '*'.
+ Try<Resources> flattened =
+ Resources(_task.resources()).flatten(frameworkInfo.role());
- Call call;
- call.set_type(Call::ACCEPT);
+ // `frameworkInfo.role()` must be valid as it's allowed to register.
+ CHECK_SOME(flattened);
+ Option<Resources> resources = offered.find(flattened.get());
- CHECK(frameworkInfo.has_id());
- call.mutable_framework_id()->CopyFrom(frameworkInfo.id());
+ CHECK_SOME(resources);
- Call::Accept* accept = call.mutable_accept();
- accept->add_offer_ids()->CopyFrom(offer.id());
+ _task.mutable_resources()->CopyFrom(resources.get());
+ } else {
+ foreach (TaskInfo _task, taskGroup->tasks()) {
+ _task.mutable_agent_id()->MergeFrom(offer.agent_id());
- Offer::Operation* operation = accept->add_operations();
- operation->set_type(Offer::Operation::LAUNCH);
+ // Takes resources first from the specified role, then from '*'.
+ Try<Resources> flattened =
+ Resources(_task.resources()).flatten(frameworkInfo.role());
- operation->mutable_launch()->add_task_infos()->CopyFrom(_task);
+ // `frameworkInfo.role()` must be valid as it's allowed to
+ // register.
+ CHECK_SOME(flattened);
+ Option<Resources> resources = offered.find(flattened.get());
- mesos->send(call);
+ CHECK_SOME(resources);
+
+ _task.mutable_resources()->CopyFrom(resources.get());
+ _taskGroup.add_tasks()->CopyFrom(_task);
+ }
+ }
+ Call call;
+ call.set_type(Call::ACCEPT);
- cout << "Submitted task '" << _task.name() << "' to agent '"
- << offer.agent_id() << "'" << endl;
+ CHECK(frameworkInfo.has_id());
+ call.mutable_framework_id()->CopyFrom(frameworkInfo.id());
- launched = true;
+ Call::Accept* accept = call.mutable_accept();
+ accept->add_offer_ids()->CopyFrom(offer.id());
+
+ Offer::Operation* operation = accept->add_operations();
+
+ if (task.isSome()) {
+ operation->set_type(Offer::Operation::LAUNCH);
+ operation->mutable_launch()->add_task_infos()->CopyFrom(_task);
+ } else {
+ operation->set_type(Offer::Operation::LAUNCH_GROUP);
+
+ ExecutorInfo* executorInfo =
+ operation->mutable_launch_group()->mutable_executor();
+
+ executorInfo->set_type(ExecutorInfo::DEFAULT);
+ executorInfo->mutable_executor_id()->set_value(
+ "default-executor");
+
+ executorInfo->mutable_framework_id()->CopyFrom(frameworkInfo.id());
+ executorInfo->mutable_resources()->CopyFrom(
+ Resources::parse("cpus:0.1;mem:32;disk:32").get());
+
+ operation->mutable_launch_group()->mutable_task_group()->CopyFrom(
+ _taskGroup);
+ }
+
+ mesos->send(call);
+
+ if (task.isSome()) {
+ cout << "Submitted task '" << task.get().name() << "' to agent '"
+ << offer.agent_id() << "'" << endl;
+ } else {
+ vector<TaskID> taskIds;
+
+ foreach (const TaskInfo& _task, taskGroup->tasks()) {
+ taskIds.push_back(_task.task_id());
+ }
+
+ cout << "Submitted task group with tasks "<< taskIds
+ << " to agent '" << offer.agent_id() << "'" << endl;
+ }
+
+ launched = true;
} else {
Call call;
call.set_type(Call::DECLINE);
@@ -445,7 +554,8 @@ protected:
{
CHECK_EQ(SUBSCRIBED, state);
- CHECK_EQ(task.name(), status.task_id().value());
+ CHECK_NE(task.isSome(), taskGroup.isSome())
+ << "Either task or task group should be set but not both";
cout << "Received status update " << status.state()
<< " for task '" << status.task_id() << "'" << endl;
@@ -487,7 +597,15 @@ protected:
}
if (mesos::internal::protobuf::isTerminalState(devolve(status).state())) {
- terminate(self());
+ if (task.isSome()) {
+ terminate(self());
+ } else {
+ terminatedTaskCount++;
+
+ if (terminatedTaskCount == taskGroup->tasks().size()) {
+ terminate(self());
+ }
+ }
}
}
@@ -503,8 +621,10 @@ private:
const string master;
const Option<Duration> killAfter;
const Option<Credential> credential;
- const TaskInfo task;
+ const Option<TaskInfo> task;
+ const Option<TaskGroupInfo> taskGroup;
bool launched;
+ int terminatedTaskCount;
Owned<Mesos> mesos;
};
@@ -643,14 +763,30 @@ int main(int argc, char** argv)
return EXIT_FAILURE;
}
- if (flags.name.isNone()) {
- cerr << flags.usage("Missing required option --name") << endl;
+ if (flags.task_group.isSome() &&
+ (flags.name.isSome() ||
+ flags.command.isSome() ||
+ flags.environment.isSome() ||
+ flags.appc_image.isSome() ||
+ flags.docker_image.isSome() ||
+ flags.volumes.isSome())) {
+ cerr << flags.usage(
+ "Either task or task group should be set but not both. Provide"
+ " either '--name, --command, --env, --appc_image, --docker_image,"
+ " --volumes' OR '--task_group'") << endl;
return EXIT_FAILURE;
}
- if (flags.shell && flags.command.isNone()) {
- cerr << flags.usage("Missing required option --command") << endl;
- return EXIT_FAILURE;
+ if (flags.task_group.isNone()) {
+ if (flags.name.isNone()) {
+ cerr << flags.usage("Missing required option --name") << endl;
+ return EXIT_FAILURE;
+ }
+
+ if (flags.shell && flags.command.isNone()) {
+ cerr << flags.usage("Missing required option --command") << endl;
+ return EXIT_FAILURE;
+ }
}
Result<string> user = os::user();
@@ -809,62 +945,68 @@ int main(int argc, char** argv)
}
}
- TaskInfo task;
- task.set_name(flags.name.get());
- task.mutable_task_id()->set_value(flags.name.get());
+ Option<TaskInfo> taskInfo = None();
- static const Try<Resources> resources = Resources::parse(flags.resources);
+ if (flags.task_group.isNone()) {
+ TaskInfo task;
+ task.set_name(flags.name.get());
+ task.mutable_task_id()->set_value(flags.name.get());
- if (resources.isError()) {
- EXIT(EXIT_FAILURE)
- << "Failed to parse resources '" << flags.resources << "': "
- << resources.error();
- }
+ static const Try<Resources> resources = Resources::parse(flags.resources);
- task.mutable_resources()->CopyFrom(resources.get());
+ if (resources.isError()) {
+ EXIT(EXIT_FAILURE)
+ << "Failed to parse resources '" << flags.resources << "': "
+ << resources.error();
+ }
- CommandInfo* commandInfo = task.mutable_command();
+ task.mutable_resources()->CopyFrom(resources.get());
- if (flags.shell) {
- CHECK_SOME(flags.command);
+ CommandInfo* commandInfo = task.mutable_command();
- commandInfo->set_shell(true);
- commandInfo->set_value(flags.command.get());
- } else {
- // TODO(gilbert): Treat 'command' as executable value and arguments.
- commandInfo->set_shell(false);
- }
+ if (flags.shell) {
+ CHECK_SOME(flags.command);
- if (flags.environment.isSome()) {
- Environment* environment_ = commandInfo->mutable_environment();
- foreachpair (
- const string& name, const string& value, environment.get()) {
- Environment::Variable* environmentVariable =
- environment_->add_variables();
+ commandInfo->set_shell(true);
+ commandInfo->set_value(flags.command.get());
+ } else {
+ // TODO(gilbert): Treat 'command' as executable value and arguments.
+ commandInfo->set_shell(false);
+ }
- environmentVariable->set_name(name);
- environmentVariable->set_value(value);
+ if (flags.environment.isSome()) {
+ Environment* environment_ = commandInfo->mutable_environment();
+ foreachpair (
+ const string& name, const string& value, environment.get()) {
+ Environment::Variable* environmentVariable =
+ environment_->add_variables();
+
+ environmentVariable->set_name(name);
+ environmentVariable->set_value(value);
+ }
}
- }
- if (uri.isSome()) {
- task.mutable_command()->add_uris()->set_value(uri.get());
- }
+ if (uri.isSome()) {
+ task.mutable_command()->add_uris()->set_value(uri.get());
+ }
- Result<ContainerInfo> containerInfo =
- getContainerInfo(
- flags.containerizer,
- volumes,
- flags.networks,
- appcImage,
- dockerImage);
+ Result<ContainerInfo> containerInfo =
+ getContainerInfo(
+ flags.containerizer,
+ volumes,
+ flags.networks,
+ appcImage,
+ dockerImage);
- if (containerInfo.isError()){
- EXIT(EXIT_FAILURE) << containerInfo.error();
- }
+ if (containerInfo.isError()){
+ EXIT(EXIT_FAILURE) << containerInfo.error();
+ }
+
+ if (containerInfo.isSome()) {
+ task.mutable_container()->CopyFrom(containerInfo.get());
+ }
- if (containerInfo.isSome()) {
- task.mutable_container()->CopyFrom(containerInfo.get());
+ taskInfo = task;
}
Owned<CommandScheduler> scheduler(
@@ -873,7 +1015,8 @@ int main(int argc, char** argv)
flags.master.get(),
flags.kill_after,
credential,
- task));
+ taskInfo,
+ flags.task_group));
process::spawn(scheduler.get());
process::wait(scheduler.get());