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 0be5d4ffbefc8f2a7c71f76c2e6bd4785df2ed3c Author: Chun-Hung Hsiao <[email protected]> AuthorDate: Thu Jan 31 14:36:11 2019 -0800 Cleaned up persistent volumes on SLRP disks. This patch limits SLRP to only support persistent volumes on MOUNT disks, and makes it clean up data in persistent volumes when processing `DESTROY` operations. NOTE: Persistent volumes backed by CSI disks that are created before upgrading to a Mesos version that does not include this fix are subject to data leakage. To ensure data security, these persistent volume must be consumed by a task at least once after the upgrade before being destroyed. Review: https://reviews.apache.org/r/69893 --- src/resource_provider/storage/provider.cpp | 98 ++++++++++++++++++++-- src/resource_provider/storage/provider_process.hpp | 9 ++ 2 files changed, 101 insertions(+), 6 deletions(-) diff --git a/src/resource_provider/storage/provider.cpp b/src/resource_provider/storage/provider.cpp index a2eefaf..7a1b1d9 100644 --- a/src/resource_provider/storage/provider.cpp +++ b/src/resource_provider/storage/provider.cpp @@ -2885,16 +2885,28 @@ Future<Nothing> StorageLocalResourceProviderProcess::_applyOperation( switch (operation.info().type()) { case Offer::Operation::RESERVE: - case Offer::Operation::UNRESERVE: - case Offer::Operation::CREATE: - case Offer::Operation::DESTROY: { - // Synchronously apply the speculative operations to ensure that - // its result is reflected in the total resources before any of - // its succeeding operations is applied. + case Offer::Operation::UNRESERVE: { + // Synchronously apply the speculative operations to ensure that its + // result is reflected in the total resources before any of its succeeding + // operations is applied. return updateOperationStatus( operationUuid, getResourceConversions(operation.info())); } + case Offer::Operation::CREATE: { + // Synchronously create the persistent volumes to ensure that its result + // is reflected in the total resources before any of its succeeding + // operations is applied. + return updateOperationStatus( + operationUuid, applyCreate(operation.info())); + } + case Offer::Operation::DESTROY: { + // Synchronously clean up and destroy the persistent volumes to ensure + // that its result is reflected in the total resources before any of its + // succeeding operations is applied. + return updateOperationStatus( + operationUuid, applyDestroy(operation.info())); + } case Offer::Operation::CREATE_DISK: { CHECK(operation.info().has_create_disk()); @@ -3155,6 +3167,7 @@ Future<vector<ResourceConversion>> StorageLocalResourceProviderProcess::applyDestroyDisk( const Resource& resource) { + CHECK(!Resources::isPersistentVolume(resource)); CHECK(resource.disk().source().type() == Resource::DiskInfo::Source::MOUNT || resource.disk().source().type() == Resource::DiskInfo::Source::BLOCK); CHECK(resource.disk().source().has_id()); @@ -3215,6 +3228,79 @@ StorageLocalResourceProviderProcess::applyDestroyDisk( } +Try<vector<ResourceConversion>> +StorageLocalResourceProviderProcess::applyCreate( + const Offer::Operation& operation) const +{ + CHECK(operation.has_create()); + + foreach (const Resource& resource, operation.create().volumes()) { + CHECK(Resources::isPersistentVolume(resource)); + + // TODO(chhsiao): Support persistent BLOCK volumes. + if (resource.disk().source().type() != Resource::DiskInfo::Source::MOUNT) { + return Error( + "Cannot create persistent volume '" + + stringify(resource.disk().persistence().id()) + "' on a " + + stringify(resource.disk().source().type()) + " disk"); + } + } + + return getResourceConversions(operation); +} + + +Try<vector<ResourceConversion>> +StorageLocalResourceProviderProcess::applyDestroy( + const Offer::Operation& operation) const +{ + CHECK(operation.has_destroy()); + + foreach (const Resource& resource, operation.destroy().volumes()) { + // TODO(chhsiao): Support cleaning up persistent BLOCK volumes, presumably + // with `dd` or any other utility to zero out the block device. + CHECK(Resources::isPersistentVolume(resource)); + CHECK(resource.disk().source().type() == Resource::DiskInfo::Source::MOUNT); + CHECK(resource.disk().source().has_id()); + + const string& volumeId = resource.disk().source().id(); + CHECK(volumes.contains(volumeId)); + + const VolumeState& volumeState = volumes.at(volumeId).state; + + // NOTE: Data can only be written to the persistent volume when when it is + // in `PUBLISHED` state (i.e., mounted). Once a volume has been transitioned + // to `PUBLISHED`, we will set the `node_publish_required` field and always + // recover it back to `PUBLISHED` after a failover, until a `DESTROY_DISK` + // is applied, which only comes after `DESTROY`. So we only need to clean up + // the volume if it has the field set. + if (!volumeState.node_publish_required()) { + continue; + } + + CHECK_EQ(VolumeState::PUBLISHED, volumeState.state()); + + const string targetPath = csi::paths::getMountTargetPath( + csi::paths::getMountRootDir( + slave::paths::getCsiRootDir(workDir), + info.storage().plugin().type(), + info.storage().plugin().name()), + volumeId); + + // Only the data in the target path, but not itself, should be removed. + Try<Nothing> rmdir = os::rmdir(targetPath, true, false); + if (rmdir.isError()) { + return Error( + "Failed to remove persistent volume '" + + stringify(resource.disk().persistence().id()) + "' at '" + + targetPath + "': " + rmdir.error()); + } + } + + return getResourceConversions(operation); +} + + Try<Nothing> StorageLocalResourceProviderProcess::updateOperationStatus( const id::UUID& operationUuid, const Try<vector<ResourceConversion>>& conversions) diff --git a/src/resource_provider/storage/provider_process.hpp b/src/resource_provider/storage/provider_process.hpp index adc4651..2ba24ae 100644 --- a/src/resource_provider/storage/provider_process.hpp +++ b/src/resource_provider/storage/provider_process.hpp @@ -308,9 +308,18 @@ private: const id::UUID& operationUuid, const Resource::DiskInfo::Source::Type& targetType, const Option<std::string>& targetProfile); + process::Future<std::vector<ResourceConversion>> applyDestroyDisk( const Resource& resource); + // Synchronously creates persistent volumes. + Try<std::vector<ResourceConversion>> applyCreate( + const Offer::Operation& operation) const; + + // Synchronously cleans up and destroys persistent volumes. + Try<std::vector<ResourceConversion>> applyDestroy( + const Offer::Operation& operation) const; + // Synchronously updates `totalResources` and the operation status and // then asks the status update manager to send status updates. Try<Nothing> updateOperationStatus(
