You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by an...@apache.org on 2016/09/15 20:50:29 UTC
[1/3] mesos git commit: Removed tasks belonging to a task group from
queued tasks.
Repository: mesos
Updated Branches:
refs/heads/master b053572bc -> cf4ded54c
Removed tasks belonging to a task group from queued tasks.
When sending the `LAUNCH_GROUP` event to the executor, we were
not removing the tasks present in the task group from queued
tasks. This resulted in the tasks being erroneously still
being present in queued tasks thereby not able to receive
terminal status updates from the executor.
Review: https://reviews.apache.org/r/51884/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/8b39241e
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/8b39241e
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/8b39241e
Branch: refs/heads/master
Commit: 8b39241e5cb3537681dac9ee1820f89ccdfe83c8
Parents: b053572
Author: Anand Mazumdar <an...@apache.org>
Authored: Thu Sep 15 13:45:36 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Thu Sep 15 13:45:36 2016 -0700
----------------------------------------------------------------------
src/slave/slave.cpp | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/8b39241e/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 7bd5fad..e77231a 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -2208,9 +2208,12 @@ void Slave::__run(
continue;
}
- // Add the tasks and send the task group to the executor.
+ // Add the tasks and send the task group to the executor. Since, the
+ // queued tasks also include tasks from the queued task group, we
+ // remove them from queued tasks.
foreach (const TaskInfo& task, it->tasks()) {
executor->addTask(task);
+ executor->queuedTasks.erase(task.task_id());
}
executor->queuedTaskGroups.erase(it);
[2/3] mesos git commit: Added a basic default executor implementation.
Posted by an...@apache.org.
Added a basic default executor implementation.
This change adds a basic default executor implementation that
just sends a TASK_RUNNING update followed by a TASK_FINISHED
update upon the receipt of a LAUNCH_GROUP event from the executor.
This allows us to test the entire workflow for now. We can iterate
upon this to later send LAUNCH_CONTAINER/DESTROY_CONTAINER
to the agent to actually launch the tasks in a sub container.
Review: https://reviews.apache.org/r/51885/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/a4753dff
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/a4753dff
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/a4753dff
Branch: refs/heads/master
Commit: a4753dffbe3cd77837aed7c4378716600440aa88
Parents: 8b39241
Author: Anand Mazumdar <an...@apache.org>
Authored: Thu Sep 15 13:45:50 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Thu Sep 15 13:45:50 2016 -0700
----------------------------------------------------------------------
src/Makefile.am | 6 +
src/launcher/default_executor.cpp | 316 +++++++++++++++++++++++++++++++++
2 files changed, 322 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/a4753dff/src/Makefile.am
----------------------------------------------------------------------
diff --git a/src/Makefile.am b/src/Makefile.am
index f1d202a..c6bd5aa 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1335,6 +1335,12 @@ mesos_executor_SOURCES = \
mesos_executor_CPPFLAGS = $(MESOS_CPPFLAGS)
mesos_executor_LDADD = libmesos.la $(LDADD)
+pkglibexec_PROGRAMS += mesos-default-executor
+mesos_default_executor_SOURCES = \
+ launcher/default_executor.cpp
+mesos_default_executor_CPPFLAGS = $(MESOS_CPPFLAGS)
+mesos_default_executor_LDADD = libmesos.la $(LDADD)
+
pkglibexec_PROGRAMS += mesos-containerizer
mesos_containerizer_SOURCES = slave/containerizer/mesos/main.cpp
mesos_containerizer_CPPFLAGS = $(MESOS_CPPFLAGS)
http://git-wip-us.apache.org/repos/asf/mesos/blob/a4753dff/src/launcher/default_executor.cpp
----------------------------------------------------------------------
diff --git a/src/launcher/default_executor.cpp b/src/launcher/default_executor.cpp
new file mode 100644
index 0000000..f0feb95
--- /dev/null
+++ b/src/launcher/default_executor.cpp
@@ -0,0 +1,316 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// 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 <iostream>
+#include <queue>
+#include <string>
+
+#include <mesos/mesos.hpp>
+
+#include <mesos/executor/executor.hpp>
+
+#include <mesos/v1/executor.hpp>
+#include <mesos/v1/mesos.hpp>
+
+#include <process/defer.hpp>
+#include <process/delay.hpp>
+#include <process/id.hpp>
+#include <process/owned.hpp>
+#include <process/process.hpp>
+
+#include <stout/linkedhashmap.hpp>
+#include <stout/option.hpp>
+#include <stout/os.hpp>
+#include <stout/uuid.hpp>
+
+#include "internal/devolve.hpp"
+#include "internal/evolve.hpp"
+
+using mesos::executor::Call;
+using mesos::executor::Event;
+
+using mesos::v1::executor::Mesos;
+
+using process::Owned;
+
+using std::cout;
+using std::cerr;
+using std::endl;
+using std::queue;
+using std::string;
+
+namespace mesos {
+namespace internal {
+
+class DefaultExecutor : public process::Process<DefaultExecutor>
+{
+public:
+ DefaultExecutor(
+ const FrameworkID& _frameworkId,
+ const ExecutorID& _executorId)
+ : ProcessBase(process::ID::generate("default-executor")),
+ state(DISCONNECTED),
+ launched(false),
+ frameworkInfo(None()),
+ frameworkId(_frameworkId),
+ executorId(_executorId) {}
+
+ virtual ~DefaultExecutor() = default;
+
+ void connected()
+ {
+ state = CONNECTED;
+
+ doReliableRegistration();
+ }
+
+ void disconnected()
+ {
+ state = DISCONNECTED;
+ }
+
+ void received(const Event& event)
+ {
+ cout << "Received " << event.type() << " event" << endl;
+
+ switch (event.type()) {
+ case Event::SUBSCRIBED: {
+ cout << "Subscribed executor on "
+ << event.subscribed().slave_info().hostname() << endl;
+
+ frameworkInfo = event.subscribed().framework_info();
+ state = SUBSCRIBED;
+ break;
+ }
+
+ case Event::LAUNCH: {
+ cerr << "LAUNCH event is not supported" << endl;
+ // Shut down because this is unexpected; `LAUNCH` event
+ // should never go to the default executor.
+ //
+ // TODO(anand): Invoke `shutdown()`.
+ break;
+ }
+
+ case Event::LAUNCH_GROUP: {
+ launchGroup(event.launch_group().task_group());
+ break;
+ }
+
+ case Event::KILL: {
+ // TODO(anand): Implement this.
+ break;
+ }
+
+ case Event::ACKNOWLEDGED: {
+ // Remove the corresponding update.
+ updates.erase(UUID::fromBytes(event.acknowledged().uuid()).get());
+
+ // Remove the corresponding task.
+ tasks.erase(event.acknowledged().task_id());
+ break;
+ }
+
+ case Event::SHUTDOWN: {
+ // TODO(anand): Implement this.
+ break;
+ }
+
+ case Event::MESSAGE: {
+ break;
+ }
+
+ case Event::ERROR: {
+ cerr << "Error: " << event.error().message() << endl;
+ break;
+ }
+
+ case Event::UNKNOWN: {
+ LOG(WARNING) << "Received an UNKNOWN event and ignored";
+ break;
+ }
+ }
+ }
+
+protected:
+ virtual void initialize()
+ {
+ mesos.reset(new Mesos(
+ ContentType::PROTOBUF,
+ defer(self(), &Self::connected),
+ defer(self(), &Self::disconnected),
+ defer(self(), [this](queue<v1::executor::Event> events) {
+ while(!events.empty()) {
+ const v1::executor::Event& event = events.front();
+ received(devolve(event));
+
+ events.pop();
+ }
+ })));
+ }
+
+ void doReliableRegistration()
+ {
+ if (state == SUBSCRIBED || state == DISCONNECTED) {
+ return;
+ }
+
+ Call call;
+ call.set_type(Call::SUBSCRIBE);
+
+ call.mutable_framework_id()->CopyFrom(frameworkId);
+ call.mutable_executor_id()->CopyFrom(executorId);
+
+ Call::Subscribe* subscribe = call.mutable_subscribe();
+
+ // Send all unacknowledged updates.
+ foreach (const Call::Update& update, updates.values()) {
+ subscribe->add_unacknowledged_updates()->MergeFrom(update);
+ }
+
+ // Send the unacknowledged tasks.
+ foreach (const TaskInfo& task, tasks.values()) {
+ subscribe->add_unacknowledged_tasks()->MergeFrom(task);
+ }
+
+ mesos->send(evolve(call));
+
+ delay(Seconds(1), self(), &Self::doReliableRegistration);
+ }
+
+ void launchGroup(const TaskGroupInfo& _taskGroup)
+ {
+ CHECK_EQ(SUBSCRIBED, state);
+
+ if (launched) {
+ foreach (const TaskInfo& task, _taskGroup.tasks()) {
+ update(
+ task.task_id(),
+ TASK_FAILED,
+ "Attempted to run multiple task groups using a "
+ "\"default\" executor");
+ }
+ return;
+ }
+
+ launched = true;
+
+ foreach (const TaskInfo& task, _taskGroup.tasks()) {
+ tasks[task.task_id()] = task;
+ }
+
+ // Send a TASK_RUNNING status update followed immediately by a
+ // TASK_FINISHED update.
+ //
+ // TODO(anand): Eventually, we need to invoke the `LAUNCH_CONTAINER`
+ // call via the Agent API.
+ foreach (const TaskInfo& task, _taskGroup.tasks()) {
+ update(task.task_id(), TASK_RUNNING);
+ }
+
+ foreach (const TaskInfo& task, _taskGroup.tasks()) {
+ update(task.task_id(), TASK_FINISHED);
+ }
+
+ // TODO(qianzhang): Remove this hack since the executor now receives
+ // acknowledgements for status updates. The executor can terminate
+ // after it receives an ACK for a terminal status update.
+ os::sleep(Seconds(1));
+ terminate(self());
+ }
+
+private:
+ void update(
+ const TaskID& taskId,
+ const TaskState& state,
+ const Option<string>& message = None())
+ {
+ UUID uuid = UUID::random();
+
+ TaskStatus status;
+ status.mutable_task_id()->CopyFrom(taskId);
+ status.mutable_executor_id()->CopyFrom(executorId);
+
+ status.set_state(state);
+ status.set_source(TaskStatus::SOURCE_EXECUTOR);
+ status.set_uuid(uuid.toBytes());
+
+ if (message.isSome()) {
+ status.set_message(message.get());
+ }
+
+ Call call;
+ call.set_type(Call::UPDATE);
+
+ call.mutable_framework_id()->CopyFrom(frameworkId);
+ call.mutable_executor_id()->CopyFrom(executorId);
+
+ call.mutable_update()->mutable_status()->CopyFrom(status);
+
+ // Capture the status update.
+ updates[uuid] = call.update();
+
+ mesos->send(evolve(call));
+ }
+
+ enum State
+ {
+ CONNECTED,
+ DISCONNECTED,
+ SUBSCRIBED
+ } state;
+
+ bool launched;
+ Option<FrameworkInfo> frameworkInfo;
+ const FrameworkID frameworkId;
+ const ExecutorID executorId;
+ Owned<Mesos> mesos;
+ LinkedHashMap<UUID, Call::Update> updates; // Unacknowledged updates.
+ LinkedHashMap<TaskID, TaskInfo> tasks; // Unacknowledged tasks.
+};
+
+} // namespace internal {
+} // namespace mesos {
+
+
+int main(int argc, char** argv)
+{
+ mesos::FrameworkID frameworkId;
+ mesos::ExecutorID executorId;
+
+ Option<string> value = os::getenv("MESOS_FRAMEWORK_ID");
+ if (value.isNone()) {
+ EXIT(EXIT_FAILURE)
+ << "Expecting 'MESOS_FRAMEWORK_ID' to be set in the environment";
+ }
+ frameworkId.set_value(value.get());
+
+ value = os::getenv("MESOS_EXECUTOR_ID");
+ if (value.isNone()) {
+ EXIT(EXIT_FAILURE)
+ << "Expecting 'MESOS_EXECUTOR_ID' to be set in the environment";
+ }
+ executorId.set_value(value.get());
+
+ Owned<mesos::internal::DefaultExecutor> executor(
+ new mesos::internal::DefaultExecutor(
+ frameworkId,
+ executorId));
+
+ process::spawn(executor.get());
+ process::wait(executor.get());
+
+ return EXIT_SUCCESS;
+}
[3/3] mesos git commit: Added tests for default executor.
Posted by an...@apache.org.
Added tests for default executor.
This change adds a basic test to ensure that the basic executor
sends a TASK_RUNNING update upon launch.
Review: https://reviews.apache.org/r/51886/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/cf4ded54
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/cf4ded54
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/cf4ded54
Branch: refs/heads/master
Commit: cf4ded54c7851ec6beb66409577a6fe185b262e5
Parents: a4753df
Author: Anand Mazumdar <an...@apache.org>
Authored: Thu Sep 15 13:46:16 2016 -0700
Committer: Anand Mazumdar <an...@apache.org>
Committed: Thu Sep 15 13:46:16 2016 -0700
----------------------------------------------------------------------
src/Makefile.am | 1 +
src/tests/default_executor_tests.cpp | 172 ++++++++++++++++++++++++++++++
2 files changed, 173 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/cf4ded54/src/Makefile.am
----------------------------------------------------------------------
diff --git a/src/Makefile.am b/src/Makefile.am
index c6bd5aa..44fd8a2 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -2072,6 +2072,7 @@ mesos_tests_SOURCES = \
tests/containerizer.cpp \
tests/cram_md5_authentication_tests.cpp \
tests/credentials_tests.cpp \
+ tests/default_executor_tests.cpp \
tests/disk_quota_tests.cpp \
tests/dynamic_weights_tests.cpp \
tests/environment.cpp \
http://git-wip-us.apache.org/repos/asf/mesos/blob/cf4ded54/src/tests/default_executor_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/default_executor_tests.cpp b/src/tests/default_executor_tests.cpp
new file mode 100644
index 0000000..055a3d7
--- /dev/null
+++ b/src/tests/default_executor_tests.cpp
@@ -0,0 +1,172 @@
+// 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/v1/executor.hpp>
+#include <mesos/v1/scheduler.hpp>
+
+#include <process/owned.hpp>
+
+#include "tests/containerizer.hpp"
+#include "tests/mesos.hpp"
+
+using mesos::master::detector::MasterDetector;
+
+using mesos::v1::scheduler::Call;
+using mesos::v1::scheduler::Mesos;
+
+using process::Future;
+using process::Owned;
+
+namespace mesos {
+namespace internal {
+namespace tests {
+
+// Tests that exercise the default executor implementation
+// should be located in this file.
+
+class DefaultExecutorTest : public MesosTest {};
+
+// This test verifies that the default executor can launch a task group.
+TEST_F(DefaultExecutorTest, TaskRunning)
+{
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
+
+ Resources resources =
+ Resources::parse("cpus:0.1;mem:32;disk:32").get();
+
+ FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO;
+
+ ExecutorInfo executorInfo;
+ executorInfo.set_type(ExecutorInfo::DEFAULT);
+
+ executorInfo.mutable_executor_id()->CopyFrom(DEFAULT_EXECUTOR_ID);
+ executorInfo.mutable_resources()->CopyFrom(resources);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get());
+ ASSERT_SOME(slave);
+
+ Future<Nothing> connected;
+ EXPECT_CALL(*scheduler, connected(_))
+ .WillOnce(FutureSatisfy(&connected));
+
+ scheduler::TestV1Mesos mesos(
+ master.get()->pid, ContentType::PROTOBUF, scheduler);
+
+ AWAIT_READY(connected);
+
+ 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.
+
+ {
+ Call call;
+ call.set_type(Call::SUBSCRIBE);
+ Call::Subscribe* subscribe = call.mutable_subscribe();
+ subscribe->mutable_framework_info()->CopyFrom(evolve(frameworkInfo));
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(subscribed);
+
+ v1::FrameworkID frameworkId(subscribed->framework_id());
+
+ // Update `executorInfo` with the subscribed `frameworkId`.
+ executorInfo.mutable_framework_id()->CopyFrom(devolve(frameworkId));
+
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
+
+ Future<v1::scheduler::Event::Update> update1;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update1));
+
+ const v1::Offer& offer = offers->offers(0);
+ const SlaveID slaveId = devolve(offer.agent_id());
+
+ v1::TaskInfo taskInfo =
+ evolve(createTask(slaveId, resources, "sleep 1000"));
+
+ v1::TaskGroupInfo taskGroup;
+ taskGroup.add_tasks()->CopyFrom(taskInfo);
+
+ {
+ Call call;
+ call.mutable_framework_id()->CopyFrom(frameworkId);
+ call.set_type(Call::ACCEPT);
+
+ Call::Accept* accept = call.mutable_accept();
+ accept->add_offer_ids()->CopyFrom(offer.id());
+
+ v1::Offer::Operation* operation = accept->add_operations();
+ operation->set_type(v1::Offer::Operation::LAUNCH_GROUP);
+
+ v1::Offer::Operation::LaunchGroup* launchGroup =
+ operation->mutable_launch_group();
+
+ launchGroup->mutable_executor()->CopyFrom(evolve(executorInfo));
+ launchGroup->mutable_task_group()->CopyFrom(taskGroup);
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(update1);
+
+ ASSERT_EQ(TASK_RUNNING, update1->status().state());
+ EXPECT_EQ(taskInfo.task_id(), update1->status().task_id());
+
+ Future<v1::scheduler::Event::Update> update2;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update2));
+
+ {
+ // Acknowledge TASK_RUNNING update.
+ Call call;
+ call.mutable_framework_id()->CopyFrom(frameworkId);
+ call.set_type(Call::ACKNOWLEDGE);
+
+ Call::Acknowledge* acknowledge = call.mutable_acknowledge();
+ acknowledge->mutable_task_id()->CopyFrom(taskInfo.task_id());
+ acknowledge->mutable_agent_id()->CopyFrom(offer.agent_id());
+ acknowledge->set_uuid(update1->status().uuid());
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(update2);
+
+ ASSERT_EQ(TASK_FINISHED, update2->status().state());
+ EXPECT_EQ(taskInfo.task_id(), update2->status().task_id());
+}
+
+} // namespace tests {
+} // namespace internal {
+} // namespace mesos {