You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by jo...@apache.org on 2018/01/05 22:42:09 UTC
[1/2] mesos git commit: Renamed VolumeProfileAdaptor to
DiskProfileAdaptor.
Repository: mesos
Updated Branches:
refs/heads/master f44a2d783 -> 87629bf32
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/volume_profile.proto
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/volume_profile.proto b/src/resource_provider/storage/volume_profile.proto
deleted file mode 100644
index c0628ec..0000000
--- a/src/resource_provider/storage/volume_profile.proto
+++ /dev/null
@@ -1,43 +0,0 @@
-// 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.
-
-syntax = "proto3";
-
-import "csi.proto";
-
-package mesos.resource_provider;
-
-
-message VolumeProfileMapping {
- message CSIManifest {
- // Capabilities used for creating, publishing, and validating volumes.
- // This field is REQUIRED.
- //
- // NOTE: The name of this field is plural because some CSI requests
- // support multiple capabilities. However, Mesos currently does not
- // support this.
- .csi.VolumeCapability volume_capabilities = 1;
-
- // Parameters passed to the CSI CreateVolume RPC.
- // This field is OPTIONAL.
- map<string, string> create_parameters = 2;
- }
-
- // Each map entry associates a profile name (type string) with the CSI
- // capabilities and parameters used to make specific CSI requests.
- // This field is OPTIONAL.
- map<string, CSIManifest> profile_matrix = 1;
-}
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/volume_profile_utils.cpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/volume_profile_utils.cpp b/src/resource_provider/storage/volume_profile_utils.cpp
deleted file mode 100644
index 3d04558..0000000
--- a/src/resource_provider/storage/volume_profile_utils.cpp
+++ /dev/null
@@ -1,121 +0,0 @@
-// 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 "resource_provider/storage/volume_profile_utils.hpp"
-
-#include <google/protobuf/util/json_util.h>
-
-#include <stout/bytes.hpp>
-#include <stout/foreach.hpp>
-
-using std::string;
-
-using mesos::resource_provider::VolumeProfileMapping;
-
-namespace mesos{
-namespace internal {
-namespace profile {
-
-Try<VolumeProfileMapping> parseVolumeProfileMapping(
- const string& data)
-{
- // Use Google's JSON utility function to parse the JSON string.
- VolumeProfileMapping output;
- google::protobuf::util::JsonParseOptions options;
- options.ignore_unknown_fields = true;
-
- google::protobuf::util::Status status =
- google::protobuf::util::JsonStringToMessage(data, &output, options);
-
- if (!status.ok()) {
- return Error(
- "Failed to parse VolumeProfileMapping message: "
- + status.ToString());
- }
-
- Option<Error> validation = validate(output);
- if (validation.isSome()) {
- return Error(
- "Fetched profile mapping failed validation with: " + validation->message);
- }
-
- return output;
-}
-
-
-Option<Error> validate(const VolumeProfileMapping& mapping)
-{
- auto iterator = mapping.profile_matrix().begin();
- while (iterator != mapping.profile_matrix().end()) {
- if (!iterator->second.has_volume_capabilities()) {
- return Error(
- "Profile '" + iterator->first + "' is missing the required field "
- + "'volume_capabilities");
- }
-
- Option<Error> capabilities =
- validate(iterator->second.volume_capabilities());
-
- if (capabilities.isSome()) {
- return Error(
- "Profile '" + iterator->first + "' VolumeCapabilities are invalid: "
- + capabilities->message);
- }
-
- // NOTE: The `create_parameters` field is optional and needs no further
- // validation after parsing.
-
- iterator++;
- }
-
- return None();
-}
-
-
-// TODO(chhsiao): Move this to CSI validation implementation file.
-Option<Error> validate(const csi::VolumeCapability& capability)
-{
- if (capability.has_mount()) {
- // The total size of this repeated field may not exceed 4 KB.
- //
- // TODO(josephw): The specification does not state how this maximum
- // size is calculated. So this check is conservative and does not
- // include padding or array separators in the size calculation.
- size_t size = 0;
- foreach (const string& flag, capability.mount().mount_flags()) {
- size += flag.size();
- }
-
- if (Bytes(size) > Kilobytes(4)) {
- return Error("Size of 'mount_flags' may not exceed 4 KB");
- }
- }
-
- if (!capability.has_access_mode()) {
- return Error("'access_mode' is a required field");
- }
-
- if (capability.access_mode().mode() ==
- csi::VolumeCapability::AccessMode::UNKNOWN) {
- return Error("'access_mode.mode' is unknown or not set");
- }
-
- return None();
-}
-
-} // namespace profile {
-} // namespace internal {
-} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/volume_profile_utils.hpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/volume_profile_utils.hpp b/src/resource_provider/storage/volume_profile_utils.hpp
deleted file mode 100644
index 46afbd3..0000000
--- a/src/resource_provider/storage/volume_profile_utils.hpp
+++ /dev/null
@@ -1,48 +0,0 @@
-// 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 __RESOURCE_PROVIDER_URI_VOLUME_PROFILE_UTILS_HPP__
-#define __RESOURCE_PROVIDER_URI_VOLUME_PROFILE_UTILS_HPP__
-
-#include <stout/option.hpp>
-#include <stout/try.hpp>
-
-// ONLY USEFUL AFTER RUNNING PROTOC.
-#include "resource_provider/storage/volume_profile.pb.h"
-
-namespace mesos {
-namespace internal {
-namespace profile {
-
-// Helper for parsing a string as the expected data format.
-Try<resource_provider::VolumeProfileMapping> parseVolumeProfileMapping(
- const std::string& data);
-
-
-// Checks the fields inside a `VolumeProfileMapping` according to the
-// comments above the protobuf.
-Option<Error> validate(const resource_provider::VolumeProfileMapping& mapping);
-
-
-// Checks the fields inside a `VolumeCapability` according to the
-// comments above the protobuf.
-Option<Error> validate(const csi::VolumeCapability& capability);
-
-} // namespace profile {
-} // namespace internal {
-} // namespace mesos {
-
-#endif // __RESOURCE_PROVIDER_URI_VOLUME_PROFILE_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/slave/flags.cpp
----------------------------------------------------------------------
diff --git a/src/slave/flags.cpp b/src/slave/flags.cpp
index 48b8821..943aaaf 100644
--- a/src/slave/flags.cpp
+++ b/src/slave/flags.cpp
@@ -112,10 +112,10 @@ mesos::internal::slave::Flags::Flags()
"}");
#ifdef ENABLE_GRPC
- add(&Flags::volume_profile_adaptor,
- "volume_profile_adaptor",
- "The name of the volume profile adaptor module that storage resource\n"
- "providers should use for translating a 'volume profile' into inputs\n"
+ add(&Flags::disk_profile_adaptor,
+ "disk_profile_adaptor",
+ "The name of the disk profile adaptor module that storage resource\n"
+ "providers should use for translating a 'disk profile' into inputs\n"
"consumed by various Container Storage Interface (CSI) plugins.\n"
"If this flag is not specified, the default behavior for storage\n"
"resource providers is to only expose resources for pre-existing\n"
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/slave/flags.hpp
----------------------------------------------------------------------
diff --git a/src/slave/flags.hpp b/src/slave/flags.hpp
index 1cf5272..42c4861 100644
--- a/src/slave/flags.hpp
+++ b/src/slave/flags.hpp
@@ -47,7 +47,7 @@ public:
Option<std::string> resources;
Option<std::string> resource_provider_config_dir;
#ifdef ENABLE_GRPC
- Option<std::string> volume_profile_adaptor;
+ Option<std::string> disk_profile_adaptor;
#endif
std::string isolation;
std::string launcher;
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index cfb675d..aeb0fda 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -37,7 +37,7 @@
#include <mesos/module/authenticatee.hpp>
#ifdef ENABLE_GRPC
-#include <mesos/resource_provider/storage/volume_profile.hpp>
+#include <mesos/resource_provider/storage/disk_profile.hpp>
#endif
#include <process/after.hpp>
@@ -424,21 +424,21 @@ void Slave::initialize()
}
#ifdef ENABLE_GRPC
- // Create the VolumeProfileAdaptor module and set it globally so
+ // Create the DiskProfileAdaptor module and set it globally so
// any component that needs the module can share this instance.
- Try<VolumeProfileAdaptor*> _volumeProfileAdaptor =
- VolumeProfileAdaptor::create(flags.volume_profile_adaptor);
+ Try<DiskProfileAdaptor*> _diskProfileAdaptor =
+ DiskProfileAdaptor::create(flags.disk_profile_adaptor);
- if (_volumeProfileAdaptor.isError()) {
+ if (_diskProfileAdaptor.isError()) {
EXIT(EXIT_FAILURE)
- << "Failed to create volume profile adaptor: "
- << _volumeProfileAdaptor.error();
+ << "Failed to create disk profile adaptor: "
+ << _diskProfileAdaptor.error();
}
- volumeProfileAdaptor =
- shared_ptr<VolumeProfileAdaptor>(_volumeProfileAdaptor.get());
+ diskProfileAdaptor =
+ shared_ptr<DiskProfileAdaptor>(_diskProfileAdaptor.get());
- VolumeProfileAdaptor::setAdaptor(volumeProfileAdaptor);
+ DiskProfileAdaptor::setAdaptor(diskProfileAdaptor);
#endif
string scheme = "http";
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index b5eec33..ef0eae2 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -98,7 +98,7 @@ namespace mesos {
// Forward declarations.
class Authorizer;
-class VolumeProfileAdaptor;
+class DiskProfileAdaptor;
namespace internal {
namespace slave {
@@ -725,7 +725,7 @@ private:
mesos::slave::QoSController* qosController;
- std::shared_ptr<VolumeProfileAdaptor> volumeProfileAdaptor;
+ std::shared_ptr<DiskProfileAdaptor> diskProfileAdaptor;
mesos::SecretGenerator* secretGenerator;
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/tests/disk_profile_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/disk_profile_tests.cpp b/src/tests/disk_profile_tests.cpp
new file mode 100644
index 0000000..b6b35ef
--- /dev/null
+++ b/src/tests/disk_profile_tests.cpp
@@ -0,0 +1,515 @@
+// 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 <map>
+#include <string>
+#include <tuple>
+#include <vector>
+
+#include <mesos/module/disk_profile.hpp>
+
+#include <mesos/resource_provider/storage/disk_profile.hpp>
+
+#include <process/clock.hpp>
+#include <process/future.hpp>
+#include <process/gmock.hpp>
+#include <process/gtest.hpp>
+#include <process/owned.hpp>
+
+#include <stout/duration.hpp>
+#include <stout/gtest.hpp>
+#include <stout/hashset.hpp>
+#include <stout/path.hpp>
+#include <stout/stringify.hpp>
+#include <stout/try.hpp>
+
+#include <stout/os/write.hpp>
+
+#include "module/manager.hpp"
+
+#include "resource_provider/storage/uri_disk_profile.hpp"
+#include "resource_provider/storage/disk_profile_utils.hpp"
+
+#include "tests/flags.hpp"
+#include "tests/mesos.hpp"
+#include "tests/utils.hpp"
+
+using namespace process;
+
+using std::map;
+using std::string;
+using std::tuple;
+using std::vector;
+
+using google::protobuf::Map;
+
+using mesos::resource_provider::DiskProfileMapping;
+
+using testing::_;
+using testing::DoAll;
+using testing::Return;
+
+namespace mesos {
+namespace internal {
+namespace tests {
+
+constexpr char URI_DISK_PROFILE_ADAPTOR_NAME[] =
+ "org_apache_mesos_UriDiskProfileAdaptor";
+
+
+class UriDiskProfileTest : public MesosTest
+{
+public:
+ virtual void SetUp()
+ {
+ MesosTest::SetUp();
+
+ string libraryPath = getModulePath("uri_disk_profile");
+
+ Modules::Library* library = modules.add_libraries();
+ library->set_name("uri_disk_profile");
+ library->set_file(libraryPath);
+
+ Modules::Library::Module* module = library->add_modules();
+ module->set_name(URI_DISK_PROFILE_ADAPTOR_NAME);
+
+ ASSERT_SOME(modules::ModuleManager::load(modules));
+ }
+
+ virtual void TearDown()
+ {
+ foreach (const Modules::Library& library, modules.libraries()) {
+ foreach (const Modules::Library::Module& module, library.modules()) {
+ if (module.has_name()) {
+ ASSERT_SOME(modules::ModuleManager::unload(module.name()));
+ }
+ }
+ }
+
+ MesosTest::TearDown();
+ }
+
+protected:
+ Modules modules;
+};
+
+
+// Exercises the disk profile map parsing method with the example found
+// in the UriDiskProfile module's help string.
+TEST_F(UriDiskProfileTest, ParseExample)
+{
+ const string example = R"~(
+ {
+ "profile_matrix" : {
+ "my-profile" : {
+ "volume_capabilities" : {
+ "block" : {},
+ "access_mode" : { "mode" : "SINGLE_NODE_WRITER" }
+ },
+ "create_parameters" : {
+ "mesos-does-not" : "interpret-these",
+ "type" : "raid5",
+ "stripes" : "3",
+ "stripesize" : "64"
+ }
+ }
+ }
+ })~";
+
+ Try<DiskProfileMapping> parsed =
+ mesos::internal::profile::parseDiskProfileMapping(example);
+ ASSERT_SOME(parsed);
+
+ const string key = "my-profile";
+ ASSERT_EQ(1u, parsed->profile_matrix().count(key));
+
+ csi::VolumeCapability capability =
+ parsed->profile_matrix().at(key).volume_capabilities();
+
+ ASSERT_TRUE(capability.has_block());
+ ASSERT_TRUE(capability.has_access_mode());
+ ASSERT_EQ(
+ csi::VolumeCapability::AccessMode::SINGLE_NODE_WRITER,
+ capability.access_mode().mode());
+
+ Map<string, string> parameters =
+ parsed->profile_matrix().at(key).create_parameters();
+
+ ASSERT_EQ(4u, parameters.size());
+ ASSERT_EQ(1u, parameters.count("mesos-does-not"));
+ ASSERT_EQ(1u, parameters.count("type"));
+ ASSERT_EQ(1u, parameters.count("stripes"));
+ ASSERT_EQ(1u, parameters.count("stripesize"));
+
+ ASSERT_EQ("interpret-these", parameters.at("mesos-does-not"));
+ ASSERT_EQ("raid5", parameters.at("type"));
+ ASSERT_EQ("3", parameters.at("stripes"));
+ ASSERT_EQ("64", parameters.at("stripesize"));
+}
+
+
+// Exercises the disk profile map parsing method with some slightly incorrect
+// inputs. Each item in the array of examples should error at a different area
+// of the code (and are ordered corresponding to the code as well).
+TEST_F(UriDiskProfileTest, ParseInvalids)
+{
+ const vector<string> examples = {
+ "Not an object, but still JSON",
+
+ R"~({
+ "profile_matrix" : {
+ "profile" : "This is not an object"
+ }
+ })~",
+
+ R"~({
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : "Wrong JSON type"
+ }
+ }
+ })~",
+
+ R"~({
+ "profile_matrix" : {
+ "profile" : {
+ "not-volume_capabilities" : "Missing required key"
+ }
+ }
+ })~",
+
+ // Missing one of 'block' or 'mount'.
+ R"~({
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {}
+ }
+ }
+ })~",
+
+ R"~({
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {
+ "mount" : {
+ "fs_type" : [ "This should not be an array" ]
+ }
+ }
+ }
+ }
+ })~",
+
+ R"~({
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {
+ "block" : {},
+ "access_mode" : { "mode": "No-enum-of-this-name" }
+ }
+ }
+ }
+ })~",
+
+ R"~({
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {
+ "mount" : {
+ "mount_flags" : [ "a", "b", "c" ]
+ },
+ "access_mode" : { "mode": "SINGLE_NODE_WRITER" }
+ },
+ "create_parameters" : "Wrong JSON type"
+ }
+ }
+ })~",
+
+ R"~({
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {
+ "mount" : { "fs_type" : "abc" },
+ "access_mode" : { "mode": "SINGLE_NODE_READER_ONLY" }
+ },
+ "create_parameters" : {
+ "incorrect" : [ "JSON type of parameter" ]
+ }
+ }
+ }
+ })~",
+
+ R"~({
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {
+ "block" : {},
+ "access_mode" : { "mode": "MULTI_NODE_READER_ONLY" }
+ }
+ },
+ "first profile is fine, second profile is broken" : {}
+ }
+ })~",
+ };
+
+ hashset<string> errors;
+ for (size_t i = 0; i < examples.size(); i++) {
+ Try<DiskProfileMapping> parsed =
+ mesos::internal::profile::parseDiskProfileMapping(examples[i]);
+
+ ASSERT_ERROR(parsed) << examples[i];
+ ASSERT_EQ(0u, errors.count(parsed.error())) << parsed.error();
+
+ errors.insert(parsed.error());
+ }
+}
+
+
+// This creates a UriDiskProfile module configured to read from a file
+// and tests the basic `watch` -> `translate` workflow which callers of
+// the module are expected to follow.
+TEST_F(UriDiskProfileTest, FetchFromFile)
+{
+ Clock::pause();
+
+ const string contents =R"~(
+ {
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {
+ "block" : {},
+ "access_mode" : { "mode": "MULTI_NODE_SINGLE_WRITER" }
+ }
+ }
+ }
+ })~";
+
+ const string profileName = "profile";
+ const string profileFile = path::join(sandbox.get(), "profiles.json");
+ const Duration pollInterval = Seconds(10);
+ const string csiPluginType = "ignored";
+
+ Parameters params;
+
+ Parameter* pollIntervalFlag = params.add_parameter();
+ pollIntervalFlag->set_key("poll_interval");
+ pollIntervalFlag->set_value(stringify(pollInterval));
+
+ // NOTE: We cannot use the `file://` URI to sepcify the file location,
+ // otherwise the file contents will be prematurely read. Therefore, we
+ // specify the absolute path of the file in the `uri` flag.
+ Parameter* uriFlag = params.add_parameter();
+ uriFlag->set_key("uri");
+ uriFlag->set_value(profileFile);
+
+ // Create the module before we've written anything to the file.
+ // This means the first poll will fail, so the module believes there
+ // are no profiles at the moment.
+ Try<DiskProfileAdaptor*> module =
+ modules::ModuleManager::create<DiskProfileAdaptor>(
+ URI_DISK_PROFILE_ADAPTOR_NAME,
+ params);
+ ASSERT_SOME(module);
+
+ // Start watching for updates.
+ // By the time this returns, we'll know that the first poll has finished
+ // because when the module reads from file, it does so immediately upon
+ // being initialized.
+ Future<hashset<string>> future =
+ module.get()->watch(hashset<string>::EMPTY, csiPluginType);
+
+ // Write the single profile to the file.
+ ASSERT_SOME(os::write(profileFile, contents));
+
+ // Trigger the next poll.
+ Clock::advance(pollInterval);
+
+ AWAIT_ASSERT_READY(future);
+ ASSERT_EQ(1u, future->size());
+ EXPECT_EQ(profileName, *(future->begin()));
+
+ // Translate the profile name into the profile mapping.
+ Future<DiskProfileAdaptor::ProfileInfo> mapping =
+ module.get()->translate(profileName, csiPluginType);
+
+ AWAIT_ASSERT_READY(mapping);
+ ASSERT_TRUE(mapping.get().capability.has_block());
+ ASSERT_EQ(
+ csi::VolumeCapability::AccessMode::MULTI_NODE_SINGLE_WRITER,
+ mapping.get().capability.access_mode().mode());
+
+ Clock::resume();
+}
+
+
+// Basic helper for UriDiskProfile modules configured to fetch from HTTP URIs.
+class MockProfileServer : public Process<MockProfileServer>
+{
+public:
+ MOCK_METHOD1(profiles, Future<http::Response>(const http::Request&));
+
+protected:
+ virtual void initialize()
+ {
+ route("/profiles", None(), &MockProfileServer::profiles);
+ }
+};
+
+
+class ServerWrapper
+{
+public:
+ ServerWrapper() : process(new MockProfileServer())
+ {
+ spawn(process.get());
+ }
+
+ ~ServerWrapper()
+ {
+ terminate(process.get());
+ wait(process.get());
+ }
+
+ Owned<MockProfileServer> process;
+};
+
+
+// This creates a UriDiskProfile module configured to read from an HTTP URI.
+// The HTTP server will return a different profile mapping between each of the
+// calls. We expect the module to ignore the second call because the module
+// does not allow profiles to be renamed. This is not a fatal error however,
+// as the HTTP server can be "fixed" without restarting the agent.
+TEST_F(UriDiskProfileTest, FetchFromHTTP)
+{
+ Clock::pause();
+
+ const string contents1 =R"~(
+ {
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {
+ "block" : {},
+ "access_mode" : { "mode": "MULTI_NODE_MULTI_WRITER" }
+ }
+ }
+ }
+ })~";
+
+ const string contents2 =R"~(
+ {
+ "profile_matrix" : {
+ "renamed-profile" : {
+ "volume_capabilities" : {
+ "block" : {},
+ "access_mode" : { "mode": "SINGLE_NODE_WRITER" }
+ }
+ }
+ }
+ })~";
+
+ const string contents3 =R"~(
+ {
+ "profile_matrix" : {
+ "profile" : {
+ "volume_capabilities" : {
+ "block" : {},
+ "access_mode" : { "mode": "MULTI_NODE_MULTI_WRITER" }
+ }
+ },
+ "another-profile" : {
+ "volume_capabilities" : {
+ "block" : {},
+ "access_mode" : { "mode": "SINGLE_NODE_WRITER" }
+ }
+ }
+ }
+ })~";
+
+ const Duration pollInterval = Seconds(10);
+ const string csiPluginType = "ignored";
+
+ ServerWrapper server;
+
+ // Wait for the server to finish initializing so that the routes are ready.
+ AWAIT_READY(dispatch(server.process->self(), []() { return Nothing(); }));
+
+ // We need to intercept this call since the module is expected to
+ // ignore the result of the second call.
+ Future<Nothing> secondCall;
+
+ EXPECT_CALL(*server.process, profiles(_))
+ .WillOnce(Return(http::OK(contents1)))
+ .WillOnce(DoAll(FutureSatisfy(&secondCall), Return(http::OK(contents2))))
+ .WillOnce(Return(http::OK(contents3)));
+
+ Parameters params;
+
+ Parameter* pollIntervalFlag = params.add_parameter();
+ pollIntervalFlag->set_key("poll_interval");
+ pollIntervalFlag->set_value(stringify(pollInterval));
+
+ Parameter* uriFlag = params.add_parameter();
+ uriFlag->set_key("uri");
+ uriFlag->set_value(stringify(process::http::URL(
+ "http",
+ process::address().ip,
+ process::address().port,
+ server.process->self().id + "/profiles")));
+
+ Try<DiskProfileAdaptor*> module =
+ modules::ModuleManager::create<DiskProfileAdaptor>(
+ URI_DISK_PROFILE_ADAPTOR_NAME,
+ params);
+ ASSERT_SOME(module);
+
+ // Wait for the first HTTP poll to complete.
+ Future<hashset<string>> future =
+ module.get()->watch(hashset<string>::EMPTY, csiPluginType);
+
+ AWAIT_ASSERT_READY(future);
+ ASSERT_EQ(1u, future->size());
+ EXPECT_EQ("profile", *(future->begin()));
+
+ // Start watching for an update to the list of profiles.
+ future = module.get()->watch({"profile"}, csiPluginType);
+
+ // Trigger the second HTTP poll.
+ Clock::advance(pollInterval);
+ AWAIT_ASSERT_READY(secondCall);
+
+ // Dispatch a call to the module, which ensures that the polling has actually
+ // completed (not just the HTTP call).
+ AWAIT_ASSERT_READY(module.get()->translate("profile", csiPluginType));
+
+ // We don't expect the module to notify watcher(s) because the server's
+ // response is considered invalid (the module does not allow profiles
+ // to be renamed).
+ ASSERT_TRUE(future.isPending());
+
+ // Trigger the third HTTP poll.
+ Clock::advance(pollInterval);
+
+ // This time, the server's response is correct and also includes a second
+ // profile, which means that the watcher(s) should be notified.
+ AWAIT_ASSERT_READY(future);
+ ASSERT_EQ(2u, future->size());
+ EXPECT_EQ((hashset<string>{"profile", "another-profile"}), future.get());
+
+ Clock::resume();
+}
+
+} // namespace tests {
+} // namespace internal {
+} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/tests/storage_local_resource_provider_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/storage_local_resource_provider_tests.cpp b/src/tests/storage_local_resource_provider_tests.cpp
index e9ea021..bbfe95e 100644
--- a/src/tests/storage_local_resource_provider_tests.cpp
+++ b/src/tests/storage_local_resource_provider_tests.cpp
@@ -43,8 +43,8 @@ namespace mesos {
namespace internal {
namespace tests {
-constexpr char URI_VOLUME_PROFILE_ADAPTOR_NAME[] =
- "org_apache_mesos_UriVolumeProfileAdaptor";
+constexpr char URI_DISK_PROFILE_ADAPTOR_NAME[] =
+ "org_apache_mesos_UriDiskProfileAdaptor";
class StorageLocalResourceProviderTest : public MesosTest
@@ -112,11 +112,11 @@ public:
path::join(resourceProviderConfigDir, "test.json"),
resourceProviderConfig.get()));
- uriVolumeProfileConfigPath =
- path::join(sandbox.get(), "volume_profiles.json");
+ uriDiskProfileConfigPath =
+ path::join(sandbox.get(), "disk_profiles.json");
Try<Nothing> write = os::write(
- uriVolumeProfileConfigPath,
+ uriDiskProfileConfigPath,
R"~(
{
"profile_matrix": {
@@ -147,20 +147,20 @@ public:
MesosTest::TearDown();
}
- void loadUriVolumeProfileModule()
+ void loadUriDiskProfileModule()
{
- string libraryPath = getModulePath("uri_volume_profile");
+ string libraryPath = getModulePath("uri_disk_profile");
Modules::Library* library = modules.add_libraries();
- library->set_name("uri_volume_profile");
+ library->set_name("uri_disk_profile");
library->set_file(libraryPath);
Modules::Library::Module* module = library->add_modules();
- module->set_name(URI_VOLUME_PROFILE_ADAPTOR_NAME);
+ module->set_name(URI_DISK_PROFILE_ADAPTOR_NAME);
Parameter* parameter = module->add_parameters();
parameter->set_key("uri");
- parameter->set_value(uriVolumeProfileConfigPath);
+ parameter->set_value(uriDiskProfileConfigPath);
ASSERT_SOME(modules::ModuleManager::load(modules));
}
@@ -168,7 +168,7 @@ public:
protected:
Modules modules;
string resourceProviderConfigDir;
- string uriVolumeProfileConfigPath;
+ string uriDiskProfileConfigPath;
};
@@ -177,7 +177,7 @@ protected:
// that uses the test CSI plugin.
TEST_F(StorageLocalResourceProviderTest, ROOT_NewVolume)
{
- loadUriVolumeProfileModule();
+ loadUriDiskProfileModule();
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -201,7 +201,7 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_NewVolume)
{capabilities.begin(), capabilities.end()});
flags.resource_provider_config_dir = resourceProviderConfigDir;
- flags.volume_profile_adaptor = URI_VOLUME_PROFILE_ADAPTOR_NAME;
+ flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
Future<SlaveRegisteredMessage> slaveRegisteredMessage =
FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
@@ -354,7 +354,7 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_NewVolume)
// the test CSI plugin after recovery.
TEST_F(StorageLocalResourceProviderTest, ROOT_NewVolumeRecovery)
{
- loadUriVolumeProfileModule();
+ loadUriDiskProfileModule();
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -378,7 +378,7 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_NewVolumeRecovery)
{capabilities.begin(), capabilities.end()});
flags.resource_provider_config_dir = resourceProviderConfigDir;
- flags.volume_profile_adaptor = URI_VOLUME_PROFILE_ADAPTOR_NAME;
+ flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
Future<SlaveRegisteredMessage> slaveRegisteredMessage =
FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
@@ -547,7 +547,7 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_NewVolumeRecovery)
// plugin, then destroy the volume while it is published.
TEST_F(StorageLocalResourceProviderTest, ROOT_LaunchTask)
{
- loadUriVolumeProfileModule();
+ loadUriDiskProfileModule();
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -571,7 +571,7 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_LaunchTask)
{capabilities.begin(), capabilities.end()});
flags.resource_provider_config_dir = resourceProviderConfigDir;
- flags.volume_profile_adaptor = URI_VOLUME_PROFILE_ADAPTOR_NAME;
+ flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
Future<SlaveRegisteredMessage> slaveRegisteredMessage =
FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
@@ -763,7 +763,7 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_LaunchTask)
// and used by a task after recovery.
TEST_F(StorageLocalResourceProviderTest, ROOT_LaunchTaskRecovery)
{
- loadUriVolumeProfileModule();
+ loadUriDiskProfileModule();
Try<Owned<cluster::Master>> master = StartMaster();
ASSERT_SOME(master);
@@ -787,7 +787,7 @@ TEST_F(StorageLocalResourceProviderTest, ROOT_LaunchTaskRecovery)
{capabilities.begin(), capabilities.end()});
flags.resource_provider_config_dir = resourceProviderConfigDir;
- flags.volume_profile_adaptor = URI_VOLUME_PROFILE_ADAPTOR_NAME;
+ flags.disk_profile_adaptor = URI_DISK_PROFILE_ADAPTOR_NAME;
Future<SlaveRegisteredMessage> slaveRegisteredMessage =
FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _);
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/tests/volume_profile_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/volume_profile_tests.cpp b/src/tests/volume_profile_tests.cpp
deleted file mode 100644
index 6ed81c9..0000000
--- a/src/tests/volume_profile_tests.cpp
+++ /dev/null
@@ -1,515 +0,0 @@
-// 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 <map>
-#include <string>
-#include <tuple>
-#include <vector>
-
-#include <mesos/module/volume_profile.hpp>
-
-#include <mesos/resource_provider/storage/volume_profile.hpp>
-
-#include <process/clock.hpp>
-#include <process/future.hpp>
-#include <process/gmock.hpp>
-#include <process/gtest.hpp>
-#include <process/owned.hpp>
-
-#include <stout/duration.hpp>
-#include <stout/gtest.hpp>
-#include <stout/hashset.hpp>
-#include <stout/path.hpp>
-#include <stout/stringify.hpp>
-#include <stout/try.hpp>
-
-#include <stout/os/write.hpp>
-
-#include "module/manager.hpp"
-
-#include "resource_provider/storage/uri_volume_profile.hpp"
-#include "resource_provider/storage/volume_profile_utils.hpp"
-
-#include "tests/flags.hpp"
-#include "tests/mesos.hpp"
-#include "tests/utils.hpp"
-
-using namespace process;
-
-using std::map;
-using std::string;
-using std::tuple;
-using std::vector;
-
-using google::protobuf::Map;
-
-using mesos::resource_provider::VolumeProfileMapping;
-
-using testing::_;
-using testing::DoAll;
-using testing::Return;
-
-namespace mesos {
-namespace internal {
-namespace tests {
-
-constexpr char URI_VOLUME_PROFILE_ADAPTOR_NAME[] =
- "org_apache_mesos_UriVolumeProfileAdaptor";
-
-
-class UriVolumeProfileTest : public MesosTest
-{
-public:
- virtual void SetUp()
- {
- MesosTest::SetUp();
-
- string libraryPath = getModulePath("uri_volume_profile");
-
- Modules::Library* library = modules.add_libraries();
- library->set_name("uri_volume_profile");
- library->set_file(libraryPath);
-
- Modules::Library::Module* module = library->add_modules();
- module->set_name(URI_VOLUME_PROFILE_ADAPTOR_NAME);
-
- ASSERT_SOME(modules::ModuleManager::load(modules));
- }
-
- virtual void TearDown()
- {
- foreach (const Modules::Library& library, modules.libraries()) {
- foreach (const Modules::Library::Module& module, library.modules()) {
- if (module.has_name()) {
- ASSERT_SOME(modules::ModuleManager::unload(module.name()));
- }
- }
- }
-
- MesosTest::TearDown();
- }
-
-protected:
- Modules modules;
-};
-
-
-// Exercises the volume profile map parsing method with the example found
-// in the UriVolumeProfile module's help string.
-TEST_F(UriVolumeProfileTest, ParseExample)
-{
- const string example = R"~(
- {
- "profile_matrix" : {
- "my-profile" : {
- "volume_capabilities" : {
- "block" : {},
- "access_mode" : { "mode" : "SINGLE_NODE_WRITER" }
- },
- "create_parameters" : {
- "mesos-does-not" : "interpret-these",
- "type" : "raid5",
- "stripes" : "3",
- "stripesize" : "64"
- }
- }
- }
- })~";
-
- Try<VolumeProfileMapping> parsed =
- mesos::internal::profile::parseVolumeProfileMapping(example);
- ASSERT_SOME(parsed);
-
- const string key = "my-profile";
- ASSERT_EQ(1u, parsed->profile_matrix().count(key));
-
- csi::VolumeCapability capability =
- parsed->profile_matrix().at(key).volume_capabilities();
-
- ASSERT_TRUE(capability.has_block());
- ASSERT_TRUE(capability.has_access_mode());
- ASSERT_EQ(
- csi::VolumeCapability::AccessMode::SINGLE_NODE_WRITER,
- capability.access_mode().mode());
-
- Map<string, string> parameters =
- parsed->profile_matrix().at(key).create_parameters();
-
- ASSERT_EQ(4u, parameters.size());
- ASSERT_EQ(1u, parameters.count("mesos-does-not"));
- ASSERT_EQ(1u, parameters.count("type"));
- ASSERT_EQ(1u, parameters.count("stripes"));
- ASSERT_EQ(1u, parameters.count("stripesize"));
-
- ASSERT_EQ("interpret-these", parameters.at("mesos-does-not"));
- ASSERT_EQ("raid5", parameters.at("type"));
- ASSERT_EQ("3", parameters.at("stripes"));
- ASSERT_EQ("64", parameters.at("stripesize"));
-}
-
-
-// Exercises the volume profile map parsing method with some slightly incorrect
-// inputs. Each item in the array of examples should error at a different area
-// of the code (and are ordered corresponding to the code as well).
-TEST_F(UriVolumeProfileTest, ParseInvalids)
-{
- const vector<string> examples = {
- "Not an object, but still JSON",
-
- R"~({
- "profile_matrix" : {
- "profile" : "This is not an object"
- }
- })~",
-
- R"~({
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : "Wrong JSON type"
- }
- }
- })~",
-
- R"~({
- "profile_matrix" : {
- "profile" : {
- "not-volume_capabilities" : "Missing required key"
- }
- }
- })~",
-
- // Missing one of 'block' or 'mount'.
- R"~({
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {}
- }
- }
- })~",
-
- R"~({
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {
- "mount" : {
- "fs_type" : [ "This should not be an array" ]
- }
- }
- }
- }
- })~",
-
- R"~({
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {
- "block" : {},
- "access_mode" : { "mode": "No-enum-of-this-name" }
- }
- }
- }
- })~",
-
- R"~({
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {
- "mount" : {
- "mount_flags" : [ "a", "b", "c" ]
- },
- "access_mode" : { "mode": "SINGLE_NODE_WRITER" }
- },
- "create_parameters" : "Wrong JSON type"
- }
- }
- })~",
-
- R"~({
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {
- "mount" : { "fs_type" : "abc" },
- "access_mode" : { "mode": "SINGLE_NODE_READER_ONLY" }
- },
- "create_parameters" : {
- "incorrect" : [ "JSON type of parameter" ]
- }
- }
- }
- })~",
-
- R"~({
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {
- "block" : {},
- "access_mode" : { "mode": "MULTI_NODE_READER_ONLY" }
- }
- },
- "first profile is fine, second profile is broken" : {}
- }
- })~",
- };
-
- hashset<string> errors;
- for (size_t i = 0; i < examples.size(); i++) {
- Try<VolumeProfileMapping> parsed =
- mesos::internal::profile::parseVolumeProfileMapping(examples[i]);
-
- ASSERT_ERROR(parsed) << examples[i];
- ASSERT_EQ(0u, errors.count(parsed.error())) << parsed.error();
-
- errors.insert(parsed.error());
- }
-}
-
-
-// This creates a UriVolumeProfile module configured to read from a file
-// and tests the basic `watch` -> `translate` workflow which callers of
-// the module are expected to follow.
-TEST_F(UriVolumeProfileTest, FetchFromFile)
-{
- Clock::pause();
-
- const string contents =R"~(
- {
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {
- "block" : {},
- "access_mode" : { "mode": "MULTI_NODE_SINGLE_WRITER" }
- }
- }
- }
- })~";
-
- const string profileName = "profile";
- const string profileFile = path::join(sandbox.get(), "profiles.json");
- const Duration pollInterval = Seconds(10);
- const string csiPluginType = "ignored";
-
- Parameters params;
-
- Parameter* pollIntervalFlag = params.add_parameter();
- pollIntervalFlag->set_key("poll_interval");
- pollIntervalFlag->set_value(stringify(pollInterval));
-
- // NOTE: We cannot use the `file://` URI to sepcify the file location,
- // otherwise the file contents will be prematurely read. Therefore, we
- // specify the absolute path of the file in the `uri` flag.
- Parameter* uriFlag = params.add_parameter();
- uriFlag->set_key("uri");
- uriFlag->set_value(profileFile);
-
- // Create the module before we've written anything to the file.
- // This means the first poll will fail, so the module believes there
- // are no profiles at the moment.
- Try<VolumeProfileAdaptor*> module =
- modules::ModuleManager::create<VolumeProfileAdaptor>(
- URI_VOLUME_PROFILE_ADAPTOR_NAME,
- params);
- ASSERT_SOME(module);
-
- // Start watching for updates.
- // By the time this returns, we'll know that the first poll has finished
- // because when the module reads from file, it does so immediately upon
- // being initialized.
- Future<hashset<string>> future =
- module.get()->watch(hashset<string>::EMPTY, csiPluginType);
-
- // Write the single profile to the file.
- ASSERT_SOME(os::write(profileFile, contents));
-
- // Trigger the next poll.
- Clock::advance(pollInterval);
-
- AWAIT_ASSERT_READY(future);
- ASSERT_EQ(1u, future->size());
- EXPECT_EQ(profileName, *(future->begin()));
-
- // Translate the profile name into the profile mapping.
- Future<VolumeProfileAdaptor::ProfileInfo> mapping =
- module.get()->translate(profileName, csiPluginType);
-
- AWAIT_ASSERT_READY(mapping);
- ASSERT_TRUE(mapping.get().capability.has_block());
- ASSERT_EQ(
- csi::VolumeCapability::AccessMode::MULTI_NODE_SINGLE_WRITER,
- mapping.get().capability.access_mode().mode());
-
- Clock::resume();
-}
-
-
-// Basic helper for UriVolumeProfile modules configured to fetch from HTTP URIs.
-class MockProfileServer : public Process<MockProfileServer>
-{
-public:
- MOCK_METHOD1(profiles, Future<http::Response>(const http::Request&));
-
-protected:
- virtual void initialize()
- {
- route("/profiles", None(), &MockProfileServer::profiles);
- }
-};
-
-
-class ServerWrapper
-{
-public:
- ServerWrapper() : process(new MockProfileServer())
- {
- spawn(process.get());
- }
-
- ~ServerWrapper()
- {
- terminate(process.get());
- wait(process.get());
- }
-
- Owned<MockProfileServer> process;
-};
-
-
-// This creates a UriVolumeProfile module configured to read from an HTTP URI.
-// The HTTP server will return a different profile mapping between each of the
-// calls. We expect the module to ignore the second call because the module
-// does not allow profiles to be renamed. This is not a fatal error however,
-// as the HTTP server can be "fixed" without restarting the agent.
-TEST_F(UriVolumeProfileTest, FetchFromHTTP)
-{
- Clock::pause();
-
- const string contents1 =R"~(
- {
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {
- "block" : {},
- "access_mode" : { "mode": "MULTI_NODE_MULTI_WRITER" }
- }
- }
- }
- })~";
-
- const string contents2 =R"~(
- {
- "profile_matrix" : {
- "renamed-profile" : {
- "volume_capabilities" : {
- "block" : {},
- "access_mode" : { "mode": "SINGLE_NODE_WRITER" }
- }
- }
- }
- })~";
-
- const string contents3 =R"~(
- {
- "profile_matrix" : {
- "profile" : {
- "volume_capabilities" : {
- "block" : {},
- "access_mode" : { "mode": "MULTI_NODE_MULTI_WRITER" }
- }
- },
- "another-profile" : {
- "volume_capabilities" : {
- "block" : {},
- "access_mode" : { "mode": "SINGLE_NODE_WRITER" }
- }
- }
- }
- })~";
-
- const Duration pollInterval = Seconds(10);
- const string csiPluginType = "ignored";
-
- ServerWrapper server;
-
- // Wait for the server to finish initializing so that the routes are ready.
- AWAIT_READY(dispatch(server.process->self(), []() { return Nothing(); }));
-
- // We need to intercept this call since the module is expected to
- // ignore the result of the second call.
- Future<Nothing> secondCall;
-
- EXPECT_CALL(*server.process, profiles(_))
- .WillOnce(Return(http::OK(contents1)))
- .WillOnce(DoAll(FutureSatisfy(&secondCall), Return(http::OK(contents2))))
- .WillOnce(Return(http::OK(contents3)));
-
- Parameters params;
-
- Parameter* pollIntervalFlag = params.add_parameter();
- pollIntervalFlag->set_key("poll_interval");
- pollIntervalFlag->set_value(stringify(pollInterval));
-
- Parameter* uriFlag = params.add_parameter();
- uriFlag->set_key("uri");
- uriFlag->set_value(stringify(process::http::URL(
- "http",
- process::address().ip,
- process::address().port,
- server.process->self().id + "/profiles")));
-
- Try<VolumeProfileAdaptor*> module =
- modules::ModuleManager::create<VolumeProfileAdaptor>(
- URI_VOLUME_PROFILE_ADAPTOR_NAME,
- params);
- ASSERT_SOME(module);
-
- // Wait for the first HTTP poll to complete.
- Future<hashset<string>> future =
- module.get()->watch(hashset<string>::EMPTY, csiPluginType);
-
- AWAIT_ASSERT_READY(future);
- ASSERT_EQ(1u, future->size());
- EXPECT_EQ("profile", *(future->begin()));
-
- // Start watching for an update to the list of profiles.
- future = module.get()->watch({"profile"}, csiPluginType);
-
- // Trigger the second HTTP poll.
- Clock::advance(pollInterval);
- AWAIT_ASSERT_READY(secondCall);
-
- // Dispatch a call to the module, which ensures that the polling has actually
- // completed (not just the HTTP call).
- AWAIT_ASSERT_READY(module.get()->translate("profile", csiPluginType));
-
- // We don't expect the module to notify watcher(s) because the server's
- // response is considered invalid (the module does not allow profiles
- // to be renamed).
- ASSERT_TRUE(future.isPending());
-
- // Trigger the third HTTP poll.
- Clock::advance(pollInterval);
-
- // This time, the server's response is correct and also includes a second
- // profile, which means that the watcher(s) should be notified.
- AWAIT_ASSERT_READY(future);
- ASSERT_EQ(2u, future->size());
- EXPECT_EQ((hashset<string>{"profile", "another-profile"}), future.get());
-
- Clock::resume();
-}
-
-} // namespace tests {
-} // namespace internal {
-} // namespace mesos {
[2/2] mesos git commit: Renamed VolumeProfileAdaptor to
DiskProfileAdaptor.
Posted by jo...@apache.org.
Renamed VolumeProfileAdaptor to DiskProfileAdaptor.
"Disk Profile" is more accurate than "Volume Profile" when describing
the set of parameters passed into a Storage Resource Provider while
creating disk resources. A disk resource is not necessarily a "Volume"
but will always be a disk.
Review: https://reviews.apache.org/r/64972
Review: https://reviews.apache.org/r/64973
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/87629bf3
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/87629bf3
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/87629bf3
Branch: refs/heads/master
Commit: 87629bf323256223fa2a4cbde08ef9d45c6cfa5a
Parents: f44a2d7
Author: Joseph Wu <jo...@apache.org>
Authored: Fri Jan 5 00:25:47 2018 -0800
Committer: Joseph Wu <jo...@apache.org>
Committed: Fri Jan 5 14:10:37 2018 -0800
----------------------------------------------------------------------
docs/csi.md | 20 +-
include/mesos/mesos.proto | 2 +-
include/mesos/module/disk_profile.hpp | 63 +++
include/mesos/module/volume_profile.hpp | 63 ---
.../resource_provider/storage/disk_profile.hpp | 144 ++++++
.../storage/volume_profile.hpp | 144 ------
include/mesos/v1/mesos.proto | 2 +-
src/Makefile.am | 34 +-
src/module/manager.cpp | 2 +-
src/resource_provider/storage/disk_profile.cpp | 131 +++++
.../storage/disk_profile.proto | 43 ++
.../storage/disk_profile_utils.cpp | 121 +++++
.../storage/disk_profile_utils.hpp | 48 ++
src/resource_provider/storage/provider.cpp | 40 +-
.../storage/uri_disk_profile.cpp | 321 ++++++++++++
.../storage/uri_disk_profile.hpp | 248 +++++++++
.../storage/uri_volume_profile.cpp | 321 ------------
.../storage/uri_volume_profile.hpp | 248 ---------
.../storage/volume_profile.cpp | 131 -----
.../storage/volume_profile.proto | 43 --
.../storage/volume_profile_utils.cpp | 121 -----
.../storage/volume_profile_utils.hpp | 48 --
src/slave/flags.cpp | 8 +-
src/slave/flags.hpp | 2 +-
src/slave/slave.cpp | 20 +-
src/slave/slave.hpp | 4 +-
src/tests/disk_profile_tests.cpp | 515 +++++++++++++++++++
.../storage_local_resource_provider_tests.cpp | 38 +-
src/tests/volume_profile_tests.cpp | 515 -------------------
29 files changed, 1720 insertions(+), 1720 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/docs/csi.md
----------------------------------------------------------------------
diff --git a/docs/csi.md b/docs/csi.md
index d268df1..641a052 100644
--- a/docs/csi.md
+++ b/docs/csi.md
@@ -388,7 +388,7 @@ message Resource {
// the framework will do disk selection based on profile names,
// instead of vendor specific disk parameters.
//
- // Also see the VolumeProfile module.
+ // Also see the DiskProfile module.
optional string profile = 6;
}
}
@@ -399,14 +399,14 @@ A typical framework that needs storage is expected to perform disk
resource selection based on the `profile` of a disk resource, rather
than low-level storage vendor specific parameters.
-### Volume Profile Adaptor Module
+### Disk Profile Adaptor Module
In order to let cluster operators customize the mapping between profiles and
storage system-specific parameters, Mesos provides a [module](#modules.md)
-interface called `VolumeProfileAdaptor`.
+interface called `DiskProfileAdaptor`.
```cpp
-class VolumeProfileAdaptor
+class DiskProfileAdaptor
{
public:
struct ProfileInfo
@@ -445,10 +445,10 @@ implicit dependency between backward compatibility of the module interface and
the CSI spec version. Since CSI doesn't provide a backward compatibility
promise, modules have to be re-built against each release of Mesos.
-### URI Volume Profile Adaptor
+### URI Disk Profile Adaptor
-To demonstrate how to use the volume profile adaptor module, Mesos ships with a
-default volume profile adaptor, called `UriVolumeProfileAdaptor`. This module
+To demonstrate how to use the disk profile adaptor module, Mesos ships with a
+default disk profile adaptor, called `UriDiskProfileAdaptor`. This module
polls the profile information (in JSON) from a configurable URI. Here are the
module parameters that can be used to configure the module:
@@ -491,16 +491,16 @@ module parameters that can be used to configure the module:
To enable this module, please follow the [modules documentation](modules.md):
add the following JSON to the `--modules` agent flag, and set agent flag
-`--volume_profile_adaptor` to `org_apache_mesos_UriVolumeProfileAdaptor`.
+`--disk_profile_adaptor` to `org_apache_mesos_UriDiskProfileAdaptor`.
```json
{
"libraries": [
{
- "file": "/PATH/TO/liburi_volume_profile.so",
+ "file": "/PATH/TO/liburi_disk_profile.so",
"modules": [
{
- "name": "org_apache_mesos_UriVolumeProfileAdaptor",
+ "name": "org_apache_mesos_UriDiskProfileAdaptor",
"parameters" : [
{
"key": "uri",
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/include/mesos/mesos.proto
----------------------------------------------------------------------
diff --git a/include/mesos/mesos.proto b/include/mesos/mesos.proto
index c677a8b..01b05f9 100644
--- a/include/mesos/mesos.proto
+++ b/include/mesos/mesos.proto
@@ -1462,7 +1462,7 @@ message Resource {
// the framework will do disk selection based on profile names,
// instead of vendor specific disk parameters.
//
- // Also see the VolumeProfile module.
+ // Also see the DiskProfile module.
optional string profile = 6; // EXPERIMENTAL.
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/include/mesos/module/disk_profile.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/module/disk_profile.hpp b/include/mesos/module/disk_profile.hpp
new file mode 100644
index 0000000..3a32150
--- /dev/null
+++ b/include/mesos/module/disk_profile.hpp
@@ -0,0 +1,63 @@
+// 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 __MESOS_MODULE_DISK_PROFILE_HPP__
+#define __MESOS_MODULE_DISK_PROFILE_HPP__
+
+#include <mesos/mesos.hpp>
+#include <mesos/module.hpp>
+
+#include <mesos/resource_provider/storage/disk_profile.hpp>
+
+namespace mesos {
+namespace modules {
+
+template <>
+inline const char* kind<mesos::DiskProfileAdaptor>()
+{
+ return "DiskProfileAdaptor";
+}
+
+
+template <>
+struct Module<mesos::DiskProfileAdaptor> : ModuleBase
+{
+ Module(
+ const char* _moduleApiVersion,
+ const char* _mesosVersion,
+ const char* _authorName,
+ const char* _authorEmail,
+ const char* _description,
+ bool (*_compatible)(),
+ mesos::DiskProfileAdaptor*
+ (*_create)(const Parameters& parameters))
+ : ModuleBase(
+ _moduleApiVersion,
+ _mesosVersion,
+ mesos::modules::kind<mesos::DiskProfileAdaptor>(),
+ _authorName,
+ _authorEmail,
+ _description,
+ _compatible),
+ create(_create) {}
+
+ mesos::DiskProfileAdaptor* (*create)(const Parameters& parameters);
+};
+
+} // namespace modules {
+} // namespace mesos {
+
+#endif // __MESOS_MODULE_DISK_PROFILE_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/include/mesos/module/volume_profile.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/module/volume_profile.hpp b/include/mesos/module/volume_profile.hpp
deleted file mode 100644
index bb95b32..0000000
--- a/include/mesos/module/volume_profile.hpp
+++ /dev/null
@@ -1,63 +0,0 @@
-// 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 __MESOS_MODULE_VOLUME_PROFILE_HPP__
-#define __MESOS_MODULE_VOLUME_PROFILE_HPP__
-
-#include <mesos/mesos.hpp>
-#include <mesos/module.hpp>
-
-#include <mesos/resource_provider/storage/volume_profile.hpp>
-
-namespace mesos {
-namespace modules {
-
-template <>
-inline const char* kind<mesos::VolumeProfileAdaptor>()
-{
- return "VolumeProfileAdaptor";
-}
-
-
-template <>
-struct Module<mesos::VolumeProfileAdaptor> : ModuleBase
-{
- Module(
- const char* _moduleApiVersion,
- const char* _mesosVersion,
- const char* _authorName,
- const char* _authorEmail,
- const char* _description,
- bool (*_compatible)(),
- mesos::VolumeProfileAdaptor*
- (*_create)(const Parameters& parameters))
- : ModuleBase(
- _moduleApiVersion,
- _mesosVersion,
- mesos::modules::kind<mesos::VolumeProfileAdaptor>(),
- _authorName,
- _authorEmail,
- _description,
- _compatible),
- create(_create) {}
-
- mesos::VolumeProfileAdaptor* (*create)(const Parameters& parameters);
-};
-
-} // namespace modules {
-} // namespace mesos {
-
-#endif // __MESOS_MODULE_VOLUME_PROFILE_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/include/mesos/resource_provider/storage/disk_profile.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/resource_provider/storage/disk_profile.hpp b/include/mesos/resource_provider/storage/disk_profile.hpp
new file mode 100644
index 0000000..d1a522a
--- /dev/null
+++ b/include/mesos/resource_provider/storage/disk_profile.hpp
@@ -0,0 +1,144 @@
+// 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 __MESOS_RESOURCE_PROVIDER_DISK_PROFILE_HPP__
+#define __MESOS_RESOURCE_PROVIDER_DISK_PROFILE_HPP__
+
+#include <memory>
+#include <string>
+#include <tuple>
+
+#include <process/future.hpp>
+
+#include <stout/hashset.hpp>
+#include <stout/none.hpp>
+#include <stout/nothing.hpp>
+
+#include <csi/spec.hpp>
+
+namespace mesos {
+
+/**
+ * This module is used by Storage Resource Providers to translate the
+ * "profile" field of a `Resource::DiskInfo::Source` into fields that a
+ * Container Storage Interface (CSI) plugin can potentially
+ * understand. This affects the following CSI requests:
+ * * ControllerPublishVolumeRequest
+ * * CreateVolumeRequest
+ * * GetCapacityRequest
+ * * NodePublishVolumeRequest
+ * * ValidateVolumeCapabilitiesRequest
+ *
+ * This module is not intended to interact with any CSI plugins directly.
+ *
+ * Documentation for each of the CSI requests can be found at:
+ * https://github.com/container-storage-interface/spec/
+ */
+class DiskProfileAdaptor
+{
+public:
+ struct ProfileInfo
+ {
+ /**
+ * Corresponds to the `volume_capability` or `volume_capabilities`
+ * fields of the affected CSI requests listed in the module description.
+ *
+ * NOTE: Some CSI requests take a list of `VolumeCapability` objects.
+ * However, because there is no existing way to choose between a list
+ * of `VolumeCapability` objects, the module will only allow a single
+ * `VolumeCapability`.
+ */
+ csi::VolumeCapability capability;
+
+ /**
+ * Free-form key-value pairs which should be passed into the body
+ * of a `CreateVolumeRequest`. Neither Mesos nor this module is expected
+ * to act upon these values (except copying them into the request).
+ */
+ google::protobuf::Map<std::string, std::string> parameters;
+ };
+
+ /**
+ * Factory method used to create a DiskProfileAdaptor instance.
+ * If the `name` parameter is provided, the module is instantiated
+ * using the `ModuleManager`. Otherwise, a "default" disk profile
+ * adaptor instance (defined in `src/resource_provider/disk_profile.cpp`)
+ * is returned.
+ *
+ * NOTE: The lifecycle of the returned object is delegated to the caller.
+ */
+ static Try<DiskProfileAdaptor*> create(
+ const Option<std::string>& name = None());
+
+ /**
+ * Global methods for setting and getting a DiskProfileAdaptor instance.
+ *
+ * The agent (or test) is expected to create and set the adaptor instance
+ * and manage the pointer (this method will only keep a weak pointer).
+ * Each component that needs to use the DiskProfileAdaptor, such as the
+ * Storage Local Resource Provider, should call `getAdaptor`.
+ */
+ static void setAdaptor(const std::shared_ptr<DiskProfileAdaptor>& adaptor);
+ static std::shared_ptr<DiskProfileAdaptor> getAdaptor();
+
+ virtual ~DiskProfileAdaptor() {}
+
+ /**
+ * Called before a Storage Resource Provider makes an affected CSI request.
+ * The caller is responsible for copying the returned values into the request
+ * object.
+ *
+ * This method is expected to return a Failure if a matching "profile"
+ * cannot be found or retrieved. The caller should not proceed with
+ * any of the affected CSI requests if this method returns a failure.
+ *
+ * The `csiPluginInfoType` parameter is the `CSIPluginInfo::type` field
+ * found inside `ResourceProviderInfo::storage`. This module may choose to
+ * filter results based on the type of CSI plugin.
+ *
+ * NOTE: This module assumes that profiles are immutable after creation.
+ * Changing the `VolumeCapability` or Parameters of a profile after creation
+ * may result in undefined behavior from the SLRP or CSI plugins.
+ */
+ virtual process::Future<ProfileInfo> translate(
+ const std::string& profile,
+ const std::string& csiPluginInfoType) = 0;
+
+ /**
+ * Returns a future that will be satisifed iff the set of profiles known
+ * by the module differs from the `knownProfiles` parameter.
+ *
+ * The `csiPluginInfoType` parameter is the `CSIPluginInfo::type` field
+ * found inside `ResourceProviderInfo::storage`. This module may choose to
+ * filter results based on the type of CSI plugin.
+ *
+ * NOTE: It is highly recommended for the module to insert a random delay
+ * between discovering a different set of profiles and satisfying this
+ * future, because the SLRP is expected to update the set of offered
+ * resources based on this future. Adding a random delay may prevent
+ * a thundering herd of resource updates to the Mesos master.
+ */
+ virtual process::Future<hashset<std::string>> watch(
+ const hashset<std::string>& knownProfiles,
+ const std::string& csiPluginInfoType) = 0;
+
+protected:
+ DiskProfileAdaptor() {}
+};
+
+} // namespace mesos {
+
+#endif // __MESOS_RESOURCE_PROVIDER_DISK_PROFILE_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/include/mesos/resource_provider/storage/volume_profile.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/resource_provider/storage/volume_profile.hpp b/include/mesos/resource_provider/storage/volume_profile.hpp
deleted file mode 100644
index 7141a14..0000000
--- a/include/mesos/resource_provider/storage/volume_profile.hpp
+++ /dev/null
@@ -1,144 +0,0 @@
-// 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 __MESOS_RESOURCE_PROVIDER_VOLUME_PROFILE_HPP__
-#define __MESOS_RESOURCE_PROVIDER_VOLUME_PROFILE_HPP__
-
-#include <memory>
-#include <string>
-#include <tuple>
-
-#include <process/future.hpp>
-
-#include <stout/hashset.hpp>
-#include <stout/none.hpp>
-#include <stout/nothing.hpp>
-
-#include <csi/spec.hpp>
-
-namespace mesos {
-
-/**
- * This module is used by Storage Resource Providers to translate the
- * "profile" field of a `Resource::DiskInfo::Source` into fields that a
- * Container Storage Interface (CSI) plugin can potentially
- * understand. This affects the following CSI requests:
- * * ControllerPublishVolumeRequest
- * * CreateVolumeRequest
- * * GetCapacityRequest
- * * NodePublishVolumeRequest
- * * ValidateVolumeCapabilitiesRequest
- *
- * This module is not intended to interact with any CSI plugins directly.
- *
- * Documentation for each of the CSI requests can be found at:
- * https://github.com/container-storage-interface/spec/
- */
-class VolumeProfileAdaptor
-{
-public:
- struct ProfileInfo
- {
- /**
- * Corresponds to the `volume_capability` or `volume_capabilities`
- * fields of the affected CSI requests listed in the module description.
- *
- * NOTE: Some CSI requests take a list of `VolumeCapability` objects.
- * However, because there is no existing way to choose between a list
- * of `VolumeCapability` objects, the module will only allow a single
- * `VolumeCapability`.
- */
- csi::VolumeCapability capability;
-
- /**
- * Free-form key-value pairs which should be passed into the body
- * of a `CreateVolumeRequest`. Neither Mesos nor this module is expected
- * to act upon these values (except copying them into the request).
- */
- google::protobuf::Map<std::string, std::string> parameters;
- };
-
- /**
- * Factory method used to create a VolumeProfileAdaptor instance.
- * If the `name` parameter is provided, the module is instantiated
- * using the `ModuleManager`. Otherwise, a "default" volume profile
- * adaptor instance (defined in `src/resource_provider/volume_profile.cpp`)
- * is returned.
- *
- * NOTE: The lifecycle of the returned object is delegated to the caller.
- */
- static Try<VolumeProfileAdaptor*> create(
- const Option<std::string>& name = None());
-
- /**
- * Global methods for setting and getting a VolumeProfileAdaptor instance.
- *
- * The agent (or test) is expected to create and set the adaptor instance
- * and manage the pointer (this method will only keep a weak pointer).
- * Each component that needs to use the VolumeProfileAdaptor, such as the
- * Storage Local Resource Provider, should call `getAdaptor`.
- */
- static void setAdaptor(const std::shared_ptr<VolumeProfileAdaptor>& adaptor);
- static std::shared_ptr<VolumeProfileAdaptor> getAdaptor();
-
- virtual ~VolumeProfileAdaptor() {}
-
- /**
- * Called before a Storage Resource Provider makes an affected CSI request.
- * The caller is responsible for copying the returned values into the request
- * object.
- *
- * This method is expected to return a Failure if a matching "profile"
- * cannot be found or retrieved. The caller should not proceed with
- * any of the affected CSI requests if this method returns a failure.
- *
- * The `csiPluginInfoType` parameter is the `CSIPluginInfo::type` field
- * found inside `ResourceProviderInfo::storage`. This module may choose to
- * filter results based on the type of CSI plugin.
- *
- * NOTE: This module assumes that profiles are immutable after creation.
- * Changing the `VolumeCapability` or Parameters of a profile after creation
- * may result in undefined behavior from the SLRP or CSI plugins.
- */
- virtual process::Future<ProfileInfo> translate(
- const std::string& profile,
- const std::string& csiPluginInfoType) = 0;
-
- /**
- * Returns a future that will be satisifed iff the set of profiles known
- * by the module differs from the `knownProfiles` parameter.
- *
- * The `csiPluginInfoType` parameter is the `CSIPluginInfo::type` field
- * found inside `ResourceProviderInfo::storage`. This module may choose to
- * filter results based on the type of CSI plugin.
- *
- * NOTE: It is highly recommended for the module to insert a random delay
- * between discovering a different set of profiles and satisfying this
- * future, because the SLRP is expected to update the set of offered
- * resources based on this future. Adding a random delay may prevent
- * a thundering herd of resource updates to the Mesos master.
- */
- virtual process::Future<hashset<std::string>> watch(
- const hashset<std::string>& knownProfiles,
- const std::string& csiPluginInfoType) = 0;
-
-protected:
- VolumeProfileAdaptor() {}
-};
-
-} // namespace mesos {
-
-#endif // __MESOS_RESOURCE_PROVIDER_VOLUME_PROFILE_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/include/mesos/v1/mesos.proto
----------------------------------------------------------------------
diff --git a/include/mesos/v1/mesos.proto b/include/mesos/v1/mesos.proto
index da7b458..b35b3f8 100644
--- a/include/mesos/v1/mesos.proto
+++ b/include/mesos/v1/mesos.proto
@@ -1443,7 +1443,7 @@ message Resource {
// the framework will do disk selection based on profile names,
// instead of vendor specific disk parameters.
//
- // Also see the VolumeProfile module.
+ // Also see the DiskProfile module.
optional string profile = 6; // EXPERIMENTAL.
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/Makefile.am
----------------------------------------------------------------------
diff --git a/src/Makefile.am b/src/Makefile.am
index 30cd4d4..bfe9eb1 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -383,8 +383,8 @@ CXX_PROTOS += \
if ENABLE_GRPC
CXX_PROTOS += \
- resource_provider/storage/volume_profile.pb.cc \
- resource_provider/storage/volume_profile.pb.h
+ resource_provider/storage/disk_profile.pb.cc \
+ resource_provider/storage/disk_profile.pb.h
CXX_CSI_PROTOS = \
../include/csi/csi.grpc.pb.cc \
@@ -748,7 +748,7 @@ module_HEADERS = \
if ENABLE_GRPC
module_HEADERS += \
- $(top_srcdir)/include/mesos/module/volume_profile.hpp
+ $(top_srcdir)/include/mesos/module/disk_profile.hpp
endif
nodist_module_HEADERS = \
@@ -781,7 +781,7 @@ resourceprovider_HEADERS = \
if ENABLE_GRPC
resourceprovider_HEADERS += \
- $(top_srcdir)/include/mesos/resource_provider/storage/volume_profile.hpp
+ $(top_srcdir)/include/mesos/resource_provider/storage/disk_profile.hpp
endif
nodist_resourceprovider_HEADERS = \
@@ -1473,10 +1473,10 @@ endif
if ENABLE_GRPC
libmesos_no_3rdparty_la_SOURCES += \
- resource_provider/storage/volume_profile.cpp \
- resource_provider/storage/volume_profile.proto \
- resource_provider/storage/volume_profile_utils.cpp \
- resource_provider/storage/volume_profile_utils.hpp
+ resource_provider/storage/disk_profile.cpp \
+ resource_provider/storage/disk_profile.proto \
+ resource_provider/storage/disk_profile_utils.cpp \
+ resource_provider/storage/disk_profile_utils.hpp
if OS_LINUX
libmesos_no_3rdparty_la_SOURCES += \
@@ -2344,15 +2344,15 @@ libload_qos_controller_la_SOURCES += slave/qos_controllers/load.cpp
libload_qos_controller_la_CPPFLAGS = $(MESOS_CPPFLAGS)
libload_qos_controller_la_LDFLAGS = $(MESOS_MODULE_LDFLAGS)
-# Library containing the URI volume profile module.
+# Library containing the URI disk profile module.
if ENABLE_GRPC
-pkgmodule_LTLIBRARIES += liburi_volume_profile.la
-liburi_volume_profile_la_SOURCES = \
- resource_provider/storage/uri_volume_profile.cpp \
- resource_provider/storage/uri_volume_profile.hpp \
- resource_provider/storage/volume_profile_utils.hpp
-liburi_volume_profile_la_CPPFLAGS = $(MESOS_CPPFLAGS)
-liburi_volume_profile_la_LDFLAGS = $(MESOS_MODULE_LDFLAGS)
+pkgmodule_LTLIBRARIES += liburi_disk_profile.la
+liburi_disk_profile_la_SOURCES = \
+ resource_provider/storage/uri_disk_profile.cpp \
+ resource_provider/storage/uri_disk_profile.hpp \
+ resource_provider/storage/disk_profile_utils.hpp
+liburi_disk_profile_la_CPPFLAGS = $(MESOS_CPPFLAGS)
+liburi_disk_profile_la_LDFLAGS = $(MESOS_MODULE_LDFLAGS)
endif
MESOS_TEST_MODULE_LDFLAGS = $(MESOS_MODULE_LDFLAGS)
@@ -2658,7 +2658,7 @@ mesos_tests_SOURCES += \
tests/csi_client_tests.cpp \
tests/mock_csi_plugin.cpp \
tests/mock_csi_plugin.hpp \
- tests/volume_profile_tests.cpp
+ tests/disk_profile_tests.cpp
if OS_LINUX
mesos_tests_SOURCES += \
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/module/manager.cpp
----------------------------------------------------------------------
diff --git a/src/module/manager.cpp b/src/module/manager.cpp
index f038f05..75f70a7 100644
--- a/src/module/manager.cpp
+++ b/src/module/manager.cpp
@@ -81,7 +81,7 @@ void ModuleManager::initialize()
kindToVersion["ResourceEstimator"] = MESOS_VERSION;
kindToVersion["SecretResolver"] = MESOS_VERSION;
kindToVersion["TestModule"] = MESOS_VERSION;
- kindToVersion["VolumeProfileAdaptor"] = MESOS_VERSION;
+ kindToVersion["DiskProfileAdaptor"] = MESOS_VERSION;
// What happens then when Mesos is built with a certain version,
// 'kindToVersion' states a certain other minimum version, and a
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/disk_profile.cpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/disk_profile.cpp b/src/resource_provider/storage/disk_profile.cpp
new file mode 100644
index 0000000..053fb7a
--- /dev/null
+++ b/src/resource_provider/storage/disk_profile.cpp
@@ -0,0 +1,131 @@
+// 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 <string>
+
+#include <mesos/mesos.hpp>
+
+#include <mesos/module/disk_profile.hpp>
+
+#include <mesos/resource_provider/storage/disk_profile.hpp>
+
+#include <process/future.hpp>
+
+#include <stout/hashset.hpp>
+
+#include <csi/spec.hpp>
+
+#include "module/manager.hpp"
+
+using std::string;
+using std::tuple;
+
+using process::Failure;
+using process::Future;
+
+using google::protobuf::Map;
+
+namespace mesos {
+namespace internal {
+
+// The default implementation does nothing and always returns a Failure
+// whenever called.
+class DefaultDiskProfileAdaptor : public DiskProfileAdaptor
+{
+public:
+ DefaultDiskProfileAdaptor() {}
+
+ ~DefaultDiskProfileAdaptor() {}
+
+ virtual Future<DiskProfileAdaptor::ProfileInfo> translate(
+ const string& profile,
+ const string& csiPluginInfoType) override
+ {
+ return Failure("By default, disk profiles are not supported");
+ }
+
+ virtual Future<hashset<string>> watch(
+ const hashset<string>& knownProfiles,
+ const string& csiPluginInfoType) override
+ {
+ // If the input set of profiles is empty, that means the caller is in sync
+ // with this module. Hence, we return a future that will never be satisified
+ // because this module will never return a non-empty set of profiles.
+ if (knownProfiles.empty()) {
+ return Future<hashset<string>>();
+ }
+
+ return hashset<string>::EMPTY;
+ }
+};
+
+} // namespace internal {
+
+
+Try<DiskProfileAdaptor*> DiskProfileAdaptor::create(
+ const Option<string>& moduleName)
+{
+ if (moduleName.isNone()) {
+ LOG(INFO) << "Creating default disk profile adaptor module";
+ return new internal::DefaultDiskProfileAdaptor();
+ }
+
+ LOG(INFO)
+ << "Creating disk profile adaptor module '" << moduleName.get() << "'";
+
+ Try<DiskProfileAdaptor*> result =
+ modules::ModuleManager::create<DiskProfileAdaptor>(moduleName.get());
+
+ if (result.isError()) {
+ return Error(
+ "Failed to initialize disk profile adaptor module: "
+ + result.error());
+ }
+
+ return result;
+}
+
+
+// NOTE: This is a pointer because we avoid using non-POD types
+// as global variables.
+//
+// NOTE: This is a `weak_ptr` because the ownership of the module should
+// belong to the caller of the `create` method above. This will, for example,
+// allow tests to instantiate an Agent and subsequently destruct the Agent
+// without leaving a module behind in a global variable.
+static std::weak_ptr<DiskProfileAdaptor>* currentAdaptor = nullptr;
+
+
+void DiskProfileAdaptor::setAdaptor(
+ const std::shared_ptr<DiskProfileAdaptor>& adaptor)
+{
+ if (currentAdaptor != nullptr) {
+ delete currentAdaptor;
+ }
+
+ currentAdaptor = new std::weak_ptr<DiskProfileAdaptor>(adaptor);
+}
+
+
+std::shared_ptr<DiskProfileAdaptor> DiskProfileAdaptor::getAdaptor()
+{
+ // This method should never be called before `setAdaptor` has been called.
+ CHECK_NOTNULL(currentAdaptor);
+
+ return currentAdaptor->lock();
+}
+
+} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/disk_profile.proto
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/disk_profile.proto b/src/resource_provider/storage/disk_profile.proto
new file mode 100644
index 0000000..6cf1f8a
--- /dev/null
+++ b/src/resource_provider/storage/disk_profile.proto
@@ -0,0 +1,43 @@
+// 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.
+
+syntax = "proto3";
+
+import "csi.proto";
+
+package mesos.resource_provider;
+
+
+message DiskProfileMapping {
+ message CSIManifest {
+ // Capabilities used for creating, publishing, and validating volumes.
+ // This field is REQUIRED.
+ //
+ // NOTE: The name of this field is plural because some CSI requests
+ // support multiple capabilities. However, Mesos currently does not
+ // support this.
+ .csi.VolumeCapability volume_capabilities = 1;
+
+ // Parameters passed to the CSI CreateVolume RPC.
+ // This field is OPTIONAL.
+ map<string, string> create_parameters = 2;
+ }
+
+ // Each map entry associates a profile name (type string) with the CSI
+ // capabilities and parameters used to make specific CSI requests.
+ // This field is OPTIONAL.
+ map<string, CSIManifest> profile_matrix = 1;
+}
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/disk_profile_utils.cpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/disk_profile_utils.cpp b/src/resource_provider/storage/disk_profile_utils.cpp
new file mode 100644
index 0000000..c84b6e1
--- /dev/null
+++ b/src/resource_provider/storage/disk_profile_utils.cpp
@@ -0,0 +1,121 @@
+// 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 "resource_provider/storage/disk_profile_utils.hpp"
+
+#include <google/protobuf/util/json_util.h>
+
+#include <stout/bytes.hpp>
+#include <stout/foreach.hpp>
+
+using std::string;
+
+using mesos::resource_provider::DiskProfileMapping;
+
+namespace mesos{
+namespace internal {
+namespace profile {
+
+Try<DiskProfileMapping> parseDiskProfileMapping(
+ const string& data)
+{
+ // Use Google's JSON utility function to parse the JSON string.
+ DiskProfileMapping output;
+ google::protobuf::util::JsonParseOptions options;
+ options.ignore_unknown_fields = true;
+
+ google::protobuf::util::Status status =
+ google::protobuf::util::JsonStringToMessage(data, &output, options);
+
+ if (!status.ok()) {
+ return Error(
+ "Failed to parse DiskProfileMapping message: "
+ + status.ToString());
+ }
+
+ Option<Error> validation = validate(output);
+ if (validation.isSome()) {
+ return Error(
+ "Fetched profile mapping failed validation with: " + validation->message);
+ }
+
+ return output;
+}
+
+
+Option<Error> validate(const DiskProfileMapping& mapping)
+{
+ auto iterator = mapping.profile_matrix().begin();
+ while (iterator != mapping.profile_matrix().end()) {
+ if (!iterator->second.has_volume_capabilities()) {
+ return Error(
+ "Profile '" + iterator->first + "' is missing the required field "
+ + "'volume_capabilities");
+ }
+
+ Option<Error> capabilities =
+ validate(iterator->second.volume_capabilities());
+
+ if (capabilities.isSome()) {
+ return Error(
+ "Profile '" + iterator->first + "' VolumeCapabilities are invalid: "
+ + capabilities->message);
+ }
+
+ // NOTE: The `create_parameters` field is optional and needs no further
+ // validation after parsing.
+
+ iterator++;
+ }
+
+ return None();
+}
+
+
+// TODO(chhsiao): Move this to CSI validation implementation file.
+Option<Error> validate(const csi::VolumeCapability& capability)
+{
+ if (capability.has_mount()) {
+ // The total size of this repeated field may not exceed 4 KB.
+ //
+ // TODO(josephw): The specification does not state how this maximum
+ // size is calculated. So this check is conservative and does not
+ // include padding or array separators in the size calculation.
+ size_t size = 0;
+ foreach (const string& flag, capability.mount().mount_flags()) {
+ size += flag.size();
+ }
+
+ if (Bytes(size) > Kilobytes(4)) {
+ return Error("Size of 'mount_flags' may not exceed 4 KB");
+ }
+ }
+
+ if (!capability.has_access_mode()) {
+ return Error("'access_mode' is a required field");
+ }
+
+ if (capability.access_mode().mode() ==
+ csi::VolumeCapability::AccessMode::UNKNOWN) {
+ return Error("'access_mode.mode' is unknown or not set");
+ }
+
+ return None();
+}
+
+} // namespace profile {
+} // namespace internal {
+} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/disk_profile_utils.hpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/disk_profile_utils.hpp b/src/resource_provider/storage/disk_profile_utils.hpp
new file mode 100644
index 0000000..fa2c7ec
--- /dev/null
+++ b/src/resource_provider/storage/disk_profile_utils.hpp
@@ -0,0 +1,48 @@
+// 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 __RESOURCE_PROVIDER_URI_DISK_PROFILE_UTILS_HPP__
+#define __RESOURCE_PROVIDER_URI_DISK_PROFILE_UTILS_HPP__
+
+#include <stout/option.hpp>
+#include <stout/try.hpp>
+
+// ONLY USEFUL AFTER RUNNING PROTOC.
+#include "resource_provider/storage/disk_profile.pb.h"
+
+namespace mesos {
+namespace internal {
+namespace profile {
+
+// Helper for parsing a string as the expected data format.
+Try<resource_provider::DiskProfileMapping> parseDiskProfileMapping(
+ const std::string& data);
+
+
+// Checks the fields inside a `DiskProfileMapping` according to the
+// comments above the protobuf.
+Option<Error> validate(const resource_provider::DiskProfileMapping& mapping);
+
+
+// Checks the fields inside a `VolumeCapability` according to the
+// comments above the protobuf.
+Option<Error> validate(const csi::VolumeCapability& capability);
+
+} // namespace profile {
+} // namespace internal {
+} // namespace mesos {
+
+#endif // __RESOURCE_PROVIDER_URI_DISK_PROFILE_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/provider.cpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/provider.cpp b/src/resource_provider/storage/provider.cpp
index a9f007a..55f2e66 100644
--- a/src/resource_provider/storage/provider.cpp
+++ b/src/resource_provider/storage/provider.cpp
@@ -37,7 +37,7 @@
#include <mesos/type_utils.hpp>
#include <mesos/resource_provider/resource_provider.hpp>
-#include <mesos/resource_provider/storage/volume_profile.hpp>
+#include <mesos/resource_provider/storage/disk_profile.hpp>
#include <mesos/v1/resource_provider.hpp>
@@ -302,8 +302,8 @@ public:
resourceVersion(id::UUID::random()),
operationSequence("operation-sequence")
{
- volumeProfileAdaptor = VolumeProfileAdaptor::getAdaptor();
- CHECK_NOTNULL(volumeProfileAdaptor.get());
+ diskProfileAdaptor = DiskProfileAdaptor::getAdaptor();
+ CHECK_NOTNULL(diskProfileAdaptor.get());
}
StorageLocalResourceProviderProcess(
@@ -345,7 +345,7 @@ private:
const Resources& discovered);
// Helper for updating the profiles mapping upon receiving an updated
- // set of profiles from the VolumeProfileAdaptor module.
+ // set of profiles from the DiskProfileAdaptor module.
Future<Nothing> updateProfiles();
// Reconcile the storage pools upon profile updates.
@@ -373,7 +373,7 @@ private:
Future<string> createVolume(
const string& name,
const Bytes& capacity,
- const VolumeProfileAdaptor::ProfileInfo& profileInfo);
+ const DiskProfileAdaptor::ProfileInfo& profileInfo);
Future<Nothing> deleteVolume(const string& volumeId, bool preExisting);
Future<string> validateCapability(
const string& volumeId,
@@ -428,7 +428,7 @@ private:
const Option<string> authToken;
const bool strict;
- shared_ptr<VolumeProfileAdaptor> volumeProfileAdaptor;
+ shared_ptr<DiskProfileAdaptor> diskProfileAdaptor;
csi::Version csiVersion;
csi::VolumeCapability defaultMountCapability;
@@ -438,10 +438,10 @@ private:
Owned<v1::resource_provider::Driver> driver;
OperationStatusUpdateManager statusUpdateManager;
- // The mapping of known profiles fetched from the VolumeProfileAdaptor.
- hashmap<string, VolumeProfileAdaptor::ProfileInfo> profileInfos;
+ // The mapping of known profiles fetched from the DiskProfileAdaptor.
+ hashmap<string, DiskProfileAdaptor::ProfileInfo> profileInfos;
- // The last set of profile names fetched from the VolumeProfileAdaptor.
+ // The last set of profile names fetched from the DiskProfileAdaptor.
hashset<string> knownProfiles;
// True if a reconciliation of storage pools is happening.
@@ -631,12 +631,12 @@ Future<Nothing> StorageLocalResourceProviderProcess::recover()
<< "Failed to watch for VolumeprofileAdaptor: " << message;
};
- // Start watching the VolumeProfileAdaptor.
+ // Start watching the DiskProfileAdaptor.
// TODO(chhsiao): Consider retrying with backoff.
loop(
self(),
[=] {
- return volumeProfileAdaptor->watch(
+ return diskProfileAdaptor->watch(
knownProfiles,
info.storage().plugin().type())
.then(defer(self(), [=](const hashset<string>& profiles) {
@@ -961,24 +961,24 @@ Future<Nothing> StorageLocalResourceProviderProcess::recoverProfiles()
}
// If no pending operation uses any profile, there is no need to
- // recover any profile. Watching the VolumeProfileAdaptor will be
+ // recover any profile. Watching the DiskProfileAdaptor will be
// initiated later.
if (requiredProfiles.empty()) {
return Nothing();
}
LOG(INFO)
- << "Waiting for VolumeProfileAdaptor to recover profiles: "
+ << "Waiting for DiskProfileAdaptor to recover profiles: "
<< stringify(requiredProfiles);
- // The VolumeProfileAdapter module must at least have knowledge of
+ // The DiskProfileAdapter module must at least have knowledge of
// the required profiles. Because the module is initialized separately
// from this resource provider, we must watch the module until all
// required profiles have been recovered.
return loop(
self(),
[=] {
- return volumeProfileAdaptor->watch(
+ return diskProfileAdaptor->watch(
knownProfiles,
info.storage().plugin().type());
},
@@ -1265,9 +1265,9 @@ Future<Nothing> StorageLocalResourceProviderProcess::updateProfiles()
continue;
}
- futures.push_back(volumeProfileAdaptor->translate(
+ futures.push_back(diskProfileAdaptor->translate(
profile, info.storage().plugin().type())
- .then(defer(self(), [=](const VolumeProfileAdaptor::ProfileInfo& info) {
+ .then(defer(self(), [=](const DiskProfileAdaptor::ProfileInfo& info) {
profileInfos.put(profile, info);
return Nothing();
})));
@@ -2287,7 +2287,7 @@ Future<Nothing> StorageLocalResourceProviderProcess::nodeUnpublish(
Future<string> StorageLocalResourceProviderProcess::createVolume(
const string& name,
const Bytes& capacity,
- const VolumeProfileAdaptor::ProfileInfo& profileInfo)
+ const DiskProfileAdaptor::ProfileInfo& profileInfo)
{
// NOTE: This can only be called after `prepareControllerService`.
CHECK_SOME(controllerCapabilities);
@@ -2489,7 +2489,7 @@ Future<Resources> StorageLocalResourceProviderProcess::listVolumes()
.then(defer(self(), [=](const csi::ListVolumesResponse& response) {
Resources resources;
- // Recover volume profiles from the checkpointed state.
+ // Recover disk profiles from the checkpointed state.
hashmap<string, string> volumesToProfiles;
foreach (const Resource& resource, totalResources) {
if (resource.disk().source().has_id() &&
@@ -2541,7 +2541,7 @@ Future<Resources> StorageLocalResourceProviderProcess::getCapacities()
// TODO(chhsiao): Skip inactive profiles.
- const VolumeProfileAdaptor::ProfileInfo& profileInfo =
+ const DiskProfileAdaptor::ProfileInfo& profileInfo =
profileInfos.at(profile);
csi::GetCapacityRequest request;
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/uri_disk_profile.cpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/uri_disk_profile.cpp b/src/resource_provider/storage/uri_disk_profile.cpp
new file mode 100644
index 0000000..7372b29
--- /dev/null
+++ b/src/resource_provider/storage/uri_disk_profile.cpp
@@ -0,0 +1,321 @@
+// 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 "resource_provider/storage/uri_disk_profile.hpp"
+
+#include <map>
+#include <string>
+#include <tuple>
+
+#include <mesos/mesos.hpp>
+
+#include <mesos/module/disk_profile.hpp>
+
+#include <mesos/resource_provider/storage/disk_profile.hpp>
+
+#include <process/defer.hpp>
+#include <process/delay.hpp>
+#include <process/dispatch.hpp>
+#include <process/future.hpp>
+#include <process/owned.hpp>
+#include <process/socket.hpp>
+
+#include <stout/duration.hpp>
+#include <stout/error.hpp>
+#include <stout/json.hpp>
+#include <stout/option.hpp>
+#include <stout/protobuf.hpp>
+#include <stout/result.hpp>
+#include <stout/strings.hpp>
+
+#include <csi/spec.hpp>
+#include <csi/utils.hpp>
+
+#include "resource_provider/storage/disk_profile_utils.hpp"
+
+using namespace mesos;
+using namespace process;
+
+using std::map;
+using std::string;
+using std::tuple;
+
+using google::protobuf::Map;
+
+using mesos::resource_provider::DiskProfileMapping;
+
+namespace mesos {
+namespace internal {
+namespace profile {
+
+bool operator==(
+ const Map<string, string>& left,
+ const Map<string, string>& right) {
+ if (left.size() != right.size()) {
+ return false;
+ }
+
+ typename Map<string, string>::const_iterator iterator = left.begin();
+ while (iterator != left.end()) {
+ if (right.count(iterator->first) != 1) {
+ return false;
+ }
+
+ if (iterator->second != right.at(iterator->first)) {
+ return false;
+ }
+ }
+
+ return true;
+}
+
+
+UriDiskProfileAdaptor::UriDiskProfileAdaptor(const Flags& _flags)
+ : flags(_flags),
+ process(new UriDiskProfileAdaptorProcess(flags))
+{
+ spawn(process.get());
+}
+
+
+UriDiskProfileAdaptor::~UriDiskProfileAdaptor()
+{
+ terminate(process.get());
+ wait(process.get());
+}
+
+
+Future<DiskProfileAdaptor::ProfileInfo> UriDiskProfileAdaptor::translate(
+ const string& profile,
+ const std::string& csiPluginInfoType)
+{
+ return dispatch(
+ process.get(),
+ &UriDiskProfileAdaptorProcess::translate,
+ profile,
+ csiPluginInfoType);
+}
+
+
+Future<hashset<string>> UriDiskProfileAdaptor::watch(
+ const hashset<string>& knownProfiles,
+ const std::string& csiPluginInfoType)
+{
+ return dispatch(
+ process.get(),
+ &UriDiskProfileAdaptorProcess::watch,
+ knownProfiles,
+ csiPluginInfoType);
+}
+
+
+UriDiskProfileAdaptorProcess::UriDiskProfileAdaptorProcess(
+ const Flags& _flags)
+ : ProcessBase(ID::generate("uri-volume-profile")),
+ flags(_flags),
+ watchPromise(new Promise<hashset<string>>()) {}
+
+
+void UriDiskProfileAdaptorProcess::initialize()
+{
+ poll();
+}
+
+
+Future<DiskProfileAdaptor::ProfileInfo>
+ UriDiskProfileAdaptorProcess::translate(
+ const string& profile,
+ const std::string& csiPluginInfoType)
+{
+ if (data.count(profile) != 1) {
+ return Failure("Profile '" + profile + "' not found");
+ }
+
+ return data.at(profile);
+}
+
+
+Future<hashset<string>> UriDiskProfileAdaptorProcess::watch(
+ const hashset<string>& knownProfiles,
+ const std::string& csiPluginInfoType)
+{
+ if (profiles != knownProfiles) {
+ return profiles;
+ }
+
+ return watchPromise->future();
+}
+
+
+void UriDiskProfileAdaptorProcess::poll()
+{
+ // NOTE: The flags do not allow relative paths, so this is guaranteed to
+ // be either 'http://' or 'https://'.
+ if (strings::startsWith(flags.uri, "http")) {
+ // NOTE: We already validated that this URI is parsable in the flags.
+ Try<http::URL> url = http::URL::parse(flags.uri.string());
+ CHECK_SOME(url);
+
+ http::get(url.get())
+ .onAny(defer(self(), [=](const Future<http::Response>& future) {
+ if (future.isReady()) {
+ // NOTE: We don't check the HTTP status code because we don't know
+ // what potential codes are considered successful.
+ _poll(future->body);
+ } else if (future.isFailed()) {
+ _poll(Error(future.failure()));
+ } else {
+ _poll(Error("Future discarded or abandoned"));
+ }
+ }));
+ } else {
+ _poll(os::read(flags.uri.string()));
+ }
+}
+
+
+void UriDiskProfileAdaptorProcess::_poll(const Try<string>& fetched)
+{
+ if (fetched.isSome()) {
+ Try<DiskProfileMapping> parsed = parseDiskProfileMapping(fetched.get());
+
+ if (parsed.isSome()) {
+ notify(parsed.get());
+ } else {
+ LOG(ERROR) << "Failed to parse result: " << parsed.error();
+ }
+ } else {
+ LOG(WARNING) << "Failed to poll URI: " << fetched.error();
+ }
+
+ // TODO(josephw): Do we want to retry if polling fails and no polling
+ // interval is set? Or perhaps we should exit in that case?
+ if (flags.poll_interval.isSome()) {
+ delay(flags.poll_interval.get(), self(), &Self::poll);
+ }
+}
+
+
+void UriDiskProfileAdaptorProcess::notify(
+ const DiskProfileMapping& parsed)
+{
+ bool hasErrors = false;
+
+ foreachkey (const string& profile, data) {
+ if (parsed.profile_matrix().count(profile) != 1) {
+ hasErrors = true;
+
+ LOG(WARNING)
+ << "Fetched profile mapping does not contain profile '" << profile
+ << "'. The fetched mapping will be ignored entirely";
+ continue;
+ }
+
+ bool matchingCapability =
+ data.at(profile).capability ==
+ parsed.profile_matrix().at(profile).volume_capabilities();
+
+ bool matchingParameters =
+ data.at(profile).parameters ==
+ parsed.profile_matrix().at(profile).create_parameters();
+
+ if (!matchingCapability || !matchingParameters) {
+ hasErrors = true;
+
+ LOG(WARNING)
+ << "Fetched profile mapping for profile '" << profile << "'"
+ << " does not match earlier data."
+ << " The fetched mapping will be ignored entirely";
+ }
+ }
+
+ // When encountering a data conflict, this module assumes there is a
+ // problem upstream (i.e. in the `--uri`). It is up to the operator
+ // to notice and resolve this.
+ if (hasErrors) {
+ return;
+ }
+
+ // Profiles can only be added, so if the parsed data is the same size,
+ // nothing has changed and no notifications need to be sent.
+ if (parsed.profile_matrix().size() <= data.size()) {
+ return;
+ }
+
+ // The fetched mapping satisfies our invariants.
+
+ // Save the protobuf as a map we can expose through the module interface.
+ // And update the convenience set of profile names.
+ profiles.clear();
+ auto iterator = parsed.profile_matrix().begin();
+ while (iterator != parsed.profile_matrix().end()) {
+ data[iterator->first] = {
+ iterator->second.volume_capabilities(),
+ iterator->second.create_parameters()
+ };
+
+ profiles.insert(iterator->first);
+ iterator++;
+ }
+
+ // Notify any watchers and then prepare a new promise for the next
+ // iteration of polling.
+ //
+ // TODO(josephw): Delay this based on the `--max_random_wait` option.
+ watchPromise->set(profiles);
+ watchPromise.reset(new Promise<hashset<string>>());
+
+ LOG(INFO)
+ << "Updated disk profile mapping to " << profiles.size()
+ << " total profiles";
+}
+
+} // namespace profile {
+} // namespace internal {
+} // namespace mesos {
+
+
+mesos::modules::Module<DiskProfileAdaptor>
+org_apache_mesos_UriDiskProfileAdaptor(
+ MESOS_MODULE_API_VERSION,
+ MESOS_VERSION,
+ "Apache Mesos",
+ "modules@mesos.apache.org",
+ "URI Disk Profile Adaptor module.",
+ nullptr,
+ [](const Parameters& parameters) -> DiskProfileAdaptor* {
+ // Convert `parameters` into a map.
+ map<string, string> values;
+ foreach (const Parameter& parameter, parameters.parameter()) {
+ values[parameter.key()] = parameter.value();
+ }
+
+ // Load and validate flags from the map.
+ mesos::internal::profile::Flags flags;
+ Try<flags::Warnings> load = flags.load(values);
+
+ if (load.isError()) {
+ LOG(ERROR) << "Failed to parse parameters: " << load.error();
+ return nullptr;
+ }
+
+ // Log any flag warnings.
+ foreach (const flags::Warning& warning, load->warnings) {
+ LOG(WARNING) << warning.message;
+ }
+
+ return new mesos::internal::profile::UriDiskProfileAdaptor(flags);
+ });
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/uri_disk_profile.hpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/uri_disk_profile.hpp b/src/resource_provider/storage/uri_disk_profile.hpp
new file mode 100644
index 0000000..2f4fc7c
--- /dev/null
+++ b/src/resource_provider/storage/uri_disk_profile.hpp
@@ -0,0 +1,248 @@
+// 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 __RESOURCE_PROVIDER_URI_DISK_PROFILE_HPP__
+#define __RESOURCE_PROVIDER_URI_DISK_PROFILE_HPP__
+
+#include <map>
+#include <string>
+#include <tuple>
+
+#include <mesos/resource_provider/storage/disk_profile.hpp>
+
+#include <process/future.hpp>
+#include <process/owned.hpp>
+#include <process/process.hpp>
+
+#include <process/ssl/flags.hpp>
+
+#include <stout/duration.hpp>
+#include <stout/error.hpp>
+#include <stout/flags.hpp>
+#include <stout/option.hpp>
+#include <stout/path.hpp>
+#include <stout/strings.hpp>
+
+#include <csi/spec.hpp>
+
+#include "resource_provider/storage/disk_profile_utils.hpp"
+
+namespace mesos {
+namespace internal {
+namespace profile {
+
+// Forward declaration.
+class UriDiskProfileAdaptorProcess;
+
+struct Flags : public virtual flags::FlagsBase
+{
+ Flags()
+ {
+ add(&Flags::uri,
+ "uri",
+ None(),
+ "URI to a JSON object containing the disk profile mapping.\n"
+ "This module supports both HTTP(s) and file URIs\n."
+ "\n"
+ "The JSON object should consist of some top-level string keys\n"
+ "corresponding to the disk profile name. Each value should\n"
+ "contain a `VolumeCapability` under a 'volume_capabilities'\n"
+ "and a free-form string-string mapping under 'create_parameters'.\n"
+ "\n"
+ "The JSON is modeled after a protobuf found in\n"
+ "`src/csi/uri_disk_profile.proto`.\n"
+ "\n"
+ "For example:\n"
+ "{\n"
+ " \"profile_matrix\" : {\n"
+ " \"my-profile\" : {\n"
+ " \"volume_capabilities\" : {\n"
+ " \"block\" : {},\n"
+ " \"access_mode\" : { \"mode\" : \"SINGLE_NODE_WRITER\" }\n"
+ " },\n"
+ " \"create_parameters\" : {\n"
+ " \"mesos-does-not\" : \"interpret-these\",\n"
+ " \"type\" : \"raid5\",\n"
+ " \"stripes\" : \"3\",\n"
+ " \"stripesize\" : \"64\"\n"
+ " }\n"
+ " }\n"
+ " }\n"
+ "}",
+ static_cast<const Path*>(nullptr),
+ [](const Path& value) -> Option<Error> {
+ // For now, just check if the URI has a supported scheme.
+ //
+ // TODO(josephw): Once we have a proper URI class and parser,
+ // consider validating this URI more thoroughly.
+ if (strings::startsWith(value.string(), "http://")
+#ifdef USE_SSL_SOCKET
+ || (process::network::openssl::flags().enabled &&
+ strings::startsWith(value.string(), "https://"))
+#endif // USE_SSL_SOCKET
+ ) {
+ Try<process::http::URL> url =
+ process::http::URL::parse(value.string());
+
+ if (url.isError()) {
+ return Error("Failed to parse URI: " + url.error());
+ }
+
+ return None();
+ }
+
+ // NOTE: The `Path` class will strip off the 'file://' prefix.
+ if (strings::contains(value.string(), "://")) {
+ return Error("--uri must use a supported scheme (file or http(s))");
+ }
+
+ // We only allow absolute paths for file paths.
+ if (!value.absolute()) {
+ return Error("--uri to a file must be an absolute path");
+ }
+
+ return None();
+ });
+
+ add(&Flags::poll_interval,
+ "poll_interval",
+ "How long to wait between polling the specified `--uri`.\n"
+ "The time is checked each time the `translate` method is called.\n"
+ "If the given time has elapsed, then the URI is re-fetched."
+ "If not specified, the URI is only fetched once.",
+ [](const Option<Duration>& value) -> Option<Error> {
+ if (value.isSome() && value.get() <= Seconds(0)) {
+ return Error("--poll_interval must be non-negative");
+ }
+
+ return None();
+ });
+
+ add(&Flags::max_random_wait,
+ "max_random_wait",
+ "How long at most to wait between discovering a new set of profiles\n"
+ "and notifying the callers of `watch`. The actual wait time is a\n"
+ "uniform random value between 0 and this value. If the `--uri` points\n"
+ "to a centralized location, it may be good to scale this number\n"
+ "according to the number of resource providers in the cluster.",
+ Seconds(0),
+ [](const Duration& value) -> Option<Error> {
+ if (value < Seconds(0)) {
+ return Error("--max_random_wait must be zero or greater");
+ }
+
+ return None();
+ });
+ }
+
+ // NOTE: We use the `Path` type here so that the stout flags parser
+ // does not attempt to read a file if given a `file://` prefixed value.
+ //
+ // TODO(josephw): Replace with a URI type when stout gets one.
+ Path uri;
+
+ Option<Duration> poll_interval;
+ Duration max_random_wait;
+};
+
+
+// The `UriDiskProfileAdaptor` is an example DiskProfile module that
+// takes a URI as a module parameter and fetches that URI periodically.
+// The fetched data is parsed into the required CSI protobufs
+// (which also acts as validation).
+//
+// If there is an error during fetching, any previously fetched results
+// will be used until fetching is successful.
+//
+// This module does not filter return results based on `CSIPluginInfo::type`
+// and assumes that all fetched profiles are meant for all resource providers.
+//
+// See `Flags` above for more information.
+class UriDiskProfileAdaptor : public mesos::DiskProfileAdaptor
+{
+public:
+ UriDiskProfileAdaptor(const Flags& _flags);
+
+ virtual ~UriDiskProfileAdaptor();
+
+ virtual process::Future<mesos::DiskProfileAdaptor::ProfileInfo> translate(
+ const std::string& profile,
+ const std::string& csiPluginInfoType) override;
+
+ virtual process::Future<hashset<std::string>> watch(
+ const hashset<std::string>& knownProfiles,
+ const std::string& csiPluginInfoType) override;
+
+protected:
+ Flags flags;
+ process::Owned<UriDiskProfileAdaptorProcess> process;
+};
+
+
+class UriDiskProfileAdaptorProcess :
+ public process::Process<UriDiskProfileAdaptorProcess>
+{
+public:
+ UriDiskProfileAdaptorProcess(const Flags& _flags);
+
+ virtual void initialize() override;
+
+ process::Future<mesos::DiskProfileAdaptor::ProfileInfo> translate(
+ const std::string& profile,
+ const std::string& csiPluginInfoType);
+
+ process::Future<hashset<std::string>> watch(
+ const hashset<std::string>& knownProfiles,
+ const std::string& csiPluginInfoType);
+
+private:
+ // Helpers for fetching the `--uri`.
+ // If `--poll_interval` is set, this method will dispatch to itself with
+ // a delay once the fetch is complete.
+ void poll();
+ void _poll(const Try<std::string>& fetched);
+
+ // Helper that is called upon successfully polling and parsing the `--uri`.
+ // This method will check the following conditions before updating the state
+ // of the module:
+ // * All known profiles must be included in the updated set.
+ // * All properties of known profiles must match those in the updated set.
+ void notify(const resource_provider::DiskProfileMapping& parsed);
+
+private:
+ Flags flags;
+
+ // The last fetched profile mapping.
+ // This module assumes that profiles can only be added and never removed.
+ // Once added, profiles cannot be changed either.
+ //
+ // TODO(josephw): Consider persisting this mapping across agent restarts.
+ std::map<std::string, DiskProfileAdaptor::ProfileInfo> data;
+
+ // Convenience set of the keys in `data` above.
+ // This module does not filter based on `CSIPluginInfo::type`, so this
+ // is valid for all input to `watch(...)`.
+ hashset<std::string> profiles;
+
+ // Will be satisfied whenever `data` is changed.
+ process::Owned<process::Promise<hashset<std::string>>> watchPromise;
+};
+
+} // namespace profile {
+} // namespace internal {
+} // namespace mesos {
+
+#endif // __RESOURCE_PROVIDER_URI_DISK_PROFILE_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/uri_volume_profile.cpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/uri_volume_profile.cpp b/src/resource_provider/storage/uri_volume_profile.cpp
deleted file mode 100644
index 4431ad1..0000000
--- a/src/resource_provider/storage/uri_volume_profile.cpp
+++ /dev/null
@@ -1,321 +0,0 @@
-// 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 "resource_provider/storage/uri_volume_profile.hpp"
-
-#include <map>
-#include <string>
-#include <tuple>
-
-#include <mesos/mesos.hpp>
-
-#include <mesos/module/volume_profile.hpp>
-
-#include <mesos/resource_provider/storage/volume_profile.hpp>
-
-#include <process/defer.hpp>
-#include <process/delay.hpp>
-#include <process/dispatch.hpp>
-#include <process/future.hpp>
-#include <process/owned.hpp>
-#include <process/socket.hpp>
-
-#include <stout/duration.hpp>
-#include <stout/error.hpp>
-#include <stout/json.hpp>
-#include <stout/option.hpp>
-#include <stout/protobuf.hpp>
-#include <stout/result.hpp>
-#include <stout/strings.hpp>
-
-#include <csi/spec.hpp>
-#include <csi/utils.hpp>
-
-#include "resource_provider/storage/volume_profile_utils.hpp"
-
-using namespace mesos;
-using namespace process;
-
-using std::map;
-using std::string;
-using std::tuple;
-
-using google::protobuf::Map;
-
-using mesos::resource_provider::VolumeProfileMapping;
-
-namespace mesos {
-namespace internal {
-namespace profile {
-
-bool operator==(
- const Map<string, string>& left,
- const Map<string, string>& right) {
- if (left.size() != right.size()) {
- return false;
- }
-
- typename Map<string, string>::const_iterator iterator = left.begin();
- while (iterator != left.end()) {
- if (right.count(iterator->first) != 1) {
- return false;
- }
-
- if (iterator->second != right.at(iterator->first)) {
- return false;
- }
- }
-
- return true;
-}
-
-
-UriVolumeProfileAdaptor::UriVolumeProfileAdaptor(const Flags& _flags)
- : flags(_flags),
- process(new UriVolumeProfileAdaptorProcess(flags))
-{
- spawn(process.get());
-}
-
-
-UriVolumeProfileAdaptor::~UriVolumeProfileAdaptor()
-{
- terminate(process.get());
- wait(process.get());
-}
-
-
-Future<VolumeProfileAdaptor::ProfileInfo> UriVolumeProfileAdaptor::translate(
- const string& profile,
- const std::string& csiPluginInfoType)
-{
- return dispatch(
- process.get(),
- &UriVolumeProfileAdaptorProcess::translate,
- profile,
- csiPluginInfoType);
-}
-
-
-Future<hashset<string>> UriVolumeProfileAdaptor::watch(
- const hashset<string>& knownProfiles,
- const std::string& csiPluginInfoType)
-{
- return dispatch(
- process.get(),
- &UriVolumeProfileAdaptorProcess::watch,
- knownProfiles,
- csiPluginInfoType);
-}
-
-
-UriVolumeProfileAdaptorProcess::UriVolumeProfileAdaptorProcess(
- const Flags& _flags)
- : ProcessBase(ID::generate("uri-volume-profile")),
- flags(_flags),
- watchPromise(new Promise<hashset<string>>()) {}
-
-
-void UriVolumeProfileAdaptorProcess::initialize()
-{
- poll();
-}
-
-
-Future<VolumeProfileAdaptor::ProfileInfo>
- UriVolumeProfileAdaptorProcess::translate(
- const string& profile,
- const std::string& csiPluginInfoType)
-{
- if (data.count(profile) != 1) {
- return Failure("Profile '" + profile + "' not found");
- }
-
- return data.at(profile);
-}
-
-
-Future<hashset<string>> UriVolumeProfileAdaptorProcess::watch(
- const hashset<string>& knownProfiles,
- const std::string& csiPluginInfoType)
-{
- if (profiles != knownProfiles) {
- return profiles;
- }
-
- return watchPromise->future();
-}
-
-
-void UriVolumeProfileAdaptorProcess::poll()
-{
- // NOTE: The flags do not allow relative paths, so this is guaranteed to
- // be either 'http://' or 'https://'.
- if (strings::startsWith(flags.uri, "http")) {
- // NOTE: We already validated that this URI is parsable in the flags.
- Try<http::URL> url = http::URL::parse(flags.uri.string());
- CHECK_SOME(url);
-
- http::get(url.get())
- .onAny(defer(self(), [=](const Future<http::Response>& future) {
- if (future.isReady()) {
- // NOTE: We don't check the HTTP status code because we don't know
- // what potential codes are considered successful.
- _poll(future->body);
- } else if (future.isFailed()) {
- _poll(Error(future.failure()));
- } else {
- _poll(Error("Future discarded or abandoned"));
- }
- }));
- } else {
- _poll(os::read(flags.uri.string()));
- }
-}
-
-
-void UriVolumeProfileAdaptorProcess::_poll(const Try<string>& fetched)
-{
- if (fetched.isSome()) {
- Try<VolumeProfileMapping> parsed = parseVolumeProfileMapping(fetched.get());
-
- if (parsed.isSome()) {
- notify(parsed.get());
- } else {
- LOG(ERROR) << "Failed to parse result: " << parsed.error();
- }
- } else {
- LOG(WARNING) << "Failed to poll URI: " << fetched.error();
- }
-
- // TODO(josephw): Do we want to retry if polling fails and no polling
- // interval is set? Or perhaps we should exit in that case?
- if (flags.poll_interval.isSome()) {
- delay(flags.poll_interval.get(), self(), &Self::poll);
- }
-}
-
-
-void UriVolumeProfileAdaptorProcess::notify(
- const VolumeProfileMapping& parsed)
-{
- bool hasErrors = false;
-
- foreachkey (const string& profile, data) {
- if (parsed.profile_matrix().count(profile) != 1) {
- hasErrors = true;
-
- LOG(WARNING)
- << "Fetched profile mapping does not contain profile '" << profile
- << "'. The fetched mapping will be ignored entirely";
- continue;
- }
-
- bool matchingCapability =
- data.at(profile).capability ==
- parsed.profile_matrix().at(profile).volume_capabilities();
-
- bool matchingParameters =
- data.at(profile).parameters ==
- parsed.profile_matrix().at(profile).create_parameters();
-
- if (!matchingCapability || !matchingParameters) {
- hasErrors = true;
-
- LOG(WARNING)
- << "Fetched profile mapping for profile '" << profile << "'"
- << " does not match earlier data."
- << " The fetched mapping will be ignored entirely";
- }
- }
-
- // When encountering a data conflict, this module assumes there is a
- // problem upstream (i.e. in the `--uri`). It is up to the operator
- // to notice and resolve this.
- if (hasErrors) {
- return;
- }
-
- // Profiles can only be added, so if the parsed data is the same size,
- // nothing has changed and no notifications need to be sent.
- if (parsed.profile_matrix().size() <= data.size()) {
- return;
- }
-
- // The fetched mapping satisfies our invariants.
-
- // Save the protobuf as a map we can expose through the module interface.
- // And update the convenience set of profile names.
- profiles.clear();
- auto iterator = parsed.profile_matrix().begin();
- while (iterator != parsed.profile_matrix().end()) {
- data[iterator->first] = {
- iterator->second.volume_capabilities(),
- iterator->second.create_parameters()
- };
-
- profiles.insert(iterator->first);
- iterator++;
- }
-
- // Notify any watchers and then prepare a new promise for the next
- // iteration of polling.
- //
- // TODO(josephw): Delay this based on the `--max_random_wait` option.
- watchPromise->set(profiles);
- watchPromise.reset(new Promise<hashset<string>>());
-
- LOG(INFO)
- << "Updated volume profile mapping to " << profiles.size()
- << " total profiles";
-}
-
-} // namespace profile {
-} // namespace internal {
-} // namespace mesos {
-
-
-mesos::modules::Module<VolumeProfileAdaptor>
-org_apache_mesos_UriVolumeProfileAdaptor(
- MESOS_MODULE_API_VERSION,
- MESOS_VERSION,
- "Apache Mesos",
- "modules@mesos.apache.org",
- "URI Volume Profile Adaptor module.",
- nullptr,
- [](const Parameters& parameters) -> VolumeProfileAdaptor* {
- // Convert `parameters` into a map.
- map<string, string> values;
- foreach (const Parameter& parameter, parameters.parameter()) {
- values[parameter.key()] = parameter.value();
- }
-
- // Load and validate flags from the map.
- mesos::internal::profile::Flags flags;
- Try<flags::Warnings> load = flags.load(values);
-
- if (load.isError()) {
- LOG(ERROR) << "Failed to parse parameters: " << load.error();
- return nullptr;
- }
-
- // Log any flag warnings.
- foreach (const flags::Warning& warning, load->warnings) {
- LOG(WARNING) << warning.message;
- }
-
- return new mesos::internal::profile::UriVolumeProfileAdaptor(flags);
- });
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/uri_volume_profile.hpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/uri_volume_profile.hpp b/src/resource_provider/storage/uri_volume_profile.hpp
deleted file mode 100644
index a58758c..0000000
--- a/src/resource_provider/storage/uri_volume_profile.hpp
+++ /dev/null
@@ -1,248 +0,0 @@
-// 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 __RESOURCE_PROVIDER_URI_VOLUME_PROFILE_HPP__
-#define __RESOURCE_PROVIDER_URI_VOLUME_PROFILE_HPP__
-
-#include <map>
-#include <string>
-#include <tuple>
-
-#include <mesos/resource_provider/storage/volume_profile.hpp>
-
-#include <process/future.hpp>
-#include <process/owned.hpp>
-#include <process/process.hpp>
-
-#include <process/ssl/flags.hpp>
-
-#include <stout/duration.hpp>
-#include <stout/error.hpp>
-#include <stout/flags.hpp>
-#include <stout/option.hpp>
-#include <stout/path.hpp>
-#include <stout/strings.hpp>
-
-#include <csi/spec.hpp>
-
-#include "resource_provider/storage/volume_profile_utils.hpp"
-
-namespace mesos {
-namespace internal {
-namespace profile {
-
-// Forward declaration.
-class UriVolumeProfileAdaptorProcess;
-
-struct Flags : public virtual flags::FlagsBase
-{
- Flags()
- {
- add(&Flags::uri,
- "uri",
- None(),
- "URI to a JSON object containing the volume profile mapping.\n"
- "This module supports both HTTP(s) and file URIs\n."
- "\n"
- "The JSON object should consist of some top-level string keys\n"
- "corresponding to the volume profile name. Each value should\n"
- "contain a `VolumeCapability` under a 'volume_capabilities'\n"
- "and a free-form string-string mapping under 'create_parameters'.\n"
- "\n"
- "The JSON is modeled after a protobuf found in\n"
- "`src/csi/uri_volume_profile.proto`.\n"
- "\n"
- "For example:\n"
- "{\n"
- " \"profile_matrix\" : {\n"
- " \"my-profile\" : {\n"
- " \"volume_capabilities\" : {\n"
- " \"block\" : {},\n"
- " \"access_mode\" : { \"mode\" : \"SINGLE_NODE_WRITER\" }\n"
- " },\n"
- " \"create_parameters\" : {\n"
- " \"mesos-does-not\" : \"interpret-these\",\n"
- " \"type\" : \"raid5\",\n"
- " \"stripes\" : \"3\",\n"
- " \"stripesize\" : \"64\"\n"
- " }\n"
- " }\n"
- " }\n"
- "}",
- static_cast<const Path*>(nullptr),
- [](const Path& value) -> Option<Error> {
- // For now, just check if the URI has a supported scheme.
- //
- // TODO(josephw): Once we have a proper URI class and parser,
- // consider validating this URI more thoroughly.
- if (strings::startsWith(value.string(), "http://")
-#ifdef USE_SSL_SOCKET
- || (process::network::openssl::flags().enabled &&
- strings::startsWith(value.string(), "https://"))
-#endif // USE_SSL_SOCKET
- ) {
- Try<process::http::URL> url =
- process::http::URL::parse(value.string());
-
- if (url.isError()) {
- return Error("Failed to parse URI: " + url.error());
- }
-
- return None();
- }
-
- // NOTE: The `Path` class will strip off the 'file://' prefix.
- if (strings::contains(value.string(), "://")) {
- return Error("--uri must use a supported scheme (file or http(s))");
- }
-
- // We only allow absolute paths for file paths.
- if (!value.absolute()) {
- return Error("--uri to a file must be an absolute path");
- }
-
- return None();
- });
-
- add(&Flags::poll_interval,
- "poll_interval",
- "How long to wait between polling the specified `--uri`.\n"
- "The time is checked each time the `translate` method is called.\n"
- "If the given time has elapsed, then the URI is re-fetched."
- "If not specified, the URI is only fetched once.",
- [](const Option<Duration>& value) -> Option<Error> {
- if (value.isSome() && value.get() <= Seconds(0)) {
- return Error("--poll_interval must be non-negative");
- }
-
- return None();
- });
-
- add(&Flags::max_random_wait,
- "max_random_wait",
- "How long at most to wait between discovering a new set of profiles\n"
- "and notifying the callers of `watch`. The actual wait time is a\n"
- "uniform random value between 0 and this value. If the `--uri` points\n"
- "to a centralized location, it may be good to scale this number\n"
- "according to the number of resource providers in the cluster.",
- Seconds(0),
- [](const Duration& value) -> Option<Error> {
- if (value < Seconds(0)) {
- return Error("--max_random_wait must be zero or greater");
- }
-
- return None();
- });
- }
-
- // NOTE: We use the `Path` type here so that the stout flags parser
- // does not attempt to read a file if given a `file://` prefixed value.
- //
- // TODO(josephw): Replace with a URI type when stout gets one.
- Path uri;
-
- Option<Duration> poll_interval;
- Duration max_random_wait;
-};
-
-
-// The `UriVolumeProfileAdaptor` is an example VolumeProfile module that
-// takes a URI as a module parameter and fetches that URI periodically.
-// The fetched data is parsed into the required CSI protobufs
-// (which also acts as validation).
-//
-// If there is an error during fetching, any previously fetched results
-// will be used until fetching is successful.
-//
-// This module does not filter return results based on `CSIPluginInfo::type`
-// and assumes that all fetched profiles are meant for all resource providers.
-//
-// See `Flags` above for more information.
-class UriVolumeProfileAdaptor : public mesos::VolumeProfileAdaptor
-{
-public:
- UriVolumeProfileAdaptor(const Flags& _flags);
-
- virtual ~UriVolumeProfileAdaptor();
-
- virtual process::Future<mesos::VolumeProfileAdaptor::ProfileInfo> translate(
- const std::string& profile,
- const std::string& csiPluginInfoType) override;
-
- virtual process::Future<hashset<std::string>> watch(
- const hashset<std::string>& knownProfiles,
- const std::string& csiPluginInfoType) override;
-
-protected:
- Flags flags;
- process::Owned<UriVolumeProfileAdaptorProcess> process;
-};
-
-
-class UriVolumeProfileAdaptorProcess :
- public process::Process<UriVolumeProfileAdaptorProcess>
-{
-public:
- UriVolumeProfileAdaptorProcess(const Flags& _flags);
-
- virtual void initialize() override;
-
- process::Future<mesos::VolumeProfileAdaptor::ProfileInfo> translate(
- const std::string& profile,
- const std::string& csiPluginInfoType);
-
- process::Future<hashset<std::string>> watch(
- const hashset<std::string>& knownProfiles,
- const std::string& csiPluginInfoType);
-
-private:
- // Helpers for fetching the `--uri`.
- // If `--poll_interval` is set, this method will dispatch to itself with
- // a delay once the fetch is complete.
- void poll();
- void _poll(const Try<std::string>& fetched);
-
- // Helper that is called upon successfully polling and parsing the `--uri`.
- // This method will check the following conditions before updating the state
- // of the module:
- // * All known profiles must be included in the updated set.
- // * All properties of known profiles must match those in the updated set.
- void notify(const resource_provider::VolumeProfileMapping& parsed);
-
-private:
- Flags flags;
-
- // The last fetched profile mapping.
- // This module assumes that profiles can only be added and never removed.
- // Once added, profiles cannot be changed either.
- //
- // TODO(josephw): Consider persisting this mapping across agent restarts.
- std::map<std::string, VolumeProfileAdaptor::ProfileInfo> data;
-
- // Convenience set of the keys in `data` above.
- // This module does not filter based on `CSIPluginInfo::type`, so this
- // is valid for all input to `watch(...)`.
- hashset<std::string> profiles;
-
- // Will be satisfied whenever `data` is changed.
- process::Owned<process::Promise<hashset<std::string>>> watchPromise;
-};
-
-} // namespace profile {
-} // namespace internal {
-} // namespace mesos {
-
-#endif // __RESOURCE_PROVIDER_URI_VOLUME_PROFILE_HPP__
http://git-wip-us.apache.org/repos/asf/mesos/blob/87629bf3/src/resource_provider/storage/volume_profile.cpp
----------------------------------------------------------------------
diff --git a/src/resource_provider/storage/volume_profile.cpp b/src/resource_provider/storage/volume_profile.cpp
deleted file mode 100644
index 336220d..0000000
--- a/src/resource_provider/storage/volume_profile.cpp
+++ /dev/null
@@ -1,131 +0,0 @@
-// 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 <string>
-
-#include <mesos/mesos.hpp>
-
-#include <mesos/module/volume_profile.hpp>
-
-#include <mesos/resource_provider/storage/volume_profile.hpp>
-
-#include <process/future.hpp>
-
-#include <stout/hashset.hpp>
-
-#include <csi/spec.hpp>
-
-#include "module/manager.hpp"
-
-using std::string;
-using std::tuple;
-
-using process::Failure;
-using process::Future;
-
-using google::protobuf::Map;
-
-namespace mesos {
-namespace internal {
-
-// The default implementation does nothing and always returns a Failure
-// whenever called.
-class DefaultVolumeProfileAdaptor : public VolumeProfileAdaptor
-{
-public:
- DefaultVolumeProfileAdaptor() {}
-
- ~DefaultVolumeProfileAdaptor() {}
-
- virtual Future<VolumeProfileAdaptor::ProfileInfo> translate(
- const string& profile,
- const string& csiPluginInfoType) override
- {
- return Failure("By default, volume profiles are not supported");
- }
-
- virtual Future<hashset<string>> watch(
- const hashset<string>& knownProfiles,
- const string& csiPluginInfoType) override
- {
- // If the input set of profiles is empty, that means the caller is in sync
- // with this module. Hence, we return a future that will never be satisified
- // because this module will never return a non-empty set of profiles.
- if (knownProfiles.empty()) {
- return Future<hashset<string>>();
- }
-
- return hashset<string>::EMPTY;
- }
-};
-
-} // namespace internal {
-
-
-Try<VolumeProfileAdaptor*> VolumeProfileAdaptor::create(
- const Option<string>& moduleName)
-{
- if (moduleName.isNone()) {
- LOG(INFO) << "Creating default volume profile adaptor module";
- return new internal::DefaultVolumeProfileAdaptor();
- }
-
- LOG(INFO)
- << "Creating volume profile adaptor module '" << moduleName.get() << "'";
-
- Try<VolumeProfileAdaptor*> result =
- modules::ModuleManager::create<VolumeProfileAdaptor>(moduleName.get());
-
- if (result.isError()) {
- return Error(
- "Failed to initialize volume profile adaptor module: "
- + result.error());
- }
-
- return result;
-}
-
-
-// NOTE: This is a pointer because we avoid using non-POD types
-// as global variables.
-//
-// NOTE: This is a `weak_ptr` because the ownership of the module should
-// belong to the caller of the `create` method above. This will, for example,
-// allow tests to instantiate an Agent and subsequently destruct the Agent
-// without leaving a module behind in a global variable.
-static std::weak_ptr<VolumeProfileAdaptor>* currentAdaptor = nullptr;
-
-
-void VolumeProfileAdaptor::setAdaptor(
- const std::shared_ptr<VolumeProfileAdaptor>& adaptor)
-{
- if (currentAdaptor != nullptr) {
- delete currentAdaptor;
- }
-
- currentAdaptor = new std::weak_ptr<VolumeProfileAdaptor>(adaptor);
-}
-
-
-std::shared_ptr<VolumeProfileAdaptor> VolumeProfileAdaptor::getAdaptor()
-{
- // This method should never be called before `setAdaptor` has been called.
- CHECK_NOTNULL(currentAdaptor);
-
- return currentAdaptor->lock();
-}
-
-} // namespace mesos {