You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by bm...@apache.org on 2013/10/29 20:33:44 UTC
[3/3] git commit: Injected Registrar into the Master and the tests.
Injected Registrar into the Master and the tests.
From: Benjamin Hindman <be...@berkeley.edu>
Review: https://reviews.apache.org/r/14382
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/137f4e86
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/137f4e86
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/137f4e86
Branch: refs/heads/master
Commit: 137f4e86ecb7f022b7fa4063431a044485904170
Parents: 7edb000
Author: Benjamin Mahler <bm...@twitter.com>
Authored: Tue Oct 29 12:10:27 2013 -0700
Committer: Benjamin Mahler <bm...@twitter.com>
Committed: Tue Oct 29 12:27:02 2013 -0700
----------------------------------------------------------------------
src/local/local.cpp | 41 ++++++++++++++++++++--
src/master/flags.hpp | 20 +++++++----
src/master/main.cpp | 32 +++++++++++++++++-
src/master/master.cpp | 29 ++++++----------
src/master/master.hpp | 8 ++++-
src/slave/slave.cpp | 7 ++--
src/tests/cluster.hpp | 60 +++++++++++++++++++++++++++++++--
src/tests/mesos.cpp | 2 ++
src/tests/slave_recovery_tests.cpp | 8 ++---
9 files changed, 169 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/local/local.cpp
----------------------------------------------------------------------
diff --git a/src/local/local.cpp b/src/local/local.cpp
index e4b5ec5..180756a 100644
--- a/src/local/local.cpp
+++ b/src/local/local.cpp
@@ -20,8 +20,10 @@
#include <sstream>
#include <vector>
-#include <stout/fatal.hpp>
+#include <stout/exit.hpp>
#include <stout/foreach.hpp>
+#include <stout/path.hpp>
+#include <stout/strings.hpp>
#include "local.hpp"
@@ -34,10 +36,15 @@
#include "master/drf_sorter.hpp"
#include "master/hierarchical_allocator_process.hpp"
#include "master/master.hpp"
+#include "master/registrar.hpp"
#include "slave/process_isolator.hpp"
#include "slave/slave.hpp"
+#include "state/leveldb.hpp"
+#include "state/protobuf.hpp"
+#include "state/storage.hpp"
+
using namespace mesos::internal;
using mesos::internal::master::allocator::Allocator;
@@ -46,6 +53,7 @@ using mesos::internal::master::allocator::DRFSorter;
using mesos::internal::master::allocator::HierarchicalDRFAllocatorProcess;
using mesos::internal::master::Master;
+using mesos::internal::master::Registrar;
using mesos::internal::slave::Slave;
using mesos::internal::slave::Isolator;
@@ -66,6 +74,9 @@ namespace local {
static Allocator* allocator = NULL;
static AllocatorProcess* allocatorProcess = NULL;
+static state::Storage* storage = NULL;
+static state::protobuf::State* state = NULL;
+static Registrar* registrar = NULL;
static Master* master = NULL;
static map<Isolator*, Slave*> slaves;
static MasterDetector* detector = NULL;
@@ -98,7 +109,24 @@ PID<Master> launch(const Flags& flags, Allocator* _allocator)
EXIT(1) << "Failed to start a local cluster while loading "
<< "master flags from the environment: " << load.error();
}
- master = new Master(_allocator, files, flags);
+
+ if (strings::startsWith(flags.registry, "zk://")) {
+ // TODO(benh):
+ EXIT(1) << "ZooKeeper based registry unimplemented";
+ } else if (flags.registry == "local") {
+ storage = new state::LevelDBStorage(
+ path::join(flags.work_dir, "registry"));
+ } else {
+ EXIT(1) << "'" << flags.registry << "' is not a supported"
+ << " option for registry persistence";
+ }
+
+ CHECK_NOTNULL(storage);
+
+ state = new state::protobuf::State(storage);
+ registrar = new Registrar(state);
+
+ master = new Master(_allocator, registrar, files, flags);
}
PID<Master> pid = process::spawn(master);
@@ -160,6 +188,15 @@ void shutdown()
delete files;
files = NULL;
+
+ delete registrar;
+ registrar = NULL;
+
+ delete state;
+ state = NULL;
+
+ delete storage;
+ storage = NULL;
}
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/master/flags.hpp
----------------------------------------------------------------------
diff --git a/src/master/flags.hpp b/src/master/flags.hpp
index 7b5f3de..16faf96 100644
--- a/src/master/flags.hpp
+++ b/src/master/flags.hpp
@@ -40,12 +40,17 @@ public:
"Can root submit frameworks?",
true);
- add(&Flags::slaves,
- "slaves",
- "Initial slaves that should be\n"
- "considered part of this cluster\n"
- "(or if using ZooKeeper a URL)",
- "*");
+ add(&Flags::work_dir,
+ "work_dir",
+ "Where to store master specific files\n",
+ "/tmp/mesos");
+
+ add(&Flags::registry,
+ "registry",
+ "Persistence strategy for the registry;\n"
+ "available options are 'local' or a ZooKeeper\n"
+ "URL (i.e., 'zk://host1:port1,host2:port2,.../path')",
+ "local");
add(&Flags::webui_dir,
"webui_dir",
@@ -111,7 +116,8 @@ public:
}
bool root_submissions;
- std::string slaves;
+ std::string work_dir;
+ std::string registry;
std::string webui_dir;
std::string whitelist;
std::string user_sorter;
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/master/main.cpp
----------------------------------------------------------------------
diff --git a/src/master/main.cpp b/src/master/main.cpp
index 19fcb9f..45caf9d 100644
--- a/src/master/main.cpp
+++ b/src/master/main.cpp
@@ -19,10 +19,13 @@
#include <mesos/mesos.hpp>
#include <stout/check.hpp>
+#include <stout/exit.hpp>
#include <stout/flags.hpp>
#include <stout/nothing.hpp>
#include <stout/os.hpp>
+#include <stout/path.hpp>
#include <stout/stringify.hpp>
+#include <stout/strings.hpp>
#include <stout/try.hpp>
#include "common/build.hpp"
@@ -36,6 +39,12 @@
#include "master/drf_sorter.hpp"
#include "master/hierarchical_allocator_process.hpp"
#include "master/master.hpp"
+#include "master/registrar.hpp"
+
+#include "state/leveldb.hpp"
+#include "state/protobuf.hpp"
+#include "state/storage.hpp"
+
using namespace mesos::internal;
using namespace mesos::internal::master;
@@ -119,8 +128,25 @@ int main(int argc, char** argv)
allocator::Allocator* allocator =
new allocator::Allocator(allocatorProcess);
+ state::Storage* storage = NULL;
+
+ if (strings::startsWith(flags.registry, "zk://")) {
+ // TODO(benh):
+ EXIT(1) << "ZooKeeper based registry unimplemented";
+ } else if (flags.registry == "local") {
+ storage = new state::LevelDBStorage(path::join(flags.work_dir, "registry"));
+ } else {
+ EXIT(1) << "'" << flags.registry << "' is not a supported"
+ << " option for registry persistence";
+ }
+
+ CHECK_NOTNULL(storage);
+
+ state::protobuf::State* state = new state::protobuf::State(storage);
+ Registrar* registrar = new Registrar(state);
+
Files files;
- Master* master = new Master(allocator, &files, flags);
+ Master* master = new Master(allocator, registrar, &files, flags);
process::spawn(master);
Try<MasterDetector*> detector =
@@ -133,6 +159,10 @@ int main(int argc, char** argv)
delete allocator;
delete allocatorProcess;
+ delete registrar;
+ delete state;
+ delete storage;
+
MasterDetector::destroy(detector.get());
return 0;
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index 1147cc6..8e14a07 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -186,20 +186,29 @@ private:
};
-Master::Master(Allocator* _allocator, Files* _files)
+Master::Master(
+ Allocator* _allocator,
+ Registrar* _registrar,
+ Files* _files)
: ProcessBase("master"),
http(*this),
flags(),
allocator(_allocator),
+ registrar(_registrar),
files(_files),
completedFrameworks(MAX_COMPLETED_FRAMEWORKS) {}
-Master::Master(Allocator* _allocator, Files* _files, const Flags& _flags)
+Master::Master(
+ Allocator* _allocator,
+ Registrar* _registrar,
+ Files* _files,
+ const Flags& _flags)
: ProcessBase("master"),
http(*this),
flags(_flags),
allocator(_allocator),
+ registrar(_registrar),
files(_files),
completedFrameworks(MAX_COMPLETED_FRAMEWORKS) {}
@@ -1223,21 +1232,7 @@ void Master::registerSlave(const UPID& from, const SlaveInfo& slaveInfo)
LOG(INFO) << "Attempting to register slave on " << slave->info.hostname()
<< " at " << slave->pid;
- // TODO(benh): We assume all slaves can register for now.
- CHECK_EQ(flags.slaves, "*");
addSlave(slave);
-
-// // Checks if this slave, or if all slaves, can be accepted.
-// if (slaveHostnamePorts.contains(slaveInfo.hostname(), from.port)) {
-// run(&SlaveRegistrar::run, slave, self());
-// } else if (flags.slaves == "*") {
-// run(&SlaveRegistrar::run, slave, self(), slavesManager->self());
-// } else {
-// LOG(WARNING) << "Cannot register slave at "
-// << slaveInfo.hostname() << ":" << from.port
-// << " because not in allocated set of slaves!";
-// reply(ShutdownMessage());
-// }
}
@@ -1326,8 +1321,6 @@ void Master::reregisterSlave(
LOG(INFO) << "Attempting to re-register slave " << slave->id << " at "
<< slave->pid << " (" << slave->info.hostname() << ")";
- // TODO(benh): We assume all slaves can register for now.
- CHECK_EQ(flags.slaves, "*");
readdSlave(slave, executorInfos, tasks);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/master/master.hpp
----------------------------------------------------------------------
diff --git a/src/master/master.hpp b/src/master/master.hpp
index 1eba03f..e377af8 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -47,6 +47,7 @@
#include "master/constants.hpp"
#include "master/flags.hpp"
+#include "master/registrar.hpp"
#include "messages/messages.hpp"
@@ -82,8 +83,12 @@ struct Role;
class Master : public ProtobufProcess<Master>
{
public:
- Master(allocator::Allocator* allocator, Files* files);
Master(allocator::Allocator* allocator,
+ Registrar* registrar,
+ Files* files);
+
+ Master(allocator::Allocator* allocator,
+ Registrar* registrar,
Files* files,
const Flags& flags);
@@ -302,6 +307,7 @@ private:
allocator::Allocator* allocator;
WhitelistWatcher* whitelistWatcher;
+ Registrar* registrar;
Files* files;
MasterInfo info;
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 06ef786..74bcbec 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -641,13 +641,14 @@ void Slave::doReliableRegistration()
// Slave started before master.
// (Vinod): Is the above comment true?
RegisterSlaveMessage message;
- message.mutable_slave()->MergeFrom(info);
+ message.mutable_slave()->CopyFrom(info);
send(master, message);
} else {
// Re-registering, so send tasks running.
ReregisterSlaveMessage message;
- message.mutable_slave_id()->MergeFrom(info.id());
- message.mutable_slave()->MergeFrom(info);
+ message.mutable_slave_id()->CopyFrom(info.id());
+ message.mutable_slave()->CopyFrom(info);
+ message.mutable_slave()->mutable_id()->CopyFrom(info.id());
foreachvalue (Framework* framework, frameworks){
foreachvalue (Executor* executor, framework->executors) {
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/tests/cluster.hpp
----------------------------------------------------------------------
diff --git a/src/tests/cluster.hpp b/src/tests/cluster.hpp
index 5621139..bea395a 100644
--- a/src/tests/cluster.hpp
+++ b/src/tests/cluster.hpp
@@ -30,6 +30,7 @@
#include <stout/nothing.hpp>
#include <stout/option.hpp>
#include <stout/owned.hpp>
+#include <stout/path.hpp>
#include <stout/try.hpp>
#include "detector/detector.hpp"
@@ -40,12 +41,17 @@
#include "master/hierarchical_allocator_process.hpp"
#include "master/flags.hpp"
#include "master/master.hpp"
+#include "master/registrar.hpp"
#include "slave/flags.hpp"
#include "slave/isolator.hpp"
#include "slave/process_isolator.hpp"
#include "slave/slave.hpp"
+#include "state/leveldb.hpp"
+#include "state/protobuf.hpp"
+#include "state/storage.hpp"
+
#include "zookeeper/url.hpp"
namespace mesos {
@@ -108,6 +114,9 @@ public:
master::Master* master;
master::allocator::Allocator* allocator;
master::allocator::AllocatorProcess* allocatorProcess;
+ state::Storage* storage;
+ state::protobuf::State* state;
+ master::Registrar* registrar;
MasterDetector* detector;
};
@@ -223,7 +232,28 @@ inline Try<process::PID<master::Master> > Cluster::Masters::start(
master.allocatorProcess = new master::allocator::HierarchicalDRFAllocatorProcess();
master.allocator = new master::allocator::Allocator(master.allocatorProcess);
- master.master = new master::Master(master.allocator, &cluster->files, flags);
+
+ if (strings::startsWith(flags.registry, "zk://")) {
+ // TODO(benh):
+ return Error("ZooKeeper based registry unimplemented");
+ } else if (flags.registry == "local") {
+ master.storage = new state::LevelDBStorage(
+ path::join(flags.work_dir, "registry"));
+ } else {
+ return Error("'" + flags.registry + "' is not a supported"
+ " option for registry persistence");
+ }
+
+ CHECK_NOTNULL(master.storage);
+
+ master.state = new state::protobuf::State(master.storage);
+ master.registrar = new master::Registrar(master.state);
+
+ master.master = new master::Master(
+ master.allocator,
+ master.registrar,
+ &cluster->files,
+ flags);
process::PID<master::Master> pid = process::spawn(master.master);
@@ -249,9 +279,31 @@ inline Try<process::PID<master::Master> > Cluster::Masters::start(
}
Master master;
+
master.allocatorProcess = NULL;
master.allocator = new master::allocator::Allocator(allocatorProcess);
- master.master = new master::Master(master.allocator, &cluster->files, flags);
+
+ if (strings::startsWith(flags.registry, "zk://")) {
+ // TODO(benh):
+ return Error("ZooKeeper based registry unimplemented");
+ } else if (flags.registry == "local") {
+ master.storage = new state::LevelDBStorage(
+ path::join(flags.work_dir, "registry"));
+ } else {
+ return Error("'" + flags.registry + "' is not a supported"
+ " option for registry persistence");
+ }
+
+ CHECK_NOTNULL(master.storage);
+
+ master.state = new state::protobuf::State(master.storage);
+ master.registrar = new master::Registrar(master.state);
+
+ master.master = new master::Master(
+ master.allocator,
+ master.registrar,
+ &cluster->files,
+ flags);
process::PID<master::Master> pid = process::spawn(master.master);
@@ -283,6 +335,10 @@ inline Try<Nothing> Cluster::Masters::stop(
delete master.allocator; // Terminates and waits for allocator process.
delete master.allocatorProcess; // May be NULL.
+ delete master.registrar;
+ delete master.state;
+ delete master.storage;
+
delete master.detector;
masters.erase(pid);
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/tests/mesos.cpp
----------------------------------------------------------------------
diff --git a/src/tests/mesos.cpp b/src/tests/mesos.cpp
index 10b3e53..351ffd6 100644
--- a/src/tests/mesos.cpp
+++ b/src/tests/mesos.cpp
@@ -32,6 +32,8 @@ master::Flags MesosTest::CreateMasterFlags()
const Try<std::string>& directory = environment->mkdtemp();
CHECK_SOME(directory) << "Failed to create temporary directory";
+ flags.work_dir = directory.get();
+
// Create a default credentials file.
const std::string& path = path::join(directory.get(), "credentials");
http://git-wip-us.apache.org/repos/asf/mesos/blob/137f4e86/src/tests/slave_recovery_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_recovery_tests.cpp b/src/tests/slave_recovery_tests.cpp
index 2812f11..37faf20 100644
--- a/src/tests/slave_recovery_tests.cpp
+++ b/src/tests/slave_recovery_tests.cpp
@@ -97,7 +97,7 @@ TEST_F(SlaveStateTest, CheckpointProtobuf)
expected.set_value("slave1");
const string& file = "slave.id";
- state::checkpoint(file, expected);
+ slave::state::checkpoint(file, expected);
const Result<SlaveID>& actual = ::protobuf::read<SlaveID>(file);
ASSERT_SOME(actual);
@@ -111,7 +111,7 @@ TEST_F(SlaveStateTest, CheckpointString)
// Checkpoint a test string.
const string expected = "test";
const string file = "test-file";
- state::checkpoint(file, expected);
+ slave::state::checkpoint(file, expected);
ASSERT_SOME_EQ(expected, os::read(file));
}
@@ -228,12 +228,12 @@ TYPED_TEST(SlaveRecoveryTest, RecoverSlaveState)
AWAIT_READY(_ack);
// Recover the state.
- Result<state::SlaveState> recover = state::recover(
+ Result<slave::state::SlaveState> recover = slave::state::recover(
paths::getMetaRootDir(flags.work_dir), true);
ASSERT_SOME(recover);
- state::SlaveState state = recover.get();
+ slave::state::SlaveState state = recover.get();
// Check slave id.
ASSERT_EQ(slaveId, state.id);