src/slave/containerizer/mesos/isolators/volume/csi/isolator.cpp - mesos - Git at Google

 // Licensed to the Apache Software Foundation (ASF) under one
 // or more contributor license agreements.  See the NOTICE file
 // distributed with this work for additional information
 // regarding copyright ownership.  The ASF licenses this file
 // to you under the Apache License, Version 2.0 (the
 // "License"); you may not use this file except in compliance
 // with the License.  You may obtain a copy of the License at
 //
 //     http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.

 #include <list>
 #include <string>
 #include <vector>

 #include <process/collect.hpp>
 #include <process/future.hpp>
 #include <process/owned.hpp>

 #include <stout/os.hpp>

 #include <stout/os/realpath.hpp>

 #include "common/protobuf_utils.hpp"

 #include "linux/fs.hpp"
 #include "linux/ns.hpp"

 #include "slave/state.hpp"

 #include "slave/containerizer/mesos/isolators/volume/csi/isolator.hpp"
 #include "slave/containerizer/mesos/isolators/volume/csi/paths.hpp"

 using std::list;
 using std::string;
 using std::vector;

 using process::Failure;
 using process::Future;
 using process::Owned;
 using process::PID;

 using mesos::slave::ContainerConfig;
 using mesos::slave::ContainerLaunchInfo;
 using mesos::slave::ContainerState;
 using mesos::slave::Isolator;

 namespace mesos {
 namespace internal {
 namespace slave {

 using AccessMode = Volume::Source::CSIVolume::VolumeCapability::AccessMode;


 Try<Isolator*> VolumeCSIIsolatorProcess::create(
     const Flags& flags,
     CSIServer* csiServer)
 {
   if (!strings::contains(flags.isolation, "filesystem/linux")) {
     return Error("'filesystem/linux' isolator must be used");
   }

   if (csiServer == nullptr) {
     return Error("No CSI server is provided");
   }

   const string csiRootDir = path::join(flags.work_dir, csi::paths::CSI_DIR);

   // Create the CSI volume information root directory if it does not exist.
   Try<Nothing> mkdir = os::mkdir(csiRootDir);
   if (mkdir.isError()) {
     return Error(
         "Failed to create CSI volume information root directory at '" +
         csiRootDir + "': " + mkdir.error());
   }

   Result<string> rootDir = os::realpath(csiRootDir);
   if (!rootDir.isSome()) {
     return Error(
         "Failed to determine canonical path of CSI volume information root"
         " directory '" + csiRootDir + "': " +
         (rootDir.isError() ? rootDir.error() : "No such file or directory"));
   }

   Owned<MesosIsolatorProcess> process(new VolumeCSIIsolatorProcess(
       flags,
       csiServer,
       rootDir.get()));

   return new MesosIsolator(process);
 }


 bool VolumeCSIIsolatorProcess::supportsNesting()
 {
   return true;
 }


 Future<Nothing> VolumeCSIIsolatorProcess::recover(
     const vector<ContainerState>& states,
     const hashset<ContainerID>& orphans)
 {
   foreach (const ContainerState& state, states) {
     const ContainerID& containerId = state.container_id();

     Try<Nothing> recover = recoverContainer(containerId);
     if (recover.isError()) {
       return Failure(
           "Failed to recover CSI volumes for container " +
           stringify(containerId) + ": " + recover.error());
     }
   }

   // Recover any orphan containers that we might have check pointed.
   // These orphan containers will be destroyed by the containerizer
   // through the regular cleanup path. See MESOS-2367 for details.
   foreach (const ContainerID& containerId, orphans) {
     Try<Nothing> recover = recoverContainer(containerId);
     if (recover.isError()) {
       return Failure(
           "Failed to recover CSI volumes for orphan container " +
           stringify(containerId) + ": " + recover.error());
     }
   }

   // Walk through all the checkpointed containers to determine if
   // there are any unknown orphan containers.
   Try<list<string>> entries = os::ls(rootDir);
   if (entries.isError()) {
     return Failure(
         "Unable to list CSI volume checkpoint directory '" +
         rootDir + "': " + entries.error());
   }

   foreach (const string& entry, entries.get()) {
     ContainerID containerId =
       protobuf::parseContainerId(Path(entry).basename());

     // Check if this container has already been recovered.
     if (infos.contains(containerId)) {
       continue;
     }

     // An unknown orphan container. Recover it and then clean it up.
     Try<Nothing> recover = recoverContainer(containerId);
     if (recover.isError()) {
       return Failure(
           "Failed to recover CSI volumes for orphan container " +
           stringify(containerId) + ": " + recover.error());
     }

     LOG(INFO) << "Cleaning up CSI volumes for unknown orphaned "
               << "container " << containerId;

     cleanup(containerId);
   }

   return Nothing();
 }


 Try<Nothing> VolumeCSIIsolatorProcess::recoverContainer(
     const ContainerID& containerId)
 {
   const string containerDir = csi::paths::getContainerDir(rootDir, containerId);
   if (!os::exists(containerDir)) {
     // This may occur in the following cases:
     //   1. The container has exited and the isolator has removed the
     //      container directory in '_cleanup()' but agent dies before
     //      noticing this.
     //   2. Agent dies before the isolator checkpoints CSI volumes for
     //      the container in 'prepare()'.
     // For the above cases, we do not need to do anything since there
     // is nothing to clean up for this container after agent restarts.
     return Nothing();
   }

   const string volumesPath = csi::paths::getVolumesPath(rootDir, containerId);
   if (!os::exists(volumesPath)) {
     // This may occur if agent dies after creating the container directory
     // but before it checkpoints anything in it.
     LOG(WARNING) << "The CSI volumes checkpoint file expected at '"
                  << volumesPath << "' for container " << containerId
                  << " does not exist";

     // Construct an info object with empty CSI volumes since no CSI volumes
     // are mounted yet for this container, and this container will be cleaned
     // up by containerizer (as known orphan container) or by `recover` (as
     // unknown orphan container).
     infos.put(containerId, Owned<Info>(new Info(hashset<CSIVolume>())));

     return Nothing();
   }

   Result<CSIVolumes> read = state::read<CSIVolumes>(volumesPath);
   if (read.isError()) {
     return Error(
         "Failed to read the CSI volumes checkpoint file '" +
         volumesPath + "': " + read.error());
   } else if (read.isNone()) {
     // This could happen if agent is hard rebooted after the checkpoint file is
     // created but before the data is synced on disk.
     LOG(WARNING) << "The CSI volumes checkpointed at '" << volumesPath
                  << "' for container " << containerId << " is empty";

     // Construct an info object with empty CSI volumes since no CSI volumes
     // are mounted yet for this container, and this container will be cleaned
     // up by containerizer (as known orphan container) or by `recover` (as
     // unknown orphan container).
     infos.put(containerId, Owned<Info>(new Info(hashset<CSIVolume>())));

     return Nothing();
   }

   hashset<CSIVolume> volumes;
   foreach (const CSIVolume& volume, read->volumes()) {
     VLOG(1) << "Recovering CSI volume with plugin '" << volume.plugin_name()
             << "' and ID '" << volume.id() << "' for container " << containerId;

     if (volumes.contains(volume)) {
       return Error(
           "Duplicate CSI volume with plugin '" + volume.plugin_name() +
           "' and ID '" + volume.id() + "'");
     }

     volumes.insert(volume);
   }

   infos.put(containerId, Owned<Info>(new Info(volumes)));

   return Nothing();
 }


 Future<Option<ContainerLaunchInfo>> VolumeCSIIsolatorProcess::prepare(
     const ContainerID& containerId,
     const ContainerConfig& containerConfig)
 {
   if (!containerConfig.has_container_info()) {
     return None();
   }

   if (containerConfig.container_info().type() != ContainerInfo::MESOS) {
     return Failure("Can only prepare CSI volumes for a MESOS container");
   }

   // The hashset is used to check if there are duplicated CSI volumes for the
   // same container.
   hashset<CSIVolume> volumeSet;

   // Represents the CSI volume mounts that we want to do for the container.
   vector<Mount> mounts;

   foreach (const Volume& _volume, containerConfig.container_info().volumes()) {
     if (!_volume.has_source() ||
         !_volume.source().has_type() ||
         _volume.source().type() != Volume::Source::CSI_VOLUME) {
       continue;
     }

     CHECK(_volume.source().has_csi_volume());
     CHECK(_volume.source().csi_volume().has_static_provisioning());

     const Volume::Source::CSIVolume& csiVolume = _volume.source().csi_volume();
     const string& pluginName = csiVolume.plugin_name();
     const string& volumeId = csiVolume.static_provisioning().volume_id();
     const AccessMode& accessMode =
       csiVolume.static_provisioning().volume_capability().access_mode();

     if ((accessMode.mode() == AccessMode::SINGLE_NODE_READER_ONLY ||
          accessMode.mode() == AccessMode::MULTI_NODE_READER_ONLY) &&
         _volume.mode() == Volume::RW) {
       return Failure(
           "Cannot use the read-only volume '" +
           volumeId + "' in read-write mode");
     }

     CSIVolume volume;
     volume.set_plugin_name(pluginName);
     volume.set_id(volumeId);

     if (volumeSet.contains(volume)) {
       return Failure(
           "Found duplicate CSI volume with plugin '" +
           pluginName + "' and volume ID '" + volumeId + "'");
     }

     // Determine the target of the mount.
     string target;

     // The logic to determine a volume mount target is identical to Linux
     // filesystem isolator, because this isolator has a dependency on that
     // isolator, and it assumes that if the container specifies a rootfs
     // the sandbox is already bind mounted into the container.
     if (path::is_absolute(_volume.container_path())) {
       // To specify a CSI volume for a container, frameworks should be allowed
       // to define the `container_path` either as an absolute path or a relative
       // path. Please see Linux filesystem isolator for details.
       if (containerConfig.has_rootfs()) {
         target = path::join(
             containerConfig.rootfs(),
             _volume.container_path());

         Try<Nothing> mkdir = os::mkdir(target);
         if (mkdir.isError()) {
           return Failure(
               "Failed to create the target of the mount at '" +
               target + "': " + mkdir.error());
         }
       } else {
         target = _volume.container_path();

         if (!os::exists(target)) {
           return Failure("Absolute container path '" + target + "' "
                          "does not exist");
         }
       }
     } else {
       if (containerConfig.has_rootfs()) {
         target = path::join(containerConfig.rootfs(),
                             flags.sandbox_directory,
                             _volume.container_path());
       } else {
         target = path::join(containerConfig.directory(),
                             _volume.container_path());
       }

       // NOTE: We cannot create the mount point at `target` if
       // container has rootfs defined. The bind mount of the sandbox
       // will hide what's inside `target`. So we should always create
       // the mount point in `directory`.
       string mountPoint = path::join(
           containerConfig.directory(),
           _volume.container_path());

       Try<Nothing> mkdir = os::mkdir(mountPoint);
       if (mkdir.isError()) {
         return Failure(
             "Failed to create the target of the mount at '" +
             mountPoint + "': " + mkdir.error());
       }
     }

     Mount mount;
     mount.volume = _volume;
     mount.csiVolume = volume;
     mount.target = target;

     mounts.push_back(mount);
     volumeSet.insert(volume);
   }

   if (volumeSet.empty()) {
     return None();
   }

   // Create the `CSIVolumes` protobuf message to checkpoint.
   CSIVolumes state;
   foreach (const CSIVolume& volume, volumeSet) {
     state.add_volumes()->CopyFrom(volume);
   }

   const string volumesPath = csi::paths::getVolumesPath(rootDir, containerId);
   Try<Nothing> checkpoint = state::checkpoint(volumesPath, state);
   if (checkpoint.isError()) {
     return Failure(
         "Failed to checkpoint CSI volumes at '" +
         volumesPath + "': " + checkpoint.error());
   }

   VLOG(1) << "Successfully created checkpoint at '" << volumesPath << "'";

   infos.put(containerId, Owned<Info>(new Info(volumeSet)));

   // Invoke CSI server to publish the volumes.
   vector<Future<string>> futures;
   futures.reserve(mounts.size());
   foreach (const Mount& mount, mounts) {
     futures.push_back(csiServer->publishVolume(mount.volume));
   }

   return await(futures)
     .then(defer(
         PID<VolumeCSIIsolatorProcess>(this),
         &VolumeCSIIsolatorProcess::_prepare,
         containerId,
         mounts,
         containerConfig.has_user()
           ? containerConfig.user()
           : Option<string>::none(),
         lambda::_1));
 }


 Future<Option<ContainerLaunchInfo>> VolumeCSIIsolatorProcess::_prepare(
     const ContainerID& containerId,
     const vector<Mount>& mounts,
     const Option<string>& user,
     const vector<Future<string>>& futures)
 {
   ContainerLaunchInfo launchInfo;
   launchInfo.add_clone_namespaces(CLONE_NEWNS);

   vector<string> messages;
   vector<string> sources;
   foreach (const Future<string>& future, futures) {
     if (!future.isReady()) {
       messages.push_back(future.isFailed() ? future.failure() : "discarded");
       continue;
     }

     sources.push_back(strings::trim(future.get()));
   }

   if (!messages.empty()) {
     return Failure(strings::join("\n", messages));
   }

   CHECK_EQ(sources.size(), mounts.size());

   for (size_t i = 0; i < sources.size(); i++) {
     const string& source = sources[i];
     const Mount& mount = mounts[i];

     if (user.isSome() && user.get() != "root") {
       bool isVolumeInUse = false;

       // Check if the volume is currently used by another container.
       foreachpair (const ContainerID& _containerId,
                    const Owned<Info>& info,
                    infos) {
         // Skip self.
         if (_containerId == containerId) {
           continue;
         }

         if (info->volumes.contains(mount.csiVolume)) {
           isVolumeInUse = true;
           break;
         }
       }

       if (!isVolumeInUse) {
         LOG(INFO) << "Changing the ownership of the CSI volume at '" << source
                   << "' to user '" << user.get() << "' for container "
                   << containerId;

         Try<Nothing> chown = os::chown(user.get(), source, false);
         if (chown.isError()) {
           return Failure(
               "Failed to set '" + user.get() + "' as the owner of the "
               "CSI volume at '" + source + "': " + chown.error());
         }
       } else {
         LOG(INFO) << "Leaving the ownership of the CSI volume at '"
                   << source << "' unchanged because it is in use";
       }
     }

     LOG(INFO) << "Mounting CSI volume mount point '" << source
               << "' to '" << mount.target << "' for container " << containerId;

     *launchInfo.add_mounts() = protobuf::slave::createContainerMount(
         source,
         mount.target,
         MS_BIND | MS_REC | (mount.volume.mode() == Volume::RO ? MS_RDONLY : 0));
   }

   return launchInfo;
 }


 Future<Nothing> VolumeCSIIsolatorProcess::cleanup(
     const ContainerID& containerId)
 {
   if (!infos.contains(containerId)) {
     VLOG(1) << "Ignoring cleanup request for unknown container " << containerId;
     return Nothing();
   }

   hashmap<CSIVolume, int> references;
   foreachvalue (const Owned<Info>& info, infos) {
     foreach (const CSIVolume& volume, info->volumes) {
       if (!references.contains(volume)) {
         references[volume] = 1;
       } else {
         references[volume]++;
       }
     }
   }

   vector<Future<Nothing>> futures;

   foreach (const CSIVolume& volume, infos[containerId]->volumes) {
     if (references.contains(volume) && references[volume] > 1) {
       VLOG(1) << "Cannot unpublish the volume with plugin '"
               << volume.plugin_name() << "' and ID '" << volume.id()
               << "' for container " << containerId
               << " since its reference count is " << references[volume];
       continue;
     }

     LOG(INFO) << "Unpublishing the volume with plugin '"
               << volume.plugin_name() << "' and ID '" << volume.id()
               << "' for container " << containerId;

     // Invoke CSI server to unpublish the volumes.
     futures.push_back(
         csiServer->unpublishVolume(volume.plugin_name(), volume.id()));
   }

   // Erase the `Info` struct of this container before unpublishing the volumes.
   // This is to ensure the reference count of the volume will not be wrongly
   // increased if unpublishing volumes fail, otherwise next time when another
   // container using the same volume is destroyed, we would NOT unpublish the
   // volume since its reference count would be larger than 1.
   infos.erase(containerId);

   return await(futures)
     .then(defer(
         PID<VolumeCSIIsolatorProcess>(this),
         &VolumeCSIIsolatorProcess::_cleanup,
         containerId,
         lambda::_1));
 }


 Future<Nothing> VolumeCSIIsolatorProcess::_cleanup(
     const ContainerID& containerId,
     const vector<Future<Nothing>>& futures)
 {
   vector<string> messages;
   foreach (const Future<Nothing>& future, futures) {
     if (!future.isReady()) {
       messages.push_back(future.isFailed() ? future.failure() : "discarded");
     }
   }

   if (!messages.empty()) {
     return Failure(strings::join("\n", messages));
   }

   const string containerDir = csi::paths::getContainerDir(rootDir, containerId);
   Try<Nothing> rmdir = os::rmdir(containerDir);
   if (rmdir.isError()) {
     return Failure(
         "Failed to remove the container directory at '" +
         containerDir + "': " + rmdir.error());
   }

   LOG(INFO) << "Removed the container directory at '" << containerDir
             << "' for container " << containerId;

   return Nothing();
 }

 } // namespace slave {
 } // namespace internal {
 } // namespace mesos {
	// Licensed to the Apache Software Foundation (ASF) under one
	// or more contributor license agreements. See the NOTICE file
	// distributed with this work for additional information
	// regarding copyright ownership. The ASF licenses this file
	// to you under the Apache License, Version 2.0 (the
	// "License"); you may not use this file except in compliance
	// with the License. You may obtain a copy of the License at
	//
	// http://www.apache.org/licenses/LICENSE-2.0
	//
	// Unless required by applicable law or agreed to in writing, software
	// distributed under the License is distributed on an "AS IS" BASIS,
	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	// See the License for the specific language governing permissions and
	// limitations under the License.

	#include <list>
	#include <string>
	#include <vector>

	#include <process/collect.hpp>
	#include <process/future.hpp>
	#include <process/owned.hpp>

	#include <stout/os.hpp>

	#include <stout/os/realpath.hpp>

	#include "common/protobuf_utils.hpp"

	#include "linux/fs.hpp"
	#include "linux/ns.hpp"

	#include "slave/state.hpp"

	#include "slave/containerizer/mesos/isolators/volume/csi/isolator.hpp"
	#include "slave/containerizer/mesos/isolators/volume/csi/paths.hpp"

	using std::list;
	using std::string;
	using std::vector;

	using process::Failure;
	using process::Future;
	using process::Owned;
	using process::PID;

	using mesos::slave::ContainerConfig;
	using mesos::slave::ContainerLaunchInfo;
	using mesos::slave::ContainerState;
	using mesos::slave::Isolator;

	namespace mesos {
	namespace internal {
	namespace slave {

	using AccessMode = Volume::Source::CSIVolume::VolumeCapability::AccessMode;


	Try<Isolator*> VolumeCSIIsolatorProcess::create(
	const Flags& flags,
	CSIServer* csiServer)
	{
	if (!strings::contains(flags.isolation, "filesystem/linux")) {
	return Error("'filesystem/linux' isolator must be used");
	}

	if (csiServer == nullptr) {
	return Error("No CSI server is provided");
	}

	const string csiRootDir = path::join(flags.work_dir, csi::paths::CSI_DIR);

	// Create the CSI volume information root directory if it does not exist.
	Try<Nothing> mkdir = os::mkdir(csiRootDir);
	if (mkdir.isError()) {
	return Error(
	"Failed to create CSI volume information root directory at '" +
	csiRootDir + "': " + mkdir.error());
	}

	Result<string> rootDir = os::realpath(csiRootDir);
	if (!rootDir.isSome()) {
	return Error(
	"Failed to determine canonical path of CSI volume information root"
	" directory '" + csiRootDir + "': " +
	(rootDir.isError() ? rootDir.error() : "No such file or directory"));
	}

	Owned<MesosIsolatorProcess> process(new VolumeCSIIsolatorProcess(
	flags,
	csiServer,
	rootDir.get()));

	return new MesosIsolator(process);
	}


	bool VolumeCSIIsolatorProcess::supportsNesting()
	{
	return true;
	}


	Future<Nothing> VolumeCSIIsolatorProcess::recover(
	const vector<ContainerState>& states,
	const hashset<ContainerID>& orphans)
	{
	foreach (const ContainerState& state, states) {
	const ContainerID& containerId = state.container_id();

	Try<Nothing> recover = recoverContainer(containerId);
	if (recover.isError()) {
	return Failure(
	"Failed to recover CSI volumes for container " +
	stringify(containerId) + ": " + recover.error());
	}
	}

	// Recover any orphan containers that we might have check pointed.
	// These orphan containers will be destroyed by the containerizer
	// through the regular cleanup path. See MESOS-2367 for details.
	foreach (const ContainerID& containerId, orphans) {
	Try<Nothing> recover = recoverContainer(containerId);
	if (recover.isError()) {
	return Failure(
	"Failed to recover CSI volumes for orphan container " +
	stringify(containerId) + ": " + recover.error());
	}
	}

	// Walk through all the checkpointed containers to determine if
	// there are any unknown orphan containers.
	Try<list<string>> entries = os::ls(rootDir);
	if (entries.isError()) {
	return Failure(
	"Unable to list CSI volume checkpoint directory '" +
	rootDir + "': " + entries.error());
	}

	foreach (const string& entry, entries.get()) {
	ContainerID containerId =
	protobuf::parseContainerId(Path(entry).basename());

	// Check if this container has already been recovered.
	if (infos.contains(containerId)) {
	continue;
	}

	// An unknown orphan container. Recover it and then clean it up.
	Try<Nothing> recover = recoverContainer(containerId);
	if (recover.isError()) {
	return Failure(
	"Failed to recover CSI volumes for orphan container " +
	stringify(containerId) + ": " + recover.error());
	}

	LOG(INFO) << "Cleaning up CSI volumes for unknown orphaned "
	<< "container " << containerId;

	cleanup(containerId);
	}

	return Nothing();
	}


	Try<Nothing> VolumeCSIIsolatorProcess::recoverContainer(
	const ContainerID& containerId)
	{
	const string containerDir = csi::paths::getContainerDir(rootDir, containerId);
	if (!os::exists(containerDir)) {
	// This may occur in the following cases:
	// 1. The container has exited and the isolator has removed the
	// container directory in '_cleanup()' but agent dies before
	// noticing this.
	// 2. Agent dies before the isolator checkpoints CSI volumes for
	// the container in 'prepare()'.
	// For the above cases, we do not need to do anything since there
	// is nothing to clean up for this container after agent restarts.
	return Nothing();
	}

	const string volumesPath = csi::paths::getVolumesPath(rootDir, containerId);
	if (!os::exists(volumesPath)) {
	// This may occur if agent dies after creating the container directory
	// but before it checkpoints anything in it.
	LOG(WARNING) << "The CSI volumes checkpoint file expected at '"
	<< volumesPath << "' for container " << containerId
	<< " does not exist";

	// Construct an info object with empty CSI volumes since no CSI volumes
	// are mounted yet for this container, and this container will be cleaned
	// up by containerizer (as known orphan container) or by `recover` (as
	// unknown orphan container).
	infos.put(containerId, Owned<Info>(new Info(hashset<CSIVolume>())));

	return Nothing();
	}

	Result<CSIVolumes> read = state::read<CSIVolumes>(volumesPath);
	if (read.isError()) {
	return Error(
	"Failed to read the CSI volumes checkpoint file '" +
	volumesPath + "': " + read.error());
	} else if (read.isNone()) {
	// This could happen if agent is hard rebooted after the checkpoint file is
	// created but before the data is synced on disk.
	LOG(WARNING) << "The CSI volumes checkpointed at '" << volumesPath
	<< "' for container " << containerId << " is empty";

	// Construct an info object with empty CSI volumes since no CSI volumes
	// are mounted yet for this container, and this container will be cleaned
	// up by containerizer (as known orphan container) or by `recover` (as
	// unknown orphan container).
	infos.put(containerId, Owned<Info>(new Info(hashset<CSIVolume>())));

	return Nothing();
	}

	hashset<CSIVolume> volumes;
	foreach (const CSIVolume& volume, read->volumes()) {
	VLOG(1) << "Recovering CSI volume with plugin '" << volume.plugin_name()
	<< "' and ID '" << volume.id() << "' for container " << containerId;

	if (volumes.contains(volume)) {
	return Error(
	"Duplicate CSI volume with plugin '" + volume.plugin_name() +
	"' and ID '" + volume.id() + "'");
	}

	volumes.insert(volume);
	}

	infos.put(containerId, Owned<Info>(new Info(volumes)));

	return Nothing();
	}


	Future<Option<ContainerLaunchInfo>> VolumeCSIIsolatorProcess::prepare(
	const ContainerID& containerId,
	const ContainerConfig& containerConfig)
	{
	if (!containerConfig.has_container_info()) {
	return None();
	}

	if (containerConfig.container_info().type() != ContainerInfo::MESOS) {
	return Failure("Can only prepare CSI volumes for a MESOS container");
	}

	// The hashset is used to check if there are duplicated CSI volumes for the
	// same container.
	hashset<CSIVolume> volumeSet;

	// Represents the CSI volume mounts that we want to do for the container.
	vector<Mount> mounts;

	foreach (const Volume& _volume, containerConfig.container_info().volumes()) {
	if (!_volume.has_source() \|\|
	!_volume.source().has_type() \|\|
	_volume.source().type() != Volume::Source::CSI_VOLUME) {
	continue;
	}

	CHECK(_volume.source().has_csi_volume());
	CHECK(_volume.source().csi_volume().has_static_provisioning());

	const Volume::Source::CSIVolume& csiVolume = _volume.source().csi_volume();
	const string& pluginName = csiVolume.plugin_name();
	const string& volumeId = csiVolume.static_provisioning().volume_id();
	const AccessMode& accessMode =
	csiVolume.static_provisioning().volume_capability().access_mode();

	if ((accessMode.mode() == AccessMode::SINGLE_NODE_READER_ONLY \|\|
	accessMode.mode() == AccessMode::MULTI_NODE_READER_ONLY) &&
	_volume.mode() == Volume::RW) {
	return Failure(
	"Cannot use the read-only volume '" +
	volumeId + "' in read-write mode");
	}

	CSIVolume volume;
	volume.set_plugin_name(pluginName);
	volume.set_id(volumeId);

	if (volumeSet.contains(volume)) {
	return Failure(
	"Found duplicate CSI volume with plugin '" +
	pluginName + "' and volume ID '" + volumeId + "'");
	}

	// Determine the target of the mount.
	string target;

	// The logic to determine a volume mount target is identical to Linux
	// filesystem isolator, because this isolator has a dependency on that
	// isolator, and it assumes that if the container specifies a rootfs
	// the sandbox is already bind mounted into the container.
	if (path::is_absolute(_volume.container_path())) {
	// To specify a CSI volume for a container, frameworks should be allowed
	// to define the `container_path` either as an absolute path or a relative
	// path. Please see Linux filesystem isolator for details.
	if (containerConfig.has_rootfs()) {
	target = path::join(
	containerConfig.rootfs(),
	_volume.container_path());

	Try<Nothing> mkdir = os::mkdir(target);
	if (mkdir.isError()) {
	return Failure(
	"Failed to create the target of the mount at '" +
	target + "': " + mkdir.error());
	}
	} else {
	target = _volume.container_path();

	if (!os::exists(target)) {
	return Failure("Absolute container path '" + target + "' "
	"does not exist");
	}
	}
	} else {
	if (containerConfig.has_rootfs()) {
	target = path::join(containerConfig.rootfs(),
	flags.sandbox_directory,
	_volume.container_path());
	} else {
	target = path::join(containerConfig.directory(),
	_volume.container_path());
	}

	// NOTE: We cannot create the mount point at `target` if
	// container has rootfs defined. The bind mount of the sandbox
	// will hide what's inside `target`. So we should always create
	// the mount point in `directory`.
	string mountPoint = path::join(
	containerConfig.directory(),
	_volume.container_path());

	Try<Nothing> mkdir = os::mkdir(mountPoint);
	if (mkdir.isError()) {
	return Failure(
	"Failed to create the target of the mount at '" +
	mountPoint + "': " + mkdir.error());
	}
	}

	Mount mount;
	mount.volume = _volume;
	mount.csiVolume = volume;
	mount.target = target;

	mounts.push_back(mount);
	volumeSet.insert(volume);
	}

	if (volumeSet.empty()) {
	return None();
	}

	// Create the `CSIVolumes` protobuf message to checkpoint.
	CSIVolumes state;
	foreach (const CSIVolume& volume, volumeSet) {
	state.add_volumes()->CopyFrom(volume);
	}

	const string volumesPath = csi::paths::getVolumesPath(rootDir, containerId);
	Try<Nothing> checkpoint = state::checkpoint(volumesPath, state);
	if (checkpoint.isError()) {
	return Failure(
	"Failed to checkpoint CSI volumes at '" +
	volumesPath + "': " + checkpoint.error());
	}

	VLOG(1) << "Successfully created checkpoint at '" << volumesPath << "'";

	infos.put(containerId, Owned<Info>(new Info(volumeSet)));

	// Invoke CSI server to publish the volumes.
	vector<Future<string>> futures;
	futures.reserve(mounts.size());
	foreach (const Mount& mount, mounts) {
	futures.push_back(csiServer->publishVolume(mount.volume));
	}

	return await(futures)
	.then(defer(
	PID<VolumeCSIIsolatorProcess>(this),
	&VolumeCSIIsolatorProcess::_prepare,
	containerId,
	mounts,
	containerConfig.has_user()
	? containerConfig.user()
	: Option<string>::none(),
	lambda::_1));
	}


	Future<Option<ContainerLaunchInfo>> VolumeCSIIsolatorProcess::_prepare(
	const ContainerID& containerId,
	const vector<Mount>& mounts,
	const Option<string>& user,
	const vector<Future<string>>& futures)
	{
	ContainerLaunchInfo launchInfo;
	launchInfo.add_clone_namespaces(CLONE_NEWNS);

	vector<string> messages;
	vector<string> sources;
	foreach (const Future<string>& future, futures) {
	if (!future.isReady()) {
	messages.push_back(future.isFailed() ? future.failure() : "discarded");
	continue;
	}

	sources.push_back(strings::trim(future.get()));
	}

	if (!messages.empty()) {
	return Failure(strings::join("\n", messages));
	}

	CHECK_EQ(sources.size(), mounts.size());

	for (size_t i = 0; i < sources.size(); i++) {
	const string& source = sources[i];
	const Mount& mount = mounts[i];

	if (user.isSome() && user.get() != "root") {
	bool isVolumeInUse = false;

	// Check if the volume is currently used by another container.
	foreachpair (const ContainerID& _containerId,
	const Owned<Info>& info,
	infos) {
	// Skip self.
	if (_containerId == containerId) {
	continue;
	}

	if (info->volumes.contains(mount.csiVolume)) {
	isVolumeInUse = true;
	break;
	}
	}

	if (!isVolumeInUse) {
	LOG(INFO) << "Changing the ownership of the CSI volume at '" << source
	<< "' to user '" << user.get() << "' for container "
	<< containerId;

	Try<Nothing> chown = os::chown(user.get(), source, false);
	if (chown.isError()) {
	return Failure(
	"Failed to set '" + user.get() + "' as the owner of the "
	"CSI volume at '" + source + "': " + chown.error());
	}
	} else {
	LOG(INFO) << "Leaving the ownership of the CSI volume at '"
	<< source << "' unchanged because it is in use";
	}
	}

	LOG(INFO) << "Mounting CSI volume mount point '" << source
	<< "' to '" << mount.target << "' for container " << containerId;

	*launchInfo.add_mounts() = protobuf::slave::createContainerMount(
	source,
	mount.target,
	MS_BIND \| MS_REC \| (mount.volume.mode() == Volume::RO ? MS_RDONLY : 0));
	}

	return launchInfo;
	}


	Future<Nothing> VolumeCSIIsolatorProcess::cleanup(
	const ContainerID& containerId)
	{
	if (!infos.contains(containerId)) {
	VLOG(1) << "Ignoring cleanup request for unknown container " << containerId;
	return Nothing();
	}

	hashmap<CSIVolume, int> references;
	foreachvalue (const Owned<Info>& info, infos) {
	foreach (const CSIVolume& volume, info->volumes) {
	if (!references.contains(volume)) {
	references[volume] = 1;
	} else {
	references[volume]++;
	}
	}
	}

	vector<Future<Nothing>> futures;

	foreach (const CSIVolume& volume, infos[containerId]->volumes) {
	if (references.contains(volume) && references[volume] > 1) {
	VLOG(1) << "Cannot unpublish the volume with plugin '"
	<< volume.plugin_name() << "' and ID '" << volume.id()
	<< "' for container " << containerId
	<< " since its reference count is " << references[volume];
	continue;
	}

	LOG(INFO) << "Unpublishing the volume with plugin '"
	<< volume.plugin_name() << "' and ID '" << volume.id()
	<< "' for container " << containerId;

	// Invoke CSI server to unpublish the volumes.
	futures.push_back(
	csiServer->unpublishVolume(volume.plugin_name(), volume.id()));
	}

	// Erase the `Info` struct of this container before unpublishing the volumes.
	// This is to ensure the reference count of the volume will not be wrongly
	// increased if unpublishing volumes fail, otherwise next time when another
	// container using the same volume is destroyed, we would NOT unpublish the
	// volume since its reference count would be larger than 1.
	infos.erase(containerId);

	return await(futures)
	.then(defer(
	PID<VolumeCSIIsolatorProcess>(this),
	&VolumeCSIIsolatorProcess::_cleanup,
	containerId,
	lambda::_1));
	}


	Future<Nothing> VolumeCSIIsolatorProcess::_cleanup(
	const ContainerID& containerId,
	const vector<Future<Nothing>>& futures)
	{
	vector<string> messages;
	foreach (const Future<Nothing>& future, futures) {
	if (!future.isReady()) {
	messages.push_back(future.isFailed() ? future.failure() : "discarded");
	}
	}

	if (!messages.empty()) {
	return Failure(strings::join("\n", messages));
	}

	const string containerDir = csi::paths::getContainerDir(rootDir, containerId);
	Try<Nothing> rmdir = os::rmdir(containerDir);
	if (rmdir.isError()) {
	return Failure(
	"Failed to remove the container directory at '" +
	containerDir + "': " + rmdir.error());
	}

	LOG(INFO) << "Removed the container directory at '" << containerDir
	<< "' for container " << containerId;

	return Nothing();
	}

	} // namespace slave {
	} // namespace internal {
	} // namespace mesos {