[mesos] 03/05: Cleaned up persistent volumes on SLRP disks.

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

chhsiao pushed a commit to branch 1.7.x
in repository https://gitbox.apache.org/repos/asf/mesos.git

commit d6c9c0377c70d47ed5ed941fbaa54bfab0a1fd8c
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
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 a0b42a2..7b2f8b9 100644
--- a/src/resource_provider/storage/provider.cpp
+++ b/src/resource_provider/storage/provider.cpp
@@ -2881,16 +2881,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());
 
@@ -3151,6 +3163,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());
@@ -3211,6 +3224,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(