You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by ji...@apache.org on 2015/01/16 22:56:34 UTC
[1/3] mesos git commit: Added three tuples overload for
process::await.
Repository: mesos
Updated Branches:
refs/heads/master 2a34f0c86 -> 7c45541a8
Added three tuples overload for process::await.
Review: https://reviews.apache.org/r/29892
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/0a08019c
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/0a08019c
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/0a08019c
Branch: refs/heads/master
Commit: 0a08019c2180777f9ab0c338372f560dd1436d62
Parents: 2a34f0c
Author: Jie Yu <yu...@gmail.com>
Authored: Wed Jan 14 10:43:54 2015 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Fri Jan 16 12:10:03 2015 -0800
----------------------------------------------------------------------
3rdparty/libprocess/include/process/collect.hpp | 136 ++++++++++++++-----
1 file changed, 99 insertions(+), 37 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/0a08019c/3rdparty/libprocess/include/process/collect.hpp
----------------------------------------------------------------------
diff --git a/3rdparty/libprocess/include/process/collect.hpp b/3rdparty/libprocess/include/process/collect.hpp
index 3bee8a6..6352883 100644
--- a/3rdparty/libprocess/include/process/collect.hpp
+++ b/3rdparty/libprocess/include/process/collect.hpp
@@ -22,32 +22,40 @@ namespace process {
// the result will be a failure. Likewise, if any future fails then
// the result future will be a failure.
template <typename T>
-Future<std::list<T> > collect(const std::list<Future<T> >& futures);
+Future<std::list<T>> collect(const std::list<Future<T>>& futures);
// Waits on each future in the specified set and returns the list of
// non-pending futures.
template <typename T>
-Future<std::list<Future<T> > > await(const std::list<Future<T> >& futures);
+Future<std::list<Future<T>>> await(const std::list<Future<T>>& futures);
// Waits on each future specified and returns the wrapping future
// typed of a tuple of futures.
+// TODO(jieyu): Investigate the use of variadic templates here.
template <typename T1, typename T2>
-Future<tuples::tuple<Future<T1>, Future<T2> > > await(
+Future<tuples::tuple<Future<T1>, Future<T2>>> await(
const Future<T1>& future1,
const Future<T2>& future2);
+template <typename T1, typename T2, typename T3>
+Future<tuples::tuple<Future<T1>, Future<T2>, Future<T3>>> await(
+ const Future<T1>& future1,
+ const Future<T2>& future2,
+ const Future<T3>& future3);
+
+
namespace internal {
template <typename T>
-class CollectProcess : public Process<CollectProcess<T> >
+class CollectProcess : public Process<CollectProcess<T>>
{
public:
CollectProcess(
- const std::list<Future<T> >& _futures,
- Promise<std::list<T> >* _promise)
+ const std::list<Future<T>>& _futures,
+ Promise<std::list<T>>* _promise)
: futures(_futures),
promise(_promise),
ready(0) {}
@@ -62,7 +70,7 @@ public:
// Stop this nonsense if nobody cares.
promise->future().onDiscard(defer(this, &CollectProcess::discarded));
- typename std::list<Future<T> >::const_iterator iterator;
+ typename std::list<Future<T>>::const_iterator iterator;
for (iterator = futures.begin(); iterator != futures.end(); ++iterator) {
(*iterator).onAny(defer(this, &CollectProcess::waited, lambda::_1));
}
@@ -97,19 +105,19 @@ private:
}
}
- const std::list<Future<T> > futures;
- Promise<std::list<T> >* promise;
+ const std::list<Future<T>> futures;
+ Promise<std::list<T>>* promise;
size_t ready;
};
template <typename T>
-class AwaitProcess : public Process<AwaitProcess<T> >
+class AwaitProcess : public Process<AwaitProcess<T>>
{
public:
AwaitProcess(
- const std::list<Future<T> >& _futures,
- Promise<std::list<Future<T> > >* _promise)
+ const std::list<Future<T>>& _futures,
+ Promise<std::list<Future<T>>>* _promise)
: futures(_futures),
promise(_promise),
ready(0) {}
@@ -124,7 +132,7 @@ public:
// Stop this nonsense if nobody cares.
promise->future().onDiscard(defer(this, &AwaitProcess::discarded));
- typename std::list<Future<T> >::const_iterator iterator;
+ typename std::list<Future<T>>::const_iterator iterator;
for (iterator = futures.begin(); iterator != futures.end(); ++iterator) {
(*iterator).onAny(defer(this, &AwaitProcess::waited, lambda::_1));
}
@@ -148,8 +156,8 @@ private:
}
}
- const std::list<Future<T> > futures;
- Promise<std::list<Future<T> > >* promise;
+ const std::list<Future<T>> futures;
+ Promise<std::list<Future<T>>>* promise;
size_t ready;
};
@@ -159,15 +167,23 @@ private:
template <typename T>
void _await(
const Future<T>& result,
- Owned<Promise<Nothing> > promise)
+ Owned<Promise<Nothing>> promise)
{
promise->set(Nothing());
}
template <typename T1, typename T2>
-Future<tuples::tuple<Future<T1>, Future<T2> > > __await(
- const tuples::tuple<Future<T1>, Future<T2> >& future)
+Future<tuples::tuple<Future<T1>, Future<T2>>> __await(
+ const tuples::tuple<Future<T1>, Future<T2>>& future)
+{
+ return future;
+}
+
+
+template <typename T1, typename T2, typename T3>
+Future<tuples::tuple<Future<T1>, Future<T2>, Future<T3>>> __await(
+ const tuples::tuple<Future<T1>, Future<T2>, Future<T3>>& future)
{
return future;
}
@@ -178,31 +194,31 @@ Future<tuples::tuple<Future<T1>, Future<T2> > > __await(
template <typename T>
-inline Future<std::list<T> > collect(
- const std::list<Future<T> >& futures)
+inline Future<std::list<T>> collect(
+ const std::list<Future<T>>& futures)
{
if (futures.empty()) {
return std::list<T>();
}
- Promise<std::list<T> >* promise = new Promise<std::list<T> >();
- Future<std::list<T> > future = promise->future();
+ Promise<std::list<T>>* promise = new Promise<std::list<T>>();
+ Future<std::list<T>> future = promise->future();
spawn(new internal::CollectProcess<T>(futures, promise), true);
return future;
}
template <typename T>
-inline Future<std::list<Future<T> > > await(
- const std::list<Future<T> >& futures)
+inline Future<std::list<Future<T>>> await(
+ const std::list<Future<T>>& futures)
{
if (futures.empty()) {
return futures;
}
- Promise<std::list<Future<T> > >* promise =
- new Promise<std::list<Future<T> > >();
- Future<std::list<Future<T> > > future = promise->future();
+ Promise<std::list<Future<T>>>* promise =
+ new Promise<std::list<Future<T>>>();
+ Future<std::list<Future<T>>> future = promise->future();
spawn(new internal::AwaitProcess<T>(futures, promise), true);
return future;
}
@@ -215,38 +231,58 @@ Future<tuples::tuple<Future<T1>, Future<T2>>> await(
const Future<T1>& future1,
const Future<T2>& future2)
{
- Owned<Promise<Nothing> > promise1(new Promise<Nothing>());
+ Owned<Promise<Nothing>> promise1(new Promise<Nothing>());
+ Owned<Promise<Nothing>> promise2(new Promise<Nothing>());
future1.onAny([=] () { promise1->set(Nothing()); });
+ future2.onAny([=] () { promise2->set(Nothing()); });
- Owned<Promise<Nothing> > promise2(new Promise<Nothing>());
+ std::list<Future<Nothing>> futures;
+ futures.push_back(promise1->future());
+ futures.push_back(promise2->future());
+
+ return await(futures)
+ .then([=] () { return tuples::make_tuple(future1, future2); });
+}
+
+
+template <typename T1, typename T2, typename T3>
+Future<tuples::tuple<Future<T1>, Future<T2>, Future<T3>>> await(
+ const Future<T1>& future1,
+ const Future<T2>& future2,
+ const Future<T3>& future3)
+{
+ Owned<Promise<Nothing>> promise1(new Promise<Nothing>());
+ Owned<Promise<Nothing>> promise2(new Promise<Nothing>());
+ Owned<Promise<Nothing>> promise3(new Promise<Nothing>());
+ future1.onAny([=] () { promise1->set(Nothing()); });
future2.onAny([=] () { promise2->set(Nothing()); });
+ future3.onAny([=] () { promise3->set(Nothing()); });
- std::list<Future<Nothing> > futures;
+ std::list<Future<Nothing>> futures;
futures.push_back(promise1->future());
futures.push_back(promise2->future());
+ futures.push_back(promise3->future());
return await(futures)
- .then([=] () { return tuples::make_tuple(future1, future2); });
+ .then([=] () { return tuples::make_tuple(future1, future2, future3); });
}
#else // __cplusplus >= 201103L
template <typename T1, typename T2>
-Future<tuples::tuple<Future<T1>, Future<T2> > > await(
+Future<tuples::tuple<Future<T1>, Future<T2>>> await(
const Future<T1>& future1,
const Future<T2>& future2)
{
- Owned<Promise<Nothing> > promise1(new Promise<Nothing>());
+ Owned<Promise<Nothing>> promise1(new Promise<Nothing>());
+ Owned<Promise<Nothing>> promise2(new Promise<Nothing>());
future1.onAny(lambda::bind(internal::_await<T1>, lambda::_1, promise1));
-
- Owned<Promise<Nothing> > promise2(new Promise<Nothing>());
-
future2.onAny(lambda::bind(internal::_await<T2>, lambda::_1, promise2));
- std::list<Future<Nothing> > futures;
+ std::list<Future<Nothing>> futures;
futures.push_back(promise1->future());
futures.push_back(promise2->future());
@@ -256,6 +292,32 @@ Future<tuples::tuple<Future<T1>, Future<T2> > > await(
tuples::make_tuple(future1, future2)));
}
+
+template <typename T1, typename T2, typename T3>
+Future<tuples::tuple<Future<T1>, Future<T2>, Future<T3>>> await(
+ const Future<T1>& future1,
+ const Future<T2>& future2,
+ const Future<T3>& future3)
+{
+ Owned<Promise<Nothing>> promise1(new Promise<Nothing>());
+ Owned<Promise<Nothing>> promise2(new Promise<Nothing>());
+ Owned<Promise<Nothing>> promise3(new Promise<Nothing>());
+
+ future1.onAny(lambda::bind(internal::_await<T1>, lambda::_1, promise1));
+ future2.onAny(lambda::bind(internal::_await<T2>, lambda::_1, promise2));
+ future3.onAny(lambda::bind(internal::_await<T3>, lambda::_1, promise3));
+
+ std::list<Future<Nothing>> futures;
+ futures.push_back(promise1->future());
+ futures.push_back(promise2->future());
+ futures.push_back(promise3->future());
+
+ return await(futures)
+ .then(lambda::bind(
+ internal::__await<T1, T2, T3>,
+ tuples::make_tuple(future1, future2, future3)));
+}
+
#endif // __cplusplus >= 201103L
} // namespace process {
[3/3] mesos git commit: Added tests for verifying PosixDiskIsolator.
Posted by ji...@apache.org.
Added tests for verifying PosixDiskIsolator.
Review: https://reviews.apache.org/r/29689
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/7c45541a
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/7c45541a
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/7c45541a
Branch: refs/heads/master
Commit: 7c45541a8fceedb6eebfbcd83e5c5c9144e9de5d
Parents: 788a509
Author: Jie Yu <yu...@gmail.com>
Authored: Wed Jan 7 16:03:29 2015 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Fri Jan 16 13:55:17 2015 -0800
----------------------------------------------------------------------
src/Makefile.am | 1 +
src/tests/disk_quota_tests.cpp | 201 ++++++++++++++++++++++++++++++++++++
2 files changed, 202 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/7c45541a/src/Makefile.am
----------------------------------------------------------------------
diff --git a/src/Makefile.am b/src/Makefile.am
index 43c693d..07bea1f 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1253,6 +1253,7 @@ mesos_tests_SOURCES = \
tests/containerizer_tests.cpp \
tests/cram_md5_authentication_tests.cpp \
tests/credentials_tests.cpp \
+ tests/disk_quota_tests.cpp \
tests/docker_containerizer_tests.cpp \
tests/docker_tests.cpp \
tests/environment.cpp \
http://git-wip-us.apache.org/repos/asf/mesos/blob/7c45541a/src/tests/disk_quota_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/disk_quota_tests.cpp b/src/tests/disk_quota_tests.cpp
new file mode 100644
index 0000000..4d23d97
--- /dev/null
+++ b/src/tests/disk_quota_tests.cpp
@@ -0,0 +1,201 @@
+/**
+ * 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 <string>
+#include <vector>
+
+#include <gmock/gmock.h>
+
+#include <mesos/mesos.hpp>
+#include <mesos/resources.hpp>
+
+#include <process/gtest.hpp>
+#include <process/pid.hpp>
+
+#include <stout/fs.hpp>
+#include <stout/gtest.hpp>
+#include <stout/os.hpp>
+#include <stout/path.hpp>
+
+#include "master/master.hpp"
+
+#include "slave/flags.hpp"
+#include "slave/slave.hpp"
+
+#include "slave/containerizer/isolators/posix/disk.hpp"
+
+#include "tests/mesos.hpp"
+#include "tests/utils.hpp"
+
+using namespace process;
+
+using namespace mesos;
+using namespace mesos::internal;
+using namespace mesos::internal::tests;
+
+using std::string;
+using std::vector;
+
+using testing::_;
+using testing::Return;
+
+using mesos::internal::master::Master;
+
+using mesos::internal::slave::DiskUsageCollector;
+using mesos::internal::slave::Slave;
+
+
+class DiskUsageCollectorTest : public TemporaryDirectoryTest {};
+
+
+// TODO(jieyu): Consider adding a test to verify that minimal check
+// interval is honored.
+
+
+// This test verifies the usage of a file.
+TEST_F(DiskUsageCollectorTest, File)
+{
+ // Create a file and write 8K bytes.
+ string path = path::join(os::getcwd(), "file");
+ ASSERT_SOME(os::write(path, string(Kilobytes(8).bytes(), 'x')));
+
+ DiskUsageCollector collector(Milliseconds(1));
+
+ Future<Bytes> usage = collector.usage(path);
+ AWAIT_READY(usage);
+
+ // NOTE: A typical file system needs more disk space to keep meta
+ // data. So the check here is not a strict equal-to check.
+ EXPECT_GE(usage.get(), Kilobytes(8));
+}
+
+
+// This test verifies the usage of a directory.
+TEST_F(DiskUsageCollectorTest, Directory)
+{
+ // Create files and subdirectories in the working directory.
+ string file1 = path::join(os::getcwd(), "file1");
+ string file2 = path::join(os::getcwd(), "file2");
+
+ string dir = path::join(os::getcwd(), "dir");
+ string file3 = path::join(dir, "file3");
+ string file4 = path::join(dir, "file4");
+
+ ASSERT_SOME(os::mkdir(dir));
+
+ ASSERT_SOME(os::write(file1, string(Kilobytes(8).bytes(), 'x')));
+ ASSERT_SOME(os::write(file2, string(Kilobytes(4).bytes(), 'y')));
+ ASSERT_SOME(os::write(file3, string(Kilobytes(1).bytes(), 'z')));
+ ASSERT_SOME(os::write(file4, string(Kilobytes(2).bytes(), '1')));
+
+ DiskUsageCollector collector(Milliseconds(1));
+
+ Future<Bytes> usage = collector.usage(os::getcwd());
+ AWAIT_READY(usage);
+
+ EXPECT_GE(usage.get(), Kilobytes(15));
+}
+
+
+// This test verifies that symbolic links are not followed.
+TEST_F(DiskUsageCollectorTest, SymbolicLink)
+{
+ string file = path::join(os::getcwd(), "file");
+ ASSERT_SOME(os::write(file, string(Kilobytes(8).bytes(), 'x')));
+
+ // Create a symbolic link to the current directory.
+ string link = path::join(os::getcwd(), "link");
+ ASSERT_SOME(fs::symlink(os::getcwd(), link));
+
+ DiskUsageCollector collector(Milliseconds(1));
+
+ Future<Bytes> usage1 = collector.usage(os::getcwd());
+ Future<Bytes> usage2 = collector.usage(link);
+
+ AWAIT_READY(usage1);
+ EXPECT_GE(usage1.get(), Kilobytes(8));
+ EXPECT_LT(usage1.get(), Kilobytes(16));
+
+ AWAIT_READY(usage2);
+ EXPECT_LT(usage2.get(), Kilobytes(8));
+}
+
+
+class DiskQuotaTest : public MesosTest {};
+
+
+// This test verifies that the container will be killed if the disk
+// usage exceeds its quota.
+TEST_F(DiskQuotaTest, DiskUsageExceedsQuota)
+{
+ Try<PID<Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ slave::Flags flags = CreateSlaveFlags();
+ flags.isolation = "posix/cpu,posix/mem,posix/disk";
+ flags.disk_quota_check_interval = Milliseconds(1);
+
+ Try<PID<Slave>> slave = StartSlave(flags);
+ ASSERT_SOME(slave);
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, DEFAULT_FRAMEWORK_INFO, master.get(), DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(&driver, _, _));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(&driver, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return()); // Ignore subsequent offers.
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ EXPECT_FALSE(offers.get().empty());
+
+ Offer offer = offers.get()[0];
+
+ // Create a task which requests 1MB disk, but actually uses more
+ // than 2MB disk.
+ TaskInfo task = createTask(
+ offer.slave_id(),
+ Resources::parse("cpus:1;mem:128;disk:1").get(),
+ "dd if=/dev/zero of=file bs=1048576 count=2 && sleep 1000");
+
+ Future<TaskStatus> status1;
+ Future<TaskStatus> status2;
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(FutureArg<1>(&status1))
+ .WillOnce(FutureArg<1>(&status2));
+
+ driver.launchTasks(offer.id(), {task});
+
+ AWAIT_READY(status1);
+ EXPECT_EQ(task.task_id(), status1.get().task_id());
+ EXPECT_EQ(TASK_RUNNING, status1.get().state());
+
+ AWAIT_READY(status2);
+ EXPECT_EQ(task.task_id(), status2.get().task_id());
+ EXPECT_EQ(TASK_FAILED, status2.get().state());
+
+ driver.stop();
+ driver.join();
+
+ Shutdown();
+}
[2/3] mesos git commit: Added PosixDiskIsolator to monitor container
disk usage and to enforce disk quota.
Posted by ji...@apache.org.
Added PosixDiskIsolator to monitor container disk usage and to enforce
disk quota.
Review: https://reviews.apache.org/r/29688
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/788a509f
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/788a509f
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/788a509f
Branch: refs/heads/master
Commit: 788a509f5bb93d28f82917320cdc2a8c7cb6b8d8
Parents: 0a08019
Author: Jie Yu <yu...@gmail.com>
Authored: Wed Jan 7 14:52:47 2015 -0800
Committer: Jie Yu <yu...@gmail.com>
Committed: Fri Jan 16 13:54:27 2015 -0800
----------------------------------------------------------------------
src/Makefile.am | 4 +-
.../containerizer/isolators/posix/disk.cpp | 492 +++++++++++++++++++
.../containerizer/isolators/posix/disk.hpp | 138 ++++++
src/slave/containerizer/mesos/containerizer.cpp | 2 +
src/slave/flags.hpp | 7 +
5 files changed, 641 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/788a509f/src/Makefile.am
----------------------------------------------------------------------
diff --git a/src/Makefile.am b/src/Makefile.am
index 256384e..43c693d 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -316,6 +316,7 @@ libmesos_no_3rdparty_la_SOURCES = \
slave/containerizer/external_containerizer.cpp \
slave/containerizer/fetcher.cpp \
slave/containerizer/isolator.cpp \
+ slave/containerizer/isolators/posix/disk.cpp \
slave/containerizer/launcher.cpp \
slave/containerizer/mesos/containerizer.cpp \
slave/containerizer/mesos/launch.cpp \
@@ -489,14 +490,13 @@ libmesos_no_3rdparty_la_SOURCES += \
slave/containerizer/mesos/containerizer.hpp \
slave/containerizer/mesos/launch.hpp \
slave/containerizer/isolators/posix.hpp \
+ slave/containerizer/isolators/posix/disk.hpp \
slave/containerizer/isolators/cgroups/constants.hpp \
slave/containerizer/isolators/cgroups/cpushare.hpp \
slave/containerizer/isolators/cgroups/mem.hpp \
slave/containerizer/isolators/cgroups/perf_event.hpp \
slave/containerizer/isolators/namespaces/pid.hpp \
- slave/containerizer/mesos/containerizer.hpp \
slave/containerizer/isolators/filesystem/shared.hpp \
- slave/containerizer/mesos/launch.hpp \
tests/cluster.hpp \
tests/containerizer.hpp \
tests/environment.hpp \
http://git-wip-us.apache.org/repos/asf/mesos/blob/788a509f/src/slave/containerizer/isolators/posix/disk.cpp
----------------------------------------------------------------------
diff --git a/src/slave/containerizer/isolators/posix/disk.cpp b/src/slave/containerizer/isolators/posix/disk.cpp
new file mode 100644
index 0000000..fcadd78
--- /dev/null
+++ b/src/slave/containerizer/isolators/posix/disk.cpp
@@ -0,0 +1,492 @@
+/**
+ * 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 <signal.h>
+
+#include <sys/types.h>
+
+#include <deque>
+
+#include <glog/logging.h>
+
+#include <process/check.hpp>
+#include <process/collect.hpp>
+#include <process/defer.hpp>
+#include <process/delay.hpp>
+#include <process/io.hpp>
+#include <process/subprocess.hpp>
+
+#include <stout/check.hpp>
+#include <stout/foreach.hpp>
+#include <stout/lambda.hpp>
+#include <stout/numify.hpp>
+#include <stout/strings.hpp>
+#include <stout/tuple.hpp>
+
+#include <stout/os/exists.hpp>
+#include <stout/os/killtree.hpp>
+
+#include "slave/containerizer/isolators/posix/disk.hpp"
+
+using namespace process;
+
+using std::deque;
+using std::list;
+using std::string;
+using std::vector;
+
+namespace mesos {
+namespace internal {
+namespace slave {
+
+Try<Isolator*> PosixDiskIsolatorProcess::create(const Flags& flags)
+{
+ // TODO(jieyu): Check the availability of command 'du'.
+
+ return new Isolator(
+ process::Owned<IsolatorProcess>(new PosixDiskIsolatorProcess(flags)));
+}
+
+
+PosixDiskIsolatorProcess::Info::~Info()
+{
+ foreachvalue (Future<Bytes> usage, usages) {
+ usage.discard();
+ }
+}
+
+
+PosixDiskIsolatorProcess::PosixDiskIsolatorProcess(const Flags& _flags)
+ : flags(_flags), collector(flags.disk_quota_check_interval) {}
+
+
+PosixDiskIsolatorProcess::~PosixDiskIsolatorProcess() {}
+
+
+Future<Nothing> PosixDiskIsolatorProcess::recover(
+ const list<state::RunState>& states)
+{
+ foreach (const state::RunState& state, states) {
+ if (state.id.isNone()) {
+ infos.clear();
+
+ return Failure("ContainerID is required for recovery");
+ }
+
+ if (state.completed) {
+ continue;
+ }
+
+ // Since we checkpoint the executor after we create its working
+ // directory, the working directory should definitely exist.
+ CHECK(os::exists(state.directory))
+ << "Executor work directory " << state.directory << " doesn't exist";
+
+ infos.put(state.id.get(), Owned<Info>(new Info(state.directory)));
+ }
+
+ return Nothing();
+}
+
+
+Future<Option<CommandInfo>> PosixDiskIsolatorProcess::prepare(
+ const ContainerID& containerId,
+ const ExecutorInfo& executorInfo,
+ const string& directory,
+ const Option<string>& user)
+{
+ if (infos.contains(containerId)) {
+ return Failure("Container has already been prepared");
+ }
+
+ infos.put(containerId, Owned<Info>(new Info(directory)));
+
+ return None();
+}
+
+
+Future<Nothing> PosixDiskIsolatorProcess::isolate(
+ const ContainerID& containerId,
+ pid_t pid)
+{
+ if (!infos.contains(containerId)) {
+ return Failure("Unknown container");
+ }
+
+ return Nothing();
+}
+
+
+Future<Limitation> PosixDiskIsolatorProcess::watch(
+ const ContainerID& containerId)
+{
+ if (!infos.contains(containerId)) {
+ return Failure("Unknown container");
+ }
+
+ return infos[containerId]->limitation.future();
+}
+
+
+Future<Nothing> PosixDiskIsolatorProcess::update(
+ const ContainerID& containerId,
+ const Resources& resources)
+{
+ if (!infos.contains(containerId)) {
+ LOG(WARNING) << "Ignoring update for unknown container " << containerId;
+ return Nothing();
+ }
+
+ LOG(INFO) << "Updating the disk resources for container "
+ << containerId << " to " << resources;
+
+ const Owned<Info>& info = infos[containerId];
+
+ // This stores the updated quotas.
+ hashmap<string, Resources> quotas;
+
+ foreach (const Resource& resource, resources) {
+ if (resource.name() != "disk") {
+ continue;
+ }
+
+ // The path at which we will collect disk usage and enforce quota.
+ string path;
+
+ // NOTE: We do not allow the case where has_disk() is true but
+ // with nothing set inside DiskInfo. The master will enforce it.
+ if (!resource.has_disk()) {
+ // Regular disk used for executor working directory.
+ path = info->directory;
+ } else {
+ // TODO(jieyu): Support persistent disk as well.
+ LOG(ERROR) << "Enforcing disk quota unsupported for " << resource;
+ continue;
+ }
+
+ quotas[path] += resource;
+ }
+
+ // Initiate disk usage collection for new paths.
+ foreachkey (const string& path, quotas) {
+ if (!info->usages.contains(path)) {
+ info->usages[path] = collector.usage(path)
+ .onAny(defer(
+ PID<PosixDiskIsolatorProcess>(this),
+ &PosixDiskIsolatorProcess::_collect,
+ containerId,
+ path,
+ lambda::_1));
+ }
+ }
+
+ // Cancel unneeded disk usage collection.
+ foreachkey (const string& path, info->quotas) {
+ if (!quotas.contains(path)) {
+ CHECK(info->usages.contains(path)) << "Unknown path " << path;
+
+ info->usages[path].discard();
+ info->usages.erase(path);
+ }
+ }
+
+ info->quotas = quotas;
+
+ CHECK_EQ(info->quotas.size(), info->usages.size());
+
+ return Nothing();
+}
+
+
+void PosixDiskIsolatorProcess::_collect(
+ const ContainerID& containerId,
+ const string& path,
+ const Future<Bytes>& future)
+{
+ if (future.isDiscarded()) {
+ LOG(INFO) << "Checking disk usage at '" << path << "' for container "
+ << containerId << " has been cancelled";
+ } else if (future.isFailed()) {
+ LOG(ERROR) << "Checking disk usage at '" << path << "' for container "
+ << containerId << " has failed: " << future.failure();
+ }
+
+ if (!infos.contains(containerId)) {
+ // The container might have just been destroyed.
+ return;
+ }
+
+ const Owned<Info>& info = infos[containerId];
+
+ if (!info->quotas.contains(path)) {
+ // The path might have just been removed from this container's
+ // resources.
+ return;
+ }
+
+ // Check if the disk usage exceeds the quota. If yes, report the
+ // limitation. We keep collecting the disk usage for 'path' by
+ // initiating another round of disk usage check. The check will be
+ // throttled by DiskUsageCollector.
+ Option<Bytes> quota = info->quotas[path].disk();
+ CHECK_SOME(quota);
+
+ if (future.isReady() && future.get() > quota.get()) {
+ info->limitation.set(Limitation(
+ info->quotas[path],
+ "Disk usage (" + stringify(future.get()) +
+ ") exceeds quota (" + stringify(quota.get()) + ")"));
+ }
+
+ CHECK(info->usages.contains(path));
+
+ info->usages[path] = collector.usage(path)
+ .onAny(defer(
+ PID<PosixDiskIsolatorProcess>(this),
+ &PosixDiskIsolatorProcess::_collect,
+ containerId,
+ path,
+ lambda::_1));
+}
+
+
+Future<ResourceStatistics> PosixDiskIsolatorProcess::usage(
+ const ContainerID& containerId)
+{
+ if (!infos.contains(containerId)) {
+ VLOG(1) << "Unkonwn container " << containerId;
+ }
+
+ // TODO(jieyu): Collect disk usage statistics.
+ return ResourceStatistics();
+}
+
+
+Future<Nothing> PosixDiskIsolatorProcess::cleanup(
+ const ContainerID& containerId)
+{
+ if (!infos.contains(containerId)) {
+ LOG(WARNING) << "Ignoring cleanup for unknown container " << containerId;
+ return Nothing();
+ }
+
+ foreachvalue (Future<Bytes> future, infos[containerId]->usages) {
+ future.discard();
+ }
+
+ infos.erase(containerId);
+
+ return Nothing();
+}
+
+
+class DiskUsageCollectorProcess : public Process<DiskUsageCollectorProcess>
+{
+public:
+ DiskUsageCollectorProcess(const Duration& _interval) : interval(_interval) {}
+ virtual ~DiskUsageCollectorProcess() {}
+
+ Future<Bytes> usage(const string& path)
+ {
+ foreach (const Owned<Entry>& entry, entries) {
+ if (entry->path == path) {
+ return entry->promise.future();
+ }
+ }
+
+ entries.push_back(Owned<Entry>(new Entry(path)));
+
+ // Install onDiscard callback.
+ Future<Bytes> future = entries.back()->promise.future();
+ future.onDiscard(defer(self(), &Self::discard, path));
+
+ return future;
+ }
+
+protected:
+ void initialize()
+ {
+ schedule();
+ }
+
+ void finalize()
+ {
+ foreach (const Owned<Entry>& entry, entries) {
+ if (entry->du.isSome() && entry->du.get().status().isPending()) {
+ os::killtree(entry->du.get().pid(), SIGKILL);
+ }
+
+ entry->promise.fail("DiskUsageCollector is destroyed");
+ }
+ }
+
+private:
+ // Describe a single pending check.
+ struct Entry
+ {
+ explicit Entry(const string& _path) : path(_path) {}
+
+ string path;
+ Option<Subprocess> du;
+ Promise<Bytes> promise;
+ };
+
+ void discard(const string& path)
+ {
+ for (auto it = entries.begin(); it != entries.end(); ++it) {
+ // We only cancel those checks whose 'du' haven't been launched.
+ if ((*it)->path == path && (*it)->du.isNone()) {
+ (*it)->promise.discard();
+ entries.erase(it);
+ break;
+ }
+ }
+ }
+
+ // Schedule a 'du' to be invoked. The current implementation does
+ // not allow multiple 'du's running concurrently. The minimal
+ // interval between two subsequent 'du's is controlled by 'interval'
+ // for throttling purpose.
+ void schedule()
+ {
+ if (entries.empty()) {
+ delay(interval, self(), &Self::schedule);
+ return;
+ }
+
+ const Owned<Entry>& entry = entries.front();
+
+ // Invoke 'du' and report number of 1K-byte blocks. We fix the
+ // block size here so that we can get consistent results on all
+ // platforms (e.g., OS X uses 512 byte blocks).
+ //
+ // NOTE: The 'du' processes are run in the slave's cgroup and it
+ // will be that cgroup that is charged for (a) memory to cache the
+ // fs data structures, (b) disk I/O to read those structures, and
+ // (c) the cpu time to traverse.
+ Try<Subprocess> s = subprocess(
+ "du -k -s " + entry->path,
+ Subprocess::PATH("/dev/null"),
+ Subprocess::PIPE(),
+ Subprocess::PIPE());
+
+ if (s.isError()) {
+ entry->promise.fail("Failed to exec 'du': " + s.error());
+
+ entries.pop_front();
+ delay(interval, self(), &Self::schedule);
+ return;
+ }
+
+ entry->du = s.get();
+
+ await(s.get().status(),
+ io::read(s.get().out().get()),
+ io::read(s.get().err().get()))
+ .onAny(defer(self(), &Self::_schedule, lambda::_1));
+ }
+
+ void _schedule(const Future<tuples::tuple<
+ Future<Option<int>>,
+ Future<string>,
+ Future<string>>>& future)
+ {
+ CHECK_READY(future);
+ CHECK(!entries.empty());
+
+ const Owned<Entry>& entry = entries.front();
+ CHECK_SOME(entry->du);
+
+ Future<Option<int>> status = tuples::get<0>(future.get());
+
+ if (!status.isReady()) {
+ entry->promise.fail(
+ "Failed to perform 'du': " +
+ (status.isFailed() ? status.failure() : "discarded"));
+ } else if (status.get().isNone()) {
+ entry->promise.fail("Failed to reap the status of 'du'");
+ } else if (status.get().get() != 0) {
+ Future<string> error = tuples::get<2>(future.get());
+ if (!error.isReady()) {
+ entry->promise.fail(
+ "Failed to perform 'du'. Reading stderr failed: " +
+ (error.isFailed() ? error.failure() : "discarded"));
+ } else {
+ entry->promise.fail("Failed to perform 'du': " + error.get());
+ }
+ } else {
+ Future<string> output = tuples::get<1>(future.get());
+ if (!output.isReady()) {
+ entry->promise.fail(
+ "Failed to read stdout from 'du': " +
+ (output.isFailed() ? output.failure() : "discarded"));
+ } else {
+ // Parsing the output from 'du'. The following is a sample
+ // output. Tab is used as the delimiter between the number of
+ // blocks and the checked path.
+ // $ du /var/lib/mesos/.../runs/container_id
+ // 1024 /var/lib/mesos/.../runs/container_id
+ vector<string> tokens = strings::tokenize(output.get(), " \t");
+ if (tokens.empty()) {
+ entry->promise.fail("Unexpected output from 'du': " + output.get());
+ } else {
+ Try<size_t> value = numify<size_t>(tokens[0]);
+ if (value.isError()) {
+ entry->promise.fail("Unexpected output from 'du': " + output.get());
+ } else {
+ // Notify the callers.
+ entry->promise.set(Kilobytes(value.get()));
+ }
+ }
+ }
+ }
+
+ entries.pop_front();
+ delay(interval, self(), &Self::schedule);
+ }
+
+ const Duration interval;
+
+ // A queue of pending checks.
+ deque<Owned<Entry>> entries;
+};
+
+
+DiskUsageCollector::DiskUsageCollector(const Duration& interval)
+{
+ process = new DiskUsageCollectorProcess(interval);
+ spawn(process);
+}
+
+
+DiskUsageCollector::~DiskUsageCollector()
+{
+ terminate(process);
+ wait(process);
+ delete process;
+}
+
+
+Future<Bytes> DiskUsageCollector::usage(const string& path)
+{
+ return dispatch(process, &DiskUsageCollectorProcess::usage, path);
+}
+
+} // namespace slave {
+} // namespace internal {
+} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/788a509f/src/slave/containerizer/isolators/posix/disk.hpp
----------------------------------------------------------------------
diff --git a/src/slave/containerizer/isolators/posix/disk.hpp b/src/slave/containerizer/isolators/posix/disk.hpp
new file mode 100644
index 0000000..a32b0e7
--- /dev/null
+++ b/src/slave/containerizer/isolators/posix/disk.hpp
@@ -0,0 +1,138 @@
+/**
+ * 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.
+ */
+
+#ifndef __POSIX_DISK_ISOLATOR_HPP__
+#define __POSIX_DISK_ISOLATOR_HPP__
+
+#include <string>
+
+#include <process/owned.hpp>
+
+#include <stout/bytes.hpp>
+#include <stout/duration.hpp>
+#include <stout/hashmap.hpp>
+
+#include "slave/containerizer/isolator.hpp"
+
+namespace mesos {
+namespace internal {
+namespace slave {
+
+// Forward declarations.
+class DiskUsageCollectorProcess;
+
+
+// Responsible for collecting disk usage for paths, while ensuring
+// that an interval elapses between each collection.
+class DiskUsageCollector
+{
+public:
+ DiskUsageCollector(const Duration& interval);
+ ~DiskUsageCollector();
+
+ // Returns the disk usage rooted at 'path'. The user can discard the
+ // returned future to cancel the check.
+ process::Future<Bytes> usage(const std::string& path);
+
+private:
+ DiskUsageCollectorProcess* process;
+};
+
+
+// This isolator monitors the disk usage for containers, and reports
+// Limitation when a container exceeds its disk quota. This leverages
+// the DiskUsageCollector to ensure that we don't induce too much CPU
+// usage and disk caching effects from running 'du' too often.
+//
+// NOTE: Currently all containers are processed in the same queue,
+// which means that when a container starts, it could take many disk
+// collection intervals until any data is available in the resource
+// usage statistics!
+//
+// TODO(jieyu): Consider handling each container independently, or
+// triggering an initial collection when the container starts, to
+// ensure that we have usage statistics without a large delay.
+class PosixDiskIsolatorProcess : public IsolatorProcess
+{
+public:
+ static Try<Isolator*> create(const Flags& flags);
+
+ virtual ~PosixDiskIsolatorProcess();
+
+ virtual process::Future<Nothing> recover(
+ const std::list<state::RunState>& states);
+
+ virtual process::Future<Option<CommandInfo>> prepare(
+ const ContainerID& containerId,
+ const ExecutorInfo& executorInfo,
+ const std::string& directory,
+ const Option<std::string>& user);
+
+ virtual process::Future<Nothing> isolate(
+ const ContainerID& containerId,
+ pid_t pid);
+
+ virtual process::Future<Limitation> watch(
+ const ContainerID& containerId);
+
+ virtual process::Future<Nothing> update(
+ const ContainerID& containerId,
+ const Resources& resources);
+
+ virtual process::Future<ResourceStatistics> usage(
+ const ContainerID& containerId);
+
+ virtual process::Future<Nothing> cleanup(
+ const ContainerID& containerId);
+
+private:
+ PosixDiskIsolatorProcess(const Flags& flags);
+
+ void _collect(
+ const ContainerID& containerId,
+ const std::string& path,
+ const process::Future<Bytes>& future);
+
+ const Flags flags;
+ DiskUsageCollector collector;
+
+ struct Info
+ {
+ explicit Info(const std::string& _directory) : directory(_directory) {}
+ ~Info();
+
+ // We save executor working directory here so that we know where
+ // to collect disk usage for disk resources without DiskInfo.
+ const std::string directory;
+
+ process::Promise<Limitation> limitation;
+
+ // The keys of the hashmaps contain the executor working directory
+ // above, and optionally paths of volumes used by the container.
+ hashmap<std::string, Resources> quotas;
+ hashmap<std::string, process::Future<Bytes>> usages;
+ };
+
+ hashmap<ContainerID, process::Owned<Info>> infos;
+};
+
+} // namespace slave {
+} // namespace internal {
+} // namespace mesos {
+
+#endif // __POSIX_DISK_ISOLATOR_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/788a509f/src/slave/containerizer/mesos/containerizer.cpp
----------------------------------------------------------------------
diff --git a/src/slave/containerizer/mesos/containerizer.cpp b/src/slave/containerizer/mesos/containerizer.cpp
index 0bcf5ce..d712278 100644
--- a/src/slave/containerizer/mesos/containerizer.cpp
+++ b/src/slave/containerizer/mesos/containerizer.cpp
@@ -39,6 +39,7 @@
#endif // __linux__
#include "slave/containerizer/isolators/posix.hpp"
+#include "slave/containerizer/isolators/posix/disk.hpp"
#ifdef __linux__
#include "slave/containerizer/isolators/cgroups/cpushare.hpp"
#include "slave/containerizer/isolators/cgroups/mem.hpp"
@@ -106,6 +107,7 @@ Try<MesosContainerizer*> MesosContainerizer::create(
creators["posix/cpu"] = &PosixCpuIsolatorProcess::create;
creators["posix/mem"] = &PosixMemIsolatorProcess::create;
+ creators["posix/disk"] = &PosixDiskIsolatorProcess::create;
#ifdef __linux__
creators["cgroups/cpu"] = &CgroupsCpushareIsolatorProcess::create;
creators["cgroups/mem"] = &CgroupsMemIsolatorProcess::create;
http://git-wip-us.apache.org/repos/asf/mesos/blob/788a509f/src/slave/flags.hpp
----------------------------------------------------------------------
diff --git a/src/slave/flags.hpp b/src/slave/flags.hpp
index a09ac73..a4498e6 100644
--- a/src/slave/flags.hpp
+++ b/src/slave/flags.hpp
@@ -381,6 +381,12 @@ public:
false);
#endif // WITH_NETWORK_ISOLATOR
+ add(&Flags::disk_quota_check_interval,
+ "disk_quota_check_interval",
+ "The interval between disk quota checks for containers. This flag is\n"
+ "used for the posix/disk' isolator.",
+ Seconds(30));
+
// This help message for --modules flag is the same for
// {master,slave,tests}/flags.hpp and should always be kept in
// sync.
@@ -491,6 +497,7 @@ public:
Option<Bytes> egress_rate_limit_per_container;
bool network_enable_socket_statistics;
#endif
+ Duration disk_quota_check_interval;
Option<Modules> modules;
std::string authenticatee;
Option<std::string> hooks;