You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by vi...@apache.org on 2016/03/04 01:10:36 UTC
[1/3] mesos git commit: Modified scheduler tests to use the callback
interface.
Repository: mesos
Updated Branches:
refs/heads/master 5387a4d8a -> 118b6ea88
Modified scheduler tests to use the callback interface.
This change modifies the existing scheduler tests to use the new callback
interface introduced in MESOS-3339.
Review: https://reviews.apache.org/r/44273/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/b446dbf2
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/b446dbf2
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/b446dbf2
Branch: refs/heads/master
Commit: b446dbf2d797168b1d48c7a9462a51101b982a1a
Parents: 5387a4d
Author: Anand Mazumdar <ma...@gmail.com>
Authored: Thu Mar 3 16:09:49 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Thu Mar 3 16:09:49 2016 -0800
----------------------------------------------------------------------
src/tests/scheduler_tests.cpp | 736 ++++++++++++++++++-------------------
1 file changed, 360 insertions(+), 376 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/b446dbf2/src/tests/scheduler_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/scheduler_tests.cpp b/src/tests/scheduler_tests.cpp
index 06cef6b..fa42fb4 100644
--- a/src/tests/scheduler_tests.cpp
+++ b/src/tests/scheduler_tests.cpp
@@ -76,19 +76,9 @@ namespace internal {
namespace tests {
-class SchedulerTest : public MesosTest, public WithParamInterface<ContentType>
-{
-protected:
- // Helper class for using EXPECT_CALL since the Mesos scheduler API
- // is callback based.
- class Callbacks
- {
- public:
- MOCK_METHOD0(connected, void());
- MOCK_METHOD0(disconnected, void());
- MOCK_METHOD1(received, void(const std::queue<Event>&));
- };
-};
+class SchedulerTest
+ : public MesosTest,
+ public WithParamInterface<ContentType> {};
// The scheduler library tests are parameterized by the content type
@@ -99,24 +89,6 @@ INSTANTIATE_TEST_CASE_P(
::testing::Values(ContentType::PROTOBUF, ContentType::JSON));
-// Enqueues all received events into a libprocess queue.
-ACTION_P(Enqueue, queue)
-{
- std::queue<Event> events = arg0;
- while (!events.empty()) {
- // Note that we currently drop HEARTBEATs because most of these tests
- // are not designed to deal with heartbeats.
- // TODO(vinod): Implement DROP_HTTP_CALLS that can filter heartbeats.
- if (events.front().type() == Event::HEARTBEAT) {
- VLOG(1) << "Ignoring HEARTBEAT event";
- } else {
- queue->put(events.front());
- }
- events.pop();
- }
-}
-
-
// This test verifies that a scheduler can subscribe with the master.
TEST_P(SchedulerTest, Subscribe)
{
@@ -126,39 +98,38 @@ TEST_P(SchedulerTest, Subscribe)
Try<PID<Master>> master = StartMaster(flags);
ASSERT_SOME(master);
- Callbacks callbacks;
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
- Call call;
- call.set_type(Call::SUBSCRIBE);
- Call::Subscribe* subscribe = call.mutable_subscribe();
- subscribe->mutable_framework_info()->CopyFrom(DEFAULT_V1_FRAMEWORK_INFO);
+ {
+ Call call;
+ call.set_type(Call::SUBSCRIBE);
+ Call::Subscribe* subscribe = call.mutable_subscribe();
+ subscribe->mutable_framework_info()->CopyFrom(DEFAULT_V1_FRAMEWORK_INFO);
- mesos.send(call);
+ mesos.send(call);
+ }
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown();
@@ -177,9 +148,6 @@ TEST_P(SchedulerTest, SchedulerFailover)
auto scheduler = std::make_shared<MockV1HTTPScheduler>();
- EXPECT_CALL(*scheduler, heartbeat(_))
- .WillRepeatedly(Return()); // Ignore heartbeats.
-
Future<Nothing> connected;
EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
@@ -195,6 +163,9 @@ TEST_P(SchedulerTest, SchedulerFailover)
EXPECT_CALL(*scheduler, subscribed(_, _))
.WillOnce(FutureArg<1>(&subscribed));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
{
Call call;
call.set_type(Call::SUBSCRIBE);
@@ -211,9 +182,6 @@ TEST_P(SchedulerTest, SchedulerFailover)
auto scheduler2 = std::make_shared<MockV1HTTPScheduler>();
- EXPECT_CALL(*scheduler2, heartbeat(_))
- .WillRepeatedly(Return()); // Ignore heartbeats.
-
EXPECT_CALL(*scheduler2, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
@@ -236,6 +204,9 @@ TEST_P(SchedulerTest, SchedulerFailover)
EXPECT_CALL(*scheduler2, subscribed(_, _))
.WillOnce(FutureArg<1>(&subscribed));
+ EXPECT_CALL(*scheduler2, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
{
Call call;
call.mutable_framework_id()->CopyFrom(frameworkId);
@@ -276,9 +247,6 @@ TEST_P(SchedulerTest, MasterFailover)
auto scheduler = std::make_shared<MockV1HTTPScheduler>();
auto detector = std::make_shared<StandaloneMasterDetector>(master.get());
- EXPECT_CALL(*scheduler, heartbeat(_))
- .WillRepeatedly(Return()); // Ignore heartbeats.
-
Future<Nothing> connected;
EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
@@ -294,6 +262,9 @@ TEST_P(SchedulerTest, MasterFailover)
EXPECT_CALL(*scheduler, subscribed(_, _))
.WillOnce(FutureArg<1>(&subscribed));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
{
Call call;
call.set_type(Call::SUBSCRIBE);
@@ -359,34 +330,36 @@ TEST_P(SchedulerTest, TaskRunning)
Try<PID<Master>> master = StartMaster(flags);
ASSERT_SOME(master);
- ExecutorID executorId = DEFAULT_EXECUTOR_ID;
-
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
auto executor = std::make_shared<MockV1HTTPExecutor>();
+
+ ExecutorID executorId = DEFAULT_EXECUTOR_ID;
TestContainerizer containerizer(executorId, executor);
Try<PID<Slave>> slave = StartSlave(&containerizer);
ASSERT_SOME(slave);
- Callbacks callbacks;
-
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
+ Future<Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
{
Call call;
@@ -398,30 +371,30 @@ TEST_P(SchedulerTest, TaskRunning)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ v1::FrameworkID frameworkId(subscribed.get().framework_id());
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
EXPECT_CALL(*executor, connected(_))
- .WillOnce(executor::SendSubscribe(id, evolve(executorId)));
+ .WillOnce(executor::SendSubscribe(frameworkId, evolve(executorId)));
EXPECT_CALL(*executor, subscribed(_, _));
EXPECT_CALL(*executor, launch(_, _))
.WillOnce(executor::SendUpdateFromTask(
- id, evolve(executorId), v1::TASK_RUNNING));
+ frameworkId, evolve(executorId), v1::TASK_RUNNING));
Future<Nothing> acknowledged;
EXPECT_CALL(*executor, acknowledged(_, _))
.WillOnce(FutureSatisfy(&acknowledged));
+ Future<Event::Update> statusUpdate;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&statusUpdate));
+
Future<Nothing> update;
EXPECT_CALL(containerizer, update(_, _))
.WillOnce(DoAll(FutureSatisfy(&update),
@@ -432,9 +405,9 @@ TEST_P(SchedulerTest, TaskRunning)
taskInfo.set_name("");
taskInfo.mutable_task_id()->set_value("1");
taskInfo.mutable_agent_id()->CopyFrom(
- event.get().offers().offers(0).agent_id());
+ offers->offers(0).agent_id());
taskInfo.mutable_resources()->CopyFrom(
- event.get().offers().offers(0).resources());
+ offers->offers(0).resources());
taskInfo.mutable_executor()->CopyFrom(DEFAULT_V1_EXECUTOR_INFO);
// TODO(benh): Enable just running a task with a command in the tests:
@@ -442,11 +415,11 @@ TEST_P(SchedulerTest, TaskRunning)
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::ACCEPT);
Call::Accept* accept = call.mutable_accept();
- accept->add_offer_ids()->CopyFrom(event.get().offers().offers(0).id());
+ accept->add_offer_ids()->CopyFrom(offers->offers(0).id());
v1::Offer::Operation* operation = accept->add_operations();
operation->set_type(v1::Offer::Operation::LAUNCH);
@@ -456,13 +429,11 @@ TEST_P(SchedulerTest, TaskRunning)
}
AWAIT_READY(acknowledged);
+ AWAIT_READY(statusUpdate);
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::UPDATE, event.get().type());
- EXPECT_EQ(v1::TASK_RUNNING, event.get().update().status().state());
- EXPECT_TRUE(event.get().update().status().has_executor_id());
- EXPECT_EQ(executorId, devolve(event.get().update().status().executor_id()));
+ EXPECT_EQ(v1::TASK_RUNNING, statusUpdate->status().state());
+ EXPECT_TRUE(statusUpdate->status().has_executor_id());
+ EXPECT_EQ(executorId, devolve(statusUpdate->status().executor_id()));
AWAIT_READY(update);
@@ -472,7 +443,7 @@ TEST_P(SchedulerTest, TaskRunning)
EXPECT_CALL(*executor, disconnected(_))
.Times(AtMost(1));
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -487,34 +458,36 @@ TEST_P(SchedulerTest, ReconcileTask)
Try<PID<Master>> master = StartMaster(flags);
ASSERT_SOME(master);
- ExecutorID executorId = DEFAULT_EXECUTOR_ID;
-
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
auto executor = std::make_shared<MockV1HTTPExecutor>();
+
+ ExecutorID executorId = DEFAULT_EXECUTOR_ID;
TestContainerizer containerizer(executorId, executor);
Try<PID<Slave>> slave = StartSlave(&containerizer);
ASSERT_SOME(slave);
- Callbacks callbacks;
-
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
+ Future<Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
{
Call call;
@@ -526,38 +499,38 @@ TEST_P(SchedulerTest, ReconcileTask)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ v1::FrameworkID frameworkId(subscribed->framework_id());
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
EXPECT_CALL(*executor, connected(_))
- .WillOnce(executor::SendSubscribe(id, evolve(executorId)));
+ .WillOnce(executor::SendSubscribe(frameworkId, evolve(executorId)));
EXPECT_CALL(*executor, subscribed(_, _));
EXPECT_CALL(*executor, launch(_, _))
.WillOnce(executor::SendUpdateFromTask(
- id, evolve(executorId), v1::TASK_RUNNING));
+ frameworkId, evolve(executorId), v1::TASK_RUNNING));
Future<Nothing> acknowledged;
EXPECT_CALL(*executor, acknowledged(_, _))
.WillOnce(FutureSatisfy(&acknowledged));
- v1::Offer offer = event.get().offers().offers(0);
+ Future<Event::Update> update1;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update1));
+
+ const v1::Offer& offer = offers->offers(0);
v1::TaskInfo taskInfo =
evolve(createTask(devolve(offer), "", DEFAULT_EXECUTOR_ID));
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::ACCEPT);
Call::Accept* accept = call.mutable_accept();
@@ -571,15 +544,17 @@ TEST_P(SchedulerTest, ReconcileTask)
}
AWAIT_READY(acknowledged);
+ AWAIT_READY(update1);
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::UPDATE, event.get().type());
- EXPECT_EQ(v1::TASK_RUNNING, event.get().update().status().state());
+ EXPECT_EQ(v1::TASK_RUNNING, update1->status().state());
+
+ Future<Event::Update> update2;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update2));
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::RECONCILE);
Call::Reconcile::Task* task = call.mutable_reconcile()->add_tasks();
@@ -588,13 +563,12 @@ TEST_P(SchedulerTest, ReconcileTask)
mesos.send(call);
}
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::UPDATE, event.get().type());
- EXPECT_FALSE(event.get().update().status().has_uuid());
- EXPECT_EQ(v1::TASK_RUNNING, event.get().update().status().state());
+ AWAIT_READY(update2);
+
+ EXPECT_FALSE(update2->status().has_uuid());
+ EXPECT_EQ(v1::TASK_RUNNING, update2->status().state());
EXPECT_EQ(v1::TaskStatus::REASON_RECONCILIATION,
- event.get().update().status().reason());
+ update2->status().reason());
EXPECT_CALL(*executor, shutdown(_))
.Times(AtMost(1));
@@ -602,7 +576,7 @@ TEST_P(SchedulerTest, ReconcileTask)
EXPECT_CALL(*executor, disconnected(_))
.Times(AtMost(1));
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -617,34 +591,36 @@ TEST_P(SchedulerTest, KillTask)
Try<PID<Master>> master = StartMaster(flags);
ASSERT_SOME(master);
- ExecutorID executorId = DEFAULT_EXECUTOR_ID;
-
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
auto executor = std::make_shared<MockV1HTTPExecutor>();
+
+ ExecutorID executorId = DEFAULT_EXECUTOR_ID;
TestContainerizer containerizer(executorId, executor);
Try<PID<Slave>> slave = StartSlave(&containerizer);
ASSERT_SOME(slave);
- Callbacks callbacks;
-
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
+ Future<Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
{
Call call;
@@ -656,39 +632,39 @@ TEST_P(SchedulerTest, KillTask)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ v1::FrameworkID frameworkId(subscribed.get().framework_id());
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
EXPECT_CALL(*executor, connected(_))
- .WillOnce(executor::SendSubscribe(id, evolve(executorId)));
+ .WillOnce(executor::SendSubscribe(frameworkId, evolve(executorId)));
EXPECT_CALL(*executor, subscribed(_, _));
EXPECT_CALL(*executor, launch(_, _))
.WillOnce(executor::SendUpdateFromTask(
- id, evolve(executorId), v1::TASK_RUNNING));
+ frameworkId, evolve(executorId), v1::TASK_RUNNING));
Future<Nothing> acknowledged;
EXPECT_CALL(*executor, acknowledged(_, _))
.WillOnce(FutureSatisfy(&acknowledged))
.WillRepeatedly(Return());
- v1::Offer offer = event.get().offers().offers(0);
+ Future<Event::Update> update1;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update1));
+
+ const v1::Offer& offer = offers->offers(0);
v1::TaskInfo taskInfo =
evolve(createTask(devolve(offer), "", DEFAULT_EXECUTOR_ID));
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::ACCEPT);
Call::Accept* accept = call.mutable_accept();
@@ -702,33 +678,35 @@ TEST_P(SchedulerTest, KillTask)
}
AWAIT_READY(acknowledged);
+ AWAIT_READY(update1);
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::UPDATE, event.get().type());
- EXPECT_EQ(v1::TASK_RUNNING, event.get().update().status().state());
+ EXPECT_EQ(v1::TASK_RUNNING, update1->status().state());
{
// Acknowledge TASK_RUNNING update.
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::ACKNOWLEDGE);
Call::Acknowledge* acknowledge = call.mutable_acknowledge();
acknowledge->mutable_task_id()->CopyFrom(taskInfo.task_id());
acknowledge->mutable_agent_id()->CopyFrom(offer.agent_id());
- acknowledge->set_uuid(event.get().update().status().uuid());
+ acknowledge->set_uuid(update1->status().uuid());
mesos.send(call);
}
+ Future<Event::Update> update2;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update2));
+
EXPECT_CALL(*executor, kill(_, _))
.WillOnce(executor::SendUpdateFromTaskID(
- id, evolve(executorId), v1::TASK_KILLED));
+ frameworkId, evolve(executorId), v1::TASK_KILLED));
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::KILL);
Call::Kill* kill = call.mutable_kill();
@@ -738,10 +716,9 @@ TEST_P(SchedulerTest, KillTask)
mesos.send(call);
}
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::UPDATE, event.get().type());
- EXPECT_EQ(v1::TASK_KILLED, event.get().update().status().state());
+ AWAIT_READY(update2);
+
+ EXPECT_EQ(v1::TASK_KILLED, update2->status().state());
EXPECT_CALL(*executor, shutdown(_))
.Times(AtMost(1));
@@ -749,7 +726,7 @@ TEST_P(SchedulerTest, KillTask)
EXPECT_CALL(*executor, disconnected(_))
.Times(AtMost(1));
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -764,34 +741,37 @@ TEST_P(SchedulerTest, ShutdownExecutor)
Try<PID<Master>> master = StartMaster(flags);
ASSERT_SOME(master);
- ExecutorID executorId = DEFAULT_EXECUTOR_ID;
-
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
auto executor = std::make_shared<MockV1HTTPExecutor>();
+
+ ExecutorID executorId = DEFAULT_EXECUTOR_ID;
TestContainerizer containerizer(executorId, executor);
Try<PID<Slave>> slave = StartSlave(&containerizer);
ASSERT_SOME(slave);
- Callbacks callbacks;
-
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
+
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ Future<Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers))
+ .WillRepeatedly(Return());
{
Call call;
@@ -803,38 +783,38 @@ TEST_P(SchedulerTest, ShutdownExecutor)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ v1::FrameworkID frameworkId(subscribed->framework_id());
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
EXPECT_CALL(*executor, connected(_))
- .WillOnce(executor::SendSubscribe(id, evolve(executorId)));
+ .WillOnce(executor::SendSubscribe(frameworkId, evolve(executorId)));
EXPECT_CALL(*executor, subscribed(_, _));
EXPECT_CALL(*executor, launch(_, _))
.WillOnce(executor::SendUpdateFromTask(
- id, evolve(executorId), v1::TASK_FINISHED));
+ frameworkId, evolve(executorId), v1::TASK_FINISHED));
Future<Nothing> acknowledged;
EXPECT_CALL(*executor, acknowledged(_, _))
.WillOnce(FutureSatisfy(&acknowledged));
- v1::Offer offer = event.get().offers().offers(0);
+ Future<Event::Update> update;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update));
+
+ const v1::Offer& offer = offers->offers(0);
v1::TaskInfo taskInfo =
evolve(createTask(devolve(offer), "", DEFAULT_EXECUTOR_ID));
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::ACCEPT);
Call::Accept* accept = call.mutable_accept();
@@ -848,19 +828,21 @@ TEST_P(SchedulerTest, ShutdownExecutor)
}
AWAIT_READY(acknowledged);
+ AWAIT_READY(update);
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::UPDATE, event.get().type());
- EXPECT_EQ(v1::TASK_FINISHED, event.get().update().status().state());
+ EXPECT_EQ(v1::TASK_FINISHED, update->status().state());
Future<Nothing> shutdown;
EXPECT_CALL(*executor, shutdown(_))
.WillOnce(FutureSatisfy(&shutdown));
+ Future<Event::Failure> failure;
+ EXPECT_CALL(*scheduler, failure(_, _))
+ .WillOnce(FutureArg<1>(&failure));
+
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::SHUTDOWN);
Call::Shutdown* shutdown = call.mutable_shutdown();
@@ -871,15 +853,13 @@ TEST_P(SchedulerTest, ShutdownExecutor)
}
AWAIT_READY(shutdown);
- containerizer.destroy(devolve(id), executorId);
+ containerizer.destroy(devolve(frameworkId), executorId);
// Executor termination results in a 'FAILURE' event.
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::FAILURE, event.get().type());
- EXPECT_EQ(evolve(executorId), event.get().failure().executor_id());
+ AWAIT_READY(failure);
+ EXPECT_EQ(executorId, devolve(failure->executor_id()));
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -894,34 +874,36 @@ TEST_P(SchedulerTest, Teardown)
Try<PID<Master>> master = StartMaster(flags);
ASSERT_SOME(master);
- ExecutorID executorId = DEFAULT_EXECUTOR_ID;
-
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
auto executor = std::make_shared<MockV1HTTPExecutor>();
+
+ ExecutorID executorId = DEFAULT_EXECUTOR_ID;
TestContainerizer containerizer(executorId, executor);
Try<PID<Slave>> slave = StartSlave(&containerizer);
ASSERT_SOME(slave);
- Callbacks callbacks;
-
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
+ Future<Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
{
Call call;
@@ -933,38 +915,38 @@ TEST_P(SchedulerTest, Teardown)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ v1::FrameworkID frameworkId(subscribed->framework_id());
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
EXPECT_CALL(*executor, connected(_))
- .WillOnce(executor::SendSubscribe(id, evolve(executorId)));
+ .WillOnce(executor::SendSubscribe(frameworkId, evolve(executorId)));
EXPECT_CALL(*executor, subscribed(_, _));
EXPECT_CALL(*executor, launch(_, _))
.WillOnce(executor::SendUpdateFromTask(
- id, evolve(executorId), v1::TASK_RUNNING));
+ frameworkId, evolve(executorId), v1::TASK_RUNNING));
Future<Nothing> acknowledged;
EXPECT_CALL(*executor, acknowledged(_, _))
.WillOnce(FutureSatisfy(&acknowledged));
- v1::Offer offer = event.get().offers().offers(0);
+ Future<Event::Update> update;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update));
+
+ const v1::Offer& offer = offers->offers(0);
v1::TaskInfo taskInfo =
evolve(createTask(devolve(offer), "", DEFAULT_EXECUTOR_ID));
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::ACCEPT);
Call::Accept* accept = call.mutable_accept();
@@ -978,11 +960,9 @@ TEST_P(SchedulerTest, Teardown)
}
AWAIT_READY(acknowledged);
+ AWAIT_READY(update);
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::UPDATE, event.get().type());
- EXPECT_EQ(v1::TASK_RUNNING, event.get().update().status().state());
+ EXPECT_EQ(v1::TASK_RUNNING, update->status().state());
Future<Nothing> shutdown;
EXPECT_CALL(*executor, shutdown(_))
@@ -990,7 +970,7 @@ TEST_P(SchedulerTest, Teardown)
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::TEARDOWN);
mesos.send(call);
@@ -998,7 +978,7 @@ TEST_P(SchedulerTest, Teardown)
AWAIT_READY(shutdown);
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -1016,26 +996,29 @@ TEST_P(SchedulerTest, Decline)
Try<PID<Slave>> slave = StartSlave();
ASSERT_SOME(slave);
- Callbacks callbacks;
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
+
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ Future<Event::Offers> offers1;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers1));
{
Call call;
@@ -1047,21 +1030,22 @@ TEST_P(SchedulerTest, Decline)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
+
+ v1::FrameworkID frameworkId(subscribed->framework_id());
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ AWAIT_READY(offers1);
+ ASSERT_EQ(1, offers1->offers().size());
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- ASSERT_EQ(1, event.get().offers().offers().size());
+ const v1::Offer& offer = offers1->offers(0);
+
+ Future<Event::Offers> offers2;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers2));
- v1::Offer offer = event.get().offers().offers(0);
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::DECLINE);
Call::Decline* decline = call.mutable_decline();
@@ -1077,13 +1061,11 @@ TEST_P(SchedulerTest, Decline)
// If the resources were properly declined, the scheduler should
// get another offer with same amount of resources.
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- ASSERT_EQ(1, event.get().offers().offers().size());
- ASSERT_EQ(offer.resources(), event.get().offers().offers(0).resources());
+ AWAIT_READY(offers2);
+ ASSERT_EQ(1, offers2->offers().size());
+ ASSERT_EQ(offer.resources(), offers2->offers(0).resources());
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -1101,26 +1083,29 @@ TEST_P(SchedulerTest, Revive)
Try<PID<Slave>> slave = StartSlave();
ASSERT_SOME(slave);
- Callbacks callbacks;
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
+
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ Future<Event::Offers> offers1;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers1));
{
Call call;
@@ -1132,21 +1117,22 @@ TEST_P(SchedulerTest, Revive)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
+
+ v1::FrameworkID frameworkId(subscribed->framework_id());
+
+ AWAIT_READY(offers1);
+ EXPECT_NE(0, offers1->offers().size());
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ const v1::Offer& offer = offers1->offers(0);
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
+ Future<Event::Offers> offers2;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers2));
- v1::Offer offer = event.get().offers().offers(0);
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::DECLINE);
Call::Decline* decline = call.mutable_decline();
@@ -1166,25 +1152,23 @@ TEST_P(SchedulerTest, Revive)
Clock::advance(Minutes(30));
Clock::settle();
- event = events.get();
- ASSERT_TRUE(event.isPending());
+ ASSERT_TRUE(offers2.isPending());
// On revival the filters should be cleared and the scheduler should
// get another offer with same amount of resources.
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::REVIVE);
mesos.send(call);
}
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
- ASSERT_EQ(offer.resources(), event.get().offers().offers(0).resources());
+ AWAIT_READY(offers2);
+ EXPECT_NE(0, offers2->offers().size());
+ ASSERT_EQ(offer.resources(), offers2->offers(0).resources());
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -1202,26 +1186,29 @@ TEST_P(SchedulerTest, Suppress)
Try<PID<Slave>> slave = StartSlave();
ASSERT_SOME(slave);
- Callbacks callbacks;
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
+ Future<Event::Offers> offers1;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers1));
{
Call call;
@@ -1233,21 +1220,22 @@ TEST_P(SchedulerTest, Suppress)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
+
+ v1::FrameworkID frameworkId(subscribed->framework_id());
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ AWAIT_READY(offers1);
+ EXPECT_NE(0, offers1->offers().size());
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
+ const v1::Offer& offer = offers1->offers(0);
+
+ Future<Event::Offers> offers2;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers2));
- v1::Offer offer = event.get().offers().offers(0);
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::DECLINE);
Call::Decline* decline = call.mutable_decline();
@@ -1266,7 +1254,7 @@ TEST_P(SchedulerTest, Suppress)
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::SUPPRESS);
mesos.send(call);
@@ -1283,25 +1271,24 @@ TEST_P(SchedulerTest, Suppress)
Clock::advance(Minutes(100));
Clock::settle();
- event = events.get();
- ASSERT_TRUE(event.isPending());
+ ASSERT_TRUE(offers2.isPending());
// On reviving offers the scheduler should get another offer with same amount
// of resources.
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::REVIVE);
mesos.send(call);
}
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
- ASSERT_EQ(offer.resources(), event.get().offers().offers(0).resources());
+ AWAIT_READY(offers2);
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_NE(0, offers2->offers().size());
+ ASSERT_EQ(offer.resources(), offers2->offers(0).resources());
+
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -1316,34 +1303,36 @@ TEST_P(SchedulerTest, Message)
Try<PID<Master>> master = StartMaster(flags);
ASSERT_SOME(master);
- ExecutorID executorId = DEFAULT_EXECUTOR_ID;
-
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
auto executor = std::make_shared<MockV1HTTPExecutor>();
+
+ ExecutorID executorId = DEFAULT_EXECUTOR_ID;
TestContainerizer containerizer(executorId, executor);
Try<PID<Slave>> slave = StartSlave(&containerizer);
ASSERT_SOME(slave);
- Callbacks callbacks;
-
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
+
+ Future<Event::Offers> offers;
+ EXPECT_CALL(*scheduler, offers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
{
Call call;
@@ -1355,38 +1344,38 @@ TEST_P(SchedulerTest, Message)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ v1::FrameworkID frameworkId(subscribed->framework_id());
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::OFFERS, event.get().type());
- EXPECT_NE(0, event.get().offers().offers().size());
+ AWAIT_READY(offers);
+ EXPECT_NE(0, offers->offers().size());
EXPECT_CALL(*executor, connected(_))
- .WillOnce(executor::SendSubscribe(id, evolve(executorId)));
+ .WillOnce(executor::SendSubscribe(frameworkId, evolve(executorId)));
EXPECT_CALL(*executor, subscribed(_, _));
EXPECT_CALL(*executor, launch(_, _))
.WillOnce(executor::SendUpdateFromTask(
- id, evolve(executorId), v1::TASK_RUNNING));
+ frameworkId, evolve(executorId), v1::TASK_RUNNING));
Future<Nothing> acknowledged;
EXPECT_CALL(*executor, acknowledged(_, _))
.WillOnce(FutureSatisfy(&acknowledged));
- v1::Offer offer = event.get().offers().offers(0);
+ Future<Event::Update> update;
+ EXPECT_CALL(*scheduler, update(_, _))
+ .WillOnce(FutureArg<1>(&update));
+
+ const v1::Offer& offer = offers->offers(0);
v1::TaskInfo taskInfo =
evolve(createTask(devolve(offer), "", DEFAULT_EXECUTOR_ID));
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::ACCEPT);
Call::Accept* accept = call.mutable_accept();
@@ -1400,11 +1389,9 @@ TEST_P(SchedulerTest, Message)
}
AWAIT_READY(acknowledged);
+ AWAIT_READY(update);
- event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::UPDATE, event.get().type());
- EXPECT_EQ(v1::TASK_RUNNING, event.get().update().status().state());
+ EXPECT_EQ(v1::TASK_RUNNING, update->status().state());
Future<v1::executor::Event::Message> message;
EXPECT_CALL(*executor, message(_, _))
@@ -1412,7 +1399,7 @@ TEST_P(SchedulerTest, Message)
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::MESSAGE);
Call::Message* message = call.mutable_message();
@@ -1432,7 +1419,7 @@ TEST_P(SchedulerTest, Message)
EXPECT_CALL(*executor, disconnected(_))
.Times(AtMost(1));
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown(); // Must shutdown before 'containerizer' gets deallocated.
@@ -1447,26 +1434,25 @@ TEST_P(SchedulerTest, Request)
Try<PID<Master>> master = StartMaster(flags);
ASSERT_SOME(master);
- Callbacks callbacks;
+ auto scheduler = std::make_shared<MockV1HTTPScheduler>();
Future<Nothing> connected;
- EXPECT_CALL(callbacks, connected())
+ EXPECT_CALL(*scheduler, connected(_))
.WillOnce(FutureSatisfy(&connected))
.WillRepeatedly(Return()); // Ignore future invocations.
- Mesos mesos(
- master.get(),
- GetParam(),
- lambda::bind(&Callbacks::connected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::disconnected, lambda::ref(callbacks)),
- lambda::bind(&Callbacks::received, lambda::ref(callbacks), lambda::_1));
+ ContentType contentType = GetParam();
+
+ scheduler::TestV1Mesos mesos(master.get(), contentType, scheduler);
AWAIT_READY(connected);
- Queue<Event> events;
+ Future<Event::Subscribed> subscribed;
+ EXPECT_CALL(*scheduler, subscribed(_, _))
+ .WillOnce(FutureArg<1>(&subscribed));
- EXPECT_CALL(callbacks, received(_))
- .WillRepeatedly(Enqueue(&events));
+ EXPECT_CALL(*scheduler, heartbeat(_))
+ .WillRepeatedly(Return()); // Ignore heartbeats.
{
Call call;
@@ -1478,18 +1464,16 @@ TEST_P(SchedulerTest, Request)
mesos.send(call);
}
- Future<Event> event = events.get();
- AWAIT_READY(event);
- EXPECT_EQ(Event::SUBSCRIBED, event.get().type());
+ AWAIT_READY(subscribed);
- v1::FrameworkID id(event.get().subscribed().framework_id());
+ v1::FrameworkID frameworkId(subscribed->framework_id());
Future<Nothing> requestResources =
FUTURE_DISPATCH(_, &MesosAllocatorProcess::requestResources);
{
Call call;
- call.mutable_framework_id()->CopyFrom(id);
+ call.mutable_framework_id()->CopyFrom(frameworkId);
call.set_type(Call::REQUEST);
// Create a dummy request.
@@ -1501,7 +1485,7 @@ TEST_P(SchedulerTest, Request)
AWAIT_READY(requestResources);
- EXPECT_CALL(callbacks, disconnected())
+ EXPECT_CALL(*scheduler, disconnected(_))
.Times(AtMost(1));
Shutdown();
[3/3] mesos git commit: Fixed flakiness in tests using the scheduler
library.
Posted by vi...@apache.org.
Fixed flakiness in tests using the scheduler library.
This change fixes the tests using the scheduler library by
explicitly invoking `stop()` from the testing interface. This
ensures that no further callbacks are delivered to the scheduler.
For one-off async callbacks that are already on the libprocess queue
we need to do a `Clock::settle` to ensure they are executed before
the mock object goes out of scope.
Review: https://reviews.apache.org/r/44275/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/118b6ea8
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/118b6ea8
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/118b6ea8
Branch: refs/heads/master
Commit: 118b6ea88615cc95003a7ba4d0518a61e9948e14
Parents: d27b866
Author: Anand Mazumdar <ma...@gmail.com>
Authored: Thu Mar 3 16:09:58 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Thu Mar 3 16:09:58 2016 -0800
----------------------------------------------------------------------
src/tests/mesos.hpp | 28 ++++++++++++++++++++++++++--
1 file changed, 26 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/118b6ea8/src/tests/mesos.hpp
----------------------------------------------------------------------
diff --git a/src/tests/mesos.hpp b/src/tests/mesos.hpp
index 1d4f075..d36840f 100644
--- a/src/tests/mesos.hpp
+++ b/src/tests/mesos.hpp
@@ -971,11 +971,13 @@ public:
: Mesos(
master,
contentType,
+ // We don't pass the `_scheduler` shared pointer as the library
+ // interface expects a `std::function` object.
lambda::bind(&MockHTTPScheduler<Mesos, Event>::connected,
- _scheduler,
+ _scheduler.get(),
this),
lambda::bind(&MockHTTPScheduler<Mesos, Event>::disconnected,
- _scheduler,
+ _scheduler.get(),
this),
lambda::bind(&TestMesos<Mesos, Event>::events,
this,
@@ -983,6 +985,28 @@ public:
detector),
scheduler(_scheduler) {}
+ virtual ~TestMesos()
+ {
+ // Since the destructor for `TestMesos` is invoked first, the library can
+ // make more callbacks to the `scheduler` object before the `Mesos` (base
+ // class) destructor is invoked. To prevent this, we invoke `stop()` here
+ // to explicitly stop the library.
+ this->stop();
+
+ bool paused = process::Clock::paused();
+
+ // Need to settle the Clock to ensure that all the pending async callbacks
+ // with references to `this` and `scheduler` queued on libprocess are
+ // executed before the object is destructed.
+ process::Clock::pause();
+ process::Clock::settle();
+
+ // Return the Clock to its original state.
+ if (!paused) {
+ process::Clock::resume();
+ }
+ }
+
protected:
void events(std::queue<Event> events)
{
[2/3] mesos git commit: Added the ability to stop running the
scheduler library process.
Posted by vi...@apache.org.
Added the ability to stop running the scheduler library process.
This change adds the ability to stop running the scheduler library
process so that no future callbacks are delivered to the scheduler.
This helps us during testing to ensure no further callbacks happen
to stack allocated mock objects.
Review: https://reviews.apache.org/r/44274/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d27b8661
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d27b8661
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d27b8661
Branch: refs/heads/master
Commit: d27b8661654c5b02032731ec8298394381552b87
Parents: b446dbf
Author: Anand Mazumdar <ma...@gmail.com>
Authored: Thu Mar 3 16:09:54 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Thu Mar 3 16:09:54 2016 -0800
----------------------------------------------------------------------
include/mesos/v1/scheduler.hpp | 9 +++++++++
src/scheduler/scheduler.cpp | 18 +++++++++++++++---
2 files changed, 24 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d27b8661/include/mesos/v1/scheduler.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/v1/scheduler.hpp b/include/mesos/v1/scheduler.hpp
index a0fb73b..765935e 100644
--- a/include/mesos/v1/scheduler.hpp
+++ b/include/mesos/v1/scheduler.hpp
@@ -83,6 +83,15 @@ protected:
const std::function<void(const std::queue<Event>&)>& received,
const Option<std::shared_ptr<mesos::internal::MasterDetector>>& detector);
+ // Stops the library so that:
+ // - No more calls can be sent to the master.
+ // - No more callbacks can be made to the scheduler. In some cases, there
+ // may be one additional callback if the library was in the middle of
+ // processing an event.
+ //
+ // NOTE: This is used for testing.
+ virtual void stop();
+
private:
MesosProcess* process;
};
http://git-wip-us.apache.org/repos/asf/mesos/blob/d27b8661/src/scheduler/scheduler.cpp
----------------------------------------------------------------------
diff --git a/src/scheduler/scheduler.cpp b/src/scheduler/scheduler.cpp
index 2c6bf0d..b010a81 100644
--- a/src/scheduler/scheduler.cpp
+++ b/src/scheduler/scheduler.cpp
@@ -714,9 +714,9 @@ Mesos::Mesos(
Mesos::~Mesos()
{
- terminate(process);
- wait(process);
- delete process;
+ if (process != NULL) {
+ stop();
+ }
}
@@ -725,6 +725,18 @@ void Mesos::send(const Call& call)
dispatch(process, &MesosProcess::send, call);
}
+
+void Mesos::stop()
+{
+ if (process != NULL) {
+ terminate(process);
+ wait(process);
+
+ delete process;
+ process = NULL;
+ }
+}
+
} // namespace scheduler {
} // namespace v1 {
} // namespace mesos {