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/04/09 06:13:22 UTC
[mesos] branch 1.8.x updated (2c7cee0 -> 2e86f64)
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a change to branch 1.8.x
in repository https://gitbox.apache.org/repos/asf/mesos.git.
from 2c7cee0 Updated CHANGELOG for 1.8.0.
new b4cc665 Added helpers to evolve/devolve repeated CSI v0 `VolumeCapability`s.
new e22d602 Added capabilities and evolve/devolve helpers for CSI v1.
new 275059a Updated CSI volume state fields to match names in v1 spec.
new a3c2632 Implemented CSI v1 volume manager.
new eca223c Refactored the test CSI plugin to make it easy to support CSI v1.
new 2e86f64 Supported CSI v1 in the test CSI plugin.
The 6 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:
include/mesos/csi/v0.hpp | 2 +
include/mesos/csi/v1.hpp | 2 +
src/CMakeLists.txt | 2 +
src/Makefile.am | 5 +
src/csi/state.proto | 13 +-
src/csi/v0_utils.cpp | 42 +-
src/csi/v0_utils.hpp | 27 +-
src/csi/v0_volume_manager.cpp | 18 +-
src/csi/v0_volume_manager_process.hpp | 6 +-
src/csi/{v0_utils.cpp => v1_utils.cpp} | 48 +-
src/csi/v1_utils.hpp | 218 +++
...v0_volume_manager.cpp => v1_volume_manager.cpp} | 128 +-
...v0_volume_manager.hpp => v1_volume_manager.hpp} | 10 +-
...r_process.hpp => v1_volume_manager_process.hpp} | 16 +-
src/examples/test_csi_plugin.cpp | 1540 +++++++++++++++-----
src/tests/csi_utils_tests.cpp | 50 +-
16 files changed, 1648 insertions(+), 479 deletions(-)
copy src/csi/{v0_utils.cpp => v1_utils.cpp} (85%)
create mode 100644 src/csi/v1_utils.hpp
copy src/csi/{v0_volume_manager.cpp => v1_volume_manager.cpp} (90%)
copy src/csi/{v0_volume_manager.hpp => v1_volume_manager.hpp} (95%)
copy src/csi/{v0_volume_manager_process.hpp => v1_volume_manager_process.hpp} (96%)
[mesos] 06/06: Supported CSI v1 in the test CSI plugin.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch 1.8.x
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 2e86f64b9ec9c0f4e2892b4c2d0f152f821d72d9
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Apr 4 21:36:12 2019 -0700
Supported CSI v1 in the test CSI plugin.
A new `--api_version` optional flag is added to the test CSI plugin. By
default, it will serve both CSI v0 and v1 requests. When this flag is
specified, only one CSI version will be served.
Review: https://reviews.apache.org/r/70404
---
include/mesos/csi/v0.hpp | 2 +
include/mesos/csi/v1.hpp | 2 +
src/examples/test_csi_plugin.cpp | 543 +++++++++++++++++++++++++++++++++++++--
3 files changed, 531 insertions(+), 16 deletions(-)
diff --git a/include/mesos/csi/v0.hpp b/include/mesos/csi/v0.hpp
index 8be070b..eb303ff 100644
--- a/include/mesos/csi/v0.hpp
+++ b/include/mesos/csi/v0.hpp
@@ -39,6 +39,8 @@ namespace v0 {
using namespace ::csi::v0;
+constexpr char API_VERSION[] = "v0";
+
} // namespace v0 {
} // namespace csi {
} // namespace mesos {
diff --git a/include/mesos/csi/v1.hpp b/include/mesos/csi/v1.hpp
index d4ebe42..9209111 100644
--- a/include/mesos/csi/v1.hpp
+++ b/include/mesos/csi/v1.hpp
@@ -39,6 +39,8 @@ namespace v1 {
using namespace ::csi::v1;
+constexpr char API_VERSION[] = "v1";
+
} // namespace v1 {
} // namespace csi {
} // namespace mesos {
diff --git a/src/examples/test_csi_plugin.cpp b/src/examples/test_csi_plugin.cpp
index 54753d9..b54d666 100644
--- a/src/examples/test_csi_plugin.cpp
+++ b/src/examples/test_csi_plugin.cpp
@@ -33,10 +33,12 @@
#include <mesos/csi/types.hpp>
#include <mesos/csi/v0.hpp>
+#include <mesos/csi/v1.hpp>
#include <process/grpc.hpp>
#include <process/http.hpp>
+#include <stout/adaptor.hpp>
#include <stout/bytes.hpp>
#include <stout/flags.hpp>
#include <stout/foreach.hpp>
@@ -56,6 +58,7 @@
#include <stout/os/rmdir.hpp>
#include "csi/v0_utils.hpp"
+#include "csi/v1_utils.hpp"
#include "linux/fs.hpp"
@@ -106,6 +109,20 @@ class Flags : public virtual mesos::internal::logging::Flags
public:
Flags()
{
+ const hashset<string> supportedApiVersions =
+ {mesos::csi::v0::API_VERSION, mesos::csi::v1::API_VERSION};
+
+ add(&Flags::api_version,
+ "api_version",
+ "If set, the plugin would only serve CSI calls of the specified API\n"
+ "version. Otherwise, the plugin serves all supported API versions by\n"
+ "default. (Supported values: " + stringify(supportedApiVersions) + ")",
+ [=](const Option<string>& apiVersion) {
+ return apiVersion.isNone() ||
+ supportedApiVersions.contains(apiVersion.get())
+ ? Option<Error>::none() : Error("Unsupported API version");
+ });
+
add(&Flags::endpoint,
"endpoint",
"Path to the Unix domain socket the plugin should bind to.");
@@ -138,6 +155,7 @@ public:
"domain socket. (Example: 'unix:///path/to/socket')");
}
+ Option<string> api_version;
string endpoint;
string work_dir;
Bytes available_capacity;
@@ -150,17 +168,22 @@ public:
class TestCSIPlugin
: public csi::v0::Identity::Service,
public csi::v0::Controller::Service,
- public csi::v0::Node::Service
+ public csi::v0::Node::Service,
+ public csi::v1::Identity::Service,
+ public csi::v1::Controller::Service,
+ public csi::v1::Node::Service
{
public:
TestCSIPlugin(
- const string& _workDir,
+ const Option<string>& _apiVersion,
const string& _endpoint,
+ const string& _workDir,
const Bytes& _availableCapacity,
const hashmap<string, string>& _createParameters,
const hashmap<string, Bytes>& _volumes)
- : workDir(_workDir),
+ : apiVersion(_apiVersion),
endpoint(_endpoint),
+ workDir(_workDir),
availableCapacity(_availableCapacity),
createParameters(_createParameters.begin(), _createParameters.end())
{
@@ -208,6 +231,8 @@ public:
void run();
+ // CSI v0 RPCs.
+
Status GetPluginInfo(
ServerContext* context,
const csi::v0::GetPluginInfoRequest* request,
@@ -293,6 +318,93 @@ public:
const csi::v0::NodeGetCapabilitiesRequest* request,
csi::v0::NodeGetCapabilitiesResponse* response) override;
+ // CSI v1 RPCs.
+
+ Status GetPluginInfo(
+ ServerContext* context,
+ const csi::v1::GetPluginInfoRequest* request,
+ csi::v1::GetPluginInfoResponse* response) override;
+
+ Status GetPluginCapabilities(
+ ServerContext* context,
+ const csi::v1::GetPluginCapabilitiesRequest* request,
+ csi::v1::GetPluginCapabilitiesResponse* response) override;
+
+ Status Probe(
+ ServerContext* context,
+ const csi::v1::ProbeRequest* request,
+ csi::v1::ProbeResponse* response) override;
+
+ Status CreateVolume(
+ ServerContext* context,
+ const csi::v1::CreateVolumeRequest* request,
+ csi::v1::CreateVolumeResponse* response) override;
+
+ Status DeleteVolume(
+ ServerContext* context,
+ const csi::v1::DeleteVolumeRequest* request,
+ csi::v1::DeleteVolumeResponse* response) override;
+
+ Status ControllerPublishVolume(
+ ServerContext* context,
+ const csi::v1::ControllerPublishVolumeRequest* request,
+ csi::v1::ControllerPublishVolumeResponse* response) override;
+
+ Status ControllerUnpublishVolume(
+ ServerContext* context,
+ const csi::v1::ControllerUnpublishVolumeRequest* request,
+ csi::v1::ControllerUnpublishVolumeResponse* response) override;
+
+ Status ValidateVolumeCapabilities(
+ ServerContext* context,
+ const csi::v1::ValidateVolumeCapabilitiesRequest* request,
+ csi::v1::ValidateVolumeCapabilitiesResponse* response) override;
+
+ Status ListVolumes(
+ ServerContext* context,
+ const csi::v1::ListVolumesRequest* request,
+ csi::v1::ListVolumesResponse* response) override;
+
+ Status GetCapacity(
+ ServerContext* context,
+ const csi::v1::GetCapacityRequest* request,
+ csi::v1::GetCapacityResponse* response) override;
+
+ Status ControllerGetCapabilities(
+ ServerContext* context,
+ const csi::v1::ControllerGetCapabilitiesRequest* request,
+ csi::v1::ControllerGetCapabilitiesResponse* response) override;
+
+ Status NodeStageVolume(
+ ServerContext* context,
+ const csi::v1::NodeStageVolumeRequest* request,
+ csi::v1::NodeStageVolumeResponse* response) override;
+
+ Status NodeUnstageVolume(
+ ServerContext* context,
+ const csi::v1::NodeUnstageVolumeRequest* request,
+ csi::v1::NodeUnstageVolumeResponse* response) override;
+
+ Status NodePublishVolume(
+ ServerContext* context,
+ const csi::v1::NodePublishVolumeRequest* request,
+ csi::v1::NodePublishVolumeResponse* response) override;
+
+ Status NodeUnpublishVolume(
+ ServerContext* context,
+ const csi::v1::NodeUnpublishVolumeRequest* request,
+ csi::v1::NodeUnpublishVolumeResponse* response) override;
+
+ Status NodeGetCapabilities(
+ ServerContext* context,
+ const csi::v1::NodeGetCapabilitiesRequest* request,
+ csi::v1::NodeGetCapabilitiesResponse* response) override;
+
+ Status NodeGetInfo(
+ ServerContext* context,
+ const csi::v1::NodeGetInfoRequest* request,
+ csi::v1::NodeGetInfoResponse* response) override;
+
private:
struct VolumeInfo
{
@@ -360,8 +472,9 @@ private:
Try<Nothing, StatusError> nodeUnpublishVolume(
const string& volumeId, const string& targetPath);
- const string workDir;
+ const Option<string> apiVersion;
const string endpoint;
+ const string workDir;
Bytes availableCapacity;
VolumeCapability defaultVolumeCapability;
@@ -374,9 +487,18 @@ void TestCSIPlugin::run()
{
ServerBuilder builder;
builder.AddListeningPort(endpoint, grpc::InsecureServerCredentials());
- builder.RegisterService(static_cast<csi::v0::Identity::Service*>(this));
- builder.RegisterService(static_cast<csi::v0::Controller::Service*>(this));
- builder.RegisterService(static_cast<csi::v0::Node::Service*>(this));
+
+ if (apiVersion.isNone() || apiVersion.get() == mesos::csi::v0::API_VERSION) {
+ builder.RegisterService(static_cast<csi::v0::Identity::Service*>(this));
+ builder.RegisterService(static_cast<csi::v0::Controller::Service*>(this));
+ builder.RegisterService(static_cast<csi::v0::Node::Service*>(this));
+ }
+
+ if (apiVersion.isNone() || apiVersion.get() == mesos::csi::v1::API_VERSION) {
+ builder.RegisterService(static_cast<csi::v1::Identity::Service*>(this));
+ builder.RegisterService(static_cast<csi::v1::Controller::Service*>(this));
+ builder.RegisterService(static_cast<csi::v1::Node::Service*>(this));
+ }
std::unique_ptr<Server> server = builder.BuildAndStart();
server->Wait();
@@ -602,14 +724,16 @@ Status TestCSIPlugin::ControllerGetCapabilities(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- response->add_capabilities()->mutable_rpc()->set_type(
- csi::v0::ControllerServiceCapability::RPC::CREATE_DELETE_VOLUME);
- response->add_capabilities()->mutable_rpc()->set_type(
- csi::v0::ControllerServiceCapability::RPC::PUBLISH_UNPUBLISH_VOLUME);
- response->add_capabilities()->mutable_rpc()->set_type(
- csi::v0::ControllerServiceCapability::RPC::GET_CAPACITY);
- response->add_capabilities()->mutable_rpc()->set_type(
- csi::v0::ControllerServiceCapability::RPC::LIST_VOLUMES);
+ const vector<csi::v0::ControllerServiceCapability::RPC::Type> rpcs = {
+ csi::v0::ControllerServiceCapability::RPC::CREATE_DELETE_VOLUME,
+ csi::v0::ControllerServiceCapability::RPC::PUBLISH_UNPUBLISH_VOLUME,
+ csi::v0::ControllerServiceCapability::RPC::GET_CAPACITY,
+ csi::v0::ControllerServiceCapability::RPC::LIST_VOLUMES,
+ };
+
+ foreach (const csi::v0::ControllerServiceCapability::RPC::Type rpc, rpcs) {
+ response->add_capabilities()->mutable_rpc()->set_type(rpc);
+ }
return Status::OK;
}
@@ -732,6 +856,229 @@ Status TestCSIPlugin::NodeGetCapabilities(
}
+Status TestCSIPlugin::GetPluginInfo(
+ ServerContext* context,
+ const csi::v1::GetPluginInfoRequest* request,
+ csi::v1::GetPluginInfoResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // Since CSI v1 requires the plugin name to follow the (forward) domain name
+ // notation, we reverse the plugin's reverse-DNS name here.
+ response->set_name(
+ strings::join(".", adaptor::reverse(strings::split(".", PLUGIN_NAME))));
+
+ response->set_vendor_version(MESOS_VERSION);
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::GetPluginCapabilities(
+ ServerContext* context,
+ const csi::v1::GetPluginCapabilitiesRequest* request,
+ csi::v1::GetPluginCapabilitiesResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ response->add_capabilities()->mutable_service()->set_type(
+ csi::v1::PluginCapability::Service::CONTROLLER_SERVICE);
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::Probe(
+ ServerContext* context,
+ const csi::v1::ProbeRequest* request,
+ csi::v1::ProbeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ response->mutable_ready()->set_value(true);
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::CreateVolume(
+ ServerContext* context,
+ const csi::v1::CreateVolumeRequest* request,
+ csi::v1::CreateVolumeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ Try<VolumeInfo, StatusError> result = createVolume(
+ request->name(),
+ request->capacity_range().required_bytes()
+ ? request->capacity_range().required_bytes() : 1,
+ request->capacity_range().limit_bytes()
+ ? request->capacity_range().limit_bytes()
+ : std::numeric_limits<int64_t>::max(),
+ mesos::csi::v1::devolve(request->volume_capabilities()),
+ request->parameters());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ response->mutable_volume()->set_volume_id(result->id);
+ response->mutable_volume()->set_capacity_bytes(result->size.bytes());
+ (*response->mutable_volume()->mutable_volume_context())["path"] =
+ getVolumePath(result.get());
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::DeleteVolume(
+ ServerContext* context,
+ const csi::v1::DeleteVolumeRequest* request,
+ csi::v1::DeleteVolumeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ Try<Nothing, StatusError> result = deleteVolume(request->volume_id());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::ControllerPublishVolume(
+ ServerContext* context,
+ const csi::v1::ControllerPublishVolumeRequest* request,
+ csi::v1::ControllerPublishVolumeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ Try<Nothing, StatusError> result = controllerPublishVolume(
+ request->volume_id(),
+ request->node_id(),
+ mesos::csi::v1::devolve(request->volume_capability()),
+ request->readonly(),
+ request->volume_context());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::ControllerUnpublishVolume(
+ ServerContext* context,
+ const csi::v1::ControllerUnpublishVolumeRequest* request,
+ csi::v1::ControllerUnpublishVolumeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ Try<Nothing, StatusError> result =
+ controllerUnpublishVolume(request->volume_id(), request->node_id());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::ValidateVolumeCapabilities(
+ ServerContext* context,
+ const csi::v1::ValidateVolumeCapabilitiesRequest* request,
+ csi::v1::ValidateVolumeCapabilitiesResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ Try<Option<Error>, StatusError> result = validateVolumeCapabilities(
+ request->volume_id(),
+ request->volume_context(),
+ mesos::csi::v1::devolve(request->volume_capabilities()),
+ request->parameters());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ if (result->isSome()) {
+ response->set_message(result->get().message);
+ } else {
+ *response->mutable_confirmed()->mutable_volume_context() =
+ request->volume_context();
+ *response->mutable_confirmed()->mutable_volume_capabilities() =
+ request->volume_capabilities();
+ *response->mutable_confirmed()->mutable_parameters() =
+ request->parameters();
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::ListVolumes(
+ ServerContext* context,
+ const csi::v1::ListVolumesRequest* request,
+ csi::v1::ListVolumesResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ Try<vector<VolumeInfo>, StatusError> result = listVolumes(
+ request->max_entries()
+ ? request->max_entries() : Option<int32_t>::none(),
+ !request->starting_token().empty()
+ ? request->starting_token() : Option<string>::none());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ foreach (const VolumeInfo& volumeInfo, result.get()) {
+ csi::v1::Volume* volume = response->add_entries()->mutable_volume();
+ volume->set_volume_id(volumeInfo.id);
+ volume->set_capacity_bytes(volumeInfo.size.bytes());
+ (*volume->mutable_volume_context())["path"] = getVolumePath(volumeInfo);
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::GetCapacity(
+ ServerContext* context,
+ const csi::v1::GetCapacityRequest* request,
+ csi::v1::GetCapacityResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ Try<Bytes, StatusError> result = getCapacity(
+ mesos::csi::v1::devolve(request->volume_capabilities()),
+ request->parameters());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ response->set_available_capacity(result->bytes());
+
+ return Status::OK;
+}
+
+
string TestCSIPlugin::getVolumePath(const VolumeInfo& volumeInfo)
{
return path::join(
@@ -740,6 +1087,169 @@ string TestCSIPlugin::getVolumePath(const VolumeInfo& volumeInfo)
}
+Status TestCSIPlugin::ControllerGetCapabilities(
+ ServerContext* context,
+ const csi::v1::ControllerGetCapabilitiesRequest* request,
+ csi::v1::ControllerGetCapabilitiesResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ const vector<csi::v1::ControllerServiceCapability::RPC::Type> rpcs = {
+ csi::v1::ControllerServiceCapability::RPC::CREATE_DELETE_VOLUME,
+ csi::v1::ControllerServiceCapability::RPC::PUBLISH_UNPUBLISH_VOLUME,
+ csi::v1::ControllerServiceCapability::RPC::GET_CAPACITY,
+ csi::v1::ControllerServiceCapability::RPC::LIST_VOLUMES,
+ };
+
+ foreach (const csi::v1::ControllerServiceCapability::RPC::Type rpc, rpcs) {
+ response->add_capabilities()->mutable_rpc()->set_type(rpc);
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::NodeStageVolume(
+ ServerContext* context,
+ const csi::v1::NodeStageVolumeRequest* request,
+ csi::v1::NodeStageVolumeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ Try<Nothing, StatusError> result = nodeStageVolume(
+ request->volume_id(),
+ request->publish_context(),
+ request->staging_target_path(),
+ mesos::csi::v1::devolve(request->volume_capability()),
+ request->volume_context());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::NodeUnstageVolume(
+ ServerContext* context,
+ const csi::v1::NodeUnstageVolumeRequest* request,
+ csi::v1::NodeUnstageVolumeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ Try<Nothing, StatusError> result =
+ nodeUnstageVolume(request->volume_id(), request->staging_target_path());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::NodePublishVolume(
+ ServerContext* context,
+ const csi::v1::NodePublishVolumeRequest* request,
+ csi::v1::NodePublishVolumeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ // Since CSI v1 states that creation of the target path is the responsibility
+ // of the plugin, we create the mount point here if it does not exist.
+ if (!os::exists(request->target_path())) {
+ Try<Nothing> mkdir = os::mkdir(request->target_path());
+ if (mkdir.isError()) {
+ return Status(
+ grpc::INTERNAL,
+ "Failed to create target path '" + request->target_path() +
+ "' for volume '" + request->volume_id() + "': " + mkdir.error());
+ }
+ }
+
+ Try<Nothing, StatusError> result = nodePublishVolume(
+ request->volume_id(),
+ request->publish_context(),
+ request->staging_target_path(),
+ request->target_path(),
+ mesos::csi::v1::devolve(request->volume_capability()),
+ request->readonly(),
+ request->volume_context());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::NodeUnpublishVolume(
+ ServerContext* context,
+ const csi::v1::NodeUnpublishVolumeRequest* request,
+ csi::v1::NodeUnpublishVolumeResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ // TODO(chhsiao): Validate the request.
+
+ Try<Nothing, StatusError> result =
+ nodeUnpublishVolume(request->volume_id(), request->target_path());
+
+ if (result.isError()) {
+ return result.error().status;
+ }
+
+ // Since CSI v1 requires the plugin to delete the target path, we delete it
+ // here if it has not been deleted by a previous attempt.
+ if (os::exists(request->target_path())) {
+ Try<Nothing> rmdir = os::rmdir(request->target_path());
+ if (rmdir.isError()) {
+ return Status(
+ grpc::INTERNAL,
+ "Failed to remove target path '" + request->target_path() +
+ "' for volume '" + request->volume_id() + "': " + rmdir.error());
+ }
+ }
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::NodeGetCapabilities(
+ ServerContext* context,
+ const csi::v1::NodeGetCapabilitiesRequest* request,
+ csi::v1::NodeGetCapabilitiesResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ response->add_capabilities()->mutable_rpc()->set_type(
+ csi::v1::NodeServiceCapability::RPC::STAGE_UNSTAGE_VOLUME);
+
+ return Status::OK;
+}
+
+
+Status TestCSIPlugin::NodeGetInfo(
+ ServerContext* context,
+ const csi::v1::NodeGetInfoRequest* request,
+ csi::v1::NodeGetInfoResponse* response)
+{
+ LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
+
+ response->set_node_id(NODE_ID);
+
+ return Status::OK;
+}
+
+
Try<TestCSIPlugin::VolumeInfo> TestCSIPlugin::parseVolumePath(
const string& path)
{
@@ -1495,8 +2005,9 @@ int main(int argc, char** argv)
proxy.run();
} else {
TestCSIPlugin plugin(
- flags.work_dir,
+ flags.api_version,
flags.endpoint,
+ flags.work_dir,
flags.available_capacity,
createParameters,
volumes);
[mesos] 05/06: Refactored the test CSI plugin to make it easy to
support CSI v1.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch 1.8.x
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit eca223cf54fcd6a6d828548fb045ce6d150dfe72
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Apr 4 18:48:24 2019 -0700
Refactored the test CSI plugin to make it easy to support CSI v1.
Review: https://reviews.apache.org/r/70403
---
src/examples/test_csi_plugin.cpp | 951 +++++++++++++++++++++++++--------------
1 file changed, 606 insertions(+), 345 deletions(-)
diff --git a/src/examples/test_csi_plugin.cpp b/src/examples/test_csi_plugin.cpp
index 4321f8f..54753d9 100644
--- a/src/examples/test_csi_plugin.cpp
+++ b/src/examples/test_csi_plugin.cpp
@@ -14,9 +14,15 @@
// See the License for the specific language governing permissions and
// limitations under the License.
+#include <algorithm>
+#include <limits>
#include <memory>
#include <thread>
#include <utility>
+#include <vector>
+
+#include <google/protobuf/map.h>
+#include <google/protobuf/message.h>
#include <grpcpp/grpcpp.h>
@@ -25,16 +31,24 @@
#include <mesos/type_utils.hpp>
+#include <mesos/csi/types.hpp>
#include <mesos/csi/v0.hpp>
+#include <process/grpc.hpp>
+#include <process/http.hpp>
+
#include <stout/bytes.hpp>
#include <stout/flags.hpp>
#include <stout/foreach.hpp>
#include <stout/hashmap.hpp>
+#include <stout/none.hpp>
#include <stout/option.hpp>
#include <stout/path.hpp>
+#include <stout/some.hpp>
#include <stout/stringify.hpp>
#include <stout/strings.hpp>
+#include <stout/try.hpp>
+#include <stout/unreachable.hpp>
#include <stout/os/exists.hpp>
#include <stout/os/ls.hpp>
@@ -47,6 +61,7 @@
#include "logging/logging.hpp"
+namespace http = process::http;
namespace fs = mesos::internal::fs;
using std::cerr;
@@ -59,6 +74,10 @@ using std::string;
using std::unique_ptr;
using std::vector;
+using google::protobuf::Map;
+using google::protobuf::MapPair;
+using google::protobuf::RepeatedPtrField;
+
using grpc::AsyncGenericService;
using grpc::ByteBuffer;
using grpc::ClientContext;
@@ -73,6 +92,10 @@ using grpc::ServerContext;
using grpc::Status;
using grpc::WriteOptions;
+using mesos::csi::types::VolumeCapability;
+
+using process::grpc::StatusError;
+
constexpr char PLUGIN_NAME[] = "org.apache.mesos.csi.test";
constexpr char NODE_ID[] = "localhost";
constexpr Bytes DEFAULT_VOLUME_CAPACITY = Megabytes(64);
@@ -144,7 +167,7 @@ public:
// Construct the default mount volume capability.
defaultVolumeCapability.mutable_mount();
defaultVolumeCapability.mutable_access_mode()
- ->set_mode(csi::v0::VolumeCapability::AccessMode::SINGLE_NODE_WRITER);
+ ->set_mode(VolumeCapability::AccessMode::SINGLE_NODE_WRITER);
// Scan for preprovisioned volumes.
//
@@ -280,12 +303,69 @@ private:
string getVolumePath(const VolumeInfo& volumeInfo);
Try<VolumeInfo> parseVolumePath(const string& path);
+ Try<VolumeInfo, StatusError> createVolume(
+ const string& name,
+ const Bytes& requiredBytes,
+ const Bytes& limitBytes,
+ const RepeatedPtrField<VolumeCapability>& capabilities,
+ const Map<string, string> parameters);
+
+ Try<Nothing, StatusError> deleteVolume(const string& volumeId);
+
+ Try<Nothing, StatusError> controllerPublishVolume(
+ const string& volumeId,
+ const string& nodeId,
+ const VolumeCapability& capability,
+ bool readonly,
+ const Map<string, string>& volumeContext);
+
+ Try<Nothing, StatusError> controllerUnpublishVolume(
+ const string& volumeId, const string& nodeId);
+
+ // Returns `StatusError` if the volume does not exist; returns `Option<Error>`
+ // with an error set if the volume is not compatible with the given arguments.
+ Try<Option<Error>, StatusError> validateVolumeCapabilities(
+ const string& volumeId,
+ const Map<string, string>& volumeContext,
+ const RepeatedPtrField<VolumeCapability>& capabilities,
+ const Option<Map<string, string>>& parameters = None());
+
+ Try<vector<VolumeInfo>, StatusError> listVolumes(
+ const Option<int32_t>& maxEntries,
+ const Option<string>& startingToken);
+
+ Try<Bytes, StatusError> getCapacity(
+ const RepeatedPtrField<VolumeCapability>& capabilities,
+ const Map<string, string>& parameters);
+
+ Try<Nothing, StatusError> nodeStageVolume(
+ const string& volumeId,
+ const Map<string, string>& publishContext,
+ const string& stagingPath,
+ const VolumeCapability& capability,
+ const Map<string, string>& volumeContext);
+
+ Try<Nothing, StatusError> nodeUnstageVolume(
+ const string& volumeId, const string& stagingPath);
+
+ Try<Nothing, StatusError> nodePublishVolume(
+ const string& volumeId,
+ const Map<string, string>& publishContext,
+ const string& stagingPath,
+ const string& targetPath,
+ const VolumeCapability& capability,
+ bool readonly,
+ const Map<string, string>& volumeContext);
+
+ Try<Nothing, StatusError> nodeUnpublishVolume(
+ const string& volumeId, const string& targetPath);
+
const string workDir;
const string endpoint;
Bytes availableCapacity;
- csi::v0::VolumeCapability defaultVolumeCapability;
- google::protobuf::Map<string, string> createParameters;
+ VolumeCapability defaultVolumeCapability;
+ Map<string, string> createParameters;
hashmap<string, VolumeInfo> volumes;
};
@@ -349,92 +429,26 @@ Status TestCSIPlugin::CreateVolume(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Validate required fields.
+ // TODO(chhsiao): Validate the request.
- if (request->name().empty()) {
- return Status(grpc::INVALID_ARGUMENT, "Volume name cannot be empty");
- }
-
- if (strings::contains(request->name(), stringify(os::PATH_SEPARATOR))) {
- return Status(
- grpc::INVALID_ARGUMENT,
- "Volume name cannot contain '" + stringify(os::PATH_SEPARATOR) + "'");
- }
+ Try<VolumeInfo, StatusError> result = createVolume(
+ request->name(),
+ request->capacity_range().required_bytes()
+ ? request->capacity_range().required_bytes() : 1,
+ request->capacity_range().limit_bytes()
+ ? request->capacity_range().limit_bytes()
+ : std::numeric_limits<int64_t>::max(),
+ mesos::csi::v0::devolve(request->volume_capabilities()),
+ request->parameters());
- foreach (const csi::v0::VolumeCapability& capability,
- request->volume_capabilities()) {
- if (capability != defaultVolumeCapability) {
- return Status(grpc::INVALID_ARGUMENT, "Unsupported volume capabilities");
- }
+ if (result.isError()) {
+ return result.error().status;
}
- if (request->parameters() != createParameters) {
- return Status(grpc::INVALID_ARGUMENT, "Unsupported create parameters");
- }
-
- // The volume ID is determined by `name`, so we check whether the volume
- // corresponding to `name` is compatible to the request if it exists.
- if (volumes.contains(request->name())) {
- const VolumeInfo volumeInfo = volumes.at(request->name());
-
- if (request->has_capacity_range()) {
- const csi::v0::CapacityRange& range = request->capacity_range();
-
- if (range.limit_bytes() != 0 &&
- volumeInfo.size > Bytes(range.limit_bytes())) {
- return Status(grpc::ALREADY_EXISTS, "Cannot satisfy 'limit_bytes'");
- } else if (range.required_bytes() != 0 &&
- volumeInfo.size < Bytes(range.required_bytes())) {
- return Status(grpc::ALREADY_EXISTS, "Cannot satisfy 'required_bytes'");
- }
- }
- } else {
- if (availableCapacity == Bytes(0)) {
- return Status(grpc::OUT_OF_RANGE, "Insufficient capacity");
- }
-
- VolumeInfo volumeInfo;
- volumeInfo.id = request->name();
- volumeInfo.size = min(DEFAULT_VOLUME_CAPACITY, availableCapacity);
-
- if (request->has_capacity_range()) {
- const csi::v0::CapacityRange& range = request->capacity_range();
-
- // The highest we can pick.
- Bytes limit = range.limit_bytes() != 0
- ? min(availableCapacity, Bytes(range.limit_bytes()))
- : availableCapacity;
-
- if (range.required_bytes() != 0 &&
- limit < Bytes(range.required_bytes())) {
- return Status(grpc::OUT_OF_RANGE, "Cannot satisfy 'required_bytes'");
- }
-
- volumeInfo.size = min(
- limit,
- max(DEFAULT_VOLUME_CAPACITY, Bytes(range.required_bytes())));
- }
-
- const string path = getVolumePath(volumeInfo);
-
- Try<Nothing> mkdir = os::mkdir(path);
- if (mkdir.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to create volume '" + volumeInfo.id + "': " + mkdir.error());
- }
-
- CHECK_GE(availableCapacity, volumeInfo.size);
- availableCapacity -= volumeInfo.size;
- volumes.put(volumeInfo.id, std::move(volumeInfo));
- }
-
- const VolumeInfo& volumeInfo = volumes.at(request->name());
-
- response->mutable_volume()->set_id(volumeInfo.id);
- response->mutable_volume()->set_capacity_bytes(volumeInfo.size.bytes());
+ response->mutable_volume()->set_id(result->id);
+ response->mutable_volume()->set_capacity_bytes(result->size.bytes());
(*response->mutable_volume()->mutable_attributes())["path"] =
- getVolumePath(volumeInfo);
+ getVolumePath(result.get());
return Status::OK;
}
@@ -447,26 +461,14 @@ Status TestCSIPlugin::DeleteVolume(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Validate required fields.
+ // TODO(chhsiao): Validate the request.
- if (!volumes.contains(request->volume_id())) {
- return Status::OK;
- }
+ Try<Nothing, StatusError> result = deleteVolume(request->volume_id());
- const VolumeInfo& volumeInfo = volumes.at(request->volume_id());
- const string path = getVolumePath(volumeInfo);
-
- Try<Nothing> rmdir = os::rmdir(path);
- if (rmdir.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to delete volume '" + request->volume_id() + "': " +
- rmdir.error());
+ if (result.isError()) {
+ return result.error().status;
}
- availableCapacity += volumeInfo.size;
- volumes.erase(volumeInfo.id);
-
return Status::OK;
}
@@ -478,29 +480,19 @@ Status TestCSIPlugin::ControllerPublishVolume(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Validate required fields.
+ // TODO(chhsiao): Validate the request.
- if (!volumes.contains(request->volume_id())) {
- return Status(
- grpc::NOT_FOUND,
- "Volume '" + request->volume_id() + "' is not found");
- }
+ Try<Nothing, StatusError> result = controllerPublishVolume(
+ request->volume_id(),
+ request->node_id(),
+ mesos::csi::v0::devolve(request->volume_capability()),
+ request->readonly(),
+ request->volume_attributes());
- const VolumeInfo& volumeInfo = volumes.at(request->volume_id());
- const string path = getVolumePath(volumeInfo);
-
- auto it = request->volume_attributes().find("path");
- if (it == request->volume_attributes().end() || it->second != path) {
- return Status(grpc::INVALID_ARGUMENT, "Invalid volume attributes");
+ if (result.isError()) {
+ return result.error().status;
}
- if (request->node_id() != NODE_ID) {
- return Status(
- grpc::NOT_FOUND,
- "Node '" + request->node_id() + "' is not found");
- }
-
- // Do nothing.
return Status::OK;
}
@@ -512,21 +504,15 @@ Status TestCSIPlugin::ControllerUnpublishVolume(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Validate required fields.
+ // TODO(chhsiao): Validate the request.
- if (!volumes.contains(request->volume_id())) {
- return Status(
- grpc::NOT_FOUND,
- "Volume '" + request->volume_id() + "' is not found");
- }
+ Try<Nothing, StatusError> result =
+ controllerUnpublishVolume(request->volume_id(), request->node_id());
- if (request->node_id() != NODE_ID) {
- return Status(
- grpc::NOT_FOUND,
- "Node '" + request->node_id() + "' is not found");
+ if (result.isError()) {
+ return result.error().status;
}
- // Do nothing.
return Status::OK;
}
@@ -538,36 +524,24 @@ Status TestCSIPlugin::ValidateVolumeCapabilities(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Validate required fields.
-
- if (!volumes.contains(request->volume_id())) {
- return Status(
- grpc::NOT_FOUND,
- "Volume '" + request->volume_id() + "' is not found");
- }
+ // TODO(chhsiao): Validate the request.
- const VolumeInfo& volumeInfo = volumes.at(request->volume_id());
- const string path = getVolumePath(volumeInfo);
+ Try<Option<Error>, StatusError> result = validateVolumeCapabilities(
+ request->volume_id(),
+ request->volume_attributes(),
+ mesos::csi::v0::devolve(request->volume_capabilities()));
- auto it = request->volume_attributes().find("path");
- if (it == request->volume_attributes().end() || it->second != path) {
- return Status(grpc::INVALID_ARGUMENT, "Invalid volume attributes");
+ if (result.isError()) {
+ return result.error().status;
}
- foreach (const csi::v0::VolumeCapability& capability,
- request->volume_capabilities()) {
- if (capability != defaultVolumeCapability) {
- response->set_supported(false);
- response->set_message("Unsupported volume capabilities");
-
- return Status::OK;
- }
+ if (result->isSome()) {
+ response->set_supported(false);
+ response->set_message(result->get().message);
+ } else {
+ response->set_supported(true);
}
- // TODO(chhsiao): Validate the parameters once we get CSI v1.
-
- response->set_supported(true);
-
return Status::OK;
}
@@ -579,17 +553,17 @@ Status TestCSIPlugin::ListVolumes(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Support the `max_entries` field.
- if (request->max_entries() > 0) {
- return Status(grpc::ABORTED, "Field 'max_entries' is not supported");
- }
+ Try<vector<VolumeInfo>, StatusError> result = listVolumes(
+ request->max_entries()
+ ? request->max_entries() : Option<int32_t>::none(),
+ !request->starting_token().empty()
+ ? request->starting_token() : Option<string>::none());
- // TODO(chhsiao): Support the `starting_token` fields.
- if (!request->starting_token().empty()) {
- return Status(grpc::ABORTED, "Field 'starting_token' is not supported");
+ if (result.isError()) {
+ return result.error().status;
}
- foreachvalue (const VolumeInfo& volumeInfo, volumes) {
+ foreach (const VolumeInfo& volumeInfo, result.get()) {
csi::v0::Volume* volume = response->add_entries()->mutable_volume();
volume->set_id(volumeInfo.id);
volume->set_capacity_bytes(volumeInfo.size.bytes());
@@ -607,25 +581,15 @@ Status TestCSIPlugin::GetCapacity(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- foreach (const csi::v0::VolumeCapability& capability,
- request->volume_capabilities()) {
- // We report zero capacity for any capability other than the
- // default-constructed mount volume capability since this plugin
- // does not support any filesystem types and mount flags.
- if (capability != defaultVolumeCapability) {
- response->set_available_capacity(0);
-
- return Status::OK;
- }
- }
-
- if (request->parameters() != createParameters) {
- response->set_available_capacity(0);
+ Try<Bytes, StatusError> result = getCapacity(
+ mesos::csi::v0::devolve(request->volume_capabilities()),
+ request->parameters());
- return Status::OK;
+ if (result.isError()) {
+ return result.error().status;
}
- response->set_available_capacity(availableCapacity.bytes());
+ response->set_available_capacity(result->bytes());
return Status::OK;
}
@@ -658,53 +622,17 @@ Status TestCSIPlugin::NodeStageVolume(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Validate required fields.
-
- if (!volumes.contains(request->volume_id())) {
- return Status(
- grpc::NOT_FOUND,
- "Volume '" + request->volume_id() + "' is not found");
- }
-
- const VolumeInfo& volumeInfo = volumes.at(request->volume_id());
- const string path = getVolumePath(volumeInfo);
-
- auto it = request->volume_attributes().find("path");
- if (it == request->volume_attributes().end() || it->second != path) {
- return Status(grpc::INVALID_ARGUMENT, "Invalid volume attributes");
- }
-
- if (!os::exists(request->staging_target_path())) {
- return Status(
- grpc::INVALID_ARGUMENT,
- "Target path '" + request->staging_target_path() + "' is not found");
- }
+ // TODO(chhsiao): Validate the request.
- Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
- if (table.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to get mount table: " + table.error());
- }
-
- foreach (const fs::MountInfoTable::Entry& entry, table->entries) {
- if (entry.target == request->staging_target_path()) {
- return Status::OK;
- }
- }
-
- Try<Nothing> mount = fs::mount(
- path,
+ Try<Nothing, StatusError> result = nodeStageVolume(
+ request->volume_id(),
+ request->publish_info(),
request->staging_target_path(),
- None(),
- MS_BIND,
- None());
+ mesos::csi::v0::devolve(request->volume_capability()),
+ request->volume_attributes());
- if (mount.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to mount from '" + path + "' to '" +
- request->staging_target_path() + "': " + mount.error());
+ if (result.isError()) {
+ return result.error().status;
}
return Status::OK;
@@ -718,39 +646,13 @@ Status TestCSIPlugin::NodeUnstageVolume(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Validate required fields.
-
- if (!volumes.contains(request->volume_id())) {
- return Status(
- grpc::NOT_FOUND,
- "Volume '" + request->volume_id() + "' is not found");
- }
-
- Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
- if (table.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to get mount table: " + table.error());
- }
-
- bool found = false;
- foreach (const fs::MountInfoTable::Entry& entry, table->entries) {
- if (entry.target == request->staging_target_path()) {
- found = true;
- break;
- }
- }
+ // TODO(chhsiao): Validate the request.
- if (!found) {
- return Status::OK;
- }
+ Try<Nothing, StatusError> result =
+ nodeUnstageVolume(request->volume_id(), request->staging_target_path());
- Try<Nothing> unmount = fs::unmount(request->staging_target_path());
- if (unmount.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to unmount '" + request->staging_target_path() +
- "': " + unmount.error());
+ if (result.isError()) {
+ return result.error().status;
}
return Status::OK;
@@ -764,73 +666,19 @@ Status TestCSIPlugin::NodePublishVolume(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- // TODO(chhsiao): Validate required fields.
-
- if (!volumes.contains(request->volume_id())) {
- return Status(
- grpc::NOT_FOUND,
- "Volume '" + request->volume_id() + "' is not found");
- }
-
- const VolumeInfo& volumeInfo = volumes.at(request->volume_id());
- const string path = getVolumePath(volumeInfo);
-
- auto it = request->volume_attributes().find("path");
- if (it == request->volume_attributes().end() || it->second != path) {
- return Status(grpc::INVALID_ARGUMENT, "Invalid volume attributes");
- }
-
- if (!os::exists(request->target_path())) {
- return Status(
- grpc::INVALID_ARGUMENT,
- "Target path '" + request->target_path() + "' is not found");
- }
-
- if (request->staging_target_path().empty()) {
- return Status(
- grpc::FAILED_PRECONDITION,
- "Expecting 'staging_target_path' to be set");
- }
-
- Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
- if (table.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to get mount table: " + table.error());
- }
-
- bool found = false;
- foreach (const fs::MountInfoTable::Entry& entry, table->entries) {
- if (entry.target == request->staging_target_path()) {
- found = true;
- break;
- }
- }
-
- if (!found) {
- return Status(
- grpc::FAILED_PRECONDITION,
- "Volume '" + request->volume_id() + "' has not been staged yet");
- }
-
- foreach (const fs::MountInfoTable::Entry& entry, table->entries) {
- if (entry.target == request->target_path()) {
- return Status::OK;
- }
- }
+ // TODO(chhsiao): Validate the request.
- Try<Nothing> mount = fs::mount(
+ Try<Nothing, StatusError> result = nodePublishVolume(
+ request->volume_id(),
+ request->publish_info(),
request->staging_target_path(),
request->target_path(),
- None(),
- MS_BIND | (request->readonly() ? MS_RDONLY : 0),
- None());
+ mesos::csi::v0::devolve(request->volume_capability()),
+ request->readonly(),
+ request->volume_attributes());
- if (mount.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to mount from '" + path + "' to '" +
- request->target_path() + "': " + mount.error());
+ if (result.isError()) {
+ return result.error().status;
}
return Status::OK;
@@ -844,37 +692,13 @@ Status TestCSIPlugin::NodeUnpublishVolume(
{
LOG(INFO) << request->GetDescriptor()->name() << " '" << *request << "'";
- if (!volumes.contains(request->volume_id())) {
- return Status(
- grpc::NOT_FOUND,
- "Volume '" + request->volume_id() + "' is not found");
- }
-
- Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
- if (table.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to get mount table: " + table.error());
- }
-
- bool found = false;
- foreach (const fs::MountInfoTable::Entry& entry, table->entries) {
- if (entry.target == request->target_path()) {
- found = true;
- break;
- }
- }
+ // TODO(chhsiao): Validate the request.
- if (!found) {
- return Status::OK;
- }
+ Try<Nothing, StatusError> result =
+ nodeUnpublishVolume(request->volume_id(), request->target_path());
- Try<Nothing> unmount = fs::unmount(request->target_path());
- if (unmount.isError()) {
- return Status(
- grpc::INTERNAL,
- "Failed to unmount '" + request->target_path() +
- "': " + unmount.error());
+ if (result.isError()) {
+ return result.error().status;
}
return Status::OK;
@@ -940,6 +764,443 @@ Try<TestCSIPlugin::VolumeInfo> TestCSIPlugin::parseVolumePath(
}
+Try<TestCSIPlugin::VolumeInfo, StatusError> TestCSIPlugin::createVolume(
+ const string& name,
+ const Bytes& requiredBytes,
+ const Bytes& limitBytes,
+ const RepeatedPtrField<VolumeCapability>& capabilities,
+ const Map<string, string> parameters)
+{
+ // The volume ID is determined by `name`, with reserved characters escaped.
+ const string volumeId = http::encode(name);
+
+ foreach (const VolumeCapability& capability, capabilities) {
+ if (capability != defaultVolumeCapability) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Unsupported volume capabilities"));
+ }
+ }
+
+ if (parameters != createParameters) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Unsupported create parameters"));
+ }
+
+ if (volumes.contains(volumeId)) {
+ const VolumeInfo& volumeInfo = volumes.at(volumeId);
+
+ if (volumeInfo.size > limitBytes) {
+ return StatusError(Status(
+ grpc::ALREADY_EXISTS, "Cannot satisfy limit bytes"));
+ }
+
+ if (volumeInfo.size < requiredBytes) {
+ return StatusError(Status(
+ grpc::ALREADY_EXISTS, "Cannot satisfy required bytes"));
+ }
+
+ return volumeInfo;
+ } else {
+ if (availableCapacity < requiredBytes) {
+ return StatusError(Status(grpc::OUT_OF_RANGE, "Insufficient capacity"));
+ }
+
+ VolumeInfo volumeInfo;
+ volumeInfo.id = volumeId;
+
+ // We assume that `requiredBytes <= limitBytes` has been verified.
+ const Bytes defaultSize = min(availableCapacity, DEFAULT_VOLUME_CAPACITY);
+ volumeInfo.size = min(max(defaultSize, requiredBytes), limitBytes);
+
+ const string path = getVolumePath(volumeInfo);
+
+ Try<Nothing> mkdir = os::mkdir(path);
+ if (mkdir.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL,
+ "Failed to create volume '" + volumeInfo.id + "': " + mkdir.error()));
+ }
+
+ CHECK_GE(availableCapacity, volumeInfo.size);
+ availableCapacity -= volumeInfo.size;
+ volumes.put(volumeInfo.id, volumeInfo);
+
+ return volumeInfo;
+ }
+
+ UNREACHABLE();
+}
+
+
+Try<Nothing, StatusError> TestCSIPlugin::deleteVolume(const string& volumeId)
+{
+ if (!volumes.contains(volumeId)) {
+ // Return a success for idempotency.
+ return Nothing();
+ }
+
+ const VolumeInfo& volumeInfo = volumes.at(volumeId);
+ const string path = getVolumePath(volumeInfo);
+
+ Try<Nothing> rmdir = os::rmdir(path);
+ if (rmdir.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL,
+ "Failed to delete volume '" + volumeId + "': " + rmdir.error()));
+ }
+
+ availableCapacity += volumeInfo.size;
+ volumes.erase(volumeInfo.id);
+
+ return Nothing();
+}
+
+
+Try<Nothing, StatusError> TestCSIPlugin::controllerPublishVolume(
+ const string& volumeId,
+ const string& nodeId,
+ const VolumeCapability& capability,
+ bool readonly,
+ const Map<string, string>& volumeContext)
+{
+ if (!volumes.contains(volumeId)) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Volume '" + volumeId + "' does not exist"));
+ }
+
+ if (nodeId != NODE_ID) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Node '" + nodeId + "' does not exist"));
+ }
+
+ if (capability != defaultVolumeCapability) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Unsupported volume capability"));
+ }
+
+ if (readonly) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Unsupported read-only mode"));
+ }
+
+ const VolumeInfo& volumeInfo = volumes.at(volumeId);
+ const string path = getVolumePath(volumeInfo);
+
+ if (!volumeContext.count("path") || volumeContext.at("path") != path) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Invalid volume context"));
+ }
+
+ // Do nothing.
+ return Nothing();
+}
+
+
+Try<Nothing, StatusError> TestCSIPlugin::controllerUnpublishVolume(
+ const string& volumeId, const string& nodeId)
+{
+ if (!volumes.contains(volumeId)) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Volume '" + volumeId + "' does not exist"));
+ }
+
+ if (nodeId != NODE_ID) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Node '" + nodeId + "' does not exist"));
+ }
+
+ // Do nothing.
+ return Nothing();
+}
+
+
+Try<Option<Error>, StatusError> TestCSIPlugin::validateVolumeCapabilities(
+ const string& volumeId,
+ const Map<string, string>& volumeContext,
+ const RepeatedPtrField<VolumeCapability>& capabilities,
+ const Option<Map<string, string>>& parameters)
+{
+ if (!volumes.contains(volumeId)) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Volume '" + volumeId + "' does not exist"));
+ }
+
+ const VolumeInfo& volumeInfo = volumes.at(volumeId);
+ const string path = getVolumePath(volumeInfo);
+
+ if (!volumeContext.count("path") || volumeContext.at("path") != path) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Invalid volume context"));
+ }
+
+ foreach (const VolumeCapability& capability, capabilities) {
+ if (capability != defaultVolumeCapability) {
+ return Some(Error("Unsupported volume capabilities"));
+ }
+ }
+
+ if (parameters.isSome() && parameters.get() != createParameters) {
+ return Some(Error("Mismatched parameters"));
+ }
+
+ return None();
+}
+
+
+Try<vector<TestCSIPlugin::VolumeInfo>, StatusError> TestCSIPlugin::listVolumes(
+ const Option<int32_t>& maxEntries,
+ const Option<string>& startingToken)
+{
+ // TODO(chhsiao): Support max entries.
+ if (maxEntries.isSome()) {
+ return StatusError(Status(
+ grpc::ABORTED, "Specifying max entries is not supported"));
+ }
+
+ // TODO(chhsiao): Support starting token.
+ if (startingToken.isSome()) {
+ return StatusError(Status(
+ grpc::ABORTED, "Specifying starting token is not supported"));
+ }
+
+ return volumes.values();
+}
+
+
+Try<Bytes, StatusError> TestCSIPlugin::getCapacity(
+ const RepeatedPtrField<VolumeCapability>& capabilities,
+ const Map<string, string>& parameters)
+{
+ // We report zero capacity if any capability other than the default mount
+ // volume capability is given. If no capacity is given, the total available
+ // capacity will be returned.
+ foreach (const VolumeCapability& capability, capabilities) {
+ if (capability != defaultVolumeCapability) {
+ return Bytes(0);
+ }
+ }
+
+ if (parameters != createParameters) {
+ return Bytes(0);
+ }
+
+ return availableCapacity;
+}
+
+
+Try<Nothing, StatusError> TestCSIPlugin::nodeStageVolume(
+ const string& volumeId,
+ const Map<string, string>& publishContext,
+ const string& stagingPath,
+ const VolumeCapability& capability,
+ const Map<string, string>& volumeContext)
+{
+ if (!volumes.contains(volumeId)) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Volume '" + volumeId + "' does not exist"));
+ }
+
+ if (!publishContext.empty()) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Invalid publish context"));
+ }
+
+ if (!os::exists(stagingPath)) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT,
+ "Staging path '" + stagingPath + "' does not exist"));
+ }
+
+ if (capability != defaultVolumeCapability) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Unsupported volume capability"));
+ }
+
+ const VolumeInfo& volumeInfo = volumes.at(volumeId);
+ const string path = getVolumePath(volumeInfo);
+
+ if (!volumeContext.count("path") || volumeContext.at("path") != path) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Invalid volume context"));
+ }
+
+ Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
+ if (table.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL, "Failed to get mount table: " + table.error()));
+ }
+
+ if (std::any_of(
+ table->entries.begin(),
+ table->entries.end(),
+ [&](const fs::MountInfoTable::Entry& entry) {
+ return entry.target == stagingPath;
+ })) {
+ return Nothing();
+ }
+
+ Try<Nothing> mount = fs::mount(path, stagingPath, None(), MS_BIND, None());
+ if (mount.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL,
+ "Failed to mount from '" + path + "' to '" + stagingPath +
+ "': " + mount.error()));
+ }
+
+ return Nothing();
+}
+
+
+Try<Nothing, StatusError> TestCSIPlugin::nodeUnstageVolume(
+ const string& volumeId, const string& stagingPath)
+{
+ if (!volumes.contains(volumeId)) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Volume '" + volumeId + "' does not exist"));
+ }
+
+ Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
+ if (table.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL, "Failed to get mount table: " + table.error()));
+ }
+
+ if (std::none_of(
+ table->entries.begin(),
+ table->entries.end(),
+ [&](const fs::MountInfoTable::Entry& entry) {
+ return entry.target == stagingPath;
+ })) {
+ return Nothing();
+ }
+
+ Try<Nothing> unmount = fs::unmount(stagingPath);
+ if (unmount.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL,
+ "Failed to unmount '" + stagingPath + "': " + unmount.error()));
+ }
+
+ return Nothing();
+}
+
+
+Try<Nothing, StatusError> TestCSIPlugin::nodePublishVolume(
+ const string& volumeId,
+ const Map<string, string>& publishContext,
+ const string& stagingPath,
+ const string& targetPath,
+ const VolumeCapability& capability,
+ bool readonly,
+ const Map<string, string>& volumeContext)
+{
+ if (!volumes.contains(volumeId)) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Volume '" + volumeId + "' does not exist"));
+ }
+
+ if (!publishContext.empty()) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Invalid publish context"));
+ }
+
+ if (!os::exists(targetPath)) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT,
+ "Target path '" + targetPath + "' does not exist"));
+ }
+
+ if (capability != defaultVolumeCapability) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Unsupported volume capability"));
+ }
+
+ const VolumeInfo& volumeInfo = volumes.at(volumeId);
+ const string path = getVolumePath(volumeInfo);
+
+ if (!volumeContext.count("path") || volumeContext.at("path") != path) {
+ return StatusError(Status(
+ grpc::INVALID_ARGUMENT, "Invalid volume context"));
+ }
+
+ Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
+ if (table.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL, "Failed to get mount table: " + table.error()));
+ }
+
+ if (std::none_of(
+ table->entries.begin(),
+ table->entries.end(),
+ [&](const fs::MountInfoTable::Entry& entry) {
+ return entry.target == stagingPath;
+ })) {
+ return StatusError(Status(
+ grpc::FAILED_PRECONDITION,
+ "Volume '" + volumeId + "' has not been staged yet"));
+ }
+
+ if (std::any_of(
+ table->entries.begin(),
+ table->entries.end(),
+ [&](const fs::MountInfoTable::Entry& entry) {
+ return entry.target == targetPath;
+ })) {
+ return Nothing();
+ }
+
+ Try<Nothing> mount = fs::mount(
+ stagingPath,
+ targetPath,
+ None(),
+ MS_BIND | (readonly ? MS_RDONLY : 0),
+ None());
+
+ if (mount.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL,
+ "Failed to mount from '" + stagingPath + "' to '" + targetPath +
+ "': " + mount.error()));
+ }
+
+ return Nothing();
+}
+
+
+Try<Nothing, StatusError> TestCSIPlugin::nodeUnpublishVolume(
+ const string& volumeId, const string& targetPath)
+{
+ if (!volumes.contains(volumeId)) {
+ return StatusError(Status(
+ grpc::NOT_FOUND, "Volume '" + volumeId + "' does not exist"));
+ }
+
+ Try<fs::MountInfoTable> table = fs::MountInfoTable::read();
+ if (table.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL, "Failed to get mount table: " + table.error()));
+ }
+
+ if (std::none_of(
+ table->entries.begin(),
+ table->entries.end(),
+ [&](const fs::MountInfoTable::Entry& entry) {
+ return entry.target == targetPath;
+ })) {
+ return Nothing();
+ }
+
+ Try<Nothing> unmount = fs::unmount(targetPath);
+ if (unmount.isError()) {
+ return StatusError(Status(
+ grpc::INTERNAL,
+ "Failed to unmount '" + targetPath + "': " + unmount.error()));
+ }
+
+ return Nothing();
+}
+
+
// Serves CSI calls from the given endpoint through forwarding the calls to
// another CSI endpoint and returning back the results.
class CSIProxy
[mesos] 01/06: Added helpers to evolve/devolve repeated CSI v0
`VolumeCapability`s.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch 1.8.x
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit b4cc665de5ab74cb527c6fa67b2bb40fd781db5c
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Apr 4 15:37:35 2019 -0700
Added helpers to evolve/devolve repeated CSI v0 `VolumeCapability`s.
Review: https://reviews.apache.org/r/70399
---
src/csi/v0_utils.cpp | 42 +++++++++++++++++++++++++++++++++++++++++-
src/csi/v0_utils.hpp | 27 +++++++++++++++++++++++++--
2 files changed, 66 insertions(+), 3 deletions(-)
diff --git a/src/csi/v0_utils.cpp b/src/csi/v0_utils.cpp
index a95d240..248e417 100644
--- a/src/csi/v0_utils.cpp
+++ b/src/csi/v0_utils.cpp
@@ -16,12 +16,25 @@
#include "csi/v0_utils.hpp"
-#include <stout/unreachable.hpp>
+using google::protobuf::RepeatedPtrField;
namespace mesos {
namespace csi {
namespace v0 {
+// Helper for repeated field devolving to `T1` from `T2`.
+template <typename T1, typename T2>
+RepeatedPtrField<T1> devolve(RepeatedPtrField<T2> from)
+{
+ RepeatedPtrField<T1> to;
+ foreach (const T2& value, from) {
+ *to.Add() = devolve(value);
+ }
+
+ return to;
+}
+
+
types::VolumeCapability::BlockVolume devolve(
const VolumeCapability::BlockVolume& block)
{
@@ -112,6 +125,26 @@ types::VolumeCapability devolve(const VolumeCapability& capability)
}
+RepeatedPtrField<types::VolumeCapability> devolve(
+ const RepeatedPtrField<VolumeCapability>& capabilities)
+{
+ return devolve<types::VolumeCapability>(capabilities);
+}
+
+
+// Helper for repeated field evolving to `T1` from `T2`.
+template <typename T1, typename T2>
+RepeatedPtrField<T1> evolve(RepeatedPtrField<T2> from)
+{
+ RepeatedPtrField<T1> to;
+ foreach (const T2& value, from) {
+ *to.Add() = evolve(value);
+ }
+
+ return to;
+}
+
+
VolumeCapability::BlockVolume evolve(
const types::VolumeCapability::BlockVolume& block)
{
@@ -197,6 +230,13 @@ VolumeCapability evolve(const types::VolumeCapability& capability)
return result;
}
+
+RepeatedPtrField<VolumeCapability> devolve(
+ const RepeatedPtrField<types::VolumeCapability>& capabilities)
+{
+ return evolve<VolumeCapability>(capabilities);
+}
+
} // namespace v0 {
} // namespace csi {
} // namespace mesos {
diff --git a/src/csi/v0_utils.hpp b/src/csi/v0_utils.hpp
index 46a5f13..468b416 100644
--- a/src/csi/v0_utils.hpp
+++ b/src/csi/v0_utils.hpp
@@ -17,6 +17,8 @@
#ifndef __CSI_V0_UTILS_HPP__
#define __CSI_V0_UTILS_HPP__
+#include <google/protobuf/message.h>
+
#include <mesos/csi/types.hpp>
#include <mesos/csi/v0.hpp>
@@ -31,7 +33,8 @@ struct PluginCapabilities
{
PluginCapabilities() = default;
- template <typename Iterable> PluginCapabilities(const Iterable& capabilities)
+ template <typename Iterable>
+ PluginCapabilities(const Iterable& capabilities)
{
foreach (const auto& capability, capabilities) {
if (capability.has_service() &&
@@ -43,6 +46,10 @@ struct PluginCapabilities
case PluginCapability::Service::CONTROLLER_SERVICE:
controllerService = true;
break;
+
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum
+ // cases for us. See: https://github.com/google/protobuf/issues/3917
case google::protobuf::kint32min:
case google::protobuf::kint32max:
UNREACHABLE();
@@ -81,6 +88,10 @@ struct ControllerCapabilities
case ControllerServiceCapability::RPC::GET_CAPACITY:
getCapacity = true;
break;
+
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum
+ // cases for us. See: https://github.com/google/protobuf/issues/3917
case google::protobuf::kint32min:
case google::protobuf::kint32max:
UNREACHABLE();
@@ -100,7 +111,8 @@ struct NodeCapabilities
{
NodeCapabilities() = default;
- template <typename Iterable> NodeCapabilities(const Iterable& capabilities)
+ template <typename Iterable>
+ NodeCapabilities(const Iterable& capabilities)
{
foreach (const auto& capability, capabilities) {
if (capability.has_rpc() &&
@@ -111,6 +123,10 @@ struct NodeCapabilities
case NodeServiceCapability::RPC::STAGE_UNSTAGE_VOLUME:
stageUnstageVolume = true;
break;
+
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum
+ // cases for us. See: https://github.com/google/protobuf/issues/3917
case google::protobuf::kint32min:
case google::protobuf::kint32max:
UNREACHABLE();
@@ -126,10 +142,17 @@ struct NodeCapabilities
// Helpers to devolve CSI v0 protobufs to their unversioned counterparts.
types::VolumeCapability devolve(const VolumeCapability& capability);
+google::protobuf::RepeatedPtrField<types::VolumeCapability> devolve(
+ const google::protobuf::RepeatedPtrField<VolumeCapability>& capabilities);
+
// Helpers to evolve unversioned CSI protobufs to their v0 counterparts.
VolumeCapability evolve(const types::VolumeCapability& capability);
+google::protobuf::RepeatedPtrField<VolumeCapability> evolve(
+ const google::protobuf::RepeatedPtrField<types::VolumeCapability>&
+ capabilities);
+
} // namespace v0 {
} // namespace csi {
} // namespace mesos {
[mesos] 02/06: Added capabilities and evolve/devolve helpers for
CSI v1.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch 1.8.x
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit e22d60212d3859b69b64b8b7c4d063856df29665
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Apr 4 16:27:48 2019 -0700
Added capabilities and evolve/devolve helpers for CSI v1.
This patch adds capability helper structures for CSI v1 services, and
evolve/devolve helpers to convert between CSI v1 protobufs and their
unversioned counterparts.
Review: https://reviews.apache.org/r/70400
---
src/CMakeLists.txt | 1 +
src/Makefile.am | 2 +
src/csi/v1_utils.cpp | 242 ++++++++++++++++++++++++++++++++++++++++++
src/csi/v1_utils.hpp | 218 +++++++++++++++++++++++++++++++++++++
src/tests/csi_utils_tests.cpp | 50 +++++++--
5 files changed, 507 insertions(+), 6 deletions(-)
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index 51c7a04..d36d0be 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -243,6 +243,7 @@ set(CSI_SRC
csi/v0_volume_manager.cpp
csi/v1.cpp
csi/v1_client.cpp
+ csi/v1_utils.cpp
csi/volume_manager.cpp)
set(DOCKER_SRC
diff --git a/src/Makefile.am b/src/Makefile.am
index 9c0180f..64898df 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1598,6 +1598,8 @@ libcsi_la_SOURCES = \
csi/v1.cpp \
csi/v1_client.cpp \
csi/v1_client.hpp \
+ csi/v1_utils.cpp \
+ csi/v1_utils.hpp \
csi/volume_manager.cpp \
csi/volume_manager.hpp
diff --git a/src/csi/v1_utils.cpp b/src/csi/v1_utils.cpp
new file mode 100644
index 0000000..e74138b
--- /dev/null
+++ b/src/csi/v1_utils.cpp
@@ -0,0 +1,242 @@
+// 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 "csi/v1_utils.hpp"
+
+using google::protobuf::RepeatedPtrField;
+
+namespace mesos {
+namespace csi {
+namespace v1 {
+
+// Helper for repeated field devolving to `T1` from `T2`.
+template <typename T1, typename T2>
+RepeatedPtrField<T1> devolve(RepeatedPtrField<T2> from)
+{
+ RepeatedPtrField<T1> to;
+ foreach (const T2& value, from) {
+ *to.Add() = devolve(value);
+ }
+
+ return to;
+}
+
+
+types::VolumeCapability::BlockVolume devolve(
+ const VolumeCapability::BlockVolume& block)
+{
+ return types::VolumeCapability::BlockVolume();
+}
+
+
+types::VolumeCapability::MountVolume devolve(
+ const VolumeCapability::MountVolume& mount)
+{
+ types::VolumeCapability::MountVolume result;
+ result.set_fs_type(mount.fs_type());
+ *result.mutable_mount_flags() = mount.mount_flags();
+ return result;
+}
+
+
+types::VolumeCapability::AccessMode devolve(
+ const VolumeCapability::AccessMode& accessMode)
+{
+ types::VolumeCapability::AccessMode result;
+
+ switch (accessMode.mode()) {
+ case VolumeCapability::AccessMode::UNKNOWN: {
+ result.set_mode(types::VolumeCapability::AccessMode::UNKNOWN);
+ break;
+ }
+ case VolumeCapability::AccessMode::SINGLE_NODE_WRITER: {
+ result.set_mode(types::VolumeCapability::AccessMode::SINGLE_NODE_WRITER);
+ break;
+ }
+ case VolumeCapability::AccessMode::SINGLE_NODE_READER_ONLY: {
+ result.set_mode(
+ types::VolumeCapability::AccessMode::SINGLE_NODE_READER_ONLY);
+ break;
+ }
+ case VolumeCapability::AccessMode::MULTI_NODE_READER_ONLY: {
+ result.set_mode(
+ types::VolumeCapability::AccessMode::MULTI_NODE_READER_ONLY);
+ break;
+ }
+ case VolumeCapability::AccessMode::MULTI_NODE_SINGLE_WRITER: {
+ result.set_mode(
+ types::VolumeCapability::AccessMode::MULTI_NODE_SINGLE_WRITER);
+ break;
+ }
+ case VolumeCapability::AccessMode::MULTI_NODE_MULTI_WRITER: {
+ result.set_mode(
+ types::VolumeCapability::AccessMode::MULTI_NODE_MULTI_WRITER);
+ break;
+ }
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum cases
+ // for us. See: https://github.com/google/protobuf/issues/3917
+ case google::protobuf::kint32min:
+ case google::protobuf::kint32max: {
+ UNREACHABLE();
+ }
+ }
+
+ return result;
+}
+
+
+types::VolumeCapability devolve(const VolumeCapability& capability)
+{
+ types::VolumeCapability result;
+
+ switch (capability.access_type_case()) {
+ case VolumeCapability::kBlock: {
+ *result.mutable_block() = devolve(capability.block());
+ break;
+ }
+ case VolumeCapability::kMount: {
+ *result.mutable_mount() = devolve(capability.mount());
+ break;
+ }
+ case VolumeCapability::ACCESS_TYPE_NOT_SET: {
+ break;
+ }
+ }
+
+ if (capability.has_access_mode()) {
+ *result.mutable_access_mode() = devolve(capability.access_mode());
+ }
+
+ return result;
+}
+
+
+RepeatedPtrField<types::VolumeCapability> devolve(
+ const RepeatedPtrField<VolumeCapability>& capabilities)
+{
+ return devolve<types::VolumeCapability>(capabilities);
+}
+
+
+// Helper for repeated field evolving to `T1` from `T2`.
+template <typename T1, typename T2>
+RepeatedPtrField<T1> evolve(RepeatedPtrField<T2> from)
+{
+ RepeatedPtrField<T1> to;
+ foreach (const T2& value, from) {
+ *to.Add() = evolve(value);
+ }
+
+ return to;
+}
+
+
+VolumeCapability::BlockVolume evolve(
+ const types::VolumeCapability::BlockVolume& block)
+{
+ return VolumeCapability::BlockVolume();
+}
+
+
+VolumeCapability::MountVolume evolve(
+ const types::VolumeCapability::MountVolume& mount)
+{
+ VolumeCapability::MountVolume result;
+ result.set_fs_type(mount.fs_type());
+ *result.mutable_mount_flags() = mount.mount_flags();
+ return result;
+}
+
+
+VolumeCapability::AccessMode evolve(
+ const types::VolumeCapability::AccessMode& accessMode)
+{
+ VolumeCapability::AccessMode result;
+
+ switch (accessMode.mode()) {
+ case types::VolumeCapability::AccessMode::UNKNOWN: {
+ result.set_mode(VolumeCapability::AccessMode::UNKNOWN);
+ break;
+ }
+ case types::VolumeCapability::AccessMode::SINGLE_NODE_WRITER: {
+ result.set_mode(VolumeCapability::AccessMode::SINGLE_NODE_WRITER);
+ break;
+ }
+ case types::VolumeCapability::AccessMode::SINGLE_NODE_READER_ONLY: {
+ result.set_mode(VolumeCapability::AccessMode::SINGLE_NODE_READER_ONLY);
+ break;
+ }
+ case types::VolumeCapability::AccessMode::MULTI_NODE_READER_ONLY: {
+ result.set_mode(VolumeCapability::AccessMode::MULTI_NODE_READER_ONLY);
+ break;
+ }
+ case types::VolumeCapability::AccessMode::MULTI_NODE_SINGLE_WRITER: {
+ result.set_mode(VolumeCapability::AccessMode::MULTI_NODE_SINGLE_WRITER);
+ break;
+ }
+ case types::VolumeCapability::AccessMode::MULTI_NODE_MULTI_WRITER: {
+ result.set_mode(VolumeCapability::AccessMode::MULTI_NODE_MULTI_WRITER);
+ break;
+ }
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum cases
+ // for us. See: https://github.com/google/protobuf/issues/3917
+ case google::protobuf::kint32min:
+ case google::protobuf::kint32max: {
+ UNREACHABLE();
+ }
+ }
+
+ return result;
+}
+
+
+VolumeCapability evolve(const types::VolumeCapability& capability)
+{
+ VolumeCapability result;
+
+ switch (capability.access_type_case()) {
+ case types::VolumeCapability::kBlock: {
+ *result.mutable_block() = evolve(capability.block());
+ break;
+ }
+ case types::VolumeCapability::kMount: {
+ *result.mutable_mount() = evolve(capability.mount());
+ break;
+ }
+ case types::VolumeCapability::ACCESS_TYPE_NOT_SET: {
+ break;
+ }
+ }
+
+ if (capability.has_access_mode()) {
+ *result.mutable_access_mode() = evolve(capability.access_mode());
+ }
+
+ return result;
+}
+
+
+RepeatedPtrField<VolumeCapability> devolve(
+ const RepeatedPtrField<types::VolumeCapability>& capabilities)
+{
+ return evolve<VolumeCapability>(capabilities);
+}
+
+} // namespace v1 {
+} // namespace csi {
+} // namespace mesos {
diff --git a/src/csi/v1_utils.hpp b/src/csi/v1_utils.hpp
new file mode 100644
index 0000000..11a64d7
--- /dev/null
+++ b/src/csi/v1_utils.hpp
@@ -0,0 +1,218 @@
+// 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.
+
+#ifndef __CSI_V1_UTILS_HPP__
+#define __CSI_V1_UTILS_HPP__
+
+#include <google/protobuf/message.h>
+
+#include <mesos/csi/types.hpp>
+#include <mesos/csi/v1.hpp>
+
+#include <stout/foreach.hpp>
+#include <stout/unreachable.hpp>
+
+namespace mesos {
+namespace csi {
+namespace v1 {
+
+struct PluginCapabilities
+{
+ PluginCapabilities() = default;
+
+ template <typename Iterable>
+ PluginCapabilities(const Iterable& capabilities)
+ {
+ foreach (const auto& capability, capabilities) {
+ if (capability.has_service() &&
+ PluginCapability::Service::Type_IsValid(
+ capability.service().type())) {
+ switch (capability.service().type()) {
+ case PluginCapability::Service::UNKNOWN:
+ break;
+ case PluginCapability::Service::CONTROLLER_SERVICE:
+ controllerService = true;
+ break;
+ case PluginCapability::Service::VOLUME_ACCESSIBILITY_CONSTRAINTS:
+ volumeAccessibilityConstraints = true;
+ break;
+
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum
+ // cases for us. See: https://github.com/google/protobuf/issues/3917
+ case google::protobuf::kint32min:
+ case google::protobuf::kint32max:
+ UNREACHABLE();
+ }
+ }
+
+ if (capability.has_volume_expansion() &&
+ PluginCapability::VolumeExpansion::Type_IsValid(
+ capability.volume_expansion().type())) {
+ switch (capability.volume_expansion().type()) {
+ case PluginCapability::VolumeExpansion::UNKNOWN:
+ break;
+ case PluginCapability::VolumeExpansion::ONLINE:
+ volumeExpansionOnline = true;
+ volumeExpansionOffline = false;
+ break;
+ case PluginCapability::VolumeExpansion::OFFLINE:
+ volumeExpansionOnline = false;
+ volumeExpansionOffline = true;
+ break;
+
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum
+ // cases for us. See: https://github.com/google/protobuf/issues/3917
+ case google::protobuf::kint32min:
+ case google::protobuf::kint32max:
+ UNREACHABLE();
+ }
+ }
+ }
+ }
+
+ bool controllerService = false;
+ bool volumeAccessibilityConstraints = false;
+ bool volumeExpansionOnline = false;
+ bool volumeExpansionOffline = false;
+};
+
+
+struct ControllerCapabilities
+{
+ ControllerCapabilities() = default;
+
+ template <typename Iterable>
+ ControllerCapabilities(const Iterable& capabilities)
+ {
+ foreach (const auto& capability, capabilities) {
+ if (capability.has_rpc() &&
+ ControllerServiceCapability::RPC::Type_IsValid(
+ capability.rpc().type())) {
+ switch (capability.rpc().type()) {
+ case ControllerServiceCapability::RPC::UNKNOWN:
+ break;
+ case ControllerServiceCapability::RPC::CREATE_DELETE_VOLUME:
+ createDeleteVolume = true;
+ break;
+ case ControllerServiceCapability::RPC::PUBLISH_UNPUBLISH_VOLUME:
+ publishUnpublishVolume = true;
+ break;
+ case ControllerServiceCapability::RPC::LIST_VOLUMES:
+ listVolumes = true;
+ break;
+ case ControllerServiceCapability::RPC::GET_CAPACITY:
+ getCapacity = true;
+ break;
+ case ControllerServiceCapability::RPC::CREATE_DELETE_SNAPSHOT:
+ createDeleteSnapshot = true;
+ break;
+ case ControllerServiceCapability::RPC::LIST_SNAPSHOTS:
+ listSnapshots = true;
+ break;
+ case ControllerServiceCapability::RPC::CLONE_VOLUME:
+ cloneVolume = true;
+ break;
+ case ControllerServiceCapability::RPC::PUBLISH_READONLY:
+ publishReadonly = true;
+ break;
+ case ControllerServiceCapability::RPC::EXPAND_VOLUME:
+ expandVolume = true;
+ break;
+
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum
+ // cases for us. See: https://github.com/google/protobuf/issues/3917
+ case google::protobuf::kint32min:
+ case google::protobuf::kint32max:
+ UNREACHABLE();
+ }
+ }
+ }
+ }
+
+ bool createDeleteVolume = false;
+ bool publishUnpublishVolume = false;
+ bool listVolumes = false;
+ bool getCapacity = false;
+ bool createDeleteSnapshot = false;
+ bool listSnapshots = false;
+ bool cloneVolume = false;
+ bool publishReadonly = false;
+ bool expandVolume = false;
+};
+
+
+struct NodeCapabilities
+{
+ NodeCapabilities() = default;
+
+ template <typename Iterable>
+ NodeCapabilities(const Iterable& capabilities)
+ {
+ foreach (const auto& capability, capabilities) {
+ if (capability.has_rpc() &&
+ NodeServiceCapability::RPC::Type_IsValid(capability.rpc().type())) {
+ switch (capability.rpc().type()) {
+ case NodeServiceCapability::RPC::UNKNOWN:
+ break;
+ case NodeServiceCapability::RPC::STAGE_UNSTAGE_VOLUME:
+ stageUnstageVolume = true;
+ break;
+ case NodeServiceCapability::RPC::GET_VOLUME_STATS:
+ getVolumeStats = true;
+ break;
+ case NodeServiceCapability::RPC::EXPAND_VOLUME:
+ expandVolume = true;
+ break;
+
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum
+ // cases for us. See: https://github.com/google/protobuf/issues/3917
+ case google::protobuf::kint32min:
+ case google::protobuf::kint32max:
+ UNREACHABLE();
+ }
+ }
+ }
+ }
+
+ bool stageUnstageVolume = false;
+ bool getVolumeStats = false;
+ bool expandVolume = false;
+};
+
+
+// Helpers to devolve CSI v1 protobufs to their unversioned counterparts.
+types::VolumeCapability devolve(const VolumeCapability& capability);
+
+google::protobuf::RepeatedPtrField<types::VolumeCapability> devolve(
+ const google::protobuf::RepeatedPtrField<VolumeCapability>& capabilities);
+
+
+// Helpers to evolve unversioned CSI protobufs to their v1 counterparts.
+VolumeCapability evolve(const types::VolumeCapability& capability);
+
+google::protobuf::RepeatedPtrField<VolumeCapability> evolve(
+ const google::protobuf::RepeatedPtrField<types::VolumeCapability>&
+ capabilities);
+
+} // namespace v1 {
+} // namespace csi {
+} // namespace mesos {
+
+#endif // __CSI_V1_UTILS_HPP__
diff --git a/src/tests/csi_utils_tests.cpp b/src/tests/csi_utils_tests.cpp
index b9d0ca8..5ebd0e1 100644
--- a/src/tests/csi_utils_tests.cpp
+++ b/src/tests/csi_utils_tests.cpp
@@ -23,23 +23,62 @@
#include <mesos/csi/types.hpp>
#include <mesos/csi/v0.hpp>
+#include <mesos/csi/v1.hpp>
#include "csi/v0_utils.hpp"
+#include "csi/v1_utils.hpp"
namespace util = google::protobuf::util;
using std::string;
using std::vector;
+// NOTE: We define the following function templates in proper namespaces so
+// argument-dependent lookup can do the magic to pick up the right functions
+// for a given protobuf in the tests.
+namespace csi {
+namespace v0 {
+
+template <typename T>
+T devolveAndEvolve(const T& t)
+{
+ return mesos::csi::v0::evolve(mesos::csi::v0::devolve(t));
+}
+
+} // namespace v0 {
+
+
+namespace v1 {
+
+template <typename T>
+T devolveAndEvolve(const T& t)
+{
+ return mesos::csi::v1::evolve(mesos::csi::v1::devolve(t));
+}
+
+} // namespace v1 {
+} // namespace csi {
+
+
namespace mesos {
namespace internal {
namespace tests {
+template <typename T>
+class CSIUtilsTest : public testing::Test {};
+
+
+using VolumeCapabilityTypes = testing::Types<
+ csi::v0::VolumeCapability,
+ csi::v1::VolumeCapability>;
+
+
+TYPED_TEST_CASE(CSIUtilsTest, VolumeCapabilityTypes);
+
+
// This test verifies that a versioned CSI `VolumeCapability` protobuf can be
// devolved to an unversioned protobuf then be evolved back to the same one.
-//
-// TODO(chhsiao): Parameterize this test with CSI versions.
-TEST(CsiUtilsTest, VolumeCapabilityEvolve)
+TYPED_TEST(CSIUtilsTest, DevolveAndEvolve)
{
// The following JSON examples contains both valid and invalid CSI volume
// capabilities. However, no matter if the capability is valid, they should be
@@ -155,10 +194,9 @@ TEST(CsiUtilsTest, VolumeCapabilityEvolve)
foreach (const string& example, examples) {
// NOTE: We use Google's JSON utility functions for proto3.
- csi::v0::VolumeCapability versioned;
+ TypeParam versioned;
ASSERT_EQ(util::Status::OK, util::JsonStringToMessage(example, &versioned));
- EXPECT_EQ(versioned, csi::v0::evolve(csi::v0::devolve(versioned)))
- << example;
+ EXPECT_EQ(versioned, devolveAndEvolve(versioned)) << example;
}
}
[mesos] 03/06: Updated CSI volume state fields to match names in v1
spec.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch 1.8.x
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 275059a566be1cc8daecb998659767cf7572f588
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Apr 4 22:44:34 2019 -0700
Updated CSI volume state fields to match names in v1 spec.
We rename `volume_attributes` to `volume_context` and `publish_info` to
`publish_context` in `state.proto` to match the new CSI v1 standard to
avoid confusions. Since this protobuf message is only used for
checkpointing and the binary format does not change, there will be no
compatibility issue.
Review: https://reviews.apache.org/r/70401
---
src/csi/state.proto | 13 +++++++------
src/csi/v0_volume_manager.cpp | 18 +++++++++---------
src/csi/v0_volume_manager_process.hpp | 6 +++---
3 files changed, 19 insertions(+), 18 deletions(-)
diff --git a/src/csi/state.proto b/src/csi/state.proto
index b7213ed..96ba420 100644
--- a/src/csi/state.proto
+++ b/src/csi/state.proto
@@ -50,14 +50,15 @@ message VolumeState {
map<string, string> parameters = 6;
// Attributes of the volume to be used on the node. This field MUST match the
- // attributes of the `Volume` returned by `CreateVolume`. This is an OPTIONAL
- // field.
- map<string, string> volume_attributes = 3;
+ // `volume_context` field of the `Volume` returned by `CreateVolume`. This is
+ // an OPTIONAL field.
+ map<string, string> volume_context = 3;
// If the plugin has the `PUBLISH_UNPUBLISH_VOLUME` controller capability,
- // this field MUST be set to the value returned by `ControllerPublishVolume`.
- // Otherwise, this field MUST remain unset. This is an OPTIONAL field.
- map<string, string> publish_info = 4;
+ // this field MUST be set to the `publish_context` field returned by
+ // `ControllerPublishVolume`. Otherwise, this field MUST remain unset. This is
+ // an OPTIONAL field.
+ map<string, string> publish_context = 4;
// This field is used to check if the node has been rebooted since the volume
// is made publishable on the node. It MUST be set to the boot ID of the node
diff --git a/src/csi/v0_volume_manager.cpp b/src/csi/v0_volume_manager.cpp
index 7de9000..02de17e 100644
--- a/src/csi/v0_volume_manager.cpp
+++ b/src/csi/v0_volume_manager.cpp
@@ -326,7 +326,7 @@ Future<VolumeInfo> VolumeManagerProcess::createVolume(
volumeState.set_state(VolumeState::CREATED);
*volumeState.mutable_volume_capability() = capability;
*volumeState.mutable_parameters() = parameters;
- *volumeState.mutable_volume_attributes() = response.volume().attributes();
+ *volumeState.mutable_volume_context() = response.volume().attributes();
volumes.put(volumeId, std::move(volumeState));
checkpointVolumeState(volumeId);
@@ -397,7 +397,7 @@ Future<Option<Error>> VolumeManagerProcess::validateVolume(
volumeState.set_state(VolumeState::CREATED);
*volumeState.mutable_volume_capability() = capability;
*volumeState.mutable_parameters() = parameters;
- *volumeState.mutable_volume_attributes() = volumeInfo.context;
+ *volumeState.mutable_volume_context() = volumeInfo.context;
volumes.put(volumeInfo.id, std::move(volumeState));
checkpointVolumeState(volumeInfo.id);
@@ -820,7 +820,7 @@ Future<Nothing> VolumeManagerProcess::_attachVolume(const string& volumeId)
*request.mutable_volume_capability() =
evolve(volumeState.volume_capability());
request.set_readonly(false);
- *request.mutable_volume_attributes() = volumeState.volume_attributes();
+ *request.mutable_volume_attributes() = volumeState.volume_context();
return call(
CONTROLLER_SERVICE, &Client::controllerPublishVolume, std::move(request))
@@ -829,7 +829,7 @@ Future<Nothing> VolumeManagerProcess::_attachVolume(const string& volumeId)
CHECK(volumes.contains(volumeId));
VolumeState& volumeState = volumes.at(volumeId).state;
volumeState.set_state(VolumeState::NODE_READY);
- *volumeState.mutable_publish_info() = response.publish_info();
+ *volumeState.mutable_publish_context() = response.publish_info();
checkpointVolumeState(volumeId);
@@ -886,7 +886,7 @@ Future<Nothing> VolumeManagerProcess::_detachVolume(const string& volumeId)
CHECK(volumes.contains(volumeId));
VolumeState& volumeState = volumes.at(volumeId).state;
volumeState.set_state(VolumeState::CREATED);
- volumeState.mutable_publish_info()->clear();
+ volumeState.mutable_publish_context()->clear();
checkpointVolumeState(volumeId);
@@ -943,12 +943,12 @@ Future<Nothing> VolumeManagerProcess::_publishVolume(const string& volumeId)
NodePublishVolumeRequest request;
request.set_volume_id(volumeId);
- *request.mutable_publish_info() = volumeState.publish_info();
+ *request.mutable_publish_info() = volumeState.publish_context();
request.set_target_path(targetPath);
*request.mutable_volume_capability() =
evolve(volumeState.volume_capability());
request.set_readonly(false);
- *request.mutable_volume_attributes() = volumeState.volume_attributes();
+ *request.mutable_volume_attributes() = volumeState.volume_context();
if (nodeCapabilities->stageUnstageVolume) {
const string stagingPath = paths::getMountStagingPath(
@@ -1033,11 +1033,11 @@ Future<Nothing> VolumeManagerProcess::__publishVolume(const string& volumeId)
NodeStageVolumeRequest request;
request.set_volume_id(volumeId);
- *request.mutable_publish_info() = volumeState.publish_info();
+ *request.mutable_publish_info() = volumeState.publish_context();
request.set_staging_target_path(stagingPath);
*request.mutable_volume_capability() =
evolve(volumeState.volume_capability());
- *request.mutable_volume_attributes() = volumeState.volume_attributes();
+ *request.mutable_volume_attributes() = volumeState.volume_context();
return call(NODE_SERVICE, &Client::nodeStageVolume, std::move(request))
.then(process::defer(self(), [this, volumeId] {
diff --git a/src/csi/v0_volume_manager_process.hpp b/src/csi/v0_volume_manager_process.hpp
index 88073e4..c3cd6ca 100644
--- a/src/csi/v0_volume_manager_process.hpp
+++ b/src/csi/v0_volume_manager_process.hpp
@@ -14,8 +14,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
-#ifndef __CSI_VOLUME_MANAGER_PROCESS_HPP__
-#define __CSI_VOLUME_MANAGER_PROCESS_HPP__
+#ifndef __CSI_V0_VOLUME_MANAGER_PROCESS_HPP__
+#define __CSI_V0_VOLUME_MANAGER_PROCESS_HPP__
#include <string>
#include <vector>
@@ -215,4 +215,4 @@ private:
} // namespace csi {
} // namespace mesos {
-#endif // __CSI_VOLUME_MANAGER_PROCESS_HPP__
+#endif // __CSI_V0_VOLUME_MANAGER_PROCESS_HPP__
[mesos] 04/06: Implemented CSI v1 volume manager.
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chhsiao pushed a commit to branch 1.8.x
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit a3c2632f87d1ccd5275dca073f87287fe09a370b
Author: Chun-Hung Hsiao <ch...@mesosphere.io>
AuthorDate: Thu Apr 4 22:47:49 2019 -0700
Implemented CSI v1 volume manager.
The v1 Volume manager is basically copied from the v0 volume manager,
with the following changes:
* `validateVolume` will validate volumes against parameters.
* `publishVolume` will only create the parent directory of the target
path. The creation of the target path itself is the responsibility
of the plugin.
* The node ID is obtained through the `NodeGetInfo` call.
Review: https://reviews.apache.org/r/70402
---
src/CMakeLists.txt | 1 +
src/Makefile.am | 3 +
src/csi/v1_volume_manager.cpp | 1356 +++++++++++++++++++++++++++++++++
src/csi/v1_volume_manager.hpp | 116 +++
src/csi/v1_volume_manager_process.hpp | 218 ++++++
5 files changed, 1694 insertions(+)
diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt
index d36d0be..1d4f541 100644
--- a/src/CMakeLists.txt
+++ b/src/CMakeLists.txt
@@ -244,6 +244,7 @@ set(CSI_SRC
csi/v1.cpp
csi/v1_client.cpp
csi/v1_utils.cpp
+ csi/v1_volume_manager.cpp
csi/volume_manager.cpp)
set(DOCKER_SRC
diff --git a/src/Makefile.am b/src/Makefile.am
index 64898df..5f97523 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1600,6 +1600,9 @@ libcsi_la_SOURCES = \
csi/v1_client.hpp \
csi/v1_utils.cpp \
csi/v1_utils.hpp \
+ csi/v1_volume_manager.cpp \
+ csi/v1_volume_manager.hpp \
+ csi/v1_volume_manager_process.hpp \
csi/volume_manager.cpp \
csi/volume_manager.hpp
diff --git a/src/csi/v1_volume_manager.cpp b/src/csi/v1_volume_manager.cpp
new file mode 100644
index 0000000..bd334f1
--- /dev/null
+++ b/src/csi/v1_volume_manager.cpp
@@ -0,0 +1,1356 @@
+// 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 "csi/v1_volume_manager.hpp"
+
+#include <algorithm>
+#include <cstdlib>
+#include <functional>
+#include <list>
+
+#include <process/after.hpp>
+#include <process/collect.hpp>
+#include <process/defer.hpp>
+#include <process/dispatch.hpp>
+#include <process/id.hpp>
+#include <process/loop.hpp>
+#include <process/process.hpp>
+
+#include <stout/check.hpp>
+#include <stout/duration.hpp>
+#include <stout/foreach.hpp>
+#include <stout/none.hpp>
+#include <stout/os.hpp>
+#include <stout/path.hpp>
+#include <stout/result.hpp>
+#include <stout/some.hpp>
+#include <stout/stringify.hpp>
+#include <stout/try.hpp>
+#include <stout/unreachable.hpp>
+
+#include "csi/paths.hpp"
+#include "csi/v1_client.hpp"
+#include "csi/v1_utils.hpp"
+#include "csi/v1_volume_manager_process.hpp"
+
+#include "slave/state.hpp"
+
+namespace http = process::http;
+namespace slave = mesos::internal::slave;
+
+using std::list;
+using std::string;
+using std::vector;
+
+using google::protobuf::Map;
+
+using mesos::csi::state::VolumeState;
+
+using process::Break;
+using process::Continue;
+using process::ControlFlow;
+using process::Failure;
+using process::Future;
+using process::ProcessBase;
+
+using process::grpc::StatusError;
+
+using process::grpc::client::Runtime;
+
+namespace mesos{
+namespace csi {
+namespace v1 {
+
+VolumeManagerProcess::VolumeManagerProcess(
+ const http::URL& agentUrl,
+ const string& _rootDir,
+ const CSIPluginInfo& _info,
+ const hashset<Service> _services,
+ const string& containerPrefix,
+ const Option<string>& authToken,
+ const Runtime& _runtime,
+ Metrics* _metrics)
+ : ProcessBase(process::ID::generate("csi-v1-volume-manager")),
+ rootDir(_rootDir),
+ info(_info),
+ services(_services),
+ runtime(_runtime),
+ metrics(_metrics),
+ serviceManager(new ServiceManager(
+ agentUrl,
+ rootDir,
+ info,
+ services,
+ containerPrefix,
+ authToken,
+ runtime,
+ metrics))
+{
+ // This should have been validated in `VolumeManager::create`.
+ CHECK(!services.empty())
+ << "Must specify at least one service for CSI plugin type '" << info.type()
+ << "' and name '" << info.name() << "'";
+}
+
+
+Future<Nothing> VolumeManagerProcess::recover()
+{
+ Try<string> bootId_ = os::bootId();
+ if (bootId_.isError()) {
+ return Failure("Failed to get boot ID: " + bootId_.error());
+ }
+
+ bootId = bootId_.get();
+
+ return serviceManager->recover()
+ .then(process::defer(self(), &Self::prepareServices))
+ .then(process::defer(self(), [this]() -> Future<Nothing> {
+ // Recover the states of CSI volumes.
+ Try<list<string>> volumePaths =
+ paths::getVolumePaths(rootDir, info.type(), info.name());
+
+ if (volumePaths.isError()) {
+ return Failure(
+ "Failed to find volumes for CSI plugin type '" + info.type() +
+ "' and name '" + info.name() + "': " + volumePaths.error());
+ }
+
+ vector<Future<Nothing>> futures;
+
+ foreach (const string& path, volumePaths.get()) {
+ Try<paths::VolumePath> volumePath =
+ paths::parseVolumePath(rootDir, path);
+
+ if (volumePath.isError()) {
+ return Failure(
+ "Failed to parse volume path '" + path +
+ "': " + volumePath.error());
+ }
+
+ CHECK_EQ(info.type(), volumePath->type);
+ CHECK_EQ(info.name(), volumePath->name);
+
+ const string& volumeId = volumePath->volumeId;
+ const string statePath = paths::getVolumeStatePath(
+ rootDir, info.type(), info.name(), volumeId);
+
+ if (!os::exists(statePath)) {
+ continue;
+ }
+
+ Result<VolumeState> volumeState =
+ slave::state::read<VolumeState>(statePath);
+
+ if (volumeState.isError()) {
+ return Failure(
+ "Failed to read volume state from '" + statePath +
+ "': " + volumeState.error());
+ }
+
+ if (volumeState.isNone()) {
+ continue;
+ }
+
+ volumes.put(volumeId, std::move(volumeState.get()));
+ VolumeData& volume = volumes.at(volumeId);
+
+ if (!VolumeState::State_IsValid(volume.state.state())) {
+ return Failure("Volume '" + volumeId + "' is in INVALID state");
+ }
+
+ // First, if there is a node reboot after the volume is made
+ // publishable, it should be reset to `NODE_READY`.
+ switch (volume.state.state()) {
+ case VolumeState::CREATED:
+ case VolumeState::NODE_READY:
+ case VolumeState::CONTROLLER_PUBLISH:
+ case VolumeState::CONTROLLER_UNPUBLISH:
+ case VolumeState::NODE_STAGE: {
+ break;
+ }
+ case VolumeState::VOL_READY:
+ case VolumeState::PUBLISHED:
+ case VolumeState::NODE_UNSTAGE:
+ case VolumeState::NODE_PUBLISH:
+ case VolumeState::NODE_UNPUBLISH: {
+ if (bootId != volume.state.boot_id()) {
+ // Since this is a no-op, no need to checkpoint here.
+ volume.state.set_state(VolumeState::NODE_READY);
+ volume.state.clear_boot_id();
+ }
+
+ break;
+ }
+ case VolumeState::UNKNOWN: {
+ return Failure("Volume '" + volumeId + "' is in UNKNOWN state");
+ }
+
+ // NOTE: We avoid using a default clause for the following values in
+ // proto3's open enum to enable the compiler to detect missing enum
+ // cases for us. See: https://github.com/google/protobuf/issues/3917
+ case google::protobuf::kint32min:
+ case google::protobuf::kint32max: {
+ UNREACHABLE();
+ }
+ }
+
+ // Second, if the volume has been used by a container before recovery,
+ // we have to bring the volume back to `PUBLISHED` so data can be
+ // cleaned up synchronously when needed.
+ if (volume.state.node_publish_required()) {
+ futures.push_back(publishVolume(volumeId));
+ }
+ }
+
+ // Garbage collect leftover mount paths that were failed to remove before.
+ const string mountRootDir =
+ paths::getMountRootDir(rootDir, info.type(), info.name());
+
+ Try<list<string>> mountPaths = paths::getMountPaths(mountRootDir);
+ if (mountPaths.isError()) {
+ // TODO(chhsiao): This could indicate that something is seriously wrong.
+ // To help debugging the problem, we should surface the error via
+ // MESOS-8745.
+ return Failure(
+ "Failed to find mount paths for CSI plugin type '" + info.type() +
+ "' and name '" + info.name() + "': " + mountPaths.error());
+ }
+
+ foreach (const string& path, mountPaths.get()) {
+ Try<string> volumeId = paths::parseMountPath(mountRootDir, path);
+ if (volumeId.isError()) {
+ return Failure(
+ "Failed to parse mount path '" + path + "': " + volumeId.error());
+ }
+
+ if (!volumes.contains(volumeId.get())) {
+ garbageCollectMountPath(volumeId.get());
+ }
+ }
+
+ return process::collect(futures).then([] { return Nothing(); });
+ }));
+}
+
+
+Future<vector<VolumeInfo>> VolumeManagerProcess::listVolumes()
+{
+ if (!controllerCapabilities->listVolumes) {
+ return vector<VolumeInfo>();
+ }
+
+ // TODO(chhsiao): Set the max entries and use a loop to do multiple
+ // `ListVolumes` calls.
+ return call(CONTROLLER_SERVICE, &Client::listVolumes, ListVolumesRequest())
+ .then(process::defer(self(), [](const ListVolumesResponse& response) {
+ vector<VolumeInfo> result;
+ foreach (const auto& entry, response.entries()) {
+ result.push_back(VolumeInfo{Bytes(entry.volume().capacity_bytes()),
+ entry.volume().volume_id(),
+ entry.volume().volume_context()});
+ }
+
+ return result;
+ }));
+}
+
+
+Future<Bytes> VolumeManagerProcess::getCapacity(
+ const types::VolumeCapability& capability,
+ const Map<string, string>& parameters)
+{
+ if (!controllerCapabilities->getCapacity) {
+ return Bytes(0);
+ }
+
+ GetCapacityRequest request;
+ *request.add_volume_capabilities() = evolve(capability);
+ *request.mutable_parameters() = parameters;
+
+ return call(CONTROLLER_SERVICE, &Client::getCapacity, std::move(request))
+ .then([](const GetCapacityResponse& response) {
+ return Bytes(response.available_capacity());
+ });
+}
+
+
+Future<VolumeInfo> VolumeManagerProcess::createVolume(
+ const string& name,
+ const Bytes& capacity,
+ const types::VolumeCapability& capability,
+ const Map<string, string>& parameters)
+{
+ if (!controllerCapabilities->createDeleteVolume) {
+ return Failure(
+ "CREATE_DELETE_VOLUME controller capability is not supported for CSI "
+ "plugin type '" + info.type() + "' and name '" + info.name());
+ }
+
+ LOG(INFO) << "Creating volume with name '" << name << "'";
+
+ CreateVolumeRequest request;
+ request.set_name(name);
+ request.mutable_capacity_range()->set_required_bytes(capacity.bytes());
+ request.mutable_capacity_range()->set_limit_bytes(capacity.bytes());
+ *request.add_volume_capabilities() = evolve(capability);
+ *request.mutable_parameters() = parameters;
+
+ // We retry the `CreateVolume` call for MESOS-9517.
+ return call(
+ CONTROLLER_SERVICE, &Client::createVolume, std::move(request), true)
+ .then(process::defer(self(), [=](
+ const CreateVolumeResponse& response) -> Future<VolumeInfo> {
+ const string& volumeId = response.volume().volume_id();
+
+ // NOTE: If the volume is already tracked, there might already be
+ // operations running in its sequence. Since this continuation runs
+ // outside the sequence, we fail the call here to avoid any race issue.
+ // This also means that this call is not idempotent.
+ if (volumes.contains(volumeId)) {
+ return Failure("Volume with name '" + name + "' already exists");
+ }
+
+ VolumeState volumeState;
+ volumeState.set_state(VolumeState::CREATED);
+ *volumeState.mutable_volume_capability() = capability;
+ *volumeState.mutable_parameters() = parameters;
+ *volumeState.mutable_volume_context() =
+ response.volume().volume_context();
+
+ volumes.put(volumeId, std::move(volumeState));
+ checkpointVolumeState(volumeId);
+
+ return VolumeInfo{capacity, volumeId, response.volume().volume_context()};
+ }));
+}
+
+
+Future<Option<Error>> VolumeManagerProcess::validateVolume(
+ const VolumeInfo& volumeInfo,
+ const types::VolumeCapability& capability,
+ const Map<string, string>& parameters)
+{
+ // If the volume has been checkpointed, the validation succeeds only if the
+ // capability and parameters of the specified profile are the same as those in
+ // the checkpoint.
+ if (volumes.contains(volumeInfo.id)) {
+ const VolumeState& volumeState = volumes.at(volumeInfo.id).state;
+
+ if (volumeState.volume_capability() != capability) {
+ return Some(Error(
+ "Unsupported volume capability for volume '" + volumeInfo.id + "'"));
+ }
+
+ if (volumeState.parameters() != parameters) {
+ return Some(Error(
+ "Mismatched parameters for volume '" + volumeInfo.id + "'"));
+ }
+
+ return None();
+ }
+
+ LOG(INFO) << "Validating volume '" << volumeInfo.id << "'";
+
+ ValidateVolumeCapabilitiesRequest request;
+ request.set_volume_id(volumeInfo.id);
+ *request.add_volume_capabilities() = evolve(capability);
+ *request.mutable_volume_context() = volumeInfo.context;
+ *request.mutable_parameters() = parameters;
+
+ return call(
+ CONTROLLER_SERVICE,
+ &Client::validateVolumeCapabilities,
+ std::move(request))
+ .then(process::defer(self(), [=](
+ const ValidateVolumeCapabilitiesResponse& response)
+ -> Future<Option<Error>> {
+ if (!response.has_confirmed()) {
+ return Error(
+ "Validation failed for volume '" + volumeInfo.id + "': " +
+ response.message());
+ }
+
+ if (response.confirmed().volume_context() != volumeInfo.context) {
+ return Error(
+ "Validation failed for volume '" + volumeInfo.id +
+ "': Mismatched volume context");
+ }
+
+ if (std::none_of(
+ response.confirmed().volume_capabilities().begin(),
+ response.confirmed().volume_capabilities().end(),
+ [&](const csi::v1::VolumeCapability& _capability) {
+ return csi::v1::devolve(_capability) == capability;
+ })) {
+ return Error(
+ "Validation failed for volume '" + volumeInfo.id +
+ "': Unsupported volume capability");
+ }
+
+ if (response.confirmed().parameters() != parameters) {
+ return Error(
+ "Validation failed for volume '" + volumeInfo.id +
+ "': Mismatched parameters");
+ }
+
+ // NOTE: If the volume is already tracked, there might already be
+ // operations running in its sequence. Since this continuation runs
+ // outside the sequence, we fail the call here to avoid any race issue.
+ // This also means that this call is not idempotent.
+ if (volumes.contains(volumeInfo.id)) {
+ return Failure("Volume '" + volumeInfo.id + "' already validated");
+ }
+
+ VolumeState volumeState;
+ volumeState.set_state(VolumeState::CREATED);
+ *volumeState.mutable_volume_capability() = capability;
+ *volumeState.mutable_parameters() = parameters;
+ *volumeState.mutable_volume_context() = volumeInfo.context;
+
+ volumes.put(volumeInfo.id, std::move(volumeState));
+ checkpointVolumeState(volumeInfo.id);
+
+ return None();
+ }));
+}
+
+
+Future<bool> VolumeManagerProcess::deleteVolume(const string& volumeId)
+{
+ if (!volumes.contains(volumeId)) {
+ return __deleteVolume(volumeId);
+ }
+
+ VolumeData& volume = volumes.at(volumeId);
+
+ LOG(INFO) << "Deleting volume '" << volumeId << "' in "
+ << volume.state.state() << " state";
+
+ // Volume deletion is sequentialized with other operations on the same volume
+ // to avoid races.
+ return volume.sequence->add(std::function<Future<bool>()>(
+ process::defer(self(), &Self::_deleteVolume, volumeId)));
+}
+
+
+Future<Nothing> VolumeManagerProcess::attachVolume(const string& volumeId)
+{
+ if (!volumes.contains(volumeId)) {
+ return Failure("Cannot attach unknown volume '" + volumeId + "'");
+ }
+
+ VolumeData& volume = volumes.at(volumeId);
+
+ LOG(INFO) << "Attaching volume '" << volumeId << "' in "
+ << volume.state.state() << " state";
+
+ // Volume attaching is serialized with other operations on the same volume to
+ // avoid races.
+ return volume.sequence->add(std::function<Future<Nothing>()>(
+ process::defer(self(), &Self::_attachVolume, volumeId)));
+}
+
+
+Future<Nothing> VolumeManagerProcess::detachVolume(const string& volumeId)
+{
+ if (!volumes.contains(volumeId)) {
+ return Failure("Cannot detach unknown volume '" + volumeId + "'");
+ }
+
+ VolumeData& volume = volumes.at(volumeId);
+
+ LOG(INFO) << "Detaching volume '" << volumeId << "' in "
+ << volume.state.state() << " state";
+
+ // Volume detaching is serialized with other operations on the same volume to
+ // avoid races.
+ return volume.sequence->add(std::function<Future<Nothing>()>(
+ process::defer(self(), &Self::_detachVolume, volumeId)));
+}
+
+
+Future<Nothing> VolumeManagerProcess::publishVolume(const string& volumeId)
+{
+ if (!volumes.contains(volumeId)) {
+ return Failure("Cannot publish unknown volume '" + volumeId + "'");
+ }
+
+ VolumeData& volume = volumes.at(volumeId);
+
+ LOG(INFO) << "Publishing volume '" << volumeId << "' in "
+ << volume.state.state() << " state";
+
+ // Volume publishing is serialized with other operations on the same volume to
+ // avoid races.
+ return volume.sequence->add(std::function<Future<Nothing>()>(
+ process::defer(self(), &Self::_publishVolume, volumeId)));
+}
+
+
+Future<Nothing> VolumeManagerProcess::unpublishVolume(const string& volumeId)
+{
+ if (!volumes.contains(volumeId)) {
+ return Failure("Cannot unpublish unknown volume '" + volumeId + "'");
+ }
+
+ VolumeData& volume = volumes.at(volumeId);
+
+ LOG(INFO) << "Unpublishing volume '" << volumeId << "' in "
+ << volume.state.state() << " state";
+
+ // Volume unpublishing is serialized with other operations on the same volume
+ // to avoid races.
+ return volume.sequence->add(std::function<Future<Nothing>()>(
+ process::defer(self(), &Self::_unpublishVolume, volumeId)));
+}
+
+
+template <typename Request, typename Response>
+Future<Response> VolumeManagerProcess::call(
+ const Service& service,
+ Future<RPCResult<Response>> (Client::*rpc)(Request),
+ const Request& request,
+ const bool retry) // Made immutable in the following mutable lambda.
+{
+ Duration maxBackoff = DEFAULT_CSI_RETRY_BACKOFF_FACTOR;
+
+ return process::loop(
+ self(),
+ [=] {
+ // Make the call to the latest service endpoint.
+ return serviceManager->getServiceEndpoint(service)
+ .then(process::defer(
+ self(),
+ &VolumeManagerProcess::_call<Request, Response>,
+ lambda::_1,
+ rpc,
+ request));
+ },
+ [=](const RPCResult<Response>& result) mutable
+ -> Future<ControlFlow<Response>> {
+ Option<Duration> backoff = retry
+ ? maxBackoff * (static_cast<double>(os::random()) / RAND_MAX)
+ : Option<Duration>::none();
+
+ maxBackoff = std::min(maxBackoff * 2, DEFAULT_CSI_RETRY_INTERVAL_MAX);
+
+ // We dispatch `__call` for testing purpose.
+ return process::dispatch(
+ self(), &VolumeManagerProcess::__call<Response>, result, backoff);
+ });
+}
+
+
+template <typename Request, typename Response>
+Future<RPCResult<Response>> VolumeManagerProcess::_call(
+ const string& endpoint,
+ Future<RPCResult<Response>> (Client::*rpc)(Request),
+ const Request& request)
+{
+ ++metrics->csi_plugin_rpcs_pending;
+
+ return (Client(endpoint, runtime).*rpc)(request).onAny(
+ process::defer(self(), [=](const Future<RPCResult<Response>>& future) {
+ --metrics->csi_plugin_rpcs_pending;
+ if (future.isReady() && future->isSome()) {
+ ++metrics->csi_plugin_rpcs_finished;
+ } else if (future.isDiscarded()) {
+ ++metrics->csi_plugin_rpcs_cancelled;
+ } else {
+ ++metrics->csi_plugin_rpcs_failed;
+ }
+ }));
+}
+
+
+template <typename Response>
+Future<ControlFlow<Response>> VolumeManagerProcess::__call(
+ const RPCResult<Response>& result, const Option<Duration>& backoff)
+{
+ if (result.isSome()) {
+ return Break(result.get());
+ }
+
+ if (backoff.isNone()) {
+ return Failure(result.error());
+ }
+
+ // See the link below for retryable status codes:
+ // https://grpc.io/grpc/cpp/namespacegrpc.html#aff1730578c90160528f6a8d67ef5c43b // NOLINT
+ switch (result.error().status.error_code()) {
+ case grpc::DEADLINE_EXCEEDED:
+ case grpc::UNAVAILABLE: {
+ LOG(ERROR) << "Received '" << result.error() << "' while expecting "
+ << Response::descriptor()->name() << ". Retrying in "
+ << backoff.get();
+
+ return process::after(backoff.get())
+ .then([]() -> Future<ControlFlow<Response>> { return Continue(); });
+ }
+ case grpc::CANCELLED:
+ case grpc::UNKNOWN:
+ case grpc::INVALID_ARGUMENT:
+ case grpc::NOT_FOUND:
+ case grpc::ALREADY_EXISTS:
+ case grpc::PERMISSION_DENIED:
+ case grpc::UNAUTHENTICATED:
+ case grpc::RESOURCE_EXHAUSTED:
+ case grpc::FAILED_PRECONDITION:
+ case grpc::ABORTED:
+ case grpc::OUT_OF_RANGE:
+ case grpc::UNIMPLEMENTED:
+ case grpc::INTERNAL:
+ case grpc::DATA_LOSS: {
+ return Failure(result.error());
+ }
+ case grpc::OK:
+ case grpc::DO_NOT_USE: {
+ UNREACHABLE();
+ }
+ }
+
+ UNREACHABLE();
+}
+
+
+Future<Nothing> VolumeManagerProcess::prepareServices()
+{
+ CHECK(!services.empty());
+
+ // Get the plugin capabilities.
+ return call(
+ *services.begin(),
+ &Client::getPluginCapabilities,
+ GetPluginCapabilitiesRequest())
+ .then(process::defer(self(), [=](
+ const GetPluginCapabilitiesResponse& response) -> Future<Nothing> {
+ pluginCapabilities = response.capabilities();
+
+ if (services.contains(CONTROLLER_SERVICE) &&
+ !pluginCapabilities->controllerService) {
+ return Failure(
+ "CONTROLLER_SERVICE plugin capability is not supported for CSI "
+ "plugin type '" + info.type() + "' and name '" + info.name() + "'");
+ }
+
+ return Nothing();
+ }))
+ // Check if all services have consistent plugin infos.
+ .then(process::defer(self(), [this] {
+ vector<Future<GetPluginInfoResponse>> futures;
+ foreach (const Service& service, services) {
+ futures.push_back(call(
+ CONTROLLER_SERVICE, &Client::getPluginInfo, GetPluginInfoRequest())
+ .onReady([service](const GetPluginInfoResponse& response) {
+ LOG(INFO) << service << " loaded: " << stringify(response);
+ }));
+ }
+
+ return process::collect(futures)
+ .then([](const vector<GetPluginInfoResponse>& pluginInfos) {
+ for (size_t i = 1; i < pluginInfos.size(); ++i) {
+ if (pluginInfos[i].name() != pluginInfos[0].name() ||
+ pluginInfos[i].vendor_version() !=
+ pluginInfos[0].vendor_version()) {
+ LOG(WARNING) << "Inconsistent plugin services. Please check with "
+ "the plugin vendor to ensure compatibility.";
+ }
+ }
+
+ return Nothing();
+ });
+ }))
+ // Get the controller capabilities.
+ .then(process::defer(self(), [this]() -> Future<Nothing> {
+ if (!services.contains(CONTROLLER_SERVICE)) {
+ controllerCapabilities = ControllerCapabilities();
+ return Nothing();
+ }
+
+ return call(
+ CONTROLLER_SERVICE,
+ &Client::controllerGetCapabilities,
+ ControllerGetCapabilitiesRequest())
+ .then(process::defer(self(), [this](
+ const ControllerGetCapabilitiesResponse& response) {
+ controllerCapabilities = response.capabilities();
+ return Nothing();
+ }));
+ }))
+ // Get the node capabilities and ID.
+ .then(process::defer(self(), [this]() -> Future<Nothing> {
+ if (!services.contains(NODE_SERVICE)) {
+ nodeCapabilities = NodeCapabilities();
+ return Nothing();
+ }
+
+ return call(
+ NODE_SERVICE,
+ &Client::nodeGetCapabilities,
+ NodeGetCapabilitiesRequest())
+ .then(process::defer(self(), [this](
+ const NodeGetCapabilitiesResponse& response) -> Future<Nothing> {
+ nodeCapabilities = response.capabilities();
+
+ if (controllerCapabilities->publishUnpublishVolume) {
+ return call(
+ NODE_SERVICE, &Client::nodeGetInfo, NodeGetInfoRequest())
+ .then(process::defer(self(), [this](
+ const NodeGetInfoResponse& response) {
+ nodeId = response.node_id();
+ return Nothing();
+ }));
+ }
+
+ return Nothing();
+ }));
+ }));
+}
+
+
+Future<bool> VolumeManagerProcess::_deleteVolume(const std::string& volumeId)
+{
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+
+ if (volumeState.node_publish_required()) {
+ CHECK_EQ(VolumeState::PUBLISHED, volumeState.state());
+
+ const string targetPath = paths::getMountTargetPath(
+ paths::getMountRootDir(rootDir, info.type(), info.name()), volumeId);
+
+ // NOTE: Normally the volume should have been cleaned up. However this may
+ // not be true for preprovisioned volumes (e.g., leftover from a previous
+ // resource provider instance). To prevent data leakage in such cases, we
+ // clean up the data (but not the target path) here.
+ Try<Nothing> rmdir = os::rmdir(targetPath, true, false);
+ if (rmdir.isError()) {
+ return Failure(
+ "Failed to clean up volume '" + volumeId + "': " + rmdir.error());
+ }
+
+ volumeState.set_node_publish_required(false);
+ checkpointVolumeState(volumeId);
+ }
+
+ if (volumeState.state() != VolumeState::CREATED) {
+ // Retry after transitioning the volume to `CREATED` state.
+ return _detachVolume(volumeId)
+ .then(process::defer(self(), &Self::_deleteVolume, volumeId));
+ }
+
+ // NOTE: The last asynchronous continuation, which is supposed to be run in
+ // the volume's sequence, would cause the sequence to be destructed, which
+ // would in turn discard the returned future. However, since the continuation
+ // would have already been run, the returned future will become ready, making
+ // the future returned by the sequence ready as well.
+ return __deleteVolume(volumeId)
+ .then(process::defer(self(), [this, volumeId](bool deleted) {
+ volumes.erase(volumeId);
+
+ const string volumePath =
+ paths::getVolumePath(rootDir, info.type(), info.name(), volumeId);
+
+ Try<Nothing> rmdir = os::rmdir(volumePath);
+ CHECK_SOME(rmdir) << "Failed to remove checkpointed volume state at '"
+ << volumePath << "': " << rmdir.error();
+
+ garbageCollectMountPath(volumeId);
+
+ return deleted;
+ }));
+}
+
+
+Future<bool> VolumeManagerProcess::__deleteVolume(
+ const string& volumeId)
+{
+ if (!controllerCapabilities->createDeleteVolume) {
+ return false;
+ }
+
+ LOG(INFO) << "Calling '/csi.v1.Controller/DeleteVolume' for volume '"
+ << volumeId << "'";
+
+ DeleteVolumeRequest request;
+ request.set_volume_id(volumeId);
+
+ // We retry the `DeleteVolume` call for MESOS-9517.
+ return call(
+ CONTROLLER_SERVICE, &Client::deleteVolume, std::move(request), true)
+ .then([] { return true; });
+}
+
+
+Future<Nothing> VolumeManagerProcess::_attachVolume(const string& volumeId)
+{
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+
+ if (volumeState.state() == VolumeState::NODE_READY) {
+ return Nothing();
+ }
+
+ if (volumeState.state() != VolumeState::CREATED &&
+ volumeState.state() != VolumeState::CONTROLLER_PUBLISH &&
+ volumeState.state() != VolumeState::CONTROLLER_UNPUBLISH) {
+ return Failure(
+ "Cannot attach volume '" + volumeId + "' in " +
+ stringify(volumeState.state()) + " state");
+ }
+
+ if (!controllerCapabilities->publishUnpublishVolume) {
+ // Since this is a no-op, no need to checkpoint here.
+ volumeState.set_state(VolumeState::NODE_READY);
+ return Nothing();
+ }
+
+ // A previously failed `ControllerUnpublishVolume` call can be recovered
+ // through an extra `ControllerUnpublishVolume` call. See:
+ // https://github.com/container-storage-interface/spec/blob/v1.1.0/spec.md#controllerunpublishvolume // NOLINT
+ if (volumeState.state() == VolumeState::CONTROLLER_UNPUBLISH) {
+ // Retry after recovering the volume to `CREATED` state.
+ return _detachVolume(volumeId)
+ .then(process::defer(self(), &Self::_attachVolume, volumeId));
+ }
+
+ if (volumeState.state() == VolumeState::CREATED) {
+ volumeState.set_state(VolumeState::CONTROLLER_PUBLISH);
+ checkpointVolumeState(volumeId);
+ }
+
+ LOG(INFO)
+ << "Calling '/csi.v1.Controller/ControllerPublishVolume' for volume '"
+ << volumeId << "'";
+
+ ControllerPublishVolumeRequest request;
+ request.set_volume_id(volumeId);
+ request.set_node_id(CHECK_NOTNONE(nodeId));
+ *request.mutable_volume_capability() =
+ evolve(volumeState.volume_capability());
+ request.set_readonly(false);
+ *request.mutable_volume_context() = volumeState.volume_context();
+
+ return call(
+ CONTROLLER_SERVICE, &Client::controllerPublishVolume, std::move(request))
+ .then(process::defer(self(), [this, volumeId](
+ const ControllerPublishVolumeResponse& response) {
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+ volumeState.set_state(VolumeState::NODE_READY);
+ *volumeState.mutable_publish_context() = response.publish_context();
+
+ checkpointVolumeState(volumeId);
+
+ return Nothing();
+ }));
+}
+
+
+Future<Nothing> VolumeManagerProcess::_detachVolume(const string& volumeId)
+{
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+
+ if (volumeState.state() == VolumeState::CREATED) {
+ return Nothing();
+ }
+
+ if (volumeState.state() != VolumeState::NODE_READY &&
+ volumeState.state() != VolumeState::CONTROLLER_PUBLISH &&
+ volumeState.state() != VolumeState::CONTROLLER_UNPUBLISH) {
+ // Retry after transitioning the volume to `CREATED` state.
+ return _unpublishVolume(volumeId)
+ .then(process::defer(self(), &Self::_detachVolume, volumeId));
+ }
+
+ if (!controllerCapabilities->publishUnpublishVolume) {
+ // Since this is a no-op, no need to checkpoint here.
+ volumeState.set_state(VolumeState::CREATED);
+ return Nothing();
+ }
+
+ // A previously failed `ControllerPublishVolume` call can be recovered through
+ // the current `ControllerUnpublishVolume` call. See:
+ // https://github.com/container-storage-interface/spec/blob/v1.1.0/spec.md#controllerpublishvolume // NOLINT
+ if (volumeState.state() == VolumeState::NODE_READY ||
+ volumeState.state() == VolumeState::CONTROLLER_PUBLISH) {
+ volumeState.set_state(VolumeState::CONTROLLER_UNPUBLISH);
+ checkpointVolumeState(volumeId);
+ }
+
+ LOG(INFO)
+ << "Calling '/csi.v1.Controller/ControllerUnpublishVolume' for volume '"
+ << volumeId << "'";
+
+ ControllerUnpublishVolumeRequest request;
+ request.set_volume_id(volumeId);
+ request.set_node_id(CHECK_NOTNONE(nodeId));
+
+ return call(
+ CONTROLLER_SERVICE,
+ &Client::controllerUnpublishVolume,
+ std::move(request))
+ .then(process::defer(self(), [this, volumeId] {
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+ volumeState.set_state(VolumeState::CREATED);
+ volumeState.mutable_publish_context()->clear();
+
+ checkpointVolumeState(volumeId);
+
+ return Nothing();
+ }));
+}
+
+
+Future<Nothing> VolumeManagerProcess::_publishVolume(const string& volumeId)
+{
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+
+ if (volumeState.state() == VolumeState::PUBLISHED) {
+ CHECK(volumeState.node_publish_required());
+ return Nothing();
+ }
+
+ if (volumeState.state() != VolumeState::VOL_READY &&
+ volumeState.state() != VolumeState::NODE_PUBLISH &&
+ volumeState.state() != VolumeState::NODE_UNPUBLISH) {
+ // Retry after transitioning the volume to `VOL_READY` state.
+ return __publishVolume(volumeId)
+ .then(process::defer(self(), &Self::_publishVolume, volumeId));
+ }
+
+ // A previously failed `NodeUnpublishVolume` call can be recovered through an
+ // extra `NodeUnpublishVolume` call. See:
+ // https://github.com/container-storage-interface/spec/blob/v1.1.0/spec.md#nodeunpublishvolume // NOLINT
+ if (volumeState.state() == VolumeState::NODE_UNPUBLISH) {
+ // Retry after recovering the volume to `VOL_READY` state.
+ return __unpublishVolume(volumeId)
+ .then(process::defer(self(), &Self::_publishVolume, volumeId));
+ }
+
+ const string targetPath = paths::getMountTargetPath(
+ paths::getMountRootDir(rootDir, info.type(), info.name()), volumeId);
+
+ // Ensure the parent directory of the target path exists. The target path
+ // itself will be created by the plugin.
+ //
+ // NOTE: The target path will be removed by the plugin as well, and The parent
+ // directory of the target path will be cleaned up during volume removal.
+ Try<Nothing> mkdir = os::mkdir(Path(targetPath).dirname());
+ if (mkdir.isError()) {
+ return Failure(
+ "Failed to create parent directory of target path '" + targetPath +
+ "': " + mkdir.error());
+ }
+
+ if (volumeState.state() == VolumeState::VOL_READY) {
+ volumeState.set_state(VolumeState::NODE_PUBLISH);
+ checkpointVolumeState(volumeId);
+ }
+
+ LOG(INFO) << "Calling '/csi.v1.Node/NodePublishVolume' for volume '"
+ << volumeId << "'";
+
+ NodePublishVolumeRequest request;
+ request.set_volume_id(volumeId);
+ *request.mutable_publish_context() = volumeState.publish_context();
+ request.set_target_path(targetPath);
+ *request.mutable_volume_capability() =
+ evolve(volumeState.volume_capability());
+ request.set_readonly(false);
+ *request.mutable_volume_context() = volumeState.volume_context();
+
+ if (nodeCapabilities->stageUnstageVolume) {
+ const string stagingPath = paths::getMountStagingPath(
+ paths::getMountRootDir(rootDir, info.type(), info.name()), volumeId);
+
+ CHECK(os::exists(stagingPath));
+ request.set_staging_target_path(stagingPath);
+ }
+
+ return call(NODE_SERVICE, &Client::nodePublishVolume, std::move(request))
+ .then(defer(self(), [this, volumeId, targetPath] {
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+
+ volumeState.set_state(VolumeState::PUBLISHED);
+
+ // NOTE: This is the first time a container is going to consume the
+ // persistent volume, so the `node_publish_required` field is set to
+ // indicate that this volume must remain published so it can be
+ // synchronously cleaned up when the persistent volume is destroyed.
+ volumeState.set_node_publish_required(true);
+
+ checkpointVolumeState(volumeId);
+
+ return Nothing();
+ }));
+}
+
+
+Future<Nothing> VolumeManagerProcess::__publishVolume(const string& volumeId)
+{
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+
+ if (volumeState.state() == VolumeState::VOL_READY) {
+ CHECK(!volumeState.boot_id().empty());
+ return Nothing();
+ }
+
+ if (volumeState.state() != VolumeState::NODE_READY &&
+ volumeState.state() != VolumeState::NODE_STAGE &&
+ volumeState.state() != VolumeState::NODE_UNSTAGE) {
+ // Retry after transitioning the volume to `NODE_READY` state.
+ return _attachVolume(volumeId)
+ .then(process::defer(self(), &Self::__publishVolume, volumeId));
+ }
+
+ if (!nodeCapabilities->stageUnstageVolume) {
+ // Since this is a no-op, no need to checkpoint here.
+ volumeState.set_state(VolumeState::VOL_READY);
+ volumeState.set_boot_id(CHECK_NOTNONE(bootId));
+ return Nothing();
+ }
+
+ // A previously failed `NodeUnstageVolume` call can be recovered through an
+ // extra `NodeUnstageVolume` call. See:
+ // https://github.com/container-storage-interface/spec/blob/v1.1.0/spec.md#nodeunstagevolume // NOLINT
+ if (volumeState.state() == VolumeState::NODE_UNSTAGE) {
+ // Retry after recovering the volume to `NODE_READY` state.
+ return _unpublishVolume(volumeId)
+ .then(process::defer(self(), &Self::__publishVolume, volumeId));
+ }
+
+ const string stagingPath = paths::getMountStagingPath(
+ paths::getMountRootDir(rootDir, info.type(), info.name()), volumeId);
+
+ // NOTE: The staging path will be cleaned up in during volume removal.
+ Try<Nothing> mkdir = os::mkdir(stagingPath);
+ if (mkdir.isError()) {
+ return Failure(
+ "Failed to create mount staging path '" + stagingPath +
+ "': " + mkdir.error());
+ }
+
+ if (volumeState.state() == VolumeState::NODE_READY) {
+ volumeState.set_state(VolumeState::NODE_STAGE);
+ checkpointVolumeState(volumeId);
+ }
+
+ LOG(INFO) << "Calling '/csi.v1.Node/NodeStageVolume' for volume '" << volumeId
+ << "'";
+
+ NodeStageVolumeRequest request;
+ request.set_volume_id(volumeId);
+ *request.mutable_publish_context() = volumeState.publish_context();
+ request.set_staging_target_path(stagingPath);
+ *request.mutable_volume_capability() =
+ evolve(volumeState.volume_capability());
+ *request.mutable_volume_context() = volumeState.volume_context();
+
+ return call(NODE_SERVICE, &Client::nodeStageVolume, std::move(request))
+ .then(process::defer(self(), [this, volumeId] {
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+ volumeState.set_state(VolumeState::VOL_READY);
+ volumeState.set_boot_id(CHECK_NOTNONE(bootId));
+
+ checkpointVolumeState(volumeId);
+
+ return Nothing();
+ }));
+}
+
+
+Future<Nothing> VolumeManagerProcess::_unpublishVolume(const string& volumeId)
+{
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+
+ if (volumeState.state() == VolumeState::NODE_READY) {
+ CHECK(volumeState.boot_id().empty());
+ return Nothing();
+ }
+
+ if (volumeState.state() != VolumeState::VOL_READY &&
+ volumeState.state() != VolumeState::NODE_STAGE &&
+ volumeState.state() != VolumeState::NODE_UNSTAGE) {
+ // Retry after transitioning the volume to `VOL_READY` state.
+ return __unpublishVolume(volumeId)
+ .then(process::defer(self(), &Self::_unpublishVolume, volumeId));
+ }
+
+ if (!nodeCapabilities->stageUnstageVolume) {
+ // Since this is a no-op, no need to checkpoint here.
+ volumeState.set_state(VolumeState::NODE_READY);
+ volumeState.clear_boot_id();
+ return Nothing();
+ }
+
+ // A previously failed `NodeStageVolume` call can be recovered through the
+ // current `NodeUnstageVolume` call. See:
+ // https://github.com/container-storage-interface/spec/blob/v1.1.0/spec.md#nodestagevolume // NOLINT
+ if (volumeState.state() == VolumeState::VOL_READY ||
+ volumeState.state() == VolumeState::NODE_STAGE) {
+ volumeState.set_state(VolumeState::NODE_UNSTAGE);
+ checkpointVolumeState(volumeId);
+ }
+
+ const string stagingPath = paths::getMountStagingPath(
+ paths::getMountRootDir(rootDir, info.type(), info.name()), volumeId);
+
+ CHECK(os::exists(stagingPath));
+
+ LOG(INFO) << "Calling '/csi.v1.Node/NodeUnstageVolume' for volume '"
+ << volumeId << "'";
+
+ NodeUnstageVolumeRequest request;
+ request.set_volume_id(volumeId);
+ request.set_staging_target_path(stagingPath);
+
+ return call(NODE_SERVICE, &Client::nodeUnstageVolume, std::move(request))
+ .then(process::defer(self(), [this, volumeId] {
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+ volumeState.set_state(VolumeState::NODE_READY);
+ volumeState.clear_boot_id();
+
+ checkpointVolumeState(volumeId);
+
+ return Nothing();
+ }));
+}
+
+
+Future<Nothing> VolumeManagerProcess::__unpublishVolume(const string& volumeId)
+{
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+
+ if (volumeState.state() == VolumeState::VOL_READY) {
+ return Nothing();
+ }
+
+ if (volumeState.state() != VolumeState::PUBLISHED &&
+ volumeState.state() != VolumeState::NODE_PUBLISH &&
+ volumeState.state() != VolumeState::NODE_UNPUBLISH) {
+ return Failure(
+ "Cannot unpublish volume '" + volumeId + "' in " +
+ stringify(volumeState.state()) + "state");
+ }
+
+ // A previously failed `NodePublishVolume` call can be recovered through the
+ // current `NodeUnpublishVolume` call. See:
+ // https://github.com/container-storage-interface/spec/blob/v1.1.0/spec.md#nodepublishvolume // NOLINT
+ if (volumeState.state() == VolumeState::PUBLISHED ||
+ volumeState.state() == VolumeState::NODE_PUBLISH) {
+ volumeState.set_state(VolumeState::NODE_UNPUBLISH);
+ checkpointVolumeState(volumeId);
+ }
+
+ const string targetPath = paths::getMountTargetPath(
+ paths::getMountRootDir(rootDir, info.type(), info.name()), volumeId);
+
+ CHECK(os::exists(targetPath));
+
+ LOG(INFO) << "Calling '/csi.v1.Node/NodeUnpublishVolume' for volume '"
+ << volumeId << "'";
+
+ NodeUnpublishVolumeRequest request;
+ request.set_volume_id(volumeId);
+ request.set_target_path(targetPath);
+
+ return call(NODE_SERVICE, &Client::nodeUnpublishVolume, std::move(request))
+ .then(process::defer(self(), [this, volumeId] {
+ CHECK(volumes.contains(volumeId));
+ VolumeState& volumeState = volumes.at(volumeId).state;
+ volumeState.set_state(VolumeState::VOL_READY);
+
+ checkpointVolumeState(volumeId);
+
+ return Nothing();
+ }));
+}
+
+
+void VolumeManagerProcess::checkpointVolumeState(const string& volumeId)
+{
+ const string statePath =
+ paths::getVolumeStatePath(rootDir, info.type(), info.name(), volumeId);
+
+ // NOTE: We ensure the checkpoint is synced to the filesystem to avoid
+ // resulting in a stale or empty checkpoint when a system crash happens.
+ Try<Nothing> checkpoint =
+ slave::state::checkpoint(statePath, volumes.at(volumeId).state, true);
+
+ CHECK_SOME(checkpoint)
+ << "Failed to checkpoint volume state to '" << statePath << "':"
+ << checkpoint.error();
+}
+
+
+void VolumeManagerProcess::garbageCollectMountPath(const string& volumeId)
+{
+ CHECK(!volumes.contains(volumeId));
+
+ const string path = paths::getMountPath(
+ paths::getMountRootDir(rootDir, info.type(), info.name()), volumeId);
+
+ if (os::exists(path)) {
+ Try<Nothing> rmdir = os::rmdir(path);
+ if (rmdir.isError()) {
+ LOG(ERROR) << "Failed to remove directory '" << path
+ << "': " << rmdir.error();
+ }
+ }
+}
+
+
+VolumeManager::VolumeManager(
+ const http::URL& agentUrl,
+ const string& rootDir,
+ const CSIPluginInfo& info,
+ const hashset<Service>& services,
+ const string& containerPrefix,
+ const Option<string>& authToken,
+ const Runtime& runtime,
+ Metrics* metrics)
+ : process(new VolumeManagerProcess(
+ agentUrl,
+ rootDir,
+ info,
+ services,
+ containerPrefix,
+ authToken,
+ runtime,
+ metrics))
+{
+ process::spawn(CHECK_NOTNULL(process.get()));
+ recovered = process::dispatch(process.get(), &VolumeManagerProcess::recover);
+}
+
+
+VolumeManager::~VolumeManager()
+{
+ process::terminate(process.get());
+ process::wait(process.get());
+}
+
+
+Future<Nothing> VolumeManager::recover()
+{
+ return recovered;
+}
+
+
+Future<vector<VolumeInfo>> VolumeManager::listVolumes()
+{
+ return recovered
+ .then(process::defer(process.get(), &VolumeManagerProcess::listVolumes));
+}
+
+
+Future<Bytes> VolumeManager::getCapacity(
+ const types::VolumeCapability& capability,
+ const Map<string, string>& parameters)
+{
+ return recovered
+ .then(process::defer(
+ process.get(),
+ &VolumeManagerProcess::getCapacity,
+ capability,
+ parameters));
+}
+
+
+Future<VolumeInfo> VolumeManager::createVolume(
+ const string& name,
+ const Bytes& capacity,
+ const types::VolumeCapability& capability,
+ const Map<string, string>& parameters)
+{
+ return recovered
+ .then(process::defer(
+ process.get(),
+ &VolumeManagerProcess::createVolume,
+ name,
+ capacity,
+ capability,
+ parameters));
+}
+
+
+Future<Option<Error>> VolumeManager::validateVolume(
+ const VolumeInfo& volumeInfo,
+ const types::VolumeCapability& capability,
+ const Map<string, string>& parameters)
+{
+ return recovered
+ .then(process::defer(
+ process.get(),
+ &VolumeManagerProcess::validateVolume,
+ volumeInfo,
+ capability,
+ parameters));
+}
+
+
+Future<bool> VolumeManager::deleteVolume(const string& volumeId)
+{
+ return recovered
+ .then(process::defer(
+ process.get(), &VolumeManagerProcess::deleteVolume, volumeId));
+}
+
+
+Future<Nothing> VolumeManager::attachVolume(const string& volumeId)
+{
+ return recovered
+ .then(process::defer(
+ process.get(), &VolumeManagerProcess::attachVolume, volumeId));
+}
+
+
+Future<Nothing> VolumeManager::detachVolume(const string& volumeId)
+{
+ return recovered
+ .then(process::defer(
+ process.get(), &VolumeManagerProcess::detachVolume, volumeId));
+}
+
+
+Future<Nothing> VolumeManager::publishVolume(const string& volumeId)
+{
+ return recovered
+ .then(process::defer(
+ process.get(), &VolumeManagerProcess::publishVolume, volumeId));
+}
+
+
+Future<Nothing> VolumeManager::unpublishVolume(const string& volumeId)
+{
+ return recovered
+ .then(process::defer(
+ process.get(), &VolumeManagerProcess::unpublishVolume, volumeId));
+}
+
+} // namespace v1 {
+} // namespace csi {
+} // namespace mesos {
diff --git a/src/csi/v1_volume_manager.hpp b/src/csi/v1_volume_manager.hpp
new file mode 100644
index 0000000..f8e6095
--- /dev/null
+++ b/src/csi/v1_volume_manager.hpp
@@ -0,0 +1,116 @@
+// 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.
+
+#ifndef __CSI_V1_VOLUME_MANAGER_HPP__
+#define __CSI_V1_VOLUME_MANAGER_HPP__
+
+#include <string>
+#include <vector>
+
+#include <google/protobuf/map.h>
+
+#include <mesos/mesos.hpp>
+
+#include <mesos/csi/types.hpp>
+
+#include <process/future.hpp>
+#include <process/grpc.hpp>
+#include <process/http.hpp>
+#include <process/owned.hpp>
+
+#include <stout/bytes.hpp>
+#include <stout/error.hpp>
+#include <stout/hashset.hpp>
+#include <stout/nothing.hpp>
+#include <stout/option.hpp>
+
+#include "csi/metrics.hpp"
+#include "csi/service_manager.hpp"
+#include "csi/volume_manager.hpp"
+
+namespace mesos {
+namespace csi {
+namespace v1 {
+
+// Forward declarations.
+class VolumeManagerProcess;
+
+
+class VolumeManager : public csi::VolumeManager
+{
+public:
+ VolumeManager(
+ const process::http::URL& agentUrl,
+ const std::string& rootDir,
+ const CSIPluginInfo& info,
+ const hashset<Service>& services,
+ const std::string& containerPrefix,
+ const Option<std::string>& authToken,
+ const process::grpc::client::Runtime& runtime,
+ Metrics* metrics);
+
+ // Since this class contains `Owned` members which should not but can be
+ // copied, explicitly make this class non-copyable.
+ //
+ // TODO(chhsiao): Remove this once MESOS-5122 is fixed.
+ VolumeManager(const VolumeManager&) = delete;
+ VolumeManager& operator=(const VolumeManager&) = delete;
+
+ ~VolumeManager() override;
+
+ process::Future<Nothing> recover() override;
+
+ process::Future<std::vector<VolumeInfo>> listVolumes() override;
+
+ process::Future<Bytes> getCapacity(
+ const types::VolumeCapability& capability,
+ const google::protobuf::Map<std::string, std::string>& parameters)
+ override;
+
+ process::Future<VolumeInfo> createVolume(
+ const std::string& name,
+ const Bytes& capacity,
+ const types::VolumeCapability& capability,
+ const google::protobuf::Map<std::string, std::string>& parameters)
+ override;
+
+ process::Future<Option<Error>> validateVolume(
+ const VolumeInfo& volumeInfo,
+ const types::VolumeCapability& capability,
+ const google::protobuf::Map<std::string, std::string>& parameters)
+ override;
+
+ process::Future<bool> deleteVolume(const std::string& volumeId) override;
+
+ process::Future<Nothing> attachVolume(const std::string& volumeId) override;
+
+ process::Future<Nothing> detachVolume(const std::string& volumeId) override;
+
+ process::Future<Nothing> publishVolume(const std::string& volumeId) override;
+
+ process::Future<Nothing> unpublishVolume(
+ const std::string& volumeId) override;
+
+private:
+ process::Owned<VolumeManagerProcess> process;
+ process::Future<Nothing> recovered;
+};
+
+} // namespace v1 {
+} // namespace csi {
+} // namespace mesos {
+
+#endif // __CSI_V1_VOLUME_MANAGER_HPP__
diff --git a/src/csi/v1_volume_manager_process.hpp b/src/csi/v1_volume_manager_process.hpp
new file mode 100644
index 0000000..1c80399
--- /dev/null
+++ b/src/csi/v1_volume_manager_process.hpp
@@ -0,0 +1,218 @@
+// 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.
+
+#ifndef __CSI_V1_VOLUME_MANAGER_PROCESS_HPP__
+#define __CSI_V1_VOLUME_MANAGER_PROCESS_HPP__
+
+#include <string>
+#include <vector>
+
+#include <google/protobuf/map.h>
+
+#include <mesos/mesos.hpp>
+
+#include <mesos/csi/types.hpp>
+
+#include <process/future.hpp>
+#include <process/grpc.hpp>
+#include <process/http.hpp>
+#include <process/loop.hpp>
+#include <process/owned.hpp>
+#include <process/process.hpp>
+#include <process/sequence.hpp>
+
+#include <stout/bytes.hpp>
+#include <stout/duration.hpp>
+#include <stout/error.hpp>
+#include <stout/hashmap.hpp>
+#include <stout/hashset.hpp>
+#include <stout/nothing.hpp>
+#include <stout/option.hpp>
+#include <stout/try.hpp>
+
+#include "csi/metrics.hpp"
+#include "csi/service_manager.hpp"
+#include "csi/state.hpp"
+#include "csi/v1_client.hpp"
+#include "csi/v1_utils.hpp"
+#include "csi/v1_volume_manager.hpp"
+#include "csi/volume_manager.hpp"
+
+namespace mesos {
+namespace csi {
+namespace v1 {
+
+// The CSI volume manager initially picks a random amount of time between
+// `[0, b]`, where `b = DEFAULT_CSI_RETRY_BACKOFF_FACTOR`, to retry CSI calls.
+// Subsequent retries are exponentially backed off based on this interval (e.g.,
+// 2nd retry uses a random value between `[0, b * 2^1]`, 3rd retry between
+// `[0, b * 2^2]`, etc) up to a maximum of `DEFAULT_CSI_RETRY_INTERVAL_MAX`.
+//
+// TODO(chhsiao): Make the retry parameters configurable.
+constexpr Duration DEFAULT_CSI_RETRY_BACKOFF_FACTOR = Seconds(10);
+constexpr Duration DEFAULT_CSI_RETRY_INTERVAL_MAX = Minutes(10);
+
+
+class VolumeManagerProcess : public process::Process<VolumeManagerProcess>
+{
+public:
+ explicit VolumeManagerProcess(
+ const process::http::URL& agentUrl,
+ const std::string& _rootDir,
+ const CSIPluginInfo& _info,
+ const hashset<Service> _services,
+ const std::string& containerPrefix,
+ const Option<std::string>& authToken,
+ const process::grpc::client::Runtime& _runtime,
+ Metrics* _metrics);
+
+ process::Future<Nothing> recover();
+
+ process::Future<std::vector<VolumeInfo>> listVolumes();
+
+ process::Future<Bytes> getCapacity(
+ const types::VolumeCapability& capability,
+ const google::protobuf::Map<std::string, std::string>& parameters);
+
+ process::Future<VolumeInfo> createVolume(
+ const std::string& name,
+ const Bytes& capacity,
+ const types::VolumeCapability& capability,
+ const google::protobuf::Map<std::string, std::string>& parameters);
+
+ process::Future<Option<Error>> validateVolume(
+ const VolumeInfo& volumeInfo,
+ const types::VolumeCapability& capability,
+ const google::protobuf::Map<std::string, std::string>& parameters);
+
+ process::Future<bool> deleteVolume(const std::string& volumeId);
+
+ process::Future<Nothing> attachVolume(const std::string& volumeId);
+
+ process::Future<Nothing> detachVolume(const std::string& volumeId);
+
+ process::Future<Nothing> publishVolume(const std::string& volumeId);
+
+ process::Future<Nothing> unpublishVolume(const std::string& volumeId);
+
+ // Wrapper functions to make CSI calls and update RPC metrics. Made public for
+ // testing purpose.
+ //
+ // The call is made asynchronously and thus no guarantee is provided on the
+ // order in which calls are sent. Callers need to either ensure to not have
+ // multiple conflicting calls in flight, or treat results idempotently.
+ //
+ // NOTE: We currently ensure this by 1) resource locking to forbid concurrent
+ // calls on the same volume, and 2) no profile update while there are ongoing
+ // `CREATE_DISK` or `DESTROY_DISK` operations.
+ template <typename Request, typename Response>
+ process::Future<Response> call(
+ const Service& service,
+ process::Future<RPCResult<Response>> (Client::*rpc)(Request),
+ const Request& request,
+ bool retry = false);
+
+ template <typename Request, typename Response>
+ process::Future<RPCResult<Response>> _call(
+ const std::string& endpoint,
+ process::Future<RPCResult<Response>> (Client::*rpc)(Request),
+ const Request& request);
+
+ template <typename Response>
+ process::Future<process::ControlFlow<Response>> __call(
+ const RPCResult<Response>& result, const Option<Duration>& backoff);
+
+private:
+ process::Future<Nothing> prepareServices();
+
+ process::Future<bool> _deleteVolume(const std::string& volumeId);
+ process::Future<bool> __deleteVolume(const std::string& volumeId);
+
+ // The following methods are used to manage volume lifecycles. Transient
+ // states are omitted.
+ //
+ // +------------+
+ // + + + | CREATED | ^
+ // _attachVolume | | | +---+----^---+ |
+ // | | | | | | _detachVolume
+ // | | | +---v----+---+ |
+ // v + + | NODE_READY | + ^
+ // | | +---+----^---+ | |
+ // __publishVolume | | | | | | _unpublishVolume
+ // | | +---v----+---+ | |
+ // v + | VOL_READY | + + ^
+ // | +---+----^---+ | | |
+ // _publishVolume | | | | | | __unpublishVolume
+ // | +---v----+---+ | | |
+ // V | PUBLISHED | + + +
+ // +------------+
+
+ // Transition a volume to `NODE_READY` state from any state above.
+ process::Future<Nothing> _attachVolume(const std::string& volumeId);
+
+ // Transition a volume to `CREATED` state from any state below.
+ process::Future<Nothing> _detachVolume(const std::string& volumeId);
+
+ // Transition a volume to `PUBLISHED` state from any state above.
+ process::Future<Nothing> _publishVolume(const std::string& volumeId);
+
+ // Transition a volume to `VOL_READY` state from any state above.
+ process::Future<Nothing> __publishVolume(const std::string& volumeId);
+
+ // Transition a volume to `NODE_READY` state from any state below.
+ process::Future<Nothing> _unpublishVolume(const std::string& volumeId);
+
+ // Transition a volume to `VOL_READY` state from any state below.
+ process::Future<Nothing> __unpublishVolume(const std::string& volumeId);
+
+ void checkpointVolumeState(const std::string& volumeId);
+
+ void garbageCollectMountPath(const std::string& volumeId);
+
+ const std::string rootDir;
+ const CSIPluginInfo info;
+ const hashset<Service> services;
+
+ process::grpc::client::Runtime runtime;
+ Metrics* metrics;
+ process::Owned<ServiceManager> serviceManager;
+
+ Option<std::string> bootId;
+ Option<PluginCapabilities> pluginCapabilities;
+ Option<ControllerCapabilities> controllerCapabilities;
+ Option<NodeCapabilities> nodeCapabilities;
+ Option<std::string> nodeId;
+
+ struct VolumeData
+ {
+ VolumeData(state::VolumeState&& _state)
+ : state(_state), sequence(new process::Sequence("csi-volume-sequence")) {}
+
+ state::VolumeState state;
+
+ // We call all CSI operations on the same volume in a sequence to ensure
+ // that they are processed in a sequential order.
+ process::Owned<process::Sequence> sequence;
+ };
+
+ hashmap<std::string, VolumeData> volumes;
+};
+
+} // namespace v1 {
+} // namespace csi {
+} // namespace mesos {
+
+#endif // __CSI_V1_VOLUME_MANAGER_PROCESS_HPP__