You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by bb...@apache.org on 2017/12/08 10:47:37 UTC
[1/4] mesos git commit: Passed versions when launching tasks.
Repository: mesos
Updated Branches:
refs/heads/master c78496fd5 -> d142d38e3
Passed versions when launching tasks.
In this patch we inject resource versions into task launch messages
and add verification in the agent. We require that resource versions
of resource providers whose resources are used in a task have not
changed. With that we can make sure to e.g., not use resources created
in speculated operations.
Review: https://reviews.apache.org/r/64299
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/0bbdbce1
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/0bbdbce1
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/0bbdbce1
Branch: refs/heads/master
Commit: 0bbdbce1daab55dbe8d02aaa858eb2e95a3071de
Parents: ac97d76
Author: Benjamin Bannier <bb...@apache.org>
Authored: Thu Nov 30 14:45:55 2017 +0100
Committer: Benjamin Bannier <bb...@apache.org>
Committed: Fri Dec 8 11:07:44 2017 +0100
----------------------------------------------------------------------
src/master/master.cpp | 6 ++
src/slave/slave.cpp | 98 ++++++++++++++++++--
src/slave/slave.hpp | 6 +-
src/tests/mock_slave.cpp | 12 ++-
src/tests/mock_slave.hpp | 8 +-
src/tests/slave_tests.cpp | 204 ++++++++++++++++++++++++++++++++++++++---
6 files changed, 307 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/0bbdbce1/src/master/master.cpp
----------------------------------------------------------------------
diff --git a/src/master/master.cpp b/src/master/master.cpp
index 584398c..5cba506 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -4990,6 +4990,9 @@ void Master::_accept(
RunTaskMessage message;
message.mutable_framework()->MergeFrom(framework->info);
+ message.mutable_resource_version_uuids()->CopyFrom(
+ protobuf::createResourceVersions(slave->resourceVersions));
+
// TODO(anand): We set 'pid' to UPID() for http frameworks
// as 'pid' was made optional in 0.24.0. In 0.25.0, we
// no longer have to set pid here for http frameworks.
@@ -5179,6 +5182,9 @@ void Master::_accept(
message.mutable_executor()->CopyFrom(executor);
message.mutable_task_group()->CopyFrom(taskGroup);
+ message.mutable_resource_version_uuids()->CopyFrom(
+ protobuf::createResourceVersions(slave->resourceVersions));
+
set<TaskID> taskIds;
Resources totalResources;
http://git-wip-us.apache.org/repos/asf/mesos/blob/0bbdbce1/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 8bdb945..7a4a4ac 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -2011,7 +2011,8 @@ void Slave::run(
frameworkInfo,
executorInfo,
task,
- taskGroup));
+ taskGroup,
+ resourceVersionUuids));
}
@@ -2020,7 +2021,8 @@ void Slave::_run(
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
const Option<TaskInfo>& task,
- const Option<TaskGroupInfo>& taskGroup)
+ const Option<TaskGroupInfo>& taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids)
{
// TODO(anindya_sinha): Consider refactoring the initial steps common
// to `_run()` and `__run()`.
@@ -2152,7 +2154,8 @@ void Slave::_run(
frameworkInfo,
executorInfo,
task,
- taskGroup));
+ taskGroup,
+ resourceVersionUuids));
}
@@ -2161,7 +2164,8 @@ void Slave::__run(
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
const Option<TaskInfo>& task,
- const Option<TaskGroupInfo>& taskGroup)
+ const Option<TaskGroupInfo>& taskGroup,
+ const vector<ResourceVersionUUID>& resourceVersionUuids)
{
CHECK_NE(task.isSome(), taskGroup.isSome())
<< "Either task or task group should be set but not both";
@@ -2306,8 +2310,84 @@ void Slave::__run(
return;
}
- LOG(INFO) << "Launching " << taskOrTaskGroup(task, taskGroup)
- << " for framework " << frameworkId;
+ // Check task invariants.
+ //
+ // TODO(bbannier): Instead of copy-pasting identical code to deal
+ // with cases where tasks need to be terminated, consolidate code
+ // below to decouple checking from terminating.
+
+ // If the master sent resource versions, perform a best-effort check
+ // that they are consistent with the resources the task uses.
+ //
+ // TODO(bbannier): Also check executor resources.
+ bool kill = false;
+ if (!resourceVersionUuids.empty()) {
+ hashset<Option<ResourceProviderID>> usedResourceProviders;
+ foreach (const TaskInfo& _task, tasks) {
+ foreach (const Resource& resource, _task.resources()) {
+ if (resource.has_provider_id()) {
+ usedResourceProviders.insert(resource.provider_id());
+ } else {
+ usedResourceProviders.insert(None());
+ }
+ }
+ }
+
+ const hashmap<Option<ResourceProviderID>, UUID> receivedResourceVersions =
+ protobuf::parseResourceVersions(
+ {resourceVersionUuids.begin(), resourceVersionUuids.end()});
+
+ foreach (auto&& resourceProvider, usedResourceProviders) {
+ Option<Error> error = None();
+
+ if (!resourceVersions.contains(resourceProvider)) {
+ // We do not expect the agent to forget about itself.
+ CHECK_SOME(resourceProvider);
+ kill = true;
+ }
+
+ CHECK(receivedResourceVersions.contains(resourceProvider));
+
+ if (resourceVersions.at(resourceProvider) !=
+ receivedResourceVersions.at(resourceProvider)) {
+ kill = true;
+ }
+ }
+ }
+
+ if (kill) {
+ // We report TASK_DROPPED to the framework because the task was
+ // never launched. For non-partition-aware frameworks, we report
+ // TASK_LOST for backward compatibility.
+ mesos::TaskState taskState = TASK_DROPPED;
+ if (!protobuf::frameworkHasCapability(
+ frameworkInfo, FrameworkInfo::Capability::PARTITION_AWARE)) {
+ taskState = TASK_LOST;
+ }
+
+ foreach (const TaskInfo& _task, tasks) {
+ const StatusUpdate update = protobuf::createStatusUpdate(
+ frameworkId,
+ info.id(),
+ _task.task_id(),
+ taskState,
+ TaskStatus::SOURCE_SLAVE,
+ UUID::random(),
+ "Tasks assumes outdated resource state",
+ TaskStatus::REASON_INVALID_OFFERS,
+ executorId);
+
+ statusUpdate(update, UPID());
+ }
+
+ // Refer to the comment after 'framework->removePendingTask' above
+ // for why we need this.
+ if (framework->idle()) {
+ removeFramework(framework);
+ }
+
+ return;
+ }
auto unallocated = [](const Resources& resources) {
Resources result = resources;
@@ -2315,6 +2395,8 @@ void Slave::__run(
return result;
};
+ CHECK_EQ(kill, false);
+
// NOTE: If the task/task group or executor uses resources that are
// checkpointed on the slave (e.g. persistent volumes), we should
// already know about it. If the slave doesn't know about them (e.g.
@@ -2322,7 +2404,6 @@ void Slave::__run(
// send TASK_DROPPED status updates here since restarting the task
// may succeed in the event that CheckpointResourcesMessage arrives
// out of order.
- bool kill = false;
foreach (const TaskInfo& _task, tasks) {
// We must unallocate the resources to check whether they are
// contained in the unallocated total checkpointed resources.
@@ -2451,6 +2532,9 @@ void Slave::__run(
CHECK(framework->state == Framework::RUNNING) << framework->state;
+ LOG(INFO) << "Launching " << taskOrTaskGroup(task, taskGroup)
+ << " for framework " << frameworkId;
+
// Either send the task/task group to an executor or start a new executor
// and queue it until the executor has started.
Executor* executor = framework->getExecutor(executorId);
http://git-wip-us.apache.org/repos/asf/mesos/blob/0bbdbce1/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index a47f93e..297d672 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -165,7 +165,8 @@ public:
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
const Option<TaskInfo>& task,
- const Option<TaskGroupInfo>& taskGroup);
+ const Option<TaskGroupInfo>& taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids);
// Made 'virtual' for Slave mocking.
virtual void runTaskGroup(
@@ -373,7 +374,8 @@ public:
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
const Option<TaskInfo>& task,
- const Option<TaskGroupInfo>& taskGroup);
+ const Option<TaskGroupInfo>& taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids);
// This is called when the resource limits of the container have
// been updated for the given tasks and task groups. If the update is
http://git-wip-us.apache.org/repos/asf/mesos/blob/0bbdbce1/src/tests/mock_slave.cpp
----------------------------------------------------------------------
diff --git a/src/tests/mock_slave.cpp b/src/tests/mock_slave.cpp
index a43a12f..8357edc 100644
--- a/src/tests/mock_slave.cpp
+++ b/src/tests/mock_slave.cpp
@@ -120,7 +120,7 @@ MockSlave::MockSlave(
// Set up default behaviors, calling the original methods.
EXPECT_CALL(*this, runTask(_, _, _, _, _, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_runTask));
- EXPECT_CALL(*this, _run(_, _, _, _, _))
+ EXPECT_CALL(*this, _run(_, _, _, _, _, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked__run));
EXPECT_CALL(*this, runTaskGroup(_, _, _, _, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_runTaskGroup));
@@ -164,10 +164,16 @@ void MockSlave::unmocked__run(
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
const Option<TaskInfo>& taskInfo,
- const Option<TaskGroupInfo>& taskGroup)
+ const Option<TaskGroupInfo>& taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids)
{
slave::Slave::_run(
- unschedules, frameworkInfo, executorInfo, taskInfo, taskGroup);
+ unschedules,
+ frameworkInfo,
+ executorInfo,
+ taskInfo,
+ taskGroup,
+ resourceVersionUuids);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/0bbdbce1/src/tests/mock_slave.hpp
----------------------------------------------------------------------
diff --git a/src/tests/mock_slave.hpp b/src/tests/mock_slave.hpp
index cf5a581..29ce714 100644
--- a/src/tests/mock_slave.hpp
+++ b/src/tests/mock_slave.hpp
@@ -117,19 +117,21 @@ public:
const TaskInfo& task,
const std::vector<ResourceVersionUUID>& resourceVersionUuids);
- MOCK_METHOD5(_run, void(
+ MOCK_METHOD6(_run, void(
const process::Future<std::list<bool>>& unschedules,
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
const Option<TaskInfo>& task,
- const Option<TaskGroupInfo>& taskGroup));
+ const Option<TaskGroupInfo>& taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids));
void unmocked__run(
const process::Future<std::list<bool>>& unschedules,
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
const Option<TaskInfo>& task,
- const Option<TaskGroupInfo>& taskGroup);
+ const Option<TaskGroupInfo>& taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids);
MOCK_METHOD5(runTaskGroup, void(
const process::UPID& from,
http://git-wip-us.apache.org/repos/asf/mesos/blob/0bbdbce1/src/tests/slave_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_tests.cpp b/src/tests/slave_tests.cpp
index 29ab216..0714543 100644
--- a/src/tests/slave_tests.cpp
+++ b/src/tests/slave_tests.cpp
@@ -1833,7 +1833,7 @@ TEST_F(SlaveTest, GetStateTaskGroupPending)
// unmocked `_run()` method. Instead, we want to do nothing so that tasks
// remain in the framework's 'pending' list.
Future<Nothing> _run;
- EXPECT_CALL(*slave.get()->mock(), _run(_, _, _, _, _))
+ EXPECT_CALL(*slave.get()->mock(), _run(_, _, _, _, _, _))
.WillOnce(FutureSatisfy(&_run));
// The executor should not be launched.
@@ -4118,17 +4118,19 @@ TEST_F(SlaveTest, KillTaskBetweenRunTaskParts)
ExecutorInfo executorInfo;
Option<TaskGroupInfo> taskGroup;
Option<TaskInfo> task_;
+ vector<ResourceVersionUUID> resourceVersionUuids;
// Skip what Slave::_run() normally does, save its arguments for
// later, tie reaching the critical moment when to kill the task to
// a future.
Future<Nothing> _run;
- EXPECT_CALL(*slave.get()->mock(), _run(_, _, _, _, _))
+ EXPECT_CALL(*slave.get()->mock(), _run(_, _, _, _, _, _))
.WillOnce(DoAll(FutureSatisfy(&_run),
SaveArg<0>(&unschedules),
SaveArg<1>(&frameworkInfo),
SaveArg<2>(&executorInfo),
SaveArg<3>(&task_),
- SaveArg<4>(&taskGroup)));
+ SaveArg<4>(&taskGroup),
+ SaveArg<5>(&resourceVersionUuids)));
driver.launchTasks(offers.get()[0].id(), {task});
@@ -4155,7 +4157,12 @@ TEST_F(SlaveTest, KillTaskBetweenRunTaskParts)
AWAIT_READY(removeFramework);
slave.get()->mock()->unmocked__run(
- unschedules, frameworkInfo, executorInfo, task_, taskGroup);
+ unschedules,
+ frameworkInfo,
+ executorInfo,
+ task_,
+ taskGroup,
+ resourceVersionUuids);
AWAIT_READY(status);
EXPECT_EQ(TASK_KILLED, status->state());
@@ -4239,21 +4246,24 @@ TEST_F(SlaveTest, KillMultiplePendingTasks)
ExecutorInfo executorInfo1, executorInfo2;
Option<TaskGroupInfo> taskGroup1, taskGroup2;
Option<TaskInfo> task_1, task_2;
+ vector<ResourceVersionUUID> resourceVersionUuids1, resourceVersionUuids2;
Future<Nothing> _run1, _run2;
- EXPECT_CALL(*slave.get()->mock(), _run(_, _, _, _, _))
+ EXPECT_CALL(*slave.get()->mock(), _run(_, _, _, _, _, _))
.WillOnce(DoAll(FutureSatisfy(&_run1),
SaveArg<0>(&unschedules1),
SaveArg<1>(&frameworkInfo1),
SaveArg<2>(&executorInfo1),
SaveArg<3>(&task_1),
- SaveArg<4>(&taskGroup1)))
+ SaveArg<4>(&taskGroup1),
+ SaveArg<5>(&resourceVersionUuids1)))
.WillOnce(DoAll(FutureSatisfy(&_run2),
SaveArg<0>(&unschedules2),
SaveArg<1>(&frameworkInfo2),
SaveArg<2>(&executorInfo2),
SaveArg<3>(&task_2),
- SaveArg<4>(&taskGroup2)));
+ SaveArg<4>(&taskGroup2),
+ SaveArg<5>(&resourceVersionUuids2)));
driver.launchTasks(offers.get()[0].id(), {task1, task2});
@@ -4290,10 +4300,20 @@ TEST_F(SlaveTest, KillMultiplePendingTasks)
// The `__run` continuations should have no effect.
slave.get()->mock()->unmocked__run(
- unschedules1, frameworkInfo1, executorInfo1, task_1, taskGroup1);
+ unschedules1,
+ frameworkInfo1,
+ executorInfo1,
+ task_1,
+ taskGroup1,
+ resourceVersionUuids1);
slave.get()->mock()->unmocked__run(
- unschedules2, frameworkInfo2, executorInfo2, task_2, taskGroup2);
+ unschedules2,
+ frameworkInfo2,
+ executorInfo2,
+ task_2,
+ taskGroup2,
+ resourceVersionUuids2);
Clock::settle();
@@ -7176,18 +7196,20 @@ TEST_F(SlaveTest, KillTaskGroupBetweenRunTaskParts)
ExecutorInfo executorInfo_;
Option<TaskGroupInfo> taskGroup_;
Option<TaskInfo> task_;
+ vector<ResourceVersionUUID> resourceVersionUuids;
// Skip what `Slave::_run()` normally does, save its arguments for
// later, till reaching the critical moment when to kill the task
// in the future.
Future<Nothing> _run;
- EXPECT_CALL(*slave.get()->mock(), _run(_, _, _, _, _))
+ EXPECT_CALL(*slave.get()->mock(), _run(_, _, _, _, _, _))
.WillOnce(DoAll(FutureSatisfy(&_run),
SaveArg<0>(&unschedules),
SaveArg<1>(&frameworkInfo),
SaveArg<2>(&executorInfo_),
SaveArg<3>(&task_),
- SaveArg<4>(&taskGroup_)));
+ SaveArg<4>(&taskGroup_),
+ SaveArg<5>(&resourceVersionUuids)));
const v1::Offer& offer = offers->offers(0);
const SlaveID slaveId = devolve(offer.agent_id());
@@ -7248,7 +7270,12 @@ TEST_F(SlaveTest, KillTaskGroupBetweenRunTaskParts)
AWAIT_READY(removeFramework);
slave.get()->mock()->unmocked__run(
- unschedules, frameworkInfo, executorInfo_, task_, taskGroup_);
+ unschedules,
+ frameworkInfo,
+ executorInfo_,
+ task_,
+ taskGroup_,
+ resourceVersionUuids);
AWAIT_READY(update1);
AWAIT_READY(update2);
@@ -9320,6 +9347,159 @@ TEST_F(SlaveTest, ResourceProviderReconciliation)
v1::Resources(offer1.resources()), v1::Resources(offer2.resources()));
}
+
+// This test verifies that the agent checks resource versions received when
+// launching tasks against its own state of the used resource providers and
+// rejects tasks assuming incompatible state.
+TEST_F(SlaveTest, RunTaskResourceVersions)
+{
+ Clock::pause();
+
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ slave::Flags slaveFlags = CreateSlaveFlags();
+ slaveFlags.authenticate_http_readwrite = false;
+
+ // Set the resource provider capability and other required capabilities.
+ constexpr SlaveInfo::Capability::Type capabilities[] = {
+ SlaveInfo::Capability::MULTI_ROLE,
+ SlaveInfo::Capability::HIERARCHICAL_ROLE,
+ SlaveInfo::Capability::RESERVATION_REFINEMENT,
+ SlaveInfo::Capability::RESOURCE_PROVIDER};
+
+ slaveFlags.agent_features = SlaveCapabilities();
+ foreach (SlaveInfo::Capability::Type type, capabilities) {
+ SlaveInfo::Capability* capability =
+ slaveFlags.agent_features->add_capabilities();
+ capability->set_type(type);
+ }
+
+ Future<UpdateSlaveMessage> updateSlaveMessage =
+ FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
+
+ StandaloneMasterDetector detector(master.get()->pid);
+ Try<Owned<cluster::Slave>> slave = StartSlave(&detector, slaveFlags);
+ ASSERT_SOME(slave);
+
+ Clock::settle();
+ Clock::advance(slaveFlags.registration_backoff_factor);
+
+ AWAIT_READY(updateSlaveMessage);
+
+ // Register a resource provider with the agent.
+ mesos::v1::ResourceProviderInfo resourceProviderInfo;
+ resourceProviderInfo.set_type("org.apache.mesos.resource_provider.test");
+ resourceProviderInfo.set_name("test");
+
+ v1::Resources resourceProviderResources = v1::createDiskResource(
+ "200",
+ "*",
+ None(),
+ None(),
+ v1::createDiskSourceRaw());
+
+ v1::MockResourceProvider resourceProvider(
+ resourceProviderInfo,
+ resourceProviderResources);
+
+ string scheme = "http";
+
+#ifdef USE_SSL_SOCKET
+ if (process::network::openssl::flags().enabled) {
+ scheme = "https";
+ }
+#endif
+
+ process::http::URL url(
+ scheme,
+ slave.get()->pid.address.ip,
+ slave.get()->pid.address.port,
+ slave.get()->pid.id + "/api/v1/resource_provider");
+
+ Owned<EndpointDetector> endpointDetector(new ConstantEndpointDetector(url));
+
+ updateSlaveMessage = FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
+
+ resourceProvider.start(
+ endpointDetector, ContentType::PROTOBUF, v1::DEFAULT_CREDENTIAL);
+
+ AWAIT_READY(updateSlaveMessage);
+
+ // Start a framework to launch a task.
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched,
+ DEFAULT_FRAMEWORK_INFO,
+ master.get()->pid,
+ false,
+ DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(_, _, _));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(_, _))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.start();
+
+ // Below we update the agent's resource version of the registered
+ // resource provider. We prevent this update from propagating to the
+ // master to simulate a race between the agent updating its state
+ // and the master launching a task.
+ updateSlaveMessage = DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
+
+ // Update resource version of the resource provider.
+ {
+ CHECK(resourceProvider.info.has_id());
+
+ v1::Resources resourceProviderResources_;
+ foreach (v1::Resource resource, resourceProviderResources) {
+ resource.mutable_provider_id()->CopyFrom(resourceProvider.info.id());
+
+ resourceProviderResources_ += resource;
+ }
+
+ v1::resource_provider::Call call;
+ call.set_type(v1::resource_provider::Call::UPDATE_STATE);
+ call.mutable_resource_provider_id()->CopyFrom(resourceProvider.info.id());
+
+ v1::resource_provider::Call::UpdateState* updateState =
+ call.mutable_update_state();
+
+ updateState->set_resource_version_uuid(UUID::random().toBytes());
+ updateState->mutable_resources()->CopyFrom(resourceProviderResources_);
+
+ AWAIT_READY(resourceProvider.send(call));
+ }
+
+ AWAIT_READY(updateSlaveMessage);
+
+ // Launch a task on the offered resources. Since the agent will only check
+ // resource versions from resource providers used in the task launch, we
+ // explicitly confirm that the offer included resource provider resources.
+ AWAIT_READY(offers);
+ ASSERT_FALSE(offers->empty());
+ const Resources& offeredResources = offers->front().resources();
+ ASSERT_TRUE(std::any_of(
+ offeredResources.begin(), offeredResources.end(), [](const Resource& r) {
+ return r.has_provider_id();
+ }));
+
+ TaskInfo task = createTask(offers->front(), "sleep 1000");
+
+ Future<TaskStatus> statusUpdate;
+ EXPECT_CALL(sched, statusUpdate(&driver, _))
+ .WillOnce(FutureArg<1>(&statusUpdate));
+
+ driver.launchTasks(offers->front().id(), {task});
+
+ AWAIT_READY(statusUpdate);
+ EXPECT_EQ(TASK_LOST, statusUpdate->state());
+ EXPECT_EQ(TaskStatus::SOURCE_SLAVE, statusUpdate->source());
+ EXPECT_EQ(TaskStatus::REASON_INVALID_OFFERS, statusUpdate->reason());
+}
+
} // namespace tests {
} // namespace internal {
} // namespace mesos {
[2/4] mesos git commit: Added ResourceVersion to RunTask and
RunTaskGroup messages.
Posted by bb...@apache.org.
Added ResourceVersion to RunTask and RunTaskGroup messages.
This commit adds the known agent resource versions to RunTaskMessage
and RunTaskGroupMessage. We also update sites where the message is
unpacked.
In a later commit we will inject versions in the master and evaluate
them in the agent.
Review: https://reviews.apache.org/r/64298
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/ac97d764
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/ac97d764
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/ac97d764
Branch: refs/heads/master
Commit: ac97d764e55eb26484f3aad8ff8ac9730e6555b2
Parents: c78496f
Author: Benjamin Bannier <bb...@apache.org>
Authored: Thu Nov 30 12:19:45 2017 +0100
Committer: Benjamin Bannier <bb...@apache.org>
Committed: Fri Dec 8 11:07:44 2017 +0100
----------------------------------------------------------------------
src/messages/messages.proto | 26 ++++++++++++++++++++++++++
src/slave/slave.cpp | 23 +++++++++++++++++------
src/slave/slave.hpp | 7 +++++--
src/tests/mock_slave.cpp | 26 ++++++++++++++++++++------
src/tests/mock_slave.hpp | 17 +++++++++++------
src/tests/slave_tests.cpp | 6 +++---
6 files changed, 82 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac97d764/src/messages/messages.proto
----------------------------------------------------------------------
diff --git a/src/messages/messages.proto b/src/messages/messages.proto
index 44b45d5..f711784 100644
--- a/src/messages/messages.proto
+++ b/src/messages/messages.proto
@@ -306,6 +306,19 @@ message RunTaskMessage {
required FrameworkInfo framework = 2;
required TaskInfo task = 4;
+ // Used to establish the relationship between the operation and the
+ // resources that the operation is operating on. Each resource
+ // provider will keep a resource version UUID, and change it when
+ // it believes that the resources from this resource provider are
+ // out of sync from the master's view. The master will keep track
+ // of the last known resource version UUID for each resource
+ // provider, and attach the resource version UUID in each operation
+ // it sends out. The resource provider should reject operations that
+ // have a different resource version UUID than that it maintains,
+ // because this means the operation is operating on resources that
+ // might have already been invalidated.
+ repeated ResourceVersionUUID resource_version_uuids = 5;
+
// The pid of the framework. This was moved to 'optional' in
// 0.24.0 to support schedulers using the HTTP API. For now, we
// continue to always set pid since it was required in 0.23.x.
@@ -328,6 +341,19 @@ message RunTaskGroupMessage {
required FrameworkInfo framework = 1;
required ExecutorInfo executor = 2;
required TaskGroupInfo task_group = 3;
+
+ // Used to establish the relationship between the operation and the
+ // resources that the operation is operating on. Each resource
+ // provider will keep a resource version UUID, and change it when
+ // it believes that the resources from this resource provider are
+ // out of sync from the master's view. The master will keep track
+ // of the last known resource version UUID for each resource
+ // provider, and attach the resource version UUID in each operation
+ // it sends out. The resource provider should reject operations that
+ // have a different resource version UUID than that it maintains,
+ // because this means the operation is operating on resources that
+ // might have already been invalidated.
+ repeated ResourceVersionUUID resource_version_uuids = 4;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac97d764/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 98370f9..8bdb945 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -654,13 +654,15 @@ void Slave::initialize()
&RunTaskMessage::framework,
&RunTaskMessage::framework_id,
&RunTaskMessage::pid,
- &RunTaskMessage::task);
+ &RunTaskMessage::task,
+ &RunTaskMessage::resource_version_uuids);
install<RunTaskGroupMessage>(
&Slave::runTaskGroup,
&RunTaskGroupMessage::framework,
&RunTaskGroupMessage::executor,
- &RunTaskGroupMessage::task_group);
+ &RunTaskGroupMessage::task_group,
+ &RunTaskGroupMessage::resource_version_uuids);
install<KillTaskMessage>(
&Slave::killTask);
@@ -1763,7 +1765,8 @@ void Slave::runTask(
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const UPID& pid,
- const TaskInfo& task)
+ const TaskInfo& task,
+ const vector<ResourceVersionUUID>& resourceVersionUuids)
{
CHECK_NE(task.has_executor(), task.has_command())
<< "Task " << task.task_id()
@@ -1784,7 +1787,7 @@ void Slave::runTask(
const ExecutorInfo executorInfo = getExecutorInfo(frameworkInfo, task);
- run(frameworkInfo, executorInfo, task, None(), pid);
+ run(frameworkInfo, executorInfo, task, None(), resourceVersionUuids, pid);
}
@@ -1793,6 +1796,7 @@ void Slave::run(
ExecutorInfo executorInfo,
Option<TaskInfo> task,
Option<TaskGroupInfo> taskGroup,
+ const vector<ResourceVersionUUID>& resourceVersionUuids,
const UPID& pid)
{
CHECK_NE(task.isSome(), taskGroup.isSome())
@@ -3037,7 +3041,8 @@ void Slave::runTaskGroup(
const UPID& from,
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
- const TaskGroupInfo& taskGroupInfo)
+ const TaskGroupInfo& taskGroupInfo,
+ const vector<ResourceVersionUUID>& resourceVersionUuids)
{
if (master != from) {
LOG(WARNING) << "Ignoring run task group message from " << from
@@ -3059,7 +3064,13 @@ void Slave::runTaskGroup(
return;
}
- run(frameworkInfo, executorInfo, None(), taskGroupInfo, UPID());
+ run(
+ frameworkInfo,
+ executorInfo,
+ None(),
+ taskGroupInfo,
+ resourceVersionUuids,
+ UPID());
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac97d764/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index d9b0469..a47f93e 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -148,13 +148,15 @@ public:
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const process::UPID& pid,
- const TaskInfo& task);
+ const TaskInfo& task,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids);
void run(
const FrameworkInfo& frameworkInfo,
ExecutorInfo executorInfo,
Option<TaskInfo> task,
Option<TaskGroupInfo> taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids,
const process::UPID& pid);
// Made 'virtual' for Slave mocking.
@@ -170,7 +172,8 @@ public:
const process::UPID& from,
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
- const TaskGroupInfo& taskGroupInfo);
+ const TaskGroupInfo& taskGroupInfo,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids);
// Made 'virtual' for Slave mocking.
virtual void killTask(
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac97d764/src/tests/mock_slave.cpp
----------------------------------------------------------------------
diff --git a/src/tests/mock_slave.cpp b/src/tests/mock_slave.cpp
index 6d050ca..a43a12f 100644
--- a/src/tests/mock_slave.cpp
+++ b/src/tests/mock_slave.cpp
@@ -45,6 +45,7 @@ using mesos::slave::QoSController;
using std::list;
using std::string;
+using std::vector;
using process::Future;
using process::UPID;
@@ -117,11 +118,11 @@ MockSlave::MockSlave(
authorizer)
{
// Set up default behaviors, calling the original methods.
- EXPECT_CALL(*this, runTask(_, _, _, _, _))
+ EXPECT_CALL(*this, runTask(_, _, _, _, _, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_runTask));
EXPECT_CALL(*this, _run(_, _, _, _, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked__run));
- EXPECT_CALL(*this, runTaskGroup(_, _, _, _))
+ EXPECT_CALL(*this, runTaskGroup(_, _, _, _, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_runTaskGroup));
EXPECT_CALL(*this, killTask(_, _))
.WillRepeatedly(Invoke(this, &MockSlave::unmocked_killTask));
@@ -145,9 +146,16 @@ void MockSlave::unmocked_runTask(
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const UPID& pid,
- const TaskInfo& task)
+ const TaskInfo& task,
+ const vector<ResourceVersionUUID>& resourceVersionUuids)
{
- slave::Slave::runTask(from, frameworkInfo, frameworkInfo.id(), pid, task);
+ slave::Slave::runTask(
+ from,
+ frameworkInfo,
+ frameworkInfo.id(),
+ pid,
+ task,
+ resourceVersionUuids);
}
@@ -167,9 +175,15 @@ void MockSlave::unmocked_runTaskGroup(
const UPID& from,
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
- const TaskGroupInfo& taskGroup)
+ const TaskGroupInfo& taskGroup,
+ const vector<ResourceVersionUUID>& resourceVersionUuids)
{
- slave::Slave::runTaskGroup(from, frameworkInfo, executorInfo, taskGroup);
+ slave::Slave::runTaskGroup(
+ from,
+ frameworkInfo,
+ executorInfo,
+ taskGroup,
+ resourceVersionUuids);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac97d764/src/tests/mock_slave.hpp
----------------------------------------------------------------------
diff --git a/src/tests/mock_slave.hpp b/src/tests/mock_slave.hpp
index d986125..cf5a581 100644
--- a/src/tests/mock_slave.hpp
+++ b/src/tests/mock_slave.hpp
@@ -19,6 +19,7 @@
#include <list>
#include <string>
+#include <vector>
#include <gmock/gmock.h>
@@ -100,19 +101,21 @@ public:
SecretGenerator* secretGenerator,
const Option<Authorizer*>& authorizer);
- MOCK_METHOD5(runTask, void(
+ MOCK_METHOD6(runTask, void(
const process::UPID& from,
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const process::UPID& pid,
- const TaskInfo& task));
+ const TaskInfo& task,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids));
void unmocked_runTask(
const process::UPID& from,
const FrameworkInfo& frameworkInfo,
const FrameworkID& frameworkId,
const process::UPID& pid,
- const TaskInfo& task);
+ const TaskInfo& task,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids);
MOCK_METHOD5(_run, void(
const process::Future<std::list<bool>>& unschedules,
@@ -128,17 +131,19 @@ public:
const Option<TaskInfo>& task,
const Option<TaskGroupInfo>& taskGroup);
- MOCK_METHOD4(runTaskGroup, void(
+ MOCK_METHOD5(runTaskGroup, void(
const process::UPID& from,
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
- const TaskGroupInfo& taskGroup));
+ const TaskGroupInfo& taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids));
void unmocked_runTaskGroup(
const process::UPID& from,
const FrameworkInfo& frameworkInfo,
const ExecutorInfo& executorInfo,
- const TaskGroupInfo& taskGroup);
+ const TaskGroupInfo& taskGroup,
+ const std::vector<ResourceVersionUUID>& resourceVersionUuids);
MOCK_METHOD2(killTask, void(
const process::UPID& from,
http://git-wip-us.apache.org/repos/asf/mesos/blob/ac97d764/src/tests/slave_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/slave_tests.cpp b/src/tests/slave_tests.cpp
index ee490a0..29ab216 100644
--- a/src/tests/slave_tests.cpp
+++ b/src/tests/slave_tests.cpp
@@ -4109,7 +4109,7 @@ TEST_F(SlaveTest, KillTaskBetweenRunTaskParts)
EXPECT_CALL(sched, statusUpdate(&driver, _))
.WillRepeatedly(FutureArg<1>(&status));
- EXPECT_CALL(*slave.get()->mock(), runTask(_, _, _, _, _))
+ EXPECT_CALL(*slave.get()->mock(), runTask(_, _, _, _, _, _))
.WillOnce(Invoke(slave.get()->mock(), &MockSlave::unmocked_runTask));
// Saved arguments from Slave::_run().
@@ -4227,7 +4227,7 @@ TEST_F(SlaveTest, KillMultiplePendingTasks)
.WillOnce(FutureArg<1>(&status1))
.WillOnce(FutureArg<1>(&status2));
- EXPECT_CALL(*slave.get()->mock(), runTask(_, _, _, _, _))
+ EXPECT_CALL(*slave.get()->mock(), runTask(_, _, _, _, _, _))
.WillOnce(Invoke(slave.get()->mock(), &MockSlave::unmocked_runTask))
.WillOnce(Invoke(slave.get()->mock(), &MockSlave::unmocked_runTask));
@@ -7166,7 +7166,7 @@ TEST_F(SlaveTest, KillTaskGroupBetweenRunTaskParts)
.WillOnce(FutureArg<1>(&update2))
.WillRepeatedly(Return());
- EXPECT_CALL(*slave.get()->mock(), runTaskGroup(_, _, _, _))
+ EXPECT_CALL(*slave.get()->mock(), runTaskGroup(_, _, _, _, _))
.WillOnce(Invoke(slave.get()->mock(),
&MockSlave::unmocked_runTaskGroup));
[3/4] mesos git commit: Provided resource provider infos in
'UpdateState' message.
Posted by bb...@apache.org.
Provided resource provider infos in 'UpdateState' message.
To support sending 'ResourceProviderInfo's of all known resource
providers to the master as part of 'UpdateSlaveMessage', these
information has been added to the 'UpdateState' message that is sent
from resource provider to agents. Also, an agent now keeps track of all
'ResourceProviderInfo's.
Review: https://reviews.apache.org/r/64370/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/12695ec0
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/12695ec0
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/12695ec0
Branch: refs/heads/master
Commit: 12695ec010e842d54cdb15ea7add73aef6092bb0
Parents: 0bbdbce
Author: Jan Schlicht <ja...@mesosphere.io>
Authored: Fri Dec 8 11:26:18 2017 +0100
Committer: Benjamin Bannier <bb...@apache.org>
Committed: Fri Dec 8 11:26:18 2017 +0100
----------------------------------------------------------------------
src/resource_provider/manager.cpp | 2 +-
src/resource_provider/message.hpp | 4 ++--
src/slave/slave.cpp | 13 ++++++++++++-
src/slave/slave.hpp | 2 ++
src/tests/resource_provider_manager_tests.cpp | 5 ++++-
5 files changed, 21 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/12695ec0/src/resource_provider/manager.cpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/manager.cpp b/src/resource_provider/manager.cpp
index 879caba..f98611c 100644
--- a/src/resource_provider/manager.cpp
+++ b/src/resource_provider/manager.cpp
@@ -654,7 +654,7 @@ void ResourceProviderManagerProcess::updateState(
<< resourceProvider->info.id() << ": " << resourceVersionUuid.error();
ResourceProviderMessage::UpdateState updateState{
- resourceProvider->info.id(),
+ resourceProvider->info,
resourceVersionUuid.get(),
update.resources(),
{update.operations().begin(), update.operations().end()}};
http://git-wip-us.apache.org/repos/asf/mesos/blob/12695ec0/src/resource_provider/message.hpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/message.hpp b/src/resource_provider/message.hpp
index c94e9c7..bbf6bb2 100644
--- a/src/resource_provider/message.hpp
+++ b/src/resource_provider/message.hpp
@@ -45,7 +45,7 @@ struct ResourceProviderMessage
struct UpdateState
{
- ResourceProviderID id;
+ ResourceProviderInfo info;
UUID resourceVersionUuid;
Resources total;
std::vector<OfferOperation> operations;
@@ -76,7 +76,7 @@ inline std::ostream& operator<<(
return stream
<< "UPDATE_STATE: "
- << updateState->id << " "
+ << updateState->info.id() << " "
<< updateState->total;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/12695ec0/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 7a4a4ac..54d8bcc 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -7083,7 +7083,18 @@ void Slave::handleResourceProviderMessage(
const Resources& newTotal = message->updateState->total;
- const ResourceProviderID& resourceProviderId = message->updateState->id;
+ CHECK(message->updateState->info.has_id());
+
+ const ResourceProviderID& resourceProviderId =
+ message->updateState->info.id();
+
+ if (resourceProviderInfos.contains(resourceProviderId)) {
+ resourceProviderInfos[resourceProviderId] = message->updateState->info;
+ } else {
+ resourceProviderInfos.put(
+ resourceProviderId,
+ message->updateState->info);
+ }
const Resources oldTotal =
totalResources.filter([&resourceProviderId](const Resource& resource) {
http://git-wip-us.apache.org/repos/asf/mesos/blob/12695ec0/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 297d672..5cb0d55 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -725,6 +725,8 @@ private:
process::Owned<LocalResourceProviderDaemon> localResourceProviderDaemon;
hashmap<Option<ResourceProviderID>, UUID> resourceVersions;
+ hashmap<ResourceProviderID, ResourceProviderInfo> resourceProviderInfos;
+
// Pending operations or terminal operations that have
// unacknowledged status updates.
hashmap<UUID, OfferOperation*> offerOperations;
http://git-wip-us.apache.org/repos/asf/mesos/blob/12695ec0/src/tests/resource_provider_manager_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/resource_provider_manager_tests.cpp b/src/tests/resource_provider_manager_tests.cpp
index 7c34b66..a6eb4c9 100644
--- a/src/tests/resource_provider_manager_tests.cpp
+++ b/src/tests/resource_provider_manager_tests.cpp
@@ -339,7 +339,10 @@ TEST_P(ResourceProviderManagerHttpApiTest, UpdateState)
AWAIT_READY(message);
EXPECT_EQ(ResourceProviderMessage::Type::UPDATE_STATE, message->type);
- EXPECT_EQ(devolve(resourceProviderId.get()), message->updateState->id);
+ ASSERT_TRUE(message->updateState->info.has_id());
+ EXPECT_EQ(
+ devolve(resourceProviderId.get()),
+ message->updateState->info.id());
EXPECT_EQ(devolve(resources), message->updateState->total);
}
}
[4/4] mesos git commit: Added a V1 API call to list resource
providers.
Posted by bb...@apache.org.
Added a V1 API call to list resource providers.
The 'GET_RESOURCE_PROVIDERS' call will list all subscribed local
resource providers of an agent.
Review: https://reviews.apache.org/r/64151/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d142d38e
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d142d38e
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d142d38e
Branch: refs/heads/master
Commit: d142d38e3183c6b6ac56ab1c1c1629a6acde3ad5
Parents: 12695ec
Author: Jan Schlicht <ja...@mesosphere.io>
Authored: Fri Dec 8 11:26:23 2017 +0100
Committer: Benjamin Bannier <bb...@apache.org>
Committed: Fri Dec 8 11:26:23 2017 +0100
----------------------------------------------------------------------
include/mesos/agent/agent.proto | 31 ++++++++--
include/mesos/v1/agent/agent.proto | 31 ++++++++--
src/slave/http.cpp | 34 +++++++++++
src/slave/http.hpp | 5 ++
src/slave/validation.cpp | 3 +
src/tests/api_tests.cpp | 102 ++++++++++++++++++++++++++++++++
6 files changed, 198 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d142d38e/include/mesos/agent/agent.proto
----------------------------------------------------------------------
diff --git a/include/mesos/agent/agent.proto b/include/mesos/agent/agent.proto
index 0f92f73..eb5f1b4 100644
--- a/include/mesos/agent/agent.proto
+++ b/include/mesos/agent/agent.proto
@@ -53,10 +53,21 @@ message Call {
GET_STATE = 9;
GET_CONTAINERS = 10;
- GET_FRAMEWORKS = 11; // Retrieves the information about known frameworks.
- GET_EXECUTORS = 12; // Retrieves the information about known executors.
- GET_TASKS = 13; // Retrieves the information about known tasks.
- GET_AGENT = 20; // Retrieves the agent information.
+
+ // Retrieves the information about known frameworks.
+ GET_FRAMEWORKS = 11;
+
+ // Retrieves the information about known executors.
+ GET_EXECUTORS = 12;
+
+ // Retrieves the information about known tasks.
+ GET_TASKS = 13;
+
+ // Retrieves the agent information.
+ GET_AGENT = 20;
+
+ // Retrieves the information about known resource providers.
+ GET_RESOURCE_PROVIDERS = 26;
// Calls for managing nested containers underneath an executor's container.
// Some of these calls are deprecated in favor of the calls
@@ -321,6 +332,7 @@ message Response {
GET_EXECUTORS = 11; // See 'GetExecutors' below.
GET_TASKS = 12; // See 'GetTasks' below.
GET_AGENT = 14; // See 'GetAgent' below.
+ GET_RESOURCE_PROVIDERS = 16; // See 'GetResourceProviders' below.
WAIT_NESTED_CONTAINER = 13 [deprecated = true];
WAIT_CONTAINER = 15; // See 'WaitContainer' below.
@@ -437,6 +449,16 @@ message Response {
optional SlaveInfo slave_info = 1;
}
+ // Lists information about all resource providers known to the agent
+ // at the current time.
+ message GetResourceProviders {
+ message ResourceProvider {
+ required ResourceProviderInfo resource_provider_info = 1;
+ }
+
+ repeated ResourceProvider resource_providers = 1;
+ }
+
// Returns termination information about the nested container.
message WaitNestedContainer {
// Wait status of the lead process in the container. Note that this
@@ -496,6 +518,7 @@ message Response {
optional GetExecutors get_executors = 12;
optional GetTasks get_tasks = 13;
optional GetAgent get_agent = 15;
+ optional GetResourceProviders get_resource_providers = 17;
optional WaitNestedContainer wait_nested_container = 14;
optional WaitContainer wait_container = 16;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/d142d38e/include/mesos/v1/agent/agent.proto
----------------------------------------------------------------------
diff --git a/include/mesos/v1/agent/agent.proto b/include/mesos/v1/agent/agent.proto
index 012ffef..3aedfe8 100644
--- a/include/mesos/v1/agent/agent.proto
+++ b/include/mesos/v1/agent/agent.proto
@@ -53,10 +53,21 @@ message Call {
GET_STATE = 9;
GET_CONTAINERS = 10;
- GET_FRAMEWORKS = 11; // Retrieves the information about known frameworks.
- GET_EXECUTORS = 12; // Retrieves the information about known executors.
- GET_TASKS = 13; // Retrieves the information about known tasks.
- GET_AGENT = 20; // Retrieves the agent information.
+
+ // Retrieves the information about known frameworks.
+ GET_FRAMEWORKS = 11;
+
+ // Retrieves the information about known executors.
+ GET_EXECUTORS = 12;
+
+ // Retrieves the information about known tasks.
+ GET_TASKS = 13;
+
+ // Retrieves the agent information.
+ GET_AGENT = 20;
+
+ // Retrieves the information about known resource providers.
+ GET_RESOURCE_PROVIDERS = 26;
// Calls for managing nested containers underneath an executor's container.
// Some of these calls are deprecated in favor of the calls
@@ -321,6 +332,7 @@ message Response {
GET_EXECUTORS = 11; // See 'GetExecutors' below.
GET_TASKS = 12; // See 'GetTasks' below.
GET_AGENT = 14; // See 'GetAgent' below.
+ GET_RESOURCE_PROVIDERS = 16; // See 'GetResourceProviders' below.
WAIT_NESTED_CONTAINER = 13 [deprecated = true];
WAIT_CONTAINER = 15; // See 'WaitContainer' below.
@@ -437,6 +449,16 @@ message Response {
optional AgentInfo agent_info = 1;
}
+ // Lists information about all resource providers known to the agent
+ // at the current time.
+ message GetResourceProviders {
+ message ResourceProvider {
+ required ResourceProviderInfo resource_provider_info = 1;
+ }
+
+ repeated ResourceProvider resource_providers = 1;
+ }
+
// Returns termination information about the nested container.
message WaitNestedContainer {
// Wait status of the lead process in the container. Note that this
@@ -496,6 +518,7 @@ message Response {
optional GetExecutors get_executors = 12;
optional GetTasks get_tasks = 13;
optional GetAgent get_agent = 15;
+ optional GetResourceProviders get_resource_providers = 17;
optional WaitNestedContainer wait_nested_container = 14;
optional WaitContainer wait_container = 16;
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/d142d38e/src/slave/http.cpp
----------------------------------------------------------------------
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index fd0e809..49278a3 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -577,6 +577,9 @@ Future<Response> Http::_api(
case mesos::agent::Call::GET_AGENT:
return getAgent(call, mediaTypes.accept, principal);
+ case mesos::agent::Call::GET_RESOURCE_PROVIDERS:
+ return getResourceProviders(call, mediaTypes.accept, principal);
+
case mesos::agent::Call::LAUNCH_NESTED_CONTAINER:
return launchNestedContainer(call, mediaTypes.accept, principal);
@@ -1853,6 +1856,37 @@ Future<Response> Http::getAgent(
}
+Future<Response> Http::getResourceProviders(
+ const mesos::agent::Call& call,
+ ContentType acceptType,
+ const Option<Principal>& principal) const
+{
+ CHECK_EQ(mesos::agent::Call::GET_RESOURCE_PROVIDERS, call.type());
+
+ LOG(INFO) << "Processing GET_RESOURCE_PROVIDERS call";
+
+ // TODO(nfnt): Authorize this call (MESOS-8314).
+
+ agent::Response response;
+ response.set_type(mesos::agent::Response::GET_RESOURCE_PROVIDERS);
+
+ agent::Response::GetResourceProviders* resourceProviders =
+ response.mutable_get_resource_providers();
+
+ foreachvalue (
+ const ResourceProviderInfo& resourceProviderInfo,
+ slave->resourceProviderInfos) {
+ agent::Response::GetResourceProviders::ResourceProvider* resourceProvider =
+ resourceProviders->add_resource_providers();
+
+ resourceProvider->mutable_resource_provider_info()->CopyFrom(
+ resourceProviderInfo);
+ }
+
+ return OK(serialize(acceptType, evolve(response)), stringify(acceptType));
+}
+
+
Future<Response> Http::getState(
const mesos::agent::Call& call,
ContentType acceptType,
http://git-wip-us.apache.org/repos/asf/mesos/blob/d142d38e/src/slave/http.hpp
----------------------------------------------------------------------
diff --git a/src/slave/http.hpp b/src/slave/http.hpp
index a51831c..5eecb2a 100644
--- a/src/slave/http.hpp
+++ b/src/slave/http.hpp
@@ -200,6 +200,11 @@ private:
ContentType acceptType,
const Option<process::http::authentication::Principal>& principal) const;
+ process::Future<process::http::Response> getResourceProviders(
+ const mesos::agent::Call& call,
+ ContentType acceptType,
+ const Option<process::http::authentication::Principal>& principal) const;
+
process::Future<process::http::Response> getState(
const mesos::agent::Call& call,
ContentType acceptType,
http://git-wip-us.apache.org/repos/asf/mesos/blob/d142d38e/src/slave/validation.cpp
----------------------------------------------------------------------
diff --git a/src/slave/validation.cpp b/src/slave/validation.cpp
index 32781fd..4a3a78a 100644
--- a/src/slave/validation.cpp
+++ b/src/slave/validation.cpp
@@ -156,6 +156,9 @@ Option<Error> validate(
case mesos::agent::Call::GET_AGENT:
return None();
+ case mesos::agent::Call::GET_RESOURCE_PROVIDERS:
+ return None();
+
case mesos::agent::Call::LAUNCH_NESTED_CONTAINER: {
if (!call.has_launch_nested_container()) {
return Error("Expecting 'launch_nested_container' to be present");
http://git-wip-us.apache.org/repos/asf/mesos/blob/d142d38e/src/tests/api_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/api_tests.cpp b/src/tests/api_tests.cpp
index 53c705e..c8855d0 100644
--- a/src/tests/api_tests.cpp
+++ b/src/tests/api_tests.cpp
@@ -6009,6 +6009,108 @@ TEST_P(AgentAPITest, DefaultAccept)
}
+TEST_P(AgentAPITest, GetResourceProviders)
+{
+ Clock::pause();
+
+ const ContentType contentType = GetParam();
+
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ StandaloneMasterDetector detector(master.get()->pid);
+
+ Future<UpdateSlaveMessage> updateSlaveMessage =
+ FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
+
+ slave::Flags slaveFlags = CreateSlaveFlags();
+ slaveFlags.authenticate_http_readwrite = false;
+
+ constexpr SlaveInfo::Capability::Type capabilities[] = {
+ SlaveInfo::Capability::MULTI_ROLE,
+ SlaveInfo::Capability::HIERARCHICAL_ROLE,
+ SlaveInfo::Capability::RESERVATION_REFINEMENT,
+ SlaveInfo::Capability::RESOURCE_PROVIDER};
+
+ slaveFlags.agent_features = SlaveCapabilities();
+ foreach (SlaveInfo::Capability::Type type, capabilities) {
+ SlaveInfo::Capability* capability =
+ slaveFlags.agent_features->add_capabilities();
+ capability->set_type(type);
+ }
+
+ Try<Owned<cluster::Slave>> slave = StartSlave(&detector, slaveFlags);
+ ASSERT_SOME(slave);
+
+ Clock::advance(slaveFlags.registration_backoff_factor);
+ Clock::settle();
+ AWAIT_READY(updateSlaveMessage);
+
+ v1::agent::Call v1Call;
+ v1Call.set_type(v1::agent::Call::GET_RESOURCE_PROVIDERS);
+
+ Future<v1::agent::Response> v1Response =
+ post(slave.get()->pid, v1Call, contentType);
+
+ AWAIT_READY(v1Response);
+ ASSERT_TRUE(v1Response->IsInitialized());
+ ASSERT_EQ(v1::agent::Response::GET_RESOURCE_PROVIDERS, v1Response->type());
+
+ EXPECT_TRUE(
+ v1Response->get_resource_providers().resource_providers().empty());
+
+ mesos::v1::ResourceProviderInfo info;
+ info.set_type("org.apache.mesos.rp.test");
+ info.set_name("test");
+
+ v1::MockResourceProvider resourceProvider(
+ info,
+ v1::createDiskResource(
+ "200", "*", None(), None(), v1::createDiskSourceRaw()));
+
+ // Start and register a resource provider.
+ string scheme = "http";
+
+#ifdef USE_SSL_SOCKET
+ if (process::network::openssl::flags().enabled) {
+ scheme = "https";
+ }
+#endif
+
+ http::URL url(
+ scheme,
+ slave.get()->pid.address.ip,
+ slave.get()->pid.address.port,
+ slave.get()->pid.id + "/api/v1/resource_provider");
+
+ Owned<EndpointDetector> endpointDetector(new ConstantEndpointDetector(url));
+
+ updateSlaveMessage = FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
+
+ resourceProvider.start(endpointDetector, contentType, v1::DEFAULT_CREDENTIAL);
+
+ // Wait until the agent's resources have been updated to include the
+ // resource provider resources.
+ AWAIT_READY(updateSlaveMessage);
+
+ v1Response = post(slave.get()->pid, v1Call, contentType);
+
+ AWAIT_READY(v1Response);
+ ASSERT_TRUE(v1Response->IsInitialized());
+ ASSERT_EQ(v1::agent::Response::GET_RESOURCE_PROVIDERS, v1Response->type());
+
+ EXPECT_EQ(1, v1Response->get_resource_providers().resource_providers_size());
+
+ const mesos::v1::ResourceProviderInfo& responseInfo =
+ v1Response->get_resource_providers()
+ .resource_providers(0)
+ .resource_provider_info();
+
+ EXPECT_EQ(info.type(), responseInfo.type());
+ EXPECT_EQ(info.name(), responseInfo.name());
+}
+
+
class AgentAPIStreamingTest
: public MesosTest,
public WithParamInterface<ContentType> {};