You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by ya...@apache.org on 2017/12/01 23:17:49 UTC
[1/3] mesos git commit: Fixed a bug in devolving framework
subscription with suppressed roles.
Repository: mesos
Updated Branches:
refs/heads/master 6cf64ed9c -> 8c2f972b5
Fixed a bug in devolving framework subscription with suppressed roles.
Review: https://reviews.apache.org/r/63741
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/5d9209e6
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/5d9209e6
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/5d9209e6
Branch: refs/heads/master
Commit: 5d9209e69a0a9600ec8c02fbf852ab912b208a88
Parents: 6cf64ed
Author: Jiang Yan Xu <xu...@apple.com>
Authored: Fri Nov 10 12:16:45 2017 -0800
Committer: Jiang Yan Xu <xu...@apple.com>
Committed: Fri Dec 1 15:11:22 2017 -0800
----------------------------------------------------------------------
src/internal/devolve.cpp | 12 +++++++++++-
src/internal/evolve.cpp | 15 ++++++++++++++-
src/master/master.cpp | 6 +++++-
3 files changed, 30 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/5d9209e6/src/internal/devolve.cpp
----------------------------------------------------------------------
diff --git a/src/internal/devolve.cpp b/src/internal/devolve.cpp
index 3a02490..289c6e3 100644
--- a/src/internal/devolve.cpp
+++ b/src/internal/devolve.cpp
@@ -193,7 +193,17 @@ mesos::resource_provider::Event devolve(
scheduler::Call devolve(const v1::scheduler::Call& call)
{
- return devolve<scheduler::Call>(call);
+ scheduler::Call _call = devolve<scheduler::Call>(call);
+
+ // Certain conversions require special handling.
+ if (_call.type() == scheduler::Call::SUBSCRIBE) {
+ // v1 Subscribe.suppressed_roles cannot be automatically converted
+ // because its tag is used by another field in the internal Subscribe.
+ *(_call.mutable_subscribe()->mutable_suppressed_roles()) =
+ call.subscribe().suppressed_roles();
+ }
+
+ return _call;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/5d9209e6/src/internal/evolve.cpp
----------------------------------------------------------------------
diff --git a/src/internal/evolve.cpp b/src/internal/evolve.cpp
index cb1c0eb..f46f864 100644
--- a/src/internal/evolve.cpp
+++ b/src/internal/evolve.cpp
@@ -244,9 +244,22 @@ v1::resource_provider::Event evolve(
}
+// TODO(xujyan): Do we need this conversion when Mesos never sends out
+// `scheduler::Call` thus never needs to evovle internal call to a v1 call?
+// Perhaps we should remove the method so there's no need to maintain it.
v1::scheduler::Call evolve(const scheduler::Call& call)
{
- return evolve<v1::scheduler::Call>(call);
+ v1::scheduler::Call _call = evolve<v1::scheduler::Call>(call);
+
+ // Certain conversions require special handling.
+ if (_call.type() == v1::scheduler::Call::SUBSCRIBE) {
+ // v1 Subscribe.suppressed_roles cannot be automatically converted
+ // because its tag is used by another field in the internal Subscribe.
+ *(_call.mutable_subscribe()->mutable_suppressed_roles()) =
+ call.subscribe().suppressed_roles();
+ }
+
+ return _call;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/5d9209e6/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index dfe60ef..398373f 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -6947,7 +6947,8 @@ void Master::updateFramework(
const FrameworkInfo& frameworkInfo,
const set<string>& suppressedRoles)
{
- LOG(INFO) << "Updating info for framework " << framework->id();
+ LOG(INFO) << "Updating framework " << *framework << " with roles "
+ << stringify(suppressedRoles) << " suppressed";
// NOTE: The allocator takes care of activating/deactivating
// the frameworks from the added/removed roles, respectively.
@@ -8807,6 +8808,9 @@ void Master::addFramework(
CHECK(!frameworks.registered.contains(framework->id()))
<< "Framework " << *framework << " already exists!";
+ LOG(INFO) << "Adding framework " << *framework << " with roles "
+ << stringify(suppressedRoles) << " suppressed";
+
frameworks.registered[framework->id()] = framework;
if (framework->connected()) {
[2/3] mesos git commit: Fixed 'NoOffersWithAllRolesSuppressed' test.
Posted by ya...@apache.org.
Fixed 'NoOffersWithAllRolesSuppressed' test.
Review: https://reviews.apache.org/r/63830
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/3711233f
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/3711233f
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/3711233f
Branch: refs/heads/master
Commit: 3711233fcec761be8625af6a028a228fe9d8dc5a
Parents: 5d9209e
Author: Jiang Yan Xu <xu...@apple.com>
Authored: Fri Nov 10 12:15:37 2017 -0800
Committer: Jiang Yan Xu <xu...@apple.com>
Committed: Fri Dec 1 15:11:23 2017 -0800
----------------------------------------------------------------------
src/tests/scheduler_tests.cpp | 46 ++++++++++++++++++++++++++------------
1 file changed, 32 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/3711233f/src/tests/scheduler_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/scheduler_tests.cpp b/src/tests/scheduler_tests.cpp
index 45fc9c0..566ffb3 100644
--- a/src/tests/scheduler_tests.cpp
+++ b/src/tests/scheduler_tests.cpp
@@ -45,12 +45,13 @@
#include "internal/devolve.hpp"
#include "internal/evolve.hpp"
+#include "master/constants.hpp"
+#include "master/master.hpp"
+
#include "master/allocator/mesos/allocator.hpp"
#include "master/detector/standalone.hpp"
-#include "master/master.hpp"
-
#include "tests/containerizer.hpp"
#include "tests/mesos.hpp"
@@ -1450,9 +1451,9 @@ TEST_P(SchedulerTest, Suppress)
}
-// TODO(alexr): Re-enable this test after MESOS-8200 is resolved and the test
-// itself is fixed as well, see MESOS-7996.
-TEST_P(SchedulerTest, DISABLED_NoOffersWithAllRolesSuppressed)
+// This test verifies that when a framework registers with all roles
+// suppressing offers, it does not receive offers.
+TEST_P(SchedulerTest, NoOffersWithAllRolesSuppressed)
{
master::Flags flags = CreateMasterFlags();
@@ -1460,9 +1461,17 @@ TEST_P(SchedulerTest, DISABLED_NoOffersWithAllRolesSuppressed)
ASSERT_SOME(master);
Owned<MasterDetector> detector = master.get()->createDetector();
+
+ Future<SlaveRegisteredMessage> slaveRegisteredMessage =
+ FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
+
Try<Owned<cluster::Slave>> slave = StartSlave(detector.get());
ASSERT_SOME(slave);
+ AWAIT_READY(slaveRegisteredMessage);
+
+ Clock::pause();
+
auto scheduler = std::make_shared<v1::MockHTTPScheduler>();
Future<Nothing> connected;
@@ -1482,12 +1491,15 @@ TEST_P(SchedulerTest, DISABLED_NoOffersWithAllRolesSuppressed)
EXPECT_CALL(*scheduler, subscribed(_, _))
.WillOnce(FutureArg<1>(&subscribed));
+ Future<Nothing> heartbeat;
EXPECT_CALL(*scheduler, heartbeat(_))
- .WillRepeatedly(Return()); // Ignore heartbeats.
+ .WillOnce(FutureSatisfy(&heartbeat));
+ // The framework will subscribe with its role being suppressed so no
+ // offers should be received by the framework.
Future<Event::Offers> offers;
EXPECT_CALL(*scheduler, offers(_, _))
- .Times(0); // No offers extended since all roles are suppressed.
+ .Times(0);
{
Call call;
@@ -1502,19 +1514,25 @@ TEST_P(SchedulerTest, DISABLED_NoOffersWithAllRolesSuppressed)
mesos.send(call);
}
- // Since the framework is subscribed with its role being suppressed, no
- // offers should be received by the framework.
- Clock::pause();
- Clock::advance(flags.allocation_interval);
- Clock::resume();
-
AWAIT_READY(subscribed);
+ AWAIT_READY(heartbeat);
+
+ // We use an additional heartbeat as a synchronization mechanism to make
+ // sure an offer would be received by the scheduler if one was ever extended.
+ // Note that Clock::settle() wouldn't be sufficient here.
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillOnce(FutureSatisfy(&heartbeat))
+ .WillRepeatedly(Return()); // Ignore additional heartbeats.
+
+ Clock::advance(master::DEFAULT_HEARTBEAT_INTERVAL);
+ AWAIT_READY(heartbeat);
+
v1::FrameworkID frameworkId(subscribed->framework_id());
+ // On revival the scheduler should get an offer.
EXPECT_CALL(*scheduler, offers(_, _))
.WillOnce(FutureArg<1>(&offers));
- // On revival the scheduler should get an offer.
{
Call call;
call.mutable_framework_id()->CopyFrom(frameworkId);
[3/3] mesos git commit: Fixed a bug that removed the suppressed
framework from sorter.
Posted by ya...@apache.org.
Fixed a bug that removed the suppressed framework from sorter.
Review: https://reviews.apache.org/r/63831
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/8c2f972b
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/8c2f972b
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/8c2f972b
Branch: refs/heads/master
Commit: 8c2f972b5c0c42e1519d09275cc26e1765a0c5de
Parents: 3711233
Author: Jiang Yan Xu <xu...@apple.com>
Authored: Tue Nov 14 00:12:17 2017 -0800
Committer: Jiang Yan Xu <xu...@apple.com>
Committed: Fri Dec 1 15:12:57 2017 -0800
----------------------------------------------------------------------
src/master/allocator/mesos/hierarchical.cpp | 48 ++++----
src/tests/scheduler_tests.cpp | 139 +++++++++++++++++++++++
2 files changed, 164 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/8c2f972b/src/master/allocator/mesos/hierarchical.cpp
----------------------------------------------------------------------
diff --git a/src/master/allocator/mesos/hierarchical.cpp b/src/master/allocator/mesos/hierarchical.cpp
index ab2abf8..715650e 100644
--- a/src/master/allocator/mesos/hierarchical.cpp
+++ b/src/master/allocator/mesos/hierarchical.cpp
@@ -35,6 +35,7 @@
#include <stout/check.hpp>
#include <stout/hashset.hpp>
+#include <stout/set.hpp>
#include <stout/stopwatch.hpp>
#include <stout/stringify.hpp>
@@ -421,27 +422,30 @@ void HierarchicalAllocatorProcess::updateFramework(
set<string> newRoles = protobuf::framework::getRoles(frameworkInfo);
set<string> oldSuppressedRoles = framework.suppressedRoles;
- // The roles which are candidates for deactivation are the roles that are
- // removed, as well as the roles which have moved from non-suppressed
- // to suppressed mode.
- const set<string> rolesToDeactivate = [&]() {
+ // TODO(xujyan): Add a stout set difference method that wraps around
+ // `std::set_difference` for this.
+ const set<string> removedRoles = [&]() {
set<string> result = oldRoles;
foreach (const string& role, newRoles) {
result.erase(role);
}
+ return result;
+ }();
- foreach (const string& role, oldRoles) {
- if (!oldSuppressedRoles.count(role) && suppressedRoles.count(role)) {
- result.insert(role);
- }
+ const set<string> newSuppressedRoles = [&]() {
+ set<string> result = suppressedRoles;
+ foreach (const string& role, oldSuppressedRoles) {
+ result.erase(role);
}
return result;
}();
- foreach (const string& role, rolesToDeactivate) {
+ foreach (const string& role, removedRoles | newSuppressedRoles) {
CHECK(frameworkSorters.contains(role));
frameworkSorters.at(role)->deactivate(frameworkId.value());
+ }
+ foreach (const string& role, removedRoles) {
// Stop tracking the framework under this role if there are
// no longer any resources allocated to it.
if (frameworkSorters.at(role)->allocation(frameworkId.value()).empty()) {
@@ -453,36 +457,34 @@ void HierarchicalAllocatorProcess::updateFramework(
}
}
- // The roles which are candidates for activation are the roles that are
- // added, as well as the roles which have moved from suppressed to
- // non-suppressed mode.
- //
- // TODO(anindya_sinha): We should activate the roles only if the
- // framework is active (instead of always).
- const set<string> rolesToActivate = [&]() {
+ const set<string> addedRoles = [&]() {
set<string> result = newRoles;
foreach (const string& role, oldRoles) {
result.erase(role);
}
+ return result;
+ }();
- foreach (const string& role, newRoles) {
- if (!suppressedRoles.count(role) && oldSuppressedRoles.count(role)) {
- result.insert(role);
- } else if (suppressedRoles.count(role)) {
- result.erase(role);
- }
+ const set<string> newRevivedRoles = [&]() {
+ set<string> result = oldSuppressedRoles;
+ foreach (const string& role, suppressedRoles) {
+ result.erase(role);
}
return result;
}();
- foreach (const string& role, rolesToActivate) {
+ foreach (const string& role, addedRoles) {
// NOTE: It's possible that we're already tracking this framework
// under the role because a framework can unsubscribe from a role
// while it still has resources allocated to the role.
if (!isFrameworkTrackedUnderRole(frameworkId, role)) {
trackFrameworkUnderRole(frameworkId, role);
}
+ }
+ // TODO(anindya_sinha): We should activate the roles only if the
+ // framework is active (instead of always).
+ foreach (const string& role, addedRoles | newRevivedRoles) {
CHECK(frameworkSorters.contains(role));
frameworkSorters.at(role)->activate(frameworkId.value());
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/8c2f972b/src/tests/scheduler_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/scheduler_tests.cpp b/src/tests/scheduler_tests.cpp
index 566ffb3..29cab82 100644
--- a/src/tests/scheduler_tests.cpp
+++ b/src/tests/scheduler_tests.cpp
@@ -1551,6 +1551,145 @@ TEST_P(SchedulerTest, NoOffersWithAllRolesSuppressed)
}
+// This test verifies that if a framework (initially with no roles
+// suppressed) decides to suppress offers for its roles on reregisteration,
+// no offers will be made.
+TEST_P(SchedulerTest, NoOffersOnReregistrationWithAllRolesSuppressed)
+{
+ master::Flags flags = CreateMasterFlags();
+
+ Try<Owned<cluster::Master>> master = StartMaster(flags);
+ ASSERT_SOME(master);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+
+ Future<SlaveRegisteredMessage> slaveRegisteredMessage =
+ FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
+
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get());
+ ASSERT_SOME(slave);
+
+ AWAIT_READY(slaveRegisteredMessage);
+
+ Clock::pause();
+
+ auto scheduler = std::make_shared<v1::MockHTTPScheduler>();
+
+ Future<Nothing> connected;
+ EXPECT_CALL(*scheduler, connected(_))
+ .WillOnce(FutureSatisfy(&connected));
+
+ ContentType contentType = GetParam();
+
+ v1::scheduler::TestMesos mesos(
+ master.get()->pid,
+ contentType,
+ scheduler);
+
+ AWAIT_READY(connected);
+
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
+
+ Future<Nothing> heartbeat;
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillOnce(FutureSatisfy(&heartbeat));
+
+ Future<Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
+
+ {
+ Call call;
+ call.set_type(Call::SUBSCRIBE);
+
+ v1::FrameworkInfo frameworkInfo = v1::DEFAULT_FRAMEWORK_INFO;
+
+ // Enable failover.
+ frameworkInfo.set_failover_timeout(Weeks(1).secs());
+
+ Call::Subscribe* subscribe = call.mutable_subscribe();
+ *(subscribe->mutable_framework_info()) = frameworkInfo;
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(subscribed);
+ AWAIT_READY(heartbeat);
+
+ v1::FrameworkID frameworkId(subscribed->framework_id());
+
+ AWAIT_READY(offers);
+ EXPECT_FALSE(offers->offers().empty());
+
+ // Now fail over and reregister with all roles suppressed.
+ EXPECT_CALL(*scheduler, disconnected(_));
+
+ EXPECT_CALL(*scheduler, connected(_))
+ .WillOnce(FutureSatisfy(&connected));
+
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillOnce(FutureSatisfy(&heartbeat));
+
+ // The framework will subscribe with its role being suppressed so no
+ // offers should be received by the framework.
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .Times(0);
+
+ // Now fail over the scheduler.
+ mesos.reconnect();
+
+ AWAIT_READY(connected);
+
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
+
+ {
+ Call call;
+ call.set_type(Call::SUBSCRIBE);
+ *(call.mutable_framework_id()) = frameworkId;
+
+ v1::FrameworkInfo frameworkInfo = v1::DEFAULT_FRAMEWORK_INFO;
+ *(frameworkInfo.mutable_id()) = frameworkId;
+
+ Call::Subscribe* subscribe = call.mutable_subscribe();
+ *(subscribe->mutable_framework_info()) = frameworkInfo;
+ subscribe->add_suppressed_roles(frameworkInfo.role());
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(subscribed);
+ AWAIT_READY(heartbeat);
+
+ // We use an additional heartbeat as a synchronization mechanism to make
+ // sure an offer would be received by the scheduler if one was ever extended.
+ // Note that Clock::settle() wouldn't be sufficient here.
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillOnce(FutureSatisfy(&heartbeat))
+ .WillRepeatedly(Return()); // Ignore additional heartbeats.
+
+ Clock::advance(master::DEFAULT_HEARTBEAT_INTERVAL);
+ AWAIT_READY(heartbeat);
+
+ // On revival the scheduler should get an offer.
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
+
+ {
+ Call call;
+ *(call.mutable_framework_id()) = frameworkId;
+ call.set_type(Call::REVIVE);
+
+ mesos.send(call);
+ }
+
+ AWAIT_READY(offers);
+ EXPECT_FALSE(offers->offers().empty());
+}
+
+
TEST_P(SchedulerTest, Message)
{
Try<Owned<cluster::Master>> master = StartMaster();