You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by ch...@apache.org on 2019/03/05 03:28:17 UTC
[mesos] branch master updated (ccce4ec -> 5a2df2c)
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git.
from ccce4ec Removed outdated authorization logic for offer operations.
new bf3982d Made SLRP `PublishResources` test to check persistent volume cleanup.
new 8aaf502 Made SLRP `PublishResourcesReboot` test to check persistent volume cleanup.
new a6effff Made SLRP `PublishResourcesRecovery` test to check volume cleanup.
new 6b05bc8 Updated `CreateDestroyDisk*` SLRP tests to test pipelined operations.
new 43bf73c Extracted common offer matching functions from SLRP tests.
new 069ac8a Added a SLRP unit test for failed persistent volume cleanup.
new 5a2df2c Removed unnecessary filters in SLRP tests.
The 7 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../storage_local_resource_provider_tests.cpp | 2109 ++++++++++----------
1 file changed, 1004 insertions(+), 1105 deletions(-)
[mesos] 01/07: Made SLRP `PublishResources` test to check
persistent volume cleanup.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit bf3982d8d143e7a4d928047a4f7dec9d69479235
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Jan 31 14:32:54 2019 -0800
Made SLRP `PublishResources` test to check persistent volume cleanup.
This patch renames the `ROOT_PublishResources` test to
`ROOT_CreateDestroyPersistentMountVolume` and makes it verify that the
persistent volume is cleaned up after `DESTROY`.
NOTE: The `filesystem/linux` isolator has been removed from the test
because it is not necessary for the test. However, the root privilege is
still required by the test CSI plugin for bind-mounting.
Review: https://reviews.apache.org/r/69895
---
.../storage_local_resource_provider_tests.cpp | 231 ++++++++++-----------
1 file changed, 115 insertions(+), 116 deletions(-)
diff --git a/src/tests/storage_local_resource_provider_tests.cpp b/src/tests/storage_local_resource_provider_tests.cpp
index 75f0d81..679d995 100644
--- a/src/tests/storage_local_resource_provider_tests.cpp
+++ b/src/tests/storage_local_resource_provider_tests.cpp
@@ -1779,10 +1779,19 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
}
-// This test verifies that the storage local resource provider can
-// publish a volume required by a task, then destroy the published
-// volume after the task finishes.
-TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResources)
+// This test verifies that the storage local resource provider can create and
+// publish a persistent volume used by a task, and the persistent volume will be
+// cleaned up when being destroyed.
+//
+// To accomplish this:
+// 1. Create a MOUNT disk from a RAW disk resource.
+// 2. Create a persistent volume on the MOUNT disk then launches a task to
+// write a file into it.
+// 3. Destroy the persistent volume but keep the MOUNT disk. The file should
+// be deleted.
+// 4. Destroy the MOUNT disk.
+TEST_F(
+ StorageLocalResourceProviderTest, ROOT_CreateDestroyPersistentMountVolume)
{
const string profilesPath = path::join(sandbox.get(), "profiles.json");
@@ -1799,8 +1808,6 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResources)
Owned<MasterDetector> detector = master.get()->createDetector();
slave::Flags slaveFlags = CreateSlaveFlags();
- slaveFlags.isolation = "filesystem/linux";
-
slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
Future<SlaveRegisteredMessage> slaveRegisteredMessage =
@@ -1821,103 +1828,79 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResources)
EXPECT_CALL(sched, registered(&driver, _, _));
- // The framework is expected to see the following offers in sequence:
- // 1. One containing a RAW disk resource before `CREATE_DISK`.
- // 2. One containing a MOUNT disk resource after `CREATE_DISK`.
- // 3. One containing a persistent volume after `CREATE` and `LAUNCH`.
- // 4. One containing the original RAW disk resource after `DESTROY`
- // and `DESTROY_DISK`.
- //
- // We set up the expectations for these offers as the test progresses.
- Future<vector<Offer>> rawDiskOffers;
- Future<vector<Offer>> volumeCreatedOffers;
- Future<vector<Offer>> taskFinishedOffers;
- Future<vector<Offer>> volumeDestroyedOffers;
-
- Sequence offers;
-
- // We use the following filter to filter offers that do not have
- // wanted resources for 365 days (the maximum).
+ // We use the following filter to filter offers that do not have wanted
+ // resources for 365 days (the maximum).
Filters declineFilters;
declineFilters.set_refuse_seconds(Days(365).secs());
- // Decline offers that contain only the agent's default resources.
+ // Decline unwanted offers. The master can send such offers before the
+ // resource provider receives profile updates.
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We are only interested in any storage pool or volume with a "test" profile.
- auto hasSourceType = [](
- const Resource& r,
- const Resource::DiskInfo::Source::Type& type) {
+ auto isStoragePool = [](const Resource& r, const string& profile) {
return r.has_disk() &&
r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ !r.disk().source().has_id() &&
r.disk().source().has_profile() &&
- r.disk().source().profile() == "test" &&
- r.disk().source().type() == type;
+ r.disk().source().profile() == profile;
};
+ Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&rawDiskOffers));
+ std::bind(isStoragePool, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.start();
- AWAIT_READY(rawDiskOffers);
- ASSERT_FALSE(rawDiskOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- Option<Resource> source;
+ Offer offer = offers->at(0);
- foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
- source = resource;
- break;
- }
- }
+ // Create a MOUNT disk.
+ Resource raw = *Resources(offer.resources())
+ .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .begin();
- ASSERT_SOME(source);
+ auto isMountDisk = [](const Resource& r, const string& profile) {
+ return r.has_disk() &&
+ r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ r.disk().source().has_id() &&
+ r.disk().source().has_profile() &&
+ r.disk().source().profile() == profile;
+ };
- // Create a volume.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeCreatedOffers));
-
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
+ std::bind(isMountDisk, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
- {rawDiskOffers->at(0).id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
+ {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
- AWAIT_READY(volumeCreatedOffers);
- ASSERT_FALSE(volumeCreatedOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- Option<Resource> volume;
+ offer = offers->at(0);
- foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::MOUNT)) {
- volume = resource;
- break;
- }
- }
+ Resource created = *Resources(offer.resources())
+ .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .begin();
- ASSERT_SOME(volume);
- ASSERT_TRUE(volume->disk().source().has_vendor());
- EXPECT_EQ(TEST_CSI_VENDOR, volume->disk().source().vendor());
- ASSERT_TRUE(volume->disk().source().has_id());
- ASSERT_TRUE(volume->disk().source().has_metadata());
- ASSERT_TRUE(volume->disk().source().has_mount());
- ASSERT_TRUE(volume->disk().source().mount().has_root());
- EXPECT_FALSE(path::absolute(volume->disk().source().mount().root()));
+ ASSERT_TRUE(created.disk().source().has_metadata());
+ ASSERT_TRUE(created.disk().source().has_mount());
+ ASSERT_TRUE(created.disk().source().mount().has_root());
+ EXPECT_FALSE(path::absolute(created.disk().source().mount().root()));
- // Check if the volume is actually created by the test CSI plugin.
+ // Check if the CSI volume is actually created.
Option<string> volumePath;
-
- foreach (const Label& label, volume->disk().source().metadata().labels()) {
+ foreach (const Label& label, created.disk().source().metadata().labels()) {
if (label.key() == "path") {
volumePath = label.value();
break;
@@ -1927,12 +1910,8 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResources)
ASSERT_SOME(volumePath);
EXPECT_TRUE(os::exists(volumePath.get()));
- // Put a file into the volume.
- ASSERT_SOME(os::touch(path::join(volumePath.get(), "file")));
-
- // Create a persistent volume on the CSI volume, then launch a task to
- // use the persistent volume.
- Resource persistentVolume = volume.get();
+ // Create a persistent MOUNT volume then launch a task to write a file.
+ Resource persistentVolume = created;
persistentVolume.mutable_disk()->mutable_persistence()
->set_id(id::UUID::random().toString());
persistentVolume.mutable_disk()->mutable_persistence()
@@ -1941,56 +1920,76 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResources)
->set_container_path("volume");
persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
- Future<TaskStatus> taskStarting;
- Future<TaskStatus> taskRunning;
- Future<TaskStatus> taskFinished;
-
- EXPECT_CALL(sched, statusUpdate(&driver, _))
- .WillOnce(FutureArg<1>(&taskStarting))
- .WillOnce(FutureArg<1>(&taskRunning))
- .WillOnce(FutureArg<1>(&taskFinished));
+ Future<Nothing> taskFinished;
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
+ .WillOnce(FutureSatisfy(&taskFinished));
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(
- persistentVolume)))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&taskFinishedOffers));
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
- {volumeCreatedOffers->at(0).id()},
+ {offer.id()},
{CREATE(persistentVolume),
LAUNCH({createTask(
- volumeCreatedOffers->at(0).slave_id(),
+ offer.slave_id(),
persistentVolume,
- createCommandInfo("test -f " + path::join("volume", "file")))})},
- acceptFilters);
+ createCommandInfo("touch " + path::join("volume", "file")))})});
- AWAIT_READY(taskStarting);
- EXPECT_EQ(TASK_STARTING, taskStarting->state());
+ AWAIT_READY(taskFinished);
- AWAIT_READY(taskRunning);
- EXPECT_EQ(TASK_RUNNING, taskRunning->state());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- AWAIT_READY(taskFinished);
- EXPECT_EQ(TASK_FINISHED, taskFinished->state());
+ offer = offers->at(0);
- AWAIT_READY(taskFinishedOffers);
+ // Destroy the persistent volume.
+ Future<UpdateOperationStatusMessage> updateOperationStatusMessage =
+ FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
- // Destroy the persistent volume and the CSI volume.
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(source.get())))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeDestroyedOffers));
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(created)))
+ .WillOnce(FutureArg<1>(&offers));
- driver.acceptOffers(
- {taskFinishedOffers->at(0).id()},
- {DESTROY(persistentVolume),
- DESTROY_DISK(volume.get())},
- acceptFilters);
+ // TODO(chhsiao): We use the following filter so that the resources will not
+ // be filtered for 5 seconds (the default) because of MESOS-9616. Remove the
+ // filter once it is resolved.
+ Filters acceptFilters;
+ acceptFilters.set_refuse_seconds(0);
+ driver.acceptOffers({offer.id()}, {DESTROY(persistentVolume)}, acceptFilters);
- AWAIT_READY(volumeDestroyedOffers);
- ASSERT_FALSE(volumeDestroyedOffers->empty());
+ // NOTE: Since `DESTROY` would be applied by the master synchronously, we
+ // might get an offer before the persistent volume is cleaned up on the agent,
+ // so we wait for an `UpdateOperationStatusMessage` before the check below.
+ AWAIT_READY(updateOperationStatusMessage);
+ EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
- // Check if the volume is actually deleted by the test CSI plugin.
+ // Check if the CSI volume still exists but has being cleaned up.
+ EXPECT_TRUE(os::exists(volumePath.get()));
+ EXPECT_FALSE(os::exists(path::join(volumePath.get(), "file")));
+
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
+
+ offer = offers->at(0);
+
+ // Destroy the MOUNT disk.
+ updateOperationStatusMessage =
+ FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
+
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.acceptOffers({offer.id()}, {DESTROY_DISK(created)});
+
+ AWAIT_READY(updateOperationStatusMessage);
+ EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
+
+ // Check if the CSI volume is actually deleted.
EXPECT_FALSE(os::exists(volumePath.get()));
+
+ AWAIT_READY(offers);
}
[mesos] 03/07: Made SLRP `PublishResourcesRecovery` test to check
volume cleanup.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit a6effff757467d6443ccc1328a65cd3e36aaa871
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Fri Feb 1 17:01:06 2019 -0800
Made SLRP `PublishResourcesRecovery` test to check volume cleanup.
This patch renames the `ROOT_PublishResourcesRecovery` test to
`ROOT_CreateDestroyPersistentMountVolumeWithRecovery` and makes it
verify that the persistent volume is cleaned up with `DESTROY` after
recovery.
NOTE: The `filesystem/linux` isolator has been removed from the test
because it is not necessary for the test. However, the root privilege is
still required by the test CSI plugin for bind-mounting.
Review: https://reviews.apache.org/r/69897
---
.../storage_local_resource_provider_tests.cpp | 388 +++++++++++----------
1 file changed, 209 insertions(+), 179 deletions(-)
diff --git a/src/tests/storage_local_resource_provider_tests.cpp b/src/tests/storage_local_resource_provider_tests.cpp
index f446a46..575795b 100644
--- a/src/tests/storage_local_resource_provider_tests.cpp
+++ b/src/tests/storage_local_resource_provider_tests.cpp
@@ -1993,9 +1993,22 @@ TEST_F(
}
-// This test verifies that the storage local resource provider can
-// destroy a published volume after recovery.
-TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesRecovery)
+// This test verifies that the storage local resource provider can republish and
+// clean up persistent volumes after recovery.
+//
+// To accomplish this:
+// 1. Create a MOUNT disk from a RAW disk resources.
+// 2. Create a persistent volume on the MOUNT disk then launch a task to
+// write a file into it.
+// 3. Restart the agent.
+// 4. Launch another task to read the file.
+// 5. Restart the agent again.
+// 6. Destroy the persistent volume but keep the MOUNT disk. The file should
+// be deleted.
+// 7. Destroy the MOUNT disk.
+TEST_F(
+ StorageLocalResourceProviderTest,
+ ROOT_CreateDestroyPersistentMountVolumeWithRecovery)
{
const string profilesPath = path::join(sandbox.get(), "profiles.json");
@@ -2012,8 +2025,6 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesRecovery)
Owned<MasterDetector> detector = master.get()->createDetector();
slave::Flags slaveFlags = CreateSlaveFlags();
- slaveFlags.isolation = "filesystem/linux";
-
slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
Future<SlaveRegisteredMessage> slaveRegisteredMessage =
@@ -2034,120 +2045,74 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesRecovery)
EXPECT_CALL(sched, registered(&driver, _, _));
- // The framework is expected to see the following offers in sequence:
- // 1. One containing a RAW disk resource before `CREATE_DISK`.
- // 2. One containing a MOUNT disk resource after `CREATE_DISK`.
- // 3. One containing a persistent volume after `CREATE` and `LAUNCH`.
- // 4. One containing the same persistent volume after the agent
- // recovers from a failover.
- // 5. One containing the same persistent volume after another `LAUNCH`.
- // 6. One containing the original RAW disk resource after `DESTROY`
- // and `DESTROY_DISK`.
- //
- // We set up the expectations for these offers as the test progresses.
- Future<vector<Offer>> rawDiskOffers;
- Future<vector<Offer>> volumeCreatedOffers;
- Future<vector<Offer>> task1FinishedOffers;
- Future<vector<Offer>> slaveRecoveredOffers;
- Future<vector<Offer>> task2FinishedOffers;
- Future<vector<Offer>> volumeDestroyedOffers;
-
- Sequence offers;
-
- // We use the following filter to filter offers that do not have
- // wanted resources for 365 days (the maximum).
+ // We use the following filter to filter offers that do not have wanted
+ // resources for 365 days (the maximum).
Filters declineFilters;
declineFilters.set_refuse_seconds(Days(365).secs());
- // Decline offers that contain only the agent's default resources.
+ // Decline unwanted offers. The master can send such offers before the
+ // resource provider receives profile updates.
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We are only interested in any storage pool or volume with a "test" profile.
- auto hasSourceType = [](
- const Resource& r,
- const Resource::DiskInfo::Source::Type& type) {
+ auto isStoragePool = [](const Resource& r, const string& profile) {
return r.has_disk() &&
r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ !r.disk().source().has_id() &&
r.disk().source().has_profile() &&
- r.disk().source().profile() == "test" &&
- r.disk().source().type() == type;
+ r.disk().source().profile() == profile;
};
+ Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&rawDiskOffers));
+ std::bind(isStoragePool, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.start();
- AWAIT_READY(rawDiskOffers);
- ASSERT_FALSE(rawDiskOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- Option<Resource> source;
+ Offer offer = offers->at(0);
- foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
- source = resource;
- break;
- }
- }
+ // Create a MOUNT disk.
+ Resource raw = *Resources(offer.resources())
+ .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .begin();
- ASSERT_SOME(source);
+ auto isMountDisk = [](const Resource& r, const string& profile) {
+ return r.has_disk() &&
+ r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ r.disk().source().has_id() &&
+ r.disk().source().has_profile() &&
+ r.disk().source().profile() == profile &&
+ !r.disk().has_persistence();
+ };
- // Create a volume.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeCreatedOffers));
-
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
+ std::bind(isMountDisk, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
- {rawDiskOffers->at(0).id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
-
- AWAIT_READY(volumeCreatedOffers);
- ASSERT_FALSE(volumeCreatedOffers->empty());
-
- Option<Resource> volume;
-
- foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::MOUNT)) {
- volume = resource;
- break;
- }
- }
-
- ASSERT_SOME(volume);
- ASSERT_TRUE(volume->disk().source().has_vendor());
- EXPECT_EQ(TEST_CSI_VENDOR, volume->disk().source().vendor());
- ASSERT_TRUE(volume->disk().source().has_id());
- ASSERT_TRUE(volume->disk().source().has_metadata());
- ASSERT_TRUE(volume->disk().source().has_mount());
- ASSERT_TRUE(volume->disk().source().mount().has_root());
- EXPECT_FALSE(path::absolute(volume->disk().source().mount().root()));
+ {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
- // Check if the volume is actually created by the test CSI plugin.
- Option<string> volumePath;
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- foreach (const Label& label, volume->disk().source().metadata().labels()) {
- if (label.key() == "path") {
- volumePath = label.value();
- break;
- }
- }
+ offer = offers->at(0);
- ASSERT_SOME(volumePath);
- EXPECT_TRUE(os::exists(volumePath.get()));
+ Resource created = *Resources(offer.resources())
+ .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .begin();
- // Create a persistent volume on the CSI volume, then launch a task to
- // use the persistent volume.
- Resource persistentVolume = volume.get();
+ // Create a persistent MOUNT volume then launch a task to write a file.
+ Resource persistentVolume = created;
persistentVolume.mutable_disk()->mutable_persistence()
->set_id(id::UUID::random().toString());
persistentVolume.mutable_disk()->mutable_persistence()
@@ -2156,111 +2121,176 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesRecovery)
->set_container_path("volume");
persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
- {
- Future<TaskStatus> taskStarting;
- Future<TaskStatus> taskRunning;
- Future<TaskStatus> taskFinished;
-
- EXPECT_CALL(sched, statusUpdate(&driver, _))
- .WillOnce(FutureArg<1>(&taskStarting))
- .WillOnce(FutureArg<1>(&taskRunning))
- .WillOnce(FutureArg<1>(&taskFinished));
-
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(
- persistentVolume)))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&task1FinishedOffers));
-
- driver.acceptOffers(
- {volumeCreatedOffers->at(0).id()},
- {CREATE(persistentVolume),
- LAUNCH({createTask(
- volumeCreatedOffers->at(0).slave_id(),
- persistentVolume,
- createCommandInfo("touch " + path::join("volume", "file")))})},
- acceptFilters);
-
- AWAIT_READY(taskStarting);
- EXPECT_EQ(TASK_STARTING, taskStarting->state());
-
- AWAIT_READY(taskRunning);
- EXPECT_EQ(TASK_RUNNING, taskRunning->state());
-
- AWAIT_READY(taskFinished);
- EXPECT_EQ(TASK_FINISHED, taskFinished->state());
- }
+ Future<Nothing> taskFinished;
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
+ .WillOnce(FutureSatisfy(&taskFinished));
+
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.acceptOffers(
+ {offer.id()},
+ {CREATE(persistentVolume),
+ LAUNCH({createTask(
+ offer.slave_id(),
+ persistentVolume,
+ createCommandInfo("touch " + path::join("volume", "file")))})});
- AWAIT_READY(task1FinishedOffers);
+ AWAIT_READY(taskFinished);
+
+ AWAIT_READY(offers);
// Restart the agent.
EXPECT_CALL(sched, offerRescinded(_, _));
slave.get()->terminate();
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(
- persistentVolume)))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&slaveRecoveredOffers));
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
slave = StartSlave(detector.get(), slaveFlags);
ASSERT_SOME(slave);
- AWAIT_READY(slaveRecoveredOffers);
- ASSERT_FALSE(slaveRecoveredOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- // Launch another task to read the file that is created by the
- // previous task on the persistent volume.
- {
- Future<TaskStatus> taskStarting;
- Future<TaskStatus> taskRunning;
- Future<TaskStatus> taskFinished;
-
- EXPECT_CALL(sched, statusUpdate(&driver, _))
- .WillOnce(FutureArg<1>(&taskStarting))
- .WillOnce(FutureArg<1>(&taskRunning))
- .WillOnce(FutureArg<1>(&taskFinished));
-
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(
- persistentVolume)))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&task2FinishedOffers));
-
- driver.acceptOffers(
- {slaveRecoveredOffers->at(0).id()},
- {LAUNCH({createTask(
- slaveRecoveredOffers->at(0).slave_id(),
- persistentVolume,
- createCommandInfo("test -f " + path::join("volume", "file")))})},
- acceptFilters);
-
- AWAIT_READY(taskStarting);
- EXPECT_EQ(TASK_STARTING, taskStarting->state());
-
- AWAIT_READY(taskRunning);
- EXPECT_EQ(TASK_RUNNING, taskRunning->state());
-
- AWAIT_READY(taskFinished);
- EXPECT_EQ(TASK_FINISHED, taskFinished->state());
- }
+ offer = offers->at(0);
- AWAIT_READY(task2FinishedOffers);
+ // Launch another task to read the file created by the previous task.
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
+ .WillOnce(FutureSatisfy(&taskFinished));
- // Destroy the persistent volume and the CSI volume.
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(source.get())))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeDestroyedOffers));
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
- {task2FinishedOffers->at(0).id()},
- {DESTROY(persistentVolume),
- DESTROY_DISK(volume.get())},
- acceptFilters);
+ {offer.id()},
+ {LAUNCH({createTask(
+ offer.slave_id(),
+ persistentVolume,
+ createCommandInfo("test -f " + path::join("volume", "file")))})});
- AWAIT_READY(volumeDestroyedOffers);
- ASSERT_FALSE(volumeDestroyedOffers->empty());
+ AWAIT_READY(taskFinished);
- // Check if the volume is actually deleted by the test CSI plugin.
+ AWAIT_READY(offers);
+
+ // Restart the agent.
+ EXPECT_CALL(sched, offerRescinded(_, _));
+
+ slave.get()->terminate();
+
+ // NOTE: We set up the resource provider with an extra storage pool, so that
+ // when the storage pool shows up, we know that the resource provider has
+ // finished reconciling storage pools and thus operations won't be dropped.
+ //
+ // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
+ // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
+ // `UpdateSlaveMessage`s (one sent after agent reregistration and one after
+ // resource provider reregistration) and wait for the third one, which should
+ // contain the extra storage pool. We let it fall through to trigger an offer
+ // allocation for the persistent volume.
+ //
+ // Since the extra storage pool is never used, we reject the offers if only
+ // the storage pool is presented.
+ //
+ // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
+ setupResourceProviderConfig(Gigabytes(5));
+
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isStoragePool, lambda::_1, "test"))))
+ .WillRepeatedly(DeclineOffers(declineFilters));
+
+ // NOTE: The order of these expectations is reversed because Google Mock will
+ // search the expectations in reverse order.
+ Future<UpdateSlaveMessage> updateSlaveMessage =
+ FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
+
+ DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
+ DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
+ DROP_PROTOBUFS(SlaveReregisteredMessage(), _, _);
+ FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
+
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ slave = StartSlave(detector.get(), slaveFlags);
+ ASSERT_SOME(slave);
+
+ AWAIT_READY(updateSlaveMessage);
+ ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
+ ASSERT_FALSE(Resources(
+ updateSlaveMessage->resource_providers().providers(0).total_resources())
+ .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .empty());
+
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
+
+ offer = offers->at(0);
+
+ // Destroy the persistent volume.
+ Future<UpdateOperationStatusMessage> updateOperationStatusMessage =
+ FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
+
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(created)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ // TODO(chhsiao): We use the following filter so that the resources will not
+ // be filtered for 5 seconds (the default) because of MESOS-9616. Remove the
+ // filter once it is resolved.
+ Filters acceptFilters;
+ acceptFilters.set_refuse_seconds(0);
+ driver.acceptOffers({offer.id()}, {DESTROY(persistentVolume)}, acceptFilters);
+
+ // NOTE: Since `DESTROY` would be applied by the master synchronously, we
+ // might get an offer before the persistent volume is cleaned up on the agent,
+ // so we wait for an `UpdateOperationStatusMessage` before the check below.
+ AWAIT_READY(updateOperationStatusMessage);
+ EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
+
+ // Check if the CSI volume still exists but has being cleaned up.
+ Option<string> volumePath;
+
+ foreach (const Label& label, created.disk().source().metadata().labels()) {
+ if (label.key() == "path") {
+ volumePath = label.value();
+ break;
+ }
+ }
+
+ ASSERT_SOME(volumePath);
+ EXPECT_TRUE(os::exists(volumePath.get()));
+ EXPECT_FALSE(os::exists(path::join(volumePath.get(), "file")));
+
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
+
+ offer = offers->at(0);
+
+ // Destroy the MOUNT disk.
+ updateOperationStatusMessage =
+ FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
+
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.acceptOffers({offer.id()}, {DESTROY_DISK(created)});
+
+ AWAIT_READY(updateOperationStatusMessage);
+ EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
+
+ // Check if the CSI volume is actually deleted.
EXPECT_FALSE(os::exists(volumePath.get()));
+
+ AWAIT_READY(offers);
}
[mesos] 02/07: Made SLRP `PublishResourcesReboot` test to check
persistent volume cleanup.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 8aaf50291f8cc601cf1742dfca610bd27bc897f4
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Jan 31 19:05:48 2019 -0800
Made SLRP `PublishResourcesReboot` test to check persistent volume cleanup.
This patch renames the `ROOT_PublishResourcesReboot` test to
`ROOT_CreateDestroyPersistentMountVolumeWithReboot` and makes it verify
that the persistent volume is cleaned up with `DESTROY` after a reboot.
NOTE: The `filesystem/linux` isolator has been removed from the test
because it is not necessary for the test. However, the root privilege is
still required by the test CSI plugin for bind-mounting.
Review: https://reviews.apache.org/r/69896
---
.../storage_local_resource_provider_tests.cpp | 438 +++++++++++----------
1 file changed, 237 insertions(+), 201 deletions(-)
diff --git a/src/tests/storage_local_resource_provider_tests.cpp b/src/tests/storage_local_resource_provider_tests.cpp
index 679d995..f446a46 100644
--- a/src/tests/storage_local_resource_provider_tests.cpp
+++ b/src/tests/storage_local_resource_provider_tests.cpp
@@ -2264,9 +2264,22 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesRecovery)
}
-// This test verifies that the storage local resource provider can
-// destroy a published volume after agent reboot.
-TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesReboot)
+// This test verifies that the storage local resource provider can republish and
+// clean up persistent volumes after agent reboot.
+//
+// To accomplish this:
+// 1. Create a MOUNT disk from a RAW disk resource.
+// 2. Create a persistent volume on the MOUNT disk then launches a task to
+// write a file into it.
+// 3. Simulate an agent reboot.
+// 4. Launch another task to read the file.
+// 5. Simulate another agent reboot.
+// 6. Destroy the persistent volume but keep the MOUNT disk. The file should
+// be deleted.
+// 7. Destroy the MOUNT disk.
+TEST_F(
+ StorageLocalResourceProviderTest,
+ ROOT_CreateDestroyPersistentMountVolumeWithReboot)
{
const string profilesPath = path::join(sandbox.get(), "profiles.json");
@@ -2283,8 +2296,6 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesReboot)
Owned<MasterDetector> detector = master.get()->createDetector();
slave::Flags slaveFlags = CreateSlaveFlags();
- slaveFlags.isolation = "filesystem/linux";
-
slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
Future<SlaveRegisteredMessage> slaveRegisteredMessage =
@@ -2305,120 +2316,74 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesReboot)
EXPECT_CALL(sched, registered(&driver, _, _));
- // The framework is expected to see the following offers in sequence:
- // 1. One containing a RAW disk resource before `CREATE_DISK`.
- // 2. One containing a MOUNT disk resource after `CREATE_DISK`.
- // 3. One containing a persistent volume after `CREATE` and `LAUNCH`.
- // 4. One containing the same persistent volume after the agent
- // recovers from a failover.
- // 5. One containing the same persistent volume after another `LAUNCH`.
- // 6. One containing the original RAW disk resource after `DESTROY`
- // and `DESTROY_DISK`.
- //
- // We set up the expectations for these offers as the test progresses.
- Future<vector<Offer>> rawDiskOffers;
- Future<vector<Offer>> volumeCreatedOffers;
- Future<vector<Offer>> task1FinishedOffers;
- Future<vector<Offer>> slaveRecoveredOffers;
- Future<vector<Offer>> task2FinishedOffers;
- Future<vector<Offer>> volumeDestroyedOffers;
-
- Sequence offers;
-
- // We use the following filter to filter offers that do not have
- // wanted resources for 365 days (the maximum).
+ // We use the following filter to filter offers that do not have wanted
+ // resources for 365 days (the maximum).
Filters declineFilters;
declineFilters.set_refuse_seconds(Days(365).secs());
- // Decline offers that contain only the agent's default resources.
+ // Decline unwanted offers. The master can send such offers before the
+ // resource provider receives profile updates.
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We are only interested in any storage pool or volume with a "test" profile.
- auto hasSourceType = [](
- const Resource& r,
- const Resource::DiskInfo::Source::Type& type) {
+ auto isStoragePool = [](const Resource& r, const string& profile) {
return r.has_disk() &&
r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ !r.disk().source().has_id() &&
r.disk().source().has_profile() &&
- r.disk().source().profile() == "test" &&
- r.disk().source().type() == type;
+ r.disk().source().profile() == profile;
};
+ Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&rawDiskOffers));
+ std::bind(isStoragePool, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.start();
- AWAIT_READY(rawDiskOffers);
- ASSERT_FALSE(rawDiskOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- Option<Resource> source;
+ Offer offer = offers->at(0);
- foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
- source = resource;
- break;
- }
- }
+ // Create a MOUNT disk.
+ Resource raw = *Resources(offer.resources())
+ .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .begin();
- ASSERT_SOME(source);
+ auto isMountDisk = [](const Resource& r, const string& profile) {
+ return r.has_disk() &&
+ r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ r.disk().source().has_id() &&
+ r.disk().source().has_profile() &&
+ r.disk().source().profile() == profile &&
+ !r.disk().has_persistence();
+ };
- // Create a volume.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeCreatedOffers));
-
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
+ std::bind(isMountDisk, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
- {rawDiskOffers->at(0).id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
-
- AWAIT_READY(volumeCreatedOffers);
- ASSERT_FALSE(volumeCreatedOffers->empty());
-
- Option<Resource> volume;
-
- foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::MOUNT)) {
- volume = resource;
- break;
- }
- }
-
- ASSERT_SOME(volume);
- ASSERT_TRUE(volume->disk().source().has_vendor());
- EXPECT_EQ(TEST_CSI_VENDOR, volume->disk().source().vendor());
- ASSERT_TRUE(volume->disk().source().has_id());
- ASSERT_TRUE(volume->disk().source().has_metadata());
- ASSERT_TRUE(volume->disk().source().has_mount());
- ASSERT_TRUE(volume->disk().source().mount().has_root());
- EXPECT_FALSE(path::absolute(volume->disk().source().mount().root()));
+ {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
- // Check if the volume is actually created by the test CSI plugin.
- Option<string> volumePath;
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- foreach (const Label& label, volume->disk().source().metadata().labels()) {
- if (label.key() == "path") {
- volumePath = label.value();
- break;
- }
- }
+ offer = offers->at(0);
- ASSERT_SOME(volumePath);
- EXPECT_TRUE(os::exists(volumePath.get()));
+ Resource created = *Resources(offer.resources())
+ .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .begin();
- // Create a persistent volume on the CSI volume, then launch a task to
- // use the persistent volume.
- Resource persistentVolume = volume.get();
+ // Create a persistent MOUNT volume then launch a task to write a file.
+ Resource persistentVolume = created;
persistentVolume.mutable_disk()->mutable_persistence()
->set_id(id::UUID::random().toString());
persistentVolume.mutable_disk()->mutable_persistence()
@@ -2427,149 +2392,220 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_PublishResourcesReboot)
->set_container_path("volume");
persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
- {
- Future<TaskStatus> taskStarting;
- Future<TaskStatus> taskRunning;
- Future<TaskStatus> taskFinished;
+ Future<Nothing> taskFinished;
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
+ .WillOnce(FutureSatisfy(&taskFinished));
- EXPECT_CALL(sched, statusUpdate(&driver, _))
- .WillOnce(FutureArg<1>(&taskStarting))
- .WillOnce(FutureArg<1>(&taskRunning))
- .WillOnce(FutureArg<1>(&taskFinished));
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(
- persistentVolume)))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&task1FinishedOffers));
+ driver.acceptOffers(
+ {offer.id()},
+ {CREATE(persistentVolume),
+ LAUNCH({createTask(
+ offer.slave_id(),
+ persistentVolume,
+ createCommandInfo("touch " + path::join("volume", "file")))})});
- driver.acceptOffers(
- {volumeCreatedOffers->at(0).id()},
- {CREATE(persistentVolume),
- LAUNCH({createTask(
- volumeCreatedOffers->at(0).slave_id(),
- persistentVolume,
- createCommandInfo("touch " + path::join("volume", "file")))})},
- acceptFilters);
+ AWAIT_READY(taskFinished);
- AWAIT_READY(taskStarting);
- EXPECT_EQ(TASK_STARTING, taskStarting->state());
+ AWAIT_READY(offers);
- AWAIT_READY(taskRunning);
- EXPECT_EQ(TASK_RUNNING, taskRunning->state());
+ // Shutdown the agent and unmount all CSI volumes to simulate the 1st reboot.
+ EXPECT_CALL(sched, offerRescinded(_, _));
- AWAIT_READY(taskFinished);
- EXPECT_EQ(TASK_FINISHED, taskFinished->state());
- }
+ slave->reset();
- AWAIT_READY(task1FinishedOffers);
+ const string csiRootDir = slave::paths::getCsiRootDir(slaveFlags.work_dir);
+ ASSERT_SOME(fs::unmountAll(csiRootDir));
- // Destruct the agent to shut down all containers.
- EXPECT_CALL(sched, offerRescinded(_, _));
+ // Inject the boot IDs to simulate the 1st reboot.
+ auto injectBootIds = [&](const string& bootId) {
+ ASSERT_SOME(os::write(
+ slave::paths::getBootIdPath(
+ slave::paths::getMetaRootDir(slaveFlags.work_dir)),
+ bootId));
- slave->reset();
+ Try<list<string>> volumePaths =
+ csi::paths::getVolumePaths(csiRootDir, "*", "*");
+ ASSERT_SOME(volumePaths);
+ ASSERT_FALSE(volumePaths->empty());
- // Modify the boot ID to simulate a reboot.
- ASSERT_SOME(os::write(
- slave::paths::getBootIdPath(
- slave::paths::getMetaRootDir(slaveFlags.work_dir)),
- "rebooted! ;)"));
+ foreach (const string& path, volumePaths.get()) {
+ Try<csi::paths::VolumePath> volumePath =
+ csi::paths::parseVolumePath(csiRootDir, path);
+ ASSERT_SOME(volumePath);
- const string csiRootDir = slave::paths::getCsiRootDir(slaveFlags.work_dir);
+ const string volumeStatePath = csi::paths::getVolumeStatePath(
+ csiRootDir, volumePath->type, volumePath->name, volumePath->volumeId);
+
+ Result<csi::state::VolumeState> volumeState =
+ slave::state::read<csi::state::VolumeState>(volumeStatePath);
- Try<list<string>> volumePaths =
- csi::paths::getVolumePaths(csiRootDir, "*", "*");
- ASSERT_SOME(volumePaths);
- ASSERT_FALSE(volumePaths->empty());
-
- foreach (const string& path, volumePaths.get()) {
- Try<csi::paths::VolumePath> volumePath =
- csi::paths::parseVolumePath(csiRootDir, path);
- ASSERT_SOME(volumePath);
-
- const string volumeStatePath = csi::paths::getVolumeStatePath(
- csiRootDir,
- volumePath->type,
- volumePath->name,
- volumePath->volumeId);
-
- Result<csi::state::VolumeState> volumeState =
- slave::state::read<csi::state::VolumeState>(volumeStatePath);
- ASSERT_SOME(volumeState);
-
- if (volumeState->state() == csi::state::VolumeState::PUBLISHED) {
- volumeState->set_boot_id("rebooted! ;)");
- ASSERT_SOME(slave::state::checkpoint(volumeStatePath, volumeState.get()));
+ ASSERT_SOME(volumeState);
+
+ if (volumeState->state() == csi::state::VolumeState::PUBLISHED) {
+ volumeState->set_boot_id(bootId);
+ ASSERT_SOME(
+ slave::state::checkpoint(volumeStatePath, volumeState.get()));
+ }
}
- }
+ };
+
+ injectBootIds("1st reboot! ;)");
- // Unmount all CSI volumes to simulate a reboot.
+ // Restart the agent.
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ slave = StartSlave(detector.get(), slaveFlags);
+ ASSERT_SOME(slave);
+
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
+
+ offer = offers->at(0);
+
+ // Launch another task to read the file created by the previous task.
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
+ .WillOnce(FutureSatisfy(&taskFinished));
+
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.acceptOffers(
+ {offer.id()},
+ {LAUNCH({createTask(
+ offer.slave_id(),
+ persistentVolume,
+ createCommandInfo("test -f " + path::join("volume", "file")))})});
+
+ AWAIT_READY(taskFinished);
+
+ AWAIT_READY(offers);
+
+ // Shutdown the agent and unmount all CSI volumes to simulate the 2nd reboot.
+ EXPECT_CALL(sched, offerRescinded(_, _));
+
+ slave->reset();
ASSERT_SOME(fs::unmountAll(csiRootDir));
+ // Inject the boot IDs to simulate the 2nd reboot.
+ injectBootIds("2nd reboot! ;)");
+
+ // NOTE: We set up the resource provider with an extra storage pool, so that
+ // when the storage pool shows up, we know that the resource provider has
+ // finished reconciling storage pools and thus operations won't be dropped.
+ //
+ // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
+ // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
+ // `UpdateSlaveMessage`s (one sent after agent reregistration and one after
+ // resource provider reregistration) and wait for the third one, which should
+ // contain the extra storage pool. We let it fall through to trigger an offer
+ // allocation for the persistent volume.
+ //
+ // Since the extra storage pool is never used, we reject the offers if only
+ // the storage pool is presented.
+ //
+ // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
+ setupResourceProviderConfig(Gigabytes(5));
+
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isStoragePool, lambda::_1, "test"))))
+ .WillRepeatedly(DeclineOffers(declineFilters));
+
+ // NOTE: The order of these expectations is reversed because Google Mock will
+ // search the expectations in reverse order.
+ Future<UpdateSlaveMessage> updateSlaveMessage =
+ FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
+
+ DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
+ DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
+ DROP_PROTOBUFS(SlaveReregisteredMessage(), _, _);
+ FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
+
// Restart the agent.
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(
- persistentVolume)))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&slaveRecoveredOffers));
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
slave = StartSlave(detector.get(), slaveFlags);
ASSERT_SOME(slave);
- AWAIT_READY(slaveRecoveredOffers);
- ASSERT_FALSE(slaveRecoveredOffers->empty());
+ AWAIT_READY(updateSlaveMessage);
+ ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
+ ASSERT_FALSE(Resources(
+ updateSlaveMessage->resource_providers().providers(0).total_resources())
+ .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .empty());
- // Launch another task to read the file that is created by the
- // previous task on the persistent volume.
- {
- Future<TaskStatus> taskStarting;
- Future<TaskStatus> taskRunning;
- Future<TaskStatus> taskFinished;
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- EXPECT_CALL(sched, statusUpdate(&driver, _))
- .WillOnce(FutureArg<1>(&taskStarting))
- .WillOnce(FutureArg<1>(&taskRunning))
- .WillOnce(FutureArg<1>(&taskFinished));
+ offer = offers->at(0);
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(
- persistentVolume)))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&task2FinishedOffers));
+ // Destroy the persistent volume.
+ Future<UpdateOperationStatusMessage> updateOperationStatusMessage =
+ FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
- driver.acceptOffers(
- {slaveRecoveredOffers->at(0).id()},
- {LAUNCH({createTask(
- slaveRecoveredOffers->at(0).slave_id(),
- persistentVolume,
- createCommandInfo("test -f " + path::join("volume", "file")))})},
- acceptFilters);
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(created)))
+ .WillOnce(FutureArg<1>(&offers));
- AWAIT_READY(taskStarting);
- EXPECT_EQ(TASK_STARTING, taskStarting->state());
+ // TODO(chhsiao): We use the following filter so that the resources will not
+ // be filtered for 5 seconds (the default) because of MESOS-9616. Remove the
+ // filter once it is resolved.
+ Filters acceptFilters;
+ acceptFilters.set_refuse_seconds(0);
+ driver.acceptOffers({offer.id()}, {DESTROY(persistentVolume)}, acceptFilters);
- AWAIT_READY(taskRunning);
- EXPECT_EQ(TASK_RUNNING, taskRunning->state());
+ // NOTE: Since `DESTROY` would be applied by the master synchronously, we
+ // might get an offer before the persistent volume is cleaned up on the agent,
+ // so we wait for an `UpdateOperationStatusMessage` before the check below.
+ AWAIT_READY(updateOperationStatusMessage);
+ EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
- AWAIT_READY(taskFinished);
- EXPECT_EQ(TASK_FINISHED, taskFinished->state());
+ // Check if the CSI volume still exists but has being cleaned up.
+ Option<string> volumePath;
+
+ foreach (const Label& label, created.disk().source().metadata().labels()) {
+ if (label.key() == "path") {
+ volumePath = label.value();
+ break;
+ }
}
- AWAIT_READY(task2FinishedOffers);
+ ASSERT_SOME(volumePath);
+ EXPECT_TRUE(os::exists(volumePath.get()));
+ EXPECT_FALSE(os::exists(path::join(volumePath.get(), "file")));
- // Destroy the persistent volume and the CSI volume.
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(source.get())))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeDestroyedOffers));
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- driver.acceptOffers(
- {task2FinishedOffers->at(0).id()},
- {DESTROY(persistentVolume),
- DESTROY_DISK(volume.get())},
- acceptFilters);
+ offer = offers->at(0);
- AWAIT_READY(volumeDestroyedOffers);
- ASSERT_FALSE(volumeDestroyedOffers->empty());
+ // Destroy the MOUNT disk.
+ updateOperationStatusMessage =
+ FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
- // Check if the volume is actually deleted by the test CSI plugin.
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.acceptOffers({offer.id()}, {DESTROY_DISK(created)});
+
+ AWAIT_READY(updateOperationStatusMessage);
+ EXPECT_EQ(OPERATION_FINISHED, updateOperationStatusMessage->status().state());
+
+ // Check if the CSI volume is actually deleted.
EXPECT_FALSE(os::exists(volumePath.get()));
+
+ AWAIT_READY(offers);
}
[mesos] 04/07: Updated `CreateDestroyDisk*` SLRP tests to test
pipelined operations.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 6b05bc8c05c1951999ea787ffb9c1815bb5ef8b4
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Mon Feb 4 16:38:50 2019 -0800
Updated `CreateDestroyDisk*` SLRP tests to test pipelined operations.
This patch extends the code coverage of the `CreateDestroyDisk` and
`CreateDestroyDiskRecovery` tests by testing pipelined `RESERVE`,
`CREATE`, `DESTROY` and `UNRESERVE` operations along with `CREATE_DISK`
and `DESTROY_DISK`. It also renames `CreateDestroyDiskRecovery` to
`CreateDestroyDiskWithRecovery` for consistency.
Review: https://reviews.apache.org/r/69898
---
.../storage_local_resource_provider_tests.cpp | 412 ++++++++++-----------
1 file changed, 200 insertions(+), 212 deletions(-)
diff --git a/src/tests/storage_local_resource_provider_tests.cpp b/src/tests/storage_local_resource_provider_tests.cpp
index 575795b..e8b3f01 100644
--- a/src/tests/storage_local_resource_provider_tests.cpp
+++ b/src/tests/storage_local_resource_provider_tests.cpp
@@ -717,8 +717,8 @@ TEST_F(StorageLocalResourceProviderTest, ProfileAppeared)
}
-// This test verifies that the storage local resource provider can
-// create then destroy a new volume from a storage pool.
+// This test verifies that the storage local resource provider can create then
+// destroy a MOUNT disk from a storage pool with other pipelined operations.
TEST_F(StorageLocalResourceProviderTest, CreateDestroyDisk)
{
const string profilesPath = path::join(sandbox.get(), "profiles.json");
@@ -748,7 +748,7 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDisk)
// Register a framework to exercise operations.
FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
- framework.set_roles(0, "storage");
+ framework.set_roles(0, "storage/role");
MockScheduler sched;
MesosSchedulerDriver driver(
@@ -756,100 +756,86 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDisk)
EXPECT_CALL(sched, registered(&driver, _, _));
- // The framework is expected to see the following offers in sequence:
- // 1. One containing a RAW disk resource before `CREATE_DISK`.
- // 2. One containing a MOUNT disk resource after `CREATE_DISK`.
- // 3. One containing a RAW disk resource after `DESTROY_DISK`.
- //
- // We set up the expectations for these offers as the test progresses.
- Future<vector<Offer>> rawDiskOffers;
- Future<vector<Offer>> volumeCreatedOffers;
- Future<vector<Offer>> volumeDestroyedOffers;
-
- Sequence offers;
-
- // We use the following filter to filter offers that do not have
- // wanted resources for 365 days (the maximum).
+ // We use the following filter to filter offers that do not have wanted
+ // resources for 365 days (the maximum).
Filters declineFilters;
declineFilters.set_refuse_seconds(Days(365).secs());
- // Decline offers that contain only the agent's default resources.
+ // Decline unwanted offers. The master can send such offers before the
+ // resource provider receives profile updates.
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We are only interested in any storage pool or volume with a "test" profile.
- auto hasSourceType = [](
- const Resource& r,
- const Resource::DiskInfo::Source::Type& type) {
+ auto isStoragePool = [](const Resource& r, const string& profile) {
return r.has_disk() &&
r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ !r.disk().source().has_id() &&
r.disk().source().has_profile() &&
- r.disk().source().profile() == "test" &&
- r.disk().source().type() == type;
+ r.disk().source().profile() == profile;
};
+ Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&rawDiskOffers));
+ std::bind(isStoragePool, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.start();
- AWAIT_READY(rawDiskOffers);
- ASSERT_FALSE(rawDiskOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- Option<Resource> source;
+ Offer offer = offers->at(0);
- foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
- source = resource;
- break;
- }
- }
+ // Create a MOUNT disk with a pipelined `RESERVE` operation.
+ Resource raw = *Resources(offer.resources())
+ .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .begin();
- ASSERT_SOME(source);
+ Resource reserved = *Resources(raw)
+ .pushReservation(createDynamicReservationInfo(
+ framework.roles(0), framework.principal()))
+ .begin();
- // Create a volume.
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeCreatedOffers));
+ auto isMountDisk = [](const Resource& r, const string& profile) {
+ return r.has_disk() &&
+ r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ r.disk().source().has_id() &&
+ r.disk().source().has_profile() &&
+ r.disk().source().profile() == profile;
+ };
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isMountDisk, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
- {rawDiskOffers->at(0).id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
+ {offer.id()},
+ {RESERVE(reserved),
+ CREATE_DISK(reserved, Resource::DiskInfo::Source::MOUNT)});
- AWAIT_READY(volumeCreatedOffers);
- ASSERT_FALSE(volumeCreatedOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- Option<Resource> volume;
+ offer = offers->at(0);
- foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::MOUNT)) {
- volume = resource;
- break;
- }
- }
+ Resource created = *Resources(offer.resources())
+ .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .begin();
- ASSERT_SOME(volume);
- ASSERT_TRUE(volume->disk().source().has_vendor());
- EXPECT_EQ(TEST_CSI_VENDOR, volume->disk().source().vendor());
- ASSERT_TRUE(volume->disk().source().has_id());
- ASSERT_TRUE(volume->disk().source().has_metadata());
- ASSERT_TRUE(volume->disk().source().has_mount());
- ASSERT_TRUE(volume->disk().source().mount().has_root());
- EXPECT_FALSE(path::absolute(volume->disk().source().mount().root()));
+ ASSERT_TRUE(created.disk().source().has_metadata());
+ ASSERT_TRUE(created.disk().source().has_mount());
+ ASSERT_TRUE(created.disk().source().mount().has_root());
+ EXPECT_FALSE(path::absolute(created.disk().source().mount().root()));
// Check if the volume is actually created by the test CSI plugin.
Option<string> volumePath;
-
- foreach (const Label& label, volume->disk().source().metadata().labels()) {
+ foreach (const Label& label, created.disk().source().metadata().labels()) {
if (label.key() == "path") {
volumePath = label.value();
break;
@@ -859,55 +845,49 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDisk)
ASSERT_SOME(volumePath);
EXPECT_TRUE(os::exists(volumePath.get()));
- // Destroy the created volume.
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeDestroyedOffers));
-
- driver.acceptOffers(
- {volumeCreatedOffers->at(0).id()},
- {DESTROY_DISK(volume.get())},
- acceptFilters);
+ // Destroy the MOUNT disk with pipelined `CREATE`, `DESTROY` and `UNRESERVE`
+ // operations. Note that `UNRESERVE` can come before `DESTROY_DISK`.
+ Resource persistentVolume = created;
+ persistentVolume.mutable_disk()->mutable_persistence()
+ ->set_id(id::UUID::random().toString());
+ persistentVolume.mutable_disk()->mutable_persistence()
+ ->set_principal(framework.principal());
+ persistentVolume.mutable_disk()->mutable_volume()
+ ->set_container_path("volume");
+ persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
- AWAIT_READY(volumeDestroyedOffers);
- ASSERT_FALSE(volumeDestroyedOffers->empty());
+ Resource unreserved = *Resources(created)
+ .popReservation()
+ .begin();
- Option<Resource> destroyed;
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
+ .WillOnce(FutureArg<1>(&offers));
- foreach (const Resource& resource, volumeDestroyedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
- destroyed = resource;
- break;
- }
- }
+ driver.acceptOffers(
+ {offer.id()},
+ {CREATE(persistentVolume),
+ DESTROY(persistentVolume),
+ UNRESERVE(created),
+ DESTROY_DISK(unreserved)});
- ASSERT_SOME(destroyed);
- ASSERT_TRUE(destroyed->disk().source().has_vendor());
- EXPECT_EQ(TEST_CSI_VENDOR, destroyed->disk().source().vendor());
- ASSERT_FALSE(destroyed->disk().source().has_id());
- ASSERT_FALSE(destroyed->disk().source().has_metadata());
- ASSERT_FALSE(destroyed->disk().source().has_mount());
+ AWAIT_READY(offers);
// Check if the volume is actually deleted by the test CSI plugin.
EXPECT_FALSE(os::exists(volumePath.get()));
}
-// This test verifies that the storage local resource provider can
-// destroy a volume created from a storage pool after recovery.
-TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskRecovery)
+// This test verifies that the storage local resource provider can destroy a
+// MOUNT disk created from a storage pool with other pipelined operations after
+// recovery.
+TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskWithRecovery)
{
- Future<shared_ptr<TestDiskProfileServer>> server =
- TestDiskProfileServer::create();
- AWAIT_READY(server);
+ const string profilesPath = path::join(sandbox.get(), "profiles.json");
- Promise<http::Response> recoveredProfileMapping;
- EXPECT_CALL(*server.get()->process, profiles(_))
- .WillOnce(Return(http::OK(createDiskProfileMapping({{"test", None()}}))))
- .WillOnce(Return(recoveredProfileMapping.future()));
+ ASSERT_SOME(
+ os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
- loadUriDiskProfileAdaptorModule(stringify(server.get()->process->url()));
+ loadUriDiskProfileAdaptorModule(profilesPath);
setupResourceProviderConfig(Gigabytes(4));
@@ -929,7 +909,7 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskRecovery)
// Register a framework to exercise operations.
FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
- framework.set_roles(0, "storage");
+ framework.set_roles(0, "storage/role");
MockScheduler sched;
MesosSchedulerDriver driver(
@@ -937,103 +917,86 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskRecovery)
EXPECT_CALL(sched, registered(&driver, _, _));
- // The framework is expected to see the following offers in sequence:
- // 1. One containing a RAW disk resource before `CREATE_DISK`.
- // 2. One containing a MOUNT disk resource after `CREATE_DISK`.
- // 3. One containing a MOUNT disk resource after the agent recovers
- // from a failover.
- // 4. One containing a RAW disk resource after `DESTROY_DISK`.
- //
- // We set up the expectations for these offers as the test progresses.
- Future<vector<Offer>> rawDiskOffers;
- Future<vector<Offer>> volumeCreatedOffers;
- Future<vector<Offer>> slaveRecoveredOffers;
- Future<vector<Offer>> volumeDestroyedOffers;
-
- Sequence offers;
-
- // We use the following filter to filter offers that do not have
- // wanted resources for 365 days (the maximum).
+ // We use the following filter to filter offers that do not have wanted
+ // resources for 365 days (the maximum).
Filters declineFilters;
declineFilters.set_refuse_seconds(Days(365).secs());
- // Decline offers that contain only the agent's default resources.
+ // Decline unwanted offers. The master can send such offers before the
+ // resource provider receives profile updates.
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We are only interested in any storage pool or volume with a "test" profile.
- auto hasSourceType = [](
- const Resource& r,
- const Resource::DiskInfo::Source::Type& type) {
+ auto isStoragePool = [](const Resource& r, const string& profile) {
return r.has_disk() &&
r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ !r.disk().source().has_id() &&
r.disk().source().has_profile() &&
- r.disk().source().profile() == "test" &&
- r.disk().source().type() == type;
+ r.disk().source().profile() == profile;
};
+ Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&rawDiskOffers));
+ std::bind(isStoragePool, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.start();
- AWAIT_READY(rawDiskOffers);
- ASSERT_FALSE(rawDiskOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- Option<Resource> source;
+ Offer offer = offers->at(0);
- foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
- source = resource;
- break;
- }
- }
+ // Create a MOUNT disk with a pipelined `RESERVE` operation.
+ Resource raw = *Resources(offer.resources())
+ .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .begin();
- ASSERT_SOME(source);
+ Resource reserved = *Resources(raw)
+ .pushReservation(createDynamicReservationInfo(
+ framework.roles(0), framework.principal()))
+ .begin();
- // Create a volume.
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeCreatedOffers));
+ auto isMountDisk = [](const Resource& r, const string& profile) {
+ return r.has_disk() &&
+ r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ r.disk().source().has_id() &&
+ r.disk().source().has_profile() &&
+ r.disk().source().profile() == profile;
+ };
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isMountDisk, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
- {rawDiskOffers->at(0).id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
+ {offer.id()},
+ {RESERVE(reserved),
+ CREATE_DISK(reserved, Resource::DiskInfo::Source::MOUNT)});
- AWAIT_READY(volumeCreatedOffers);
- ASSERT_FALSE(volumeCreatedOffers->empty());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- Option<Resource> volume;
+ offer = offers->at(0);
- foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::MOUNT)) {
- volume = resource;
- break;
- }
- }
+ Resource created = *Resources(offer.resources())
+ .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .begin();
- ASSERT_SOME(volume);
- ASSERT_TRUE(volume->disk().source().has_vendor());
- EXPECT_EQ(TEST_CSI_VENDOR, volume->disk().source().vendor());
- ASSERT_TRUE(volume->disk().source().has_id());
- ASSERT_TRUE(volume->disk().source().has_metadata());
- ASSERT_TRUE(volume->disk().source().has_mount());
- ASSERT_TRUE(volume->disk().source().mount().has_root());
- EXPECT_FALSE(path::absolute(volume->disk().source().mount().root()));
+ ASSERT_TRUE(created.disk().source().has_metadata());
+ ASSERT_TRUE(created.disk().source().has_mount());
+ ASSERT_TRUE(created.disk().source().mount().has_root());
+ EXPECT_FALSE(path::absolute(created.disk().source().mount().root()));
// Check if the volume is actually created by the test CSI plugin.
Option<string> volumePath;
-
- foreach (const Label& label, volume->disk().source().metadata().labels()) {
+ foreach (const Label& label, created.disk().source().metadata().labels()) {
if (label.key() == "path") {
volumePath = label.value();
break;
@@ -1048,57 +1011,82 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskRecovery)
slave.get()->terminate();
+ // NOTE: We set up the resource provider with an extra storage pool, so that
+ // when the storage pool shows up, we know that the resource provider has
+ // finished reconciling storage pools and thus operations won't be dropped.
+ //
+ // To achieve this, we drop `SlaveRegisteredMessage`s other than the first one
+ // to avoid unexpected `UpdateSlaveMessage`s. Then, we also drop the first two
+ // `UpdateSlaveMessage`s (one sent after agent reregistration and one after
+ // resource provider reregistration) and wait for the third one, which should
+ // contain the extra storage pool. We let it fall through to trigger an offer
+ // allocation for the persistent volume.
+ //
+ // Since the extra storage pool is never used, we reject the offers if only
+ // the storage pool is presented.
+ //
+ // TODO(chhsiao): Remove this workaround once MESOS-9553 is done.
+ setupResourceProviderConfig(Gigabytes(5));
+
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&slaveRecoveredOffers));
+ std::bind(isStoragePool, lambda::_1, "test"))))
+ .WillRepeatedly(DeclineOffers(declineFilters));
- slave = StartSlave(detector.get(), slaveFlags);
- ASSERT_SOME(slave);
+ // NOTE: The order of these expectations is reversed because Google Mock will
+ // search the expectations in reverse order.
+ Future<UpdateSlaveMessage> updateSlaveMessage =
+ FUTURE_PROTOBUF(UpdateSlaveMessage(), _, _);
- AWAIT_READY(slaveRecoveredOffers);
- ASSERT_FALSE(slaveRecoveredOffers->empty());
+ DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
+ DROP_PROTOBUF(UpdateSlaveMessage(), _, _);
+ DROP_PROTOBUFS(SlaveReregisteredMessage(), _, _);
+ FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _);
- // Destroy the created volume.
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
- .InSequence(offers)
- .WillOnce(FutureArg<1>(&volumeDestroyedOffers));
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(created)))
+ .WillOnce(FutureArg<1>(&offers));
- Future<UpdateOperationStatusMessage> destroyDiskStatus =
- FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
+ slave = StartSlave(detector.get(), slaveFlags);
+ ASSERT_SOME(slave);
- driver.acceptOffers(
- {slaveRecoveredOffers->at(0).id()},
- {DESTROY_DISK(volume.get())},
- acceptFilters);
+ AWAIT_READY(updateSlaveMessage);
+ ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
+ ASSERT_FALSE(Resources(
+ updateSlaveMessage->resource_providers().providers(0).total_resources())
+ .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .empty());
- AWAIT_READY(destroyDiskStatus);
- EXPECT_EQ(OPERATION_FINISHED, destroyDiskStatus->status().state());
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
- // NOTE: We update the disk profile mapping after the `DESTROY_DISK` operation
- // is applied, otherwise it could be dropped due to reconciling storage pools.
- recoveredProfileMapping.set(
- http::OK(createDiskProfileMapping({{"test", None()}})));
+ offer = offers->at(0);
- AWAIT_READY(volumeDestroyedOffers);
- ASSERT_FALSE(volumeDestroyedOffers->empty());
+ // Destroy the MOUNT disk with pipelined `CREATE`, `DESTROY` and `UNRESERVE`
+ // operations. Note that `UNRESERVE` can come before `DESTROY_DISK`.
+ Resource persistentVolume = created;
+ persistentVolume.mutable_disk()->mutable_persistence()
+ ->set_id(id::UUID::random().toString());
+ persistentVolume.mutable_disk()->mutable_persistence()
+ ->set_principal(framework.principal());
+ persistentVolume.mutable_disk()->mutable_volume()
+ ->set_container_path("volume");
+ persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
- Option<Resource> destroyed;
+ Resource unreserved = *Resources(created)
+ .popReservation()
+ .begin();
- foreach (const Resource& resource, volumeDestroyedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
- destroyed = resource;
- break;
- }
- }
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveResource(raw)))
+ .WillOnce(FutureArg<1>(&offers));
- ASSERT_SOME(destroyed);
- ASSERT_TRUE(destroyed->disk().source().has_vendor());
- EXPECT_EQ(TEST_CSI_VENDOR, destroyed->disk().source().vendor());
- ASSERT_FALSE(destroyed->disk().source().has_id());
- ASSERT_FALSE(destroyed->disk().source().has_metadata());
- ASSERT_FALSE(destroyed->disk().source().has_mount());
+ driver.acceptOffers(
+ {offer.id()},
+ {CREATE(persistentVolume),
+ DESTROY(persistentVolume),
+ UNRESERVE(created),
+ DESTROY_DISK(unreserved)});
+
+ AWAIT_READY(offers);
// Check if the volume is actually deleted by the test CSI plugin.
EXPECT_FALSE(os::exists(volumePath.get()));
[mesos] 05/07: Extracted common offer matching functions from SLRP
tests.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 43bf73ca2f902356c3a5b92fb7cf535af3b7ccc4
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Tue Feb 5 21:23:23 2019 -0800
Extracted common offer matching functions from SLRP tests.
This patch extracts lambda functions `isStoragePool`, `isMountDisk` and
`isPreprovisionedVolume` from all tests and makes them common utitily
functions in the test fixture.
Review: https://reviews.apache.org/r/69904
---
.../storage_local_resource_provider_tests.cpp | 563 ++++++---------------
1 file changed, 142 insertions(+), 421 deletions(-)
diff --git a/src/tests/storage_local_resource_provider_tests.cpp b/src/tests/storage_local_resource_provider_tests.cpp
index e8b3f01..ce6eb70 100644
--- a/src/tests/storage_local_resource_provider_tests.cpp
+++ b/src/tests/storage_local_resource_provider_tests.cpp
@@ -347,6 +347,54 @@ private:
};
+// Tests whether a resource is a storage pool of a given profile. A storage pool
+// is a RAW disk resource with a profile but no source ID.
+template <typename Resource>
+static bool isStoragePool(const Resource& r, const string& profile)
+{
+ return r.has_disk() &&
+ r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ !r.disk().source().has_id() &&
+ r.disk().source().has_profile() &&
+ r.disk().source().profile() == profile;
+}
+
+
+// Tests whether a resource is a MOUNT disk of a given profile but not a
+// persistent volume. A MOUNT disk has both profile and source ID set.
+template <typename Resource>
+static bool isMountDisk(const Resource& r, const string& profile)
+{
+ return r.has_disk() &&
+ r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ r.disk().source().has_id() &&
+ r.disk().source().has_profile() &&
+ r.disk().source().profile() == profile &&
+ !r.disk().has_persistence();
+}
+
+
+// Tests whether a resource is a preprovisioned volume. A preprovisioned volume
+// is a RAW disk resource with a source ID but no profile.
+template <typename Resource>
+static bool isPreprovisionedVolume(const Resource& r)
+{
+ return r.has_disk() &&
+ r.disk().has_source() &&
+ r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
+ r.disk().source().has_vendor() &&
+ r.disk().source().vendor() == TEST_CSI_VENDOR &&
+ r.disk().source().has_id() &&
+ !r.disk().source().has_profile();
+}
+
+
// This test verifies that a storage local resource provider can report
// no resource and recover from this state.
TEST_F(StorageLocalResourceProviderTest, NoResource)
@@ -530,32 +578,11 @@ TEST_F(StorageLocalResourceProviderTest, DISABLED_SmallDisk)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We expect to receive an offer that contains a storage pool and a
- // pre-existing volume.
- auto isStoragePool = [](const Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile();
- };
-
- auto isPreExistingVolume = [](const Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- !r.disk().source().has_profile();
- };
-
// Since the resource provider always reports the pre-existing volume,
// but only reports the storage pool after it gets the profile, an
// offer containing the latter will also contain the former.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- isStoragePool)))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&rawDisksOffers));
driver.start();
@@ -566,9 +593,9 @@ TEST_F(StorageLocalResourceProviderTest, DISABLED_SmallDisk)
Option<Resource> storagePool;
Option<Resource> preExistingVolume;
foreach (const Resource& resource, rawDisksOffers->at(0).resources()) {
- if (isStoragePool(resource)) {
+ if (isStoragePool(resource, "test")) {
storagePool = resource;
- } else if (isPreExistingVolume(resource)) {
+ } else if (isPreprovisionedVolume(resource)) {
preExistingVolume = resource;
}
}
@@ -696,22 +723,10 @@ TEST_F(StorageLocalResourceProviderTest, ProfileAppeared)
AWAIT_READY(offersAfterProfileFound);
ASSERT_FALSE(offersAfterProfileFound->empty());
- // A storage pool is a RAW disk that has a profile but no ID.
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
// A new storage pool with profile "test" should show up now.
Resources storagePools =
Resources(offersAfterProfileFound->at(0).resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"));
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"));
EXPECT_FALSE(storagePools.empty());
}
@@ -766,20 +781,9 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDisk)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.start();
@@ -791,7 +795,7 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDisk)
// Create a MOUNT disk with a pipelined `RESERVE` operation.
Resource raw = *Resources(offer.resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.begin();
Resource reserved = *Resources(raw)
@@ -799,19 +803,8 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDisk)
framework.roles(0), framework.principal()))
.begin();
- auto isMountDisk = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
@@ -825,7 +818,7 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDisk)
offer = offers->at(0);
Resource created = *Resources(offer.resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
.begin();
ASSERT_TRUE(created.disk().source().has_metadata());
@@ -927,20 +920,9 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskWithRecovery)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.start();
@@ -952,7 +934,7 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskWithRecovery)
// Create a MOUNT disk with a pipelined `RESERVE` operation.
Resource raw = *Resources(offer.resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.begin();
Resource reserved = *Resources(raw)
@@ -960,19 +942,8 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskWithRecovery)
framework.roles(0), framework.principal()))
.begin();
- auto isMountDisk = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
@@ -986,7 +957,7 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskWithRecovery)
offer = offers->at(0);
Resource created = *Resources(offer.resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
.begin();
ASSERT_TRUE(created.disk().source().has_metadata());
@@ -1029,7 +1000,7 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskWithRecovery)
setupResourceProviderConfig(Gigabytes(5));
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillRepeatedly(DeclineOffers(declineFilters));
// NOTE: The order of these expectations is reversed because Google Mock will
@@ -1053,7 +1024,7 @@ TEST_F(StorageLocalResourceProviderTest, CreateDestroyDiskWithRecovery)
ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
ASSERT_FALSE(Resources(
updateSlaveMessage->resource_providers().providers(0).total_resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.empty());
AWAIT_READY(offers);
@@ -1187,22 +1158,11 @@ TEST_F(StorageLocalResourceProviderTest, ProfileDisappeared)
EXPECT_CALL(*scheduler, offers(_, _))
.WillRepeatedly(v1::scheduler::DeclineOffers());
- auto isStoragePool = [](const v1::Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == v1::Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
Future<v1::scheduler::Event::Offers> offers;
EXPECT_CALL(
*scheduler,
offers(_, v1::scheduler::OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test1"))))
+ std::bind(isStoragePool<v1::Resource>, lambda::_1, "test1"))))
.WillOnce(FutureArg<1>(&offers));
v1::scheduler::TestMesos mesos(
@@ -1228,13 +1188,13 @@ TEST_F(StorageLocalResourceProviderTest, ProfileDisappeared)
EXPECT_CALL(
*scheduler,
offers(_, v1::scheduler::OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test1"))))
+ std::bind(isStoragePool<v1::Resource>, lambda::_1, "test1"))))
.WillOnce(FutureArg<1>(&offers));
// Create a 2GB MOUNT disk of profile 'test1'.
{
v1::Resources raw = v1::Resources(offer.resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test1"));
+ .filter(std::bind(isStoragePool<v1::Resource>, lambda::_1, "test1"));
ASSERT_SOME_EQ(Gigabytes(4), raw.disk());
@@ -1275,21 +1235,10 @@ TEST_F(StorageLocalResourceProviderTest, ProfileDisappeared)
offer = offers->offers(0);
- auto isMountDisk = [](const v1::Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == v1::Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
EXPECT_CALL(
*scheduler,
offers(_, v1::scheduler::OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test1"))))
+ std::bind(isMountDisk<v1::Resource>, lambda::_1, "test1"))))
.WillOnce(FutureArg<1>(&offers));
// We drop the agent update (which is triggered by the changes in the known
@@ -1310,7 +1259,7 @@ TEST_F(StorageLocalResourceProviderTest, ProfileDisappeared)
// operation would be dropped due to a mismatched resource version.
{
v1::Resources raw = v1::Resources(offer.resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test1"));
+ .filter(std::bind(isStoragePool<v1::Resource>, lambda::_1, "test1"));
ASSERT_SOME_EQ(Gigabytes(2), raw.disk());
@@ -1349,14 +1298,14 @@ TEST_F(StorageLocalResourceProviderTest, ProfileDisappeared)
EXPECT_CALL(
*scheduler,
offers(_, v1::scheduler::OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test2"))))
+ std::bind(isStoragePool<v1::Resource>, lambda::_1, "test2"))))
.WillOnce(FutureArg<1>(&offers));
// Destroy the MOUNT disk of profile 'test1'. The returned converted resources
// should be empty.
{
v1::Resources created = v1::Resources(offer.resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test1"));
+ .filter(std::bind(isMountDisk<v1::Resource>, lambda::_1, "test1"));
ASSERT_SOME_EQ(Gigabytes(2), created.disk());
@@ -1392,7 +1341,7 @@ TEST_F(StorageLocalResourceProviderTest, ProfileDisappeared)
// Check that the freed disk shows up as of profile 'test2'.
{
v1::Resources raw = v1::Resources(offer.resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test2"));
+ .filter(std::bind(isStoragePool<v1::Resource>, lambda::_1, "test2"));
EXPECT_SOME_EQ(Gigabytes(4), raw.disk());
}
@@ -1572,20 +1521,8 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // Before the agent fails over, we are interested in the 'test1' storage pool.
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test1"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test1"))))
.WillOnce(FutureArg<1>(&rawDiskOffers));
driver.start();
@@ -1594,23 +1531,12 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
ASSERT_EQ(1u, rawDiskOffers->size());
Resource raw = *Resources(rawDiskOffers->at(0).resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test1"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test1"))
.begin();
- auto isMountDisk = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
// Create a MOUNT disk of profile 'test1'.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test1"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test1"))))
.WillOnce(FutureArg<1>(&diskCreatedOffers));
// We use the following filter so that the resources will not be
@@ -1627,7 +1553,7 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
ASSERT_EQ(1u, diskCreatedOffers->size());
Resource created = *Resources(diskCreatedOffers->at(0).resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test1"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test1"))
.begin();
ASSERT_TRUE(created.has_provider_id());
@@ -1677,25 +1603,13 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
slaveRegisteredMessage =
FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, Not(slave.get()->pid));
- // After the agent fails over, any volume created before becomes a
- // preprovisioned volume, which has an ID but no profile.
- auto isPreprovisionedVolume = [](const Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- !r.disk().source().has_profile();
- };
-
// NOTE: Instead of expecting a preprovisioned volume, we expect an offer with
// a 'test1' storage pool as an indication that the profile is known to the
// resource provider. The offer should also have the preprovisioned volume.
// But, an extra offer with the storage pool may be received as a side effect
// of this workaround, so we decline it if this happens.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test1"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test1"))))
.WillOnce(FutureArg<1>(&slaveRecoveredOffers))
.WillRepeatedly(DeclineOffers(declineFilters));
@@ -1708,7 +1622,7 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
ASSERT_EQ(1u, slaveRecoveredOffers->size());
Resources _preprovisioned = Resources(slaveRecoveredOffers->at(0).resources())
- .filter(isPreprovisionedVolume);
+ .filter(isPreprovisionedVolume<Resource>);
ASSERT_SOME_EQ(Gigabytes(2), _preprovisioned.disk());
@@ -1741,7 +1655,7 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
// Apply profile 'test1' to the preprovisioned volume, which will succeed.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test1"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test1"))))
.WillOnce(FutureArg<1>(&diskRecoveredOffers));
driver.acceptOffers(
@@ -1753,7 +1667,7 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
ASSERT_EQ(1u, diskRecoveredOffers->size());
Resource recovered = *Resources(diskRecoveredOffers->at(0).resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test1"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test1"))
.begin();
ASSERT_EQ(preprovisioned.provider_id(), recovered.provider_id());
@@ -1826,20 +1740,9 @@ TEST_F(
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.start();
@@ -1851,22 +1754,11 @@ TEST_F(
// Create a MOUNT disk.
Resource raw = *Resources(offer.resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.begin();
- auto isMountDisk = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
@@ -1878,7 +1770,7 @@ TEST_F(
offer = offers->at(0);
Resource created = *Resources(offer.resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
.begin();
ASSERT_TRUE(created.disk().source().has_metadata());
@@ -2043,20 +1935,9 @@ TEST_F(
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.start();
@@ -2068,23 +1949,11 @@ TEST_F(
// Create a MOUNT disk.
Resource raw = *Resources(offer.resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.begin();
- auto isMountDisk = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile &&
- !r.disk().has_persistence();
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
@@ -2096,7 +1965,7 @@ TEST_F(
offer = offers->at(0);
Resource created = *Resources(offer.resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
.begin();
// Create a persistent MOUNT volume then launch a task to write a file.
@@ -2192,7 +2061,7 @@ TEST_F(
setupResourceProviderConfig(Gigabytes(5));
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillRepeatedly(DeclineOffers(declineFilters));
// NOTE: The order of these expectations is reversed because Google Mock will
@@ -2216,7 +2085,7 @@ TEST_F(
ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
ASSERT_FALSE(Resources(
updateSlaveMessage->resource_providers().providers(0).total_resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.empty());
AWAIT_READY(offers);
@@ -2344,20 +2213,9 @@ TEST_F(
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
Future<vector<Offer>> offers;
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.start();
@@ -2369,23 +2227,11 @@ TEST_F(
// Create a MOUNT disk.
Resource raw = *Resources(offer.resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.begin();
- auto isMountDisk = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile &&
- !r.disk().has_persistence();
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.acceptOffers(
@@ -2397,7 +2243,7 @@ TEST_F(
offer = offers->at(0);
Resource created = *Resources(offer.resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
.begin();
// Create a persistent MOUNT volume then launch a task to write a file.
@@ -2536,7 +2382,7 @@ TEST_F(
setupResourceProviderConfig(Gigabytes(5));
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillRepeatedly(DeclineOffers(declineFilters));
// NOTE: The order of these expectations is reversed because Google Mock will
@@ -2561,7 +2407,7 @@ TEST_F(
ASSERT_EQ(1, updateSlaveMessage->resource_providers().providers_size());
ASSERT_FALSE(Resources(
updateSlaveMessage->resource_providers().providers(0).total_resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.empty());
AWAIT_READY(offers);
@@ -2704,19 +2550,8 @@ TEST_F(
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We are only interested in any storage pool or volume with a "test" profile.
- auto hasSourceType = [](
- const Resource& r,
- const Resource::DiskInfo::Source::Type& type) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == "test" &&
- r.disk().source().type() == type;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.InSequence(offers)
.WillOnce(FutureArg<1>(&rawDiskOffers));
@@ -2728,7 +2563,7 @@ TEST_F(
Option<Resource> source;
foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
+ if (isStoragePool(resource, "test")) {
source = resource;
break;
}
@@ -2758,7 +2593,7 @@ TEST_F(
// Create a volume.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.InSequence(offers)
.WillOnce(FutureArg<1>(&volumeCreatedOffers));
@@ -2778,7 +2613,7 @@ TEST_F(
Option<Resource> volume;
foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::MOUNT)) {
+ if (isMountDisk(resource, "test")) {
volume = resource;
break;
}
@@ -2957,24 +2792,13 @@ TEST_F(StorageLocalResourceProviderTest, ImportPreprovisionedVolume)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
// NOTE: Instead of expecting a preprovisioned volume, we expect an offer with
// a 'test1' storage pool as an indication that the profile is known to the
// resource provider. The offer should also have the preprovisioned volume.
// But, an extra offer with the storage pool may be received as a side effect
// of this workaround, so we decline it if this happens.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&rawDiskOffers))
.WillRepeatedly(DeclineOffers(declineFilters));
@@ -2983,18 +2807,8 @@ TEST_F(StorageLocalResourceProviderTest, ImportPreprovisionedVolume)
AWAIT_READY(rawDiskOffers);
ASSERT_EQ(1u, rawDiskOffers->size());
- auto isPreprovisionedVolume = [](const Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- !r.disk().source().has_profile();
- };
-
Resources _preprovisioned = Resources(rawDiskOffers->at(0).resources())
- .filter(isPreprovisionedVolume);
+ .filter(isPreprovisionedVolume<Resource>);
ASSERT_SOME_EQ(Gigabytes(2), _preprovisioned.disk());
@@ -3014,20 +2828,9 @@ TEST_F(StorageLocalResourceProviderTest, ImportPreprovisionedVolume)
ASSERT_SOME(volumePath);
ASSERT_TRUE(os::exists(volumePath.get()));
- auto isMountDisk = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
// Apply profile 'test' to the preprovisioned volume.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&diskCreatedOffers));
// We use the following filter so that the resources will not be
@@ -3044,12 +2847,12 @@ TEST_F(StorageLocalResourceProviderTest, ImportPreprovisionedVolume)
ASSERT_EQ(1u, diskCreatedOffers->size());
Resource created = *Resources(diskCreatedOffers->at(0).resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
.begin();
// Destroy the created disk.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&diskDestroyedOffers));
driver.acceptOffers(
@@ -3061,7 +2864,7 @@ TEST_F(StorageLocalResourceProviderTest, ImportPreprovisionedVolume)
ASSERT_EQ(1u, diskDestroyedOffers->size());
Resources raw = Resources(diskDestroyedOffers->at(0).resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"));
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"));
EXPECT_SOME_EQ(Gigabytes(4), raw.disk());
@@ -3150,14 +2953,8 @@ TEST_F(StorageLocalResourceProviderTest, RetryOperationStatusUpdate)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers());
- auto isRaw = [](const Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW;
- };
-
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(isRaw)))
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.start();
@@ -3177,7 +2974,7 @@ TEST_F(StorageLocalResourceProviderTest, RetryOperationStatusUpdate)
Option<Resource> source;
foreach (const Resource& resource, offer.resources()) {
- if (isRaw(resource)) {
+ if (isStoragePool(resource, "test")) {
source = resource;
break;
}
@@ -3308,14 +3105,8 @@ TEST_F(
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers());
- auto isRaw = [](const Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW;
- };
-
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(isRaw)))
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.start();
@@ -3335,7 +3126,7 @@ TEST_F(
Option<Resource> source;
foreach (const Resource& resource, offer.resources()) {
- if (isRaw(resource)) {
+ if (isStoragePool(resource, "test")) {
source = resource;
break;
}
@@ -3596,15 +3387,10 @@ TEST_F(StorageLocalResourceProviderTest, OperationUpdate)
Future<v1::scheduler::Event::Offers> offers;
- auto isRaw = [](const v1::Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().type() == v1::Resource::DiskInfo::Source::RAW;
- };
-
EXPECT_CALL(
- *scheduler, offers(_, v1::scheduler::OffersHaveAnyResource(isRaw)))
+ *scheduler,
+ offers(_, v1::scheduler::OffersHaveAnyResource(
+ std::bind(isStoragePool<v1::Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
v1::scheduler::TestMesos mesos(
@@ -3635,7 +3421,7 @@ TEST_F(StorageLocalResourceProviderTest, OperationUpdate)
Option<v1::Resource> source;
Option<mesos::v1::ResourceProviderID> resourceProviderId;
foreach (const v1::Resource& resource, offer.resources()) {
- if (isRaw(resource)) {
+ if (isStoragePool(resource, "test")) {
source = resource;
ASSERT_TRUE(resource.has_provider_id());
@@ -3747,19 +3533,8 @@ TEST_F(StorageLocalResourceProviderTest, OperationStateMetrics)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We are only interested in any storage pool or volume with a "test" profile.
- auto hasSourceType = [](
- const Resource& r,
- const Resource::DiskInfo::Source::Type& type) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == "test" &&
- r.disk().source().type() == type;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.InSequence(offers)
.WillOnce(FutureArg<1>(&rawDiskOffers));
@@ -3771,7 +3546,7 @@ TEST_F(StorageLocalResourceProviderTest, OperationStateMetrics)
Option<Resource> source;
foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
+ if (isStoragePool(resource, "test")) {
source = resource;
break;
}
@@ -3796,7 +3571,7 @@ TEST_F(StorageLocalResourceProviderTest, OperationStateMetrics)
// Create a volume.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.InSequence(offers)
.WillOnce(FutureArg<1>(&volumeCreatedOffers));
@@ -3816,7 +3591,7 @@ TEST_F(StorageLocalResourceProviderTest, OperationStateMetrics)
Option<Resource> volume;
foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::MOUNT)) {
+ if (isMountDisk(resource, "test")) {
volume = resource;
break;
}
@@ -3997,19 +3772,8 @@ TEST_F(StorageLocalResourceProviderTest, CsiPluginRpcMetrics)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- // We are only interested in any storage pool or volume with a "test" profile.
- auto hasSourceType = [](
- const Resource& r,
- const Resource::DiskInfo::Source::Type& type) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == "test" &&
- r.disk().source().type() == type;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::RAW))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.InSequence(offers)
.WillOnce(FutureArg<1>(&rawDiskOffers));
@@ -4021,7 +3785,7 @@ TEST_F(StorageLocalResourceProviderTest, CsiPluginRpcMetrics)
Option<Resource> source;
foreach (const Resource& resource, rawDiskOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::RAW)) {
+ if (isStoragePool(resource, "test")) {
source = resource;
break;
}
@@ -4074,7 +3838,7 @@ TEST_F(StorageLocalResourceProviderTest, CsiPluginRpcMetrics)
// Create a volume.
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(hasSourceType, lambda::_1, Resource::DiskInfo::Source::MOUNT))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.InSequence(offers)
.WillOnce(FutureArg<1>(&volumeCreatedOffers));
@@ -4094,7 +3858,7 @@ TEST_F(StorageLocalResourceProviderTest, CsiPluginRpcMetrics)
Option<Resource> volume;
foreach (const Resource& resource, volumeCreatedOffers->at(0).resources()) {
- if (hasSourceType(resource, Resource::DiskInfo::Source::MOUNT)) {
+ if (isMountDisk(resource, "test")) {
volume = resource;
break;
}
@@ -4306,16 +4070,10 @@ TEST_F(StorageLocalResourceProviderTest, ReconcileDroppedOperation)
EXPECT_CALL(sched, resourceOffers(&driver, _))
.WillRepeatedly(DeclineOffers(declineFilters));
- auto isRaw = [](const Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW;
- };
-
Future<vector<Offer>> offersBeforeOperations;
- EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(isRaw)))
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offersBeforeOperations))
.WillRepeatedly(DeclineOffers(declineFilters)); // Decline further offers.
@@ -4325,7 +4083,8 @@ TEST_F(StorageLocalResourceProviderTest, ReconcileDroppedOperation)
ASSERT_EQ(1u, offersBeforeOperations->size());
Resources raw =
- Resources(offersBeforeOperations->at(0).resources()).filter(isRaw);
+ Resources(offersBeforeOperations->at(0).resources())
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"));
// Create two MOUNT disks of 2GB each.
ASSERT_SOME_EQ(Gigabytes(4), raw.disk());
@@ -4398,15 +4157,8 @@ TEST_F(StorageLocalResourceProviderTest, ReconcileDroppedOperation)
Future<vector<Offer>> offersAfterOperations;
- auto isMountDisk = [](const Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT;
- };
-
- EXPECT_CALL(
- sched, resourceOffers(&driver, OffersHaveAnyResource(isMountDisk)))
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offersAfterOperations));
// Advance the clock to trigger a batch allocation.
@@ -4416,7 +4168,8 @@ TEST_F(StorageLocalResourceProviderTest, ReconcileDroppedOperation)
ASSERT_FALSE(offersAfterOperations->empty());
Resources converted =
- Resources(offersAfterOperations->at(0).resources()).filter(isMountDisk);
+ Resources(offersAfterOperations->at(0).resources())
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"));
ASSERT_EQ(1u, converted.size());
@@ -4508,15 +4261,10 @@ TEST_F(StorageLocalResourceProviderTest, RetryOperationStatusUpdateToScheduler)
Future<v1::scheduler::Event::Offers> offers;
- auto isRaw = [](const v1::Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().type() == v1::Resource::DiskInfo::Source::RAW;
- };
-
EXPECT_CALL(
- *scheduler, offers(_, v1::scheduler::OffersHaveAnyResource(isRaw)))
+ *scheduler,
+ offers(_, v1::scheduler::OffersHaveAnyResource(
+ std::bind(isStoragePool<v1::Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
v1::scheduler::TestMesos mesos(
@@ -4544,7 +4292,7 @@ TEST_F(StorageLocalResourceProviderTest, RetryOperationStatusUpdateToScheduler)
Option<v1::Resource> source;
Option<mesos::v1::ResourceProviderID> resourceProviderId;
foreach (const v1::Resource& resource, offer.resources()) {
- if (isRaw(resource)) {
+ if (isStoragePool(resource, "test")) {
source = resource;
ASSERT_TRUE(resource.has_provider_id());
@@ -4699,15 +4447,10 @@ TEST_F(
Future<v1::scheduler::Event::Offers> offers;
- auto isRaw = [](const v1::Resource& r) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().has_profile() &&
- r.disk().source().type() == v1::Resource::DiskInfo::Source::RAW;
- };
-
EXPECT_CALL(
- *scheduler, offers(_, v1::scheduler::OffersHaveAnyResource(isRaw)))
+ *scheduler,
+ offers(_, v1::scheduler::OffersHaveAnyResource(
+ std::bind(isStoragePool<v1::Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
v1::scheduler::TestMesos mesos(
@@ -4740,7 +4483,7 @@ TEST_F(
Option<v1::Resource> source;
Option<mesos::v1::ResourceProviderID> resourceProviderId;
foreach (const v1::Resource& resource, offer.resources()) {
- if (isRaw(resource)) {
+ if (isStoragePool(resource, "test")) {
source = resource;
ASSERT_TRUE(resource.has_provider_id());
@@ -4958,19 +4701,8 @@ TEST_F(StorageLocalResourceProviderTest, RetryRpcWithExponentialBackoff)
Future<vector<Offer>> offers;
- auto isStoragePool = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::RAW &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- !r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isStoragePool, lambda::_1, "test"))))
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
driver.start();
@@ -4979,7 +4711,7 @@ TEST_F(StorageLocalResourceProviderTest, RetryRpcWithExponentialBackoff)
ASSERT_EQ(1u, offers->size());
Resource raw = *Resources(offers->at(0).resources())
- .filter(std::bind(isStoragePool, lambda::_1, "test"))
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
.begin();
// Create a MOUNT disk.
@@ -5048,19 +4780,8 @@ TEST_F(StorageLocalResourceProviderTest, RetryRpcWithExponentialBackoff)
Clock::settle();
ASSERT_EQ(0u, createVolumeRequests.size());
- auto isMountDisk = [](const Resource& r, const string& profile) {
- return r.has_disk() &&
- r.disk().has_source() &&
- r.disk().source().type() == Resource::DiskInfo::Source::MOUNT &&
- r.disk().source().has_vendor() &&
- r.disk().source().vendor() == TEST_CSI_VENDOR &&
- r.disk().source().has_id() &&
- r.disk().source().has_profile() &&
- r.disk().source().profile() == profile;
- };
-
EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
- std::bind(isMountDisk, lambda::_1, "test"))))
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&offers));
// Return a successful response for the last `CreateVolume` call.
@@ -5081,7 +4802,7 @@ TEST_F(StorageLocalResourceProviderTest, RetryRpcWithExponentialBackoff)
ASSERT_EQ(1u, offers->size());
Resource created = *Resources(offers->at(0).resources())
- .filter(std::bind(isMountDisk, lambda::_1, "test"))
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
.begin();
// Destroy the MOUNT disk.
[mesos] 07/07: Removed unnecessary filters in SLRP tests.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 5a2df2cada9a19a6d08cbc4311afd2b7d1abba8f
Author: Chun-Hung Hsiao <ch...@apache.org>
AuthorDate: Wed Feb 27 11:39:54 2019 -0800
Removed unnecessary filters in SLRP tests.
Review: https://reviews.apache.org/r/69403/
---
.../storage_local_resource_provider_tests.cpp | 86 +++++-----------------
1 file changed, 18 insertions(+), 68 deletions(-)
diff --git a/src/tests/storage_local_resource_provider_tests.cpp b/src/tests/storage_local_resource_provider_tests.cpp
index e813c1d..7945384 100644
--- a/src/tests/storage_local_resource_provider_tests.cpp
+++ b/src/tests/storage_local_resource_provider_tests.cpp
@@ -1543,15 +1543,9 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
std::bind(isMountDisk<Resource>, lambda::_1, "test1"))))
.WillOnce(FutureArg<1>(&diskCreatedOffers));
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
-
driver.acceptOffers(
{rawDiskOffers->at(0).id()},
- {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
+ {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
AWAIT_READY(diskCreatedOffers);
ASSERT_EQ(1u, diskCreatedOffers->size());
@@ -1648,8 +1642,8 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
driver.acceptOffers(
{slaveRecoveredOffers->at(0).id()},
- {CREATE_DISK(preprovisioned, Resource::DiskInfo::Source::MOUNT, "test2")},
- acceptFilters);
+ {CREATE_DISK(
+ preprovisioned, Resource::DiskInfo::Source::MOUNT, "test2")});
AWAIT_READY(operationFailedStatus);
EXPECT_EQ(OPERATION_FAILED, operationFailedStatus->status().state());
@@ -1664,8 +1658,8 @@ TEST_F(StorageLocalResourceProviderTest, AgentRegisteredWithNewId)
driver.acceptOffers(
{operationFailedOffers->at(0).id()},
- {CREATE_DISK(preprovisioned, Resource::DiskInfo::Source::MOUNT, "test1")},
- acceptFilters);
+ {CREATE_DISK(
+ preprovisioned, Resource::DiskInfo::Source::MOUNT, "test1")});
AWAIT_READY(diskRecoveredOffers);
ASSERT_EQ(1u, diskRecoveredOffers->size());
@@ -2601,15 +2595,9 @@ TEST_F(
.InSequence(offers)
.WillOnce(FutureArg<1>(&volumeCreatedOffers));
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
-
driver.acceptOffers(
{rawDiskOffers->at(0).id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
+ {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
AWAIT_READY(volumeCreatedOffers);
ASSERT_FALSE(volumeCreatedOffers->empty());
@@ -2691,8 +2679,7 @@ TEST_F(
LAUNCH({createTask(
volumeCreatedOffers->at(0).slave_id(),
persistentVolume,
- createCommandInfo("test -f " + path::join("volume", "file")))})},
- acceptFilters);
+ createCommandInfo("test -f " + path::join("volume", "file")))})});
AWAIT_READY(taskStarting);
EXPECT_EQ(TASK_STARTING, taskStarting->state());
@@ -2724,9 +2711,7 @@ TEST_F(
driver.acceptOffers(
{taskFinishedOffers->at(0).id()},
- {DESTROY(persistentVolume),
- DESTROY_DISK(volume.get())},
- acceptFilters);
+ {DESTROY(persistentVolume), DESTROY_DISK(volume.get())});
AWAIT_READY(volumeDestroyedOffers);
ASSERT_FALSE(volumeDestroyedOffers->empty());
@@ -3008,15 +2993,9 @@ TEST_F(StorageLocalResourceProviderTest, ImportPreprovisionedVolume)
std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&diskCreatedOffers));
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
-
driver.acceptOffers(
{rawDiskOffers->at(0).id()},
- {CREATE_DISK(preprovisioned, Resource::DiskInfo::Source::MOUNT, "test")},
- acceptFilters);
+ {CREATE_DISK(preprovisioned, Resource::DiskInfo::Source::MOUNT, "test")});
AWAIT_READY(diskCreatedOffers);
ASSERT_EQ(1u, diskCreatedOffers->size());
@@ -3030,10 +3009,7 @@ TEST_F(StorageLocalResourceProviderTest, ImportPreprovisionedVolume)
std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
.WillOnce(FutureArg<1>(&diskDestroyedOffers));
- driver.acceptOffers(
- {diskCreatedOffers->at(0).id()},
- {DESTROY_DISK(created)},
- acceptFilters);
+ driver.acceptOffers({diskCreatedOffers->at(0).id()}, {DESTROY_DISK(created)});
AWAIT_READY(diskDestroyedOffers);
ASSERT_EQ(1u, diskDestroyedOffers->size());
@@ -3165,8 +3141,7 @@ TEST_F(StorageLocalResourceProviderTest, RetryOperationStatusUpdate)
// Create a volume.
driver.acceptOffers(
{offer.id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- {});
+ {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
AWAIT_READY(droppedUpdateOperationStatusMessage);
@@ -3317,8 +3292,7 @@ TEST_F(
// Create a volume.
driver.acceptOffers(
{offer.id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- {});
+ {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
AWAIT_READY(droppedUpdateOperationStatusMessage);
@@ -3750,15 +3724,9 @@ TEST_F(StorageLocalResourceProviderTest, OperationStateMetrics)
.InSequence(offers)
.WillOnce(FutureArg<1>(&volumeCreatedOffers));
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
-
driver.acceptOffers(
{rawDiskOffers->at(0).id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
+ {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
AWAIT_READY(volumeCreatedOffers);
ASSERT_FALSE(volumeCreatedOffers->empty());
@@ -3816,9 +3784,7 @@ TEST_F(StorageLocalResourceProviderTest, OperationStateMetrics)
.WillRepeatedly(DeclineOffers(declineFilters)); // Decline further offers.
driver.acceptOffers(
- {volumeCreatedOffers->at(0).id()},
- {DESTROY_DISK(volume.get())},
- acceptFilters);
+ {volumeCreatedOffers->at(0).id()}, {DESTROY_DISK(volume.get())});
AWAIT_READY(operationFailedOffers);
ASSERT_FALSE(operationFailedOffers->empty());
@@ -3843,9 +3809,7 @@ TEST_F(StorageLocalResourceProviderTest, OperationStateMetrics)
DROP_PROTOBUF(ApplyOperationMessage(), _, _);
driver.acceptOffers(
- {operationFailedOffers->at(0).id()},
- {DESTROY_DISK(volume.get())},
- acceptFilters);
+ {operationFailedOffers->at(0).id()}, {DESTROY_DISK(volume.get())});
AWAIT_READY(applyOperationMessage);
ASSERT_TRUE(applyOperationMessage
@@ -4017,15 +3981,9 @@ TEST_F(StorageLocalResourceProviderTest, CsiPluginRpcMetrics)
.InSequence(offers)
.WillOnce(FutureArg<1>(&volumeCreatedOffers));
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
-
driver.acceptOffers(
{rawDiskOffers->at(0).id()},
- {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
+ {CREATE_DISK(source.get(), Resource::DiskInfo::Source::MOUNT)});
AWAIT_READY(volumeCreatedOffers);
ASSERT_FALSE(volumeCreatedOffers->empty());
@@ -4111,9 +4069,7 @@ TEST_F(StorageLocalResourceProviderTest, CsiPluginRpcMetrics)
.WillRepeatedly(DeclineOffers(declineFilters)); // Decline further offers.
driver.acceptOffers(
- {volumeCreatedOffers->at(0).id()},
- {DESTROY_DISK(volume.get())},
- acceptFilters);
+ {volumeCreatedOffers->at(0).id()}, {DESTROY_DISK(volume.get())});
AWAIT_READY(operationFailedOffers);
ASSERT_FALSE(operationFailedOffers->empty());
@@ -4276,17 +4232,11 @@ TEST_F(StorageLocalResourceProviderTest, ReconcileDroppedOperation)
Future<UpdateOperationStatusMessage> operationFinishedStatus =
FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
- // We use the following filter so that the resources will not be
- // filtered for 5 seconds (the default).
- Filters acceptFilters;
- acceptFilters.set_refuse_seconds(0);
-
// Attempt the creation of two volumes.
driver.acceptOffers(
{offersBeforeOperations->at(0).id()},
{CREATE_DISK(source1, Resource::DiskInfo::Source::MOUNT),
- CREATE_DISK(source2, Resource::DiskInfo::Source::MOUNT)},
- acceptFilters);
+ CREATE_DISK(source2, Resource::DiskInfo::Source::MOUNT)});
// Ensure that the operations are processed.
Clock::settle();
[mesos] 06/07: Added a SLRP unit test for failed persistent volume
cleanup.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 069ac8ab9f298d11344accc964120103ba84ddbd
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Jan 31 16:19:43 2019 -0800
Added a SLRP unit test for failed persistent volume cleanup.
Test `ROOT_CreateDestroyPersistentMountVolumeFailed` verifies that if
SLRP fails to clean up a persistent volume, `DESTROY` would fail, the
persistent volume would remain, and depended operations will be dropped.
Review: https://reviews.apache.org/r/69905
---
.../storage_local_resource_provider_tests.cpp | 183 ++++++++++++++++++++-
1 file changed, 179 insertions(+), 4 deletions(-)
diff --git a/src/tests/storage_local_resource_provider_tests.cpp b/src/tests/storage_local_resource_provider_tests.cpp
index ce6eb70..e813c1d 100644
--- a/src/tests/storage_local_resource_provider_tests.cpp
+++ b/src/tests/storage_local_resource_provider_tests.cpp
@@ -55,6 +55,7 @@
#include "slave/containerizer/mesos/containerizer.hpp"
#include "tests/disk_profile_server.hpp"
+#include "tests/environment.hpp"
#include "tests/flags.hpp"
#include "tests/mesos.hpp"
#include "tests/mock_csi_plugin.hpp"
@@ -113,6 +114,11 @@ public:
path::join(sandbox.get(), "resource_provider_configs");
ASSERT_SOME(os::mkdir(resourceProviderConfigDir.get()));
+
+ Try<string> mkdtemp = environment->mkdtemp();
+ ASSERT_SOME(mkdtemp);
+
+ testCsiPluginWorkDir = mkdtemp.get();
}
void TearDown() override
@@ -218,9 +224,6 @@ public:
const string testCsiPluginPath =
path::join(tests::flags.build_dir, "src", "test-csi-plugin");
- const string testCsiPluginWorkDir = path::join(sandbox.get(), "storage");
- ASSERT_SOME(os::mkdir(testCsiPluginWorkDir));
-
Try<string> resourceProviderConfig = strings::format(
R"~(
{
@@ -282,7 +285,7 @@ public:
TEST_CSI_PLUGIN_NAME,
testCsiPluginPath,
testCsiPluginPath,
- testCsiPluginWorkDir,
+ testCsiPluginWorkDir.get(),
stringify(capacity),
createParameters.isSome()
? "--create_parameters=" + createParameters.get() : "",
@@ -344,6 +347,7 @@ private:
Modules modules;
vector<string> slaveWorkDirs;
Option<string> resourceProviderConfigDir;
+ Option<string> testCsiPluginWorkDir;
};
@@ -2732,6 +2736,177 @@ TEST_F(
}
+// This test verifies that if the storage local resource provider fails to clean
+// up a persistent volume, the volume will not be destroyed.
+//
+// To accomplish this:
+// 1. Creates a MOUNT disk from a RAW disk resource.
+// 2. Creates a persistent volume on the MOUNT disk then launches a task to
+// write a file into it.
+// 3. Bind-mounts a file in the CSI volume so the persistent volume cleanup
+// would fail with EBUSY.
+// 4. Destroys the persistent volume and the MOUNT disk. `DESTROY` would fail
+// and `DESTROY_DISK` would be dropped.
+TEST_F(
+ StorageLocalResourceProviderTest, ROOT_DestroyPersistentMountVolumeFailed)
+{
+ const string profilesPath = path::join(sandbox.get(), "profiles.json");
+
+ ASSERT_SOME(
+ os::write(profilesPath, createDiskProfileMapping({{"test", None()}})));
+
+ loadUriDiskProfileAdaptorModule(profilesPath);
+
+ setupResourceProviderConfig(Gigabytes(4));
+
+ Try<Owned<cluster::Master>> master = StartMaster();
+ ASSERT_SOME(master);
+
+ Owned<MasterDetector> detector = master.get()->createDetector();
+
+ slave::Flags slaveFlags = CreateSlaveFlags();
+ slaveFlags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
+
+ Future<SlaveRegisteredMessage> slaveRegisteredMessage =
+ FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
+
+ Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags);
+ ASSERT_SOME(slave);
+
+ AWAIT_READY(slaveRegisteredMessage);
+
+ // Register a framework to exercise operations.
+ FrameworkInfo framework = DEFAULT_FRAMEWORK_INFO;
+ framework.set_roles(0, "storage");
+
+ MockScheduler sched;
+ MesosSchedulerDriver driver(
+ &sched, framework, master.get()->pid, DEFAULT_CREDENTIAL);
+
+ EXPECT_CALL(sched, registered(&driver, _, _));
+
+ // We use the following filter to filter offers that do not have wanted
+ // resources for 365 days (the maximum).
+ Filters declineFilters;
+ declineFilters.set_refuse_seconds(Days(365).secs());
+
+ // Decline unwanted offers. The master can send such offers before the
+ // resource provider receives profile updates.
+ EXPECT_CALL(sched, resourceOffers(&driver, _))
+ .WillRepeatedly(DeclineOffers(declineFilters));
+
+ Future<vector<Offer>> offers;
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isStoragePool<Resource>, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.start();
+
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
+
+ Offer offer = offers->at(0);
+
+ // Create a MOUNT disk.
+ Resource raw = *Resources(offer.resources())
+ .filter(std::bind(isStoragePool<Resource>, lambda::_1, "test"))
+ .begin();
+
+ EXPECT_CALL(sched, resourceOffers(&driver, OffersHaveAnyResource(
+ std::bind(isMountDisk<Resource>, lambda::_1, "test"))))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.acceptOffers(
+ {offer.id()}, {CREATE_DISK(raw, Resource::DiskInfo::Source::MOUNT)});
+
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
+
+ offer = offers->at(0);
+
+ Resource created = *Resources(offer.resources())
+ .filter(std::bind(isMountDisk<Resource>, lambda::_1, "test"))
+ .begin();
+
+ // Create a persistent MOUNT volume then launch a task to write a file.
+ Resource persistentVolume = created;
+ persistentVolume.mutable_disk()->mutable_persistence()
+ ->set_id(id::UUID::random().toString());
+ persistentVolume.mutable_disk()->mutable_persistence()
+ ->set_principal(framework.principal());
+ persistentVolume.mutable_disk()->mutable_volume()
+ ->set_container_path("volume");
+ persistentVolume.mutable_disk()->mutable_volume()->set_mode(Volume::RW);
+
+ Future<Nothing> taskFinished;
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_STARTING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_RUNNING)));
+ EXPECT_CALL(sched, statusUpdate(&driver, TaskStatusStateEq(TASK_FINISHED)))
+ .WillOnce(FutureSatisfy(&taskFinished));
+
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.acceptOffers(
+ {offer.id()},
+ {CREATE(persistentVolume),
+ LAUNCH({createTask(
+ offer.slave_id(),
+ persistentVolume,
+ createCommandInfo("touch " + path::join("volume", "file")))})});
+
+ AWAIT_READY(taskFinished);
+
+ AWAIT_READY(offers);
+ ASSERT_EQ(1u, offers->size());
+
+ offer = offers->at(0);
+
+ // Bind-mount the file in the CSI volume to fail persistent volume cleanup.
+ Option<string> volumePath;
+ foreach (const Label& label, created.disk().source().metadata().labels()) {
+ if (label.key() == "path") {
+ volumePath = label.value();
+ break;
+ }
+ }
+
+ ASSERT_SOME(volumePath);
+
+ const string filePath = path::join(volumePath.get(), "file");
+ ASSERT_TRUE(os::exists(filePath));
+ ASSERT_SOME(fs::mount(filePath, filePath, None(), MS_BIND, None()));
+
+ // Destroy the persistent volume and the MOUNT disk, but none will succeed.
+ //
+ // NOTE: The order of the two `FUTURE_PROTOBUF`s is reversed because
+ // Google Mock will search the expectations in reverse order.
+ Future<UpdateOperationStatusMessage> destroyDiskOperationStatus =
+ FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
+ Future<UpdateOperationStatusMessage> destroyOperationStatus =
+ FUTURE_PROTOBUF(UpdateOperationStatusMessage(), _, _);
+
+ EXPECT_CALL(
+ sched, resourceOffers(&driver, OffersHaveResource(persistentVolume)))
+ .WillOnce(FutureArg<1>(&offers));
+
+ driver.acceptOffers(
+ {offer.id()}, {DESTROY(persistentVolume), DESTROY_DISK(created)});
+
+ AWAIT_READY(destroyOperationStatus);
+ EXPECT_EQ(OPERATION_FAILED, destroyOperationStatus->status().state());
+
+ AWAIT_READY(destroyDiskOperationStatus);
+ EXPECT_EQ(OPERATION_DROPPED, destroyDiskOperationStatus->status().state());
+
+ AWAIT_READY(offers);
+
+ // Check if the MOUNT disk still exists and not being cleaned up.
+ EXPECT_TRUE(os::exists(filePath));
+}
+
+
// This test verifies that the storage local resource provider can import a
// preprovisioned CSI volume as a MOUNT disk of a given profile, and return the
// space back to the storage pool after destroying the volume.