You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by be...@apache.org on 2015/05/08 00:22:10 UTC
[1/2] mesos git commit: Add `hashset::EMPTY` constant in stout.
Repository: mesos
Updated Branches:
refs/heads/master da0880035 -> d37889da1
Add `hashset<T>::EMPTY` constant in stout.
Review: https://reviews.apache.org/r/33643
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/ce293d3d
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/ce293d3d
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/ce293d3d
Branch: refs/heads/master
Commit: ce293d3d5ad2a57edf72d0f3b8e6bb48397ad6e8
Parents: da08800
Author: Joris Van Remoortere <jo...@gmail.com>
Authored: Thu May 7 14:56:04 2015 -0700
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Thu May 7 14:56:04 2015 -0700
----------------------------------------------------------------------
3rdparty/libprocess/3rdparty/stout/include/stout/hashset.hpp | 7 +++++++
1 file changed, 7 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/ce293d3d/3rdparty/libprocess/3rdparty/stout/include/stout/hashset.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/3rdparty/stout/include/stout/hashset.hpp b/3rdparty/libprocess/3rdparty/stout/include/stout/hashset.hpp
index d2b7439..21b0599 100644
--- a/3rdparty/libprocess/3rdparty/stout/include/stout/hashset.hpp
+++ b/3rdparty/libprocess/3rdparty/stout/include/stout/hashset.hpp
@@ -30,6 +30,8 @@ template <typename Elem>
class hashset : public boost::unordered_set<Elem>
{
public:
+ static const hashset<Elem>& EMPTY;
+
// An explicit default constructor is needed so
// 'const hashset<T> map;' is not an error.
hashset() {}
@@ -68,6 +70,11 @@ public:
};
+// TODO(jmlvanre): Possibly remove this reference as per MESOS-2694.
+template <typename Elem>
+const hashset<Elem>& hashset<Elem>::EMPTY = *new hashset<Elem>();
+
+
// Union operator.
template <typename Elem>
hashset<Elem> operator | (const hashset<Elem>& left, const hashset<Elem>& right)
[2/2] mesos git commit: Add '/state-summary' endpoint to master.
Posted by be...@apache.org.
Add '/state-summary' endpoint to master.
This exposes framework and slave statistics that are of high
importance and aggregated per framework or slave respectively. The
statistics will not be exact due to the circular buffer that stores
completed tasks, but this was also an issue with '/state.json'.
Review: https://reviews.apache.org/r/33505
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d37889da
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d37889da
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d37889da
Branch: refs/heads/master
Commit: d37889da16d4449afe2903dd0d0dc7a60e735769
Parents: ce293d3
Author: Joris Van Remoortere <jo...@gmail.com>
Authored: Thu May 7 14:57:46 2015 -0700
Committer: Benjamin Hindman <be...@gmail.com>
Committed: Thu May 7 14:57:50 2015 -0700
----------------------------------------------------------------------
src/master/http.cpp | 305 ++++++++++++++++++++++++++++++++++++++--
src/master/master.cpp | 3 +
src/master/master.hpp | 4 +
src/tests/master_tests.cpp | 109 ++++++++++++++
4 files changed, 411 insertions(+), 10 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d37889da/src/master/http.cpp
----------------------------------------------------------------------
diff --git a/src/master/http.cpp b/src/master/http.cpp
index fb44825..f0668aa 100644
--- a/src/master/http.cpp
+++ b/src/master/http.cpp
@@ -105,12 +105,31 @@ JSON::Object model(const Offer& offer)
}
-// Returns a JSON object modeled on a Framework.
-JSON::Object model(const Framework& framework)
+// Returns a JSON object summarizing some important fields in a
+// Framework.
+JSON::Object summarize(const Framework& framework)
{
JSON::Object object;
object.values["id"] = framework.id().value();
object.values["name"] = framework.info.name();
+
+ // TODO(bmahler): Use these in the webui.
+ object.values["used_resources"] = model(framework.totalUsedResources);
+ object.values["offered_resources"] = model(framework.totalOfferedResources);
+
+ object.values["hostname"] = framework.info.hostname();
+ object.values["webui_url"] = framework.info.webui_url();
+
+ return object;
+}
+
+
+// Returns a JSON object modeled on a Framework.
+JSON::Object model(const Framework& framework)
+{
+ // Add additional fields to those generated by 'summarize'.
+ JSON::Object object = summarize(framework);
+
object.values["user"] = framework.info.user();
object.values["failover_timeout"] = framework.info.failover_timeout();
object.values["checkpoint"] = framework.info.checkpoint();
@@ -120,17 +139,10 @@ JSON::Object model(const Framework& framework)
object.values["active"] = framework.active;
// TODO(bmahler): Consider deprecating this in favor of the split
- // used and offered resources below.
+ // used and offered resources added in 'summarize'.
object.values["resources"] =
model(framework.totalUsedResources + framework.totalOfferedResources);
- // TODO(bmahler): Use these in the webui.
- object.values["used_resources"] = model(framework.totalUsedResources);
- object.values["offered_resources"] = model(framework.totalOfferedResources);
-
- object.values["hostname"] = framework.info.hostname();
- object.values["webui_url"] = framework.info.webui_url();
-
// TODO(benh): Consider making reregisteredTime an Option.
if (framework.registeredTime != framework.reregisteredTime) {
object.values["reregistered_time"] = framework.reregisteredTime.secs();
@@ -183,6 +195,20 @@ JSON::Object model(const Framework& framework)
}
+// Forward declaration for 'summarize(Slave)'.
+JSON::Object model(const Slave& slave);
+
+
+// Returns a JSON object summarizing some important fields in a Slave.
+// For now this just calls 'model(slave)' because all the fields in
+// 'model' are of value, and the model for a slave is not really heavy
+// weight.
+JSON::Object summarize(const Slave& slave)
+{
+ return model(slave);
+}
+
+
// Returns a JSON object modeled after a Slave.
JSON::Object model(const Slave& slave)
{
@@ -197,6 +223,9 @@ JSON::Object model(const Slave& slave)
}
object.values["resources"] = model(slave.info.resources());
+ object.values["used_resources"] = model(Resources::sum(slave.usedResources));
+ object.values["offered_resources"] = model(slave.offeredResources);
+
object.values["attributes"] = model(slave.info.attributes());
object.values["active"] = slave.active;
return object;
@@ -550,6 +579,262 @@ Future<Response> Master::Http::state(const Request& request)
}
+// This abstraction has no side-effects. It factors out computing the
+// mapping from 'slaves' to 'frameworks' to answer the questions 'what
+// frameworks are running on a given slave?' and 'what slaves are
+// running the given framework?'.
+class SlaveFrameworkMapping
+{
+public:
+ SlaveFrameworkMapping(const hashmap<FrameworkID, Framework*>& frameworks)
+ {
+ foreachpair (const FrameworkID& frameworkId,
+ const Framework* framework,
+ frameworks) {
+ foreachvalue (const TaskInfo& taskInfo, framework->pendingTasks) {
+ frameworksToSlaves[frameworkId].insert(taskInfo.slave_id());
+ slavesToFrameworks[taskInfo.slave_id()].insert(frameworkId);
+ }
+
+ foreachvalue (const Task* task, framework->tasks) {
+ frameworksToSlaves[frameworkId].insert(task->slave_id());
+ slavesToFrameworks[task->slave_id()].insert(frameworkId);
+ }
+
+ foreach (const std::shared_ptr<Task>& task, framework->completedTasks) {
+ frameworksToSlaves[frameworkId].insert(task->slave_id());
+ slavesToFrameworks[task->slave_id()].insert(frameworkId);
+ }
+ }
+ }
+
+ const hashset<FrameworkID>& frameworks(const SlaveID& slaveId) const
+ {
+ const auto iterator = slavesToFrameworks.find(slaveId);
+ return iterator != slavesToFrameworks.end() ?
+ iterator->second : hashset<FrameworkID>::EMPTY;
+ }
+
+ const hashset<SlaveID>& slaves(const FrameworkID& frameworkId) const
+ {
+ const auto iterator = frameworksToSlaves.find(frameworkId);
+ return iterator != frameworksToSlaves.end() ?
+ iterator->second : hashset<SlaveID>::EMPTY;
+ }
+
+private:
+ hashmap<SlaveID, hashset<FrameworkID>> slavesToFrameworks;
+ hashmap<FrameworkID, hashset<SlaveID>> frameworksToSlaves;
+};
+
+
+// This abstraction has no side-effects. It factors out the accounting
+// for a 'TaskState' summary. We use this to summarize 'TaskState's
+// for both frameworks as well as slaves.
+struct TaskStateSummary
+{
+ // TODO(jmlvanre): Possibly clean this up as per MESOS-2694.
+ const static TaskStateSummary EMPTY;
+
+ TaskStateSummary()
+ : staging(0),
+ starting(0),
+ running(0),
+ finished(0),
+ killed(0),
+ failed(0),
+ lost(0),
+ error(0) {}
+
+ // Account for the state of the given task.
+ void count(const Task& task)
+ {
+ switch (task.state()) {
+ case TASK_STAGING: { ++staging; break; }
+ case TASK_STARTING: { ++starting; break; }
+ case TASK_RUNNING: { ++running; break; }
+ case TASK_FINISHED: { ++finished; break; }
+ case TASK_KILLED: { ++killed; break; }
+ case TASK_FAILED: { ++failed; break; }
+ case TASK_LOST: { ++lost; break; }
+ case TASK_ERROR: { ++error; break; }
+ // No default case allows for a helpful compiler error if we
+ // introduce a new state.
+ }
+ }
+
+ size_t staging;
+ size_t starting;
+ size_t running;
+ size_t finished;
+ size_t killed;
+ size_t failed;
+ size_t lost;
+ size_t error;
+};
+
+
+const TaskStateSummary TaskStateSummary::EMPTY;
+
+
+// This abstraction has no side-effects. It factors out computing the
+// 'TaskState' sumaries for frameworks and slaves. This answers the
+// questions 'How many tasks are in each state for a given framework?'
+// and 'How many tasks are in each state for a given slave?'.
+class TaskStateSummaries
+{
+public:
+ TaskStateSummaries(const hashmap<FrameworkID, Framework*>& frameworks)
+ {
+ foreachpair (const FrameworkID& frameworkId,
+ const Framework* framework,
+ frameworks) {
+ foreachvalue (const TaskInfo& taskInfo, framework->pendingTasks) {
+ frameworkTaskSummaries[frameworkId].staging++;
+ slaveTaskSummaries[taskInfo.slave_id()].staging++;
+ }
+
+ foreachvalue (const Task* task, framework->tasks) {
+ frameworkTaskSummaries[frameworkId].count(*task);
+ slaveTaskSummaries[task->slave_id()].count(*task);
+ }
+
+ foreach (const std::shared_ptr<Task>& task, framework->completedTasks) {
+ frameworkTaskSummaries[frameworkId].count(*task);
+ slaveTaskSummaries[task->slave_id()].count(*task);
+ }
+ }
+ }
+
+ const TaskStateSummary& framework(const FrameworkID& frameworkId) const
+ {
+ const auto iterator = frameworkTaskSummaries.find(frameworkId);
+ return iterator != frameworkTaskSummaries.end() ?
+ iterator->second : TaskStateSummary::EMPTY;
+ }
+
+ const TaskStateSummary& slave(const SlaveID& slaveId) const
+ {
+ const auto iterator = slaveTaskSummaries.find(slaveId);
+ return iterator != slaveTaskSummaries.end() ?
+ iterator->second : TaskStateSummary::EMPTY;
+ }
+private:
+ hashmap<FrameworkID, TaskStateSummary> frameworkTaskSummaries;
+ hashmap<SlaveID, TaskStateSummary> slaveTaskSummaries;
+};
+
+
+Future<Response> Master::Http::stateSummary(const Request& request)
+{
+ LOG(INFO) << "HTTP request for '" << request.path << "'";
+
+ JSON::Object object;
+
+ object.values["hostname"] = master->info().hostname();
+
+ if (master->flags.cluster.isSome()) {
+ object.values["cluster"] = master->flags.cluster.get();
+ }
+
+ // We use the tasks in the 'Frameworks' struct to compute summaries
+ // for this endpoint. This is done 1) for consistency between the
+ // 'slaves' and 'frameworks' subsections below 2) because we want to
+ // provide summary information for frameworks that are currently
+ // registered 3) the frameworks keep a circular buffer of completed
+ // tasks that we can use to keep a limited view on the history of
+ // recent completed / failed tasks.
+
+ // Generate mappings from 'slave' to 'framework' and reverse.
+ SlaveFrameworkMapping slaveFrameworkMapping(master->frameworks.registered);
+
+ // Generate 'TaskState' summaries for all framework and slave ids.
+ TaskStateSummaries taskStateSummaries(master->frameworks.registered);
+
+ // Model all of the slaves.
+ {
+ JSON::Array array;
+ array.values.reserve(master->slaves.registered.size()); // MESOS-2353.
+
+ foreachvalue (Slave* slave, master->slaves.registered) {
+ JSON::Object json = summarize(*slave);
+
+ // Add the 'TaskState' summary for this slave.
+ const TaskStateSummary& summary = taskStateSummaries.slave(slave->id);
+
+ json.values["TASK_STAGING"] = summary.staging;
+ json.values["TASK_STARTING"] = summary.starting;
+ json.values["TASK_RUNNING"] = summary.running;
+ json.values["TASK_FINISHED"] = summary.finished;
+ json.values["TASK_KILLED"] = summary.killed;
+ json.values["TASK_FAILED"] = summary.failed;
+ json.values["TASK_LOST"] = summary.lost;
+ json.values["TASK_ERROR"] = summary.error;
+
+ // Add the ids of all the frameworks running on this slave.
+ const hashset<FrameworkID>& frameworks =
+ slaveFrameworkMapping.frameworks(slave->id);
+
+ JSON::Array frameworkIdArray;
+ frameworkIdArray.values.reserve(frameworks.size()); // MESOS-2353.
+
+ foreach (const FrameworkID& frameworkId, frameworks) {
+ frameworkIdArray.values.push_back(frameworkId.value());
+ }
+
+ json.values["framework_ids"] = std::move(frameworkIdArray);
+
+ array.values.push_back(std::move(json));
+ }
+
+ object.values["slaves"] = std::move(array);
+ }
+
+ // Model all of the frameworks.
+ {
+ JSON::Array array;
+ array.values.reserve(master->frameworks.registered.size()); // MESOS-2353.
+
+ foreachpair (const FrameworkID& frameworkId,
+ Framework* framework,
+ master->frameworks.registered) {
+ JSON::Object json = summarize(*framework);
+
+ // Add the 'TaskState' summary for this framework.
+ const TaskStateSummary& summary =
+ taskStateSummaries.framework(frameworkId);
+ json.values["TASK_STAGING"] = summary.staging;
+ json.values["TASK_STARTING"] = summary.starting;
+ json.values["TASK_RUNNING"] = summary.running;
+ json.values["TASK_FINISHED"] = summary.finished;
+ json.values["TASK_KILLED"] = summary.killed;
+ json.values["TASK_FAILED"] = summary.failed;
+ json.values["TASK_LOST"] = summary.lost;
+ json.values["TASK_ERROR"] = summary.error;
+
+ // Add the ids of all the slaves running this framework.
+ const hashset<SlaveID>& slaves =
+ slaveFrameworkMapping.slaves(frameworkId);
+
+ JSON::Array slaveIdArray;
+ slaveIdArray.values.reserve(slaves.size()); // MESOS-2353.
+
+ foreach (const SlaveID& slaveId, slaves) {
+ slaveIdArray.values.push_back(slaveId.value());
+ }
+
+ json.values["slave_ids"] = std::move(slaveIdArray);
+
+ array.values.push_back(std::move(json));
+ }
+
+ object.values["frameworks"] = std::move(array);
+ }
+
+ return OK(object, request.query.get("jsonp"));
+}
+
+
Future<Response> Master::Http::roles(const Request& request)
{
LOG(INFO) << "HTTP request for '" << request.path << "'";
http://git-wip-us.apache.org/repos/asf/mesos/blob/d37889da/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index bee8425..49638ce 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -742,6 +742,9 @@ void Master::initialize()
route("/state.json",
None(),
lambda::bind(&Http::state, http, lambda::_1));
+ route("/state-summary",
+ None(),
+ lambda::bind(&Http::stateSummary, http, lambda::_1));
route("/tasks.json",
Http::TASKS_HELP,
lambda::bind(&Http::tasks, http, lambda::_1));
http://git-wip-us.apache.org/repos/asf/mesos/blob/d37889da/src/master/master.hpp
----------------------------------------------------------------------
diff --git a/src/master/master.hpp b/src/master/master.hpp
index 49ee050..d00be17 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -505,6 +505,10 @@ private:
process::Future<process::http::Response> state(
const process::http::Request& request);
+ // /master/state-summary
+ process::Future<process::http::Response> stateSummary(
+ const process::http::Request& request);
+
// /master/tasks.json
process::Future<process::http::Response> tasks(
const process::http::Request& request);
http://git-wip-us.apache.org/repos/asf/mesos/blob/d37889da/src/tests/master_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/master_tests.cpp b/src/tests/master_tests.cpp
index bdfccb2..75ffada 100644
--- a/src/tests/master_tests.cpp
+++ b/src/tests/master_tests.cpp
@@ -2826,6 +2826,115 @@ TEST_F(MasterTest, StateEndpoint)
}
+TEST_F(MasterTest, StateSummaryEndpoint)
+{
+ master::Flags flags = CreateMasterFlags();
+
+ flags.hostname = "localhost";
+ flags.cluster = "test-cluster";
+
+ Try<PID<Master>> master = StartMaster(flags);
+ ASSERT_SOME(master);
+
+ MockExecutor exec(DEFAULT_EXECUTOR_ID);
+
+ Try<PID<Slave>> slave = StartSlave(&exec);
+ ASSERT_SOME(slave);
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, DEFAULT_FRAMEWORK_INFO, master.get(), DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(&driver, _, _))
+ .Times(1);
+
+ Future<vector<Offer> > offers;
+ EXPECT_CALL(sched, resourceOffers(&driver, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ EXPECT_NE(0u, offers.get().size());
+
+ TaskID taskId;
+ taskId.set_value("1");
+
+ TaskInfo task;
+ task.set_name("");
+ task.mutable_task_id()->MergeFrom(taskId);
+ task.mutable_slave_id()->MergeFrom(offers.get()[0].slave_id());
+ task.mutable_resources()->MergeFrom(offers.get()[0].resources());
+ task.mutable_executor()->MergeFrom(DEFAULT_EXECUTOR_INFO);
+
+ vector<TaskInfo> tasks;
+ tasks.push_back(task);
+
+ EXPECT_CALL(exec, registered(_, _, _, _))
+ .Times(1);
+
+ EXPECT_CALL(exec, launchTask(_, _))
+ .WillOnce(SendStatusUpdateFromTask(TASK_RUNNING));
+
+ Future<TaskStatus> status;
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ driver.launchTasks(offers.get()[0].id(), tasks);
+
+ AWAIT_READY(status);
+ EXPECT_EQ(TASK_RUNNING, status.get().state());
+
+ EXPECT_CALL(exec, killTask(_, _))
+ .WillOnce(SendStatusUpdateFromTaskID(TASK_KILLED));
+
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(FutureArg<1>(&status));
+
+ driver.killTask(taskId);
+
+ AWAIT_READY(status);
+ EXPECT_EQ(TASK_KILLED, status.get().state());
+
+ EXPECT_CALL(exec, shutdown(_))
+ .Times(AtMost(1));
+
+ Future<http::Response> response =
+ http::get(master.get(), "state-summary");
+
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
+
+ EXPECT_SOME_EQ(
+ "application/json",
+ response.get().headers.get("Content-Type"));
+
+ Try<JSON::Object> parse = JSON::parse<JSON::Object>(response.get().body);
+ ASSERT_SOME(parse);
+
+ JSON::Object state = parse.get();
+
+ EXPECT_EQ(flags.hostname.get(), state.values["hostname"]);
+
+ EXPECT_EQ(flags.cluster.get(), state.values["cluster"]);
+
+ ASSERT_TRUE(state.values["slaves"].is<JSON::Array>());
+ ASSERT_EQ(1u, state.values["slaves"].as<JSON::Array>().values.size());
+ ASSERT_SOME_EQ(0u, state.find<JSON::Number>("slaves[0].TASK_RUNNING"));
+ ASSERT_SOME_EQ(1u, state.find<JSON::Number>("slaves[0].TASK_KILLED"));
+
+ ASSERT_TRUE(state.values["frameworks"].is<JSON::Array>());
+ ASSERT_EQ(1u, state.values["frameworks"].as<JSON::Array>().values.size());
+ ASSERT_SOME_EQ(0u, state.find<JSON::Number>("frameworks[0].TASK_RUNNING"));
+ ASSERT_SOME_EQ(1u, state.find<JSON::Number>("frameworks[0].TASK_KILLED"));
+
+ driver.stop();
+ driver.join();
+
+ Shutdown();
+}
+
+
// This test ensures that the web UI of a framework is included in the
// state.json endpoint, if provided by the framework.
TEST_F(MasterTest, FrameworkWebUIUrl)