This is an automated email from the ASF dual-hosted git repository. mzhu pushed a commit to branch 1.6.x in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 360e0023fc89d64d67e3a5effc6faa28923818b8 Author: Meng Zhu <[email protected]> AuthorDate: Sun Dec 30 20:55:06 2018 -0800 Extended `min_allocatable_resources` flag to cover non-scalar resources. The master flag min_allocatable_resources currently only supports scalar resources. Sometimes it is desirable to control the quantities of non-scalar resources in the offer as well, for example, to ensure offers always contain port resources. This patch extended the flag to cover non-scalar resources. For RANGES and SET type resources, their quantities are the number of different instances in the range or set. For example, range:[1-5] has a quantity of 5 and set:{a,b} has a quantity of 2. This patch also updated affected tests and modified the existing test for min_allocatable_resources to cover the updated use cases. Review: https://reviews.apache.org/r/69603 --- docs/configuration/master.md | 22 +++--- include/mesos/allocator/allocator.hpp | 4 +- src/master/allocator/mesos/allocator.hpp | 25 +++---- src/master/allocator/mesos/hierarchical.cpp | 11 +-- src/master/allocator/mesos/hierarchical.hpp | 5 +- src/master/flags.cpp | 22 +++--- src/master/master.cpp | 54 +++++++------- src/tests/allocator.hpp | 2 +- src/tests/hierarchical_allocator_tests.cpp | 13 ++-- src/tests/master_allocator_tests.cpp | 106 ++++++++++++++++++++++++++++ 10 files changed, 192 insertions(+), 72 deletions(-) diff --git a/docs/configuration/master.md b/docs/configuration/master.md index eac505b..528ec14 100644 --- a/docs/configuration/master.md +++ b/docs/configuration/master.md @@ -197,15 +197,19 @@ load an alternate allocator module using <code>--modules</code>. --min_allocatable_resources=VALUE </td> <td> -One or more sets of resources that define the minimum allocatable -resources for the allocator. The allocator will only offer resources -that contain at least one of the specified sets. The resources in each -set should be delimited by semicolons, and the sets should be delimited -by the pipe character. -(Example: <code>disk:1|cpu:1;mem:32</code> -configures the allocator to only offer resources if they contain a disk -resource of at least 1 megabyte, or if they contain both 1 cpu and -32 megabytes of memory.) (default: cpus:0.01|mem:32). +One or more sets of resource quantities that define the minimum allocatable +resource for the allocator. The allocator will only offer resources that +meets the quantity requirement of at least one of the specified sets. For +`SCALAR` type resources, its quantity is its scalar value. For `RANGES` and +`SET` type, their quantities are the number of different instances in the +range or set. For example, `range:[1-5]` has a quantity of 5 and `set:{a,b}` +has a quantity of 2. The resources in each set should be delimited by +semicolons (acting as logical AND), and each set should be delimited by the +pipe character (acting as logical OR). +(Example: `disk:1|cpu:1;mem:32;port:1` configures the allocator to only offer +resources if they contain a disk resource of at least 1 megabyte, or if they +at least contain 1 cpu, 32 megabytes of memory and 1 port.) +(default: cpus:0.01|mem:32). </td> </tr> diff --git a/include/mesos/allocator/allocator.hpp b/include/mesos/allocator/allocator.hpp index c19ab64..df64faf 100644 --- a/include/mesos/allocator/allocator.hpp +++ b/include/mesos/allocator/allocator.hpp @@ -38,6 +38,8 @@ #include <stout/option.hpp> #include <stout/try.hpp> +#include "common/resource_quantities.hpp" + namespace mesos { namespace allocator { @@ -102,7 +104,7 @@ public: fairnessExcludeResourceNames = None(), bool filterGpuResources = true, const Option<DomainInfo>& domain = None(), - const Option<std::vector<Resources>>& + const Option<std::vector<mesos::internal::ResourceQuantities>>& minAllocatableResources = None()) = 0; /** diff --git a/src/master/allocator/mesos/allocator.hpp b/src/master/allocator/mesos/allocator.hpp index 900c8ee..32dd40c 100644 --- a/src/master/allocator/mesos/allocator.hpp +++ b/src/master/allocator/mesos/allocator.hpp @@ -48,19 +48,19 @@ public: void initialize( const Duration& allocationInterval, - const lambda::function< - void(const FrameworkID&, - const hashmap<std::string, hashmap<SlaveID, Resources>>&)>& - offerCallback, - const lambda::function< - void(const FrameworkID&, - const hashmap<SlaveID, UnavailableResources>&)>& + const lambda::function<void( + const FrameworkID&, + const hashmap<std::string, hashmap<SlaveID, Resources>>&)>& + offerCallback, + const lambda::function<void( + const FrameworkID&, const hashmap<SlaveID, UnavailableResources>&)>& inverseOfferCallback, - const Option<std::set<std::string>>& - fairnessExcludeResourceNames = None(), + const Option<std::set<std::string>>& fairnessExcludeResourceNames = + None(), bool filterGpuResources = true, const Option<DomainInfo>& domain = None(), - const Option<std::vector<Resources>>& minAllocatableResources = None()); + const Option<std::vector<mesos::internal::ResourceQuantities>>& + minAllocatableResources = None()); void recover( const int expectedAgentCount, @@ -207,7 +207,7 @@ public: fairnessExcludeResourceNames = None(), bool filterGpuResources = true, const Option<DomainInfo>& domain = None(), - const Option<std::vector<Resources>>& + const Option<std::vector<mesos::internal::ResourceQuantities>>& minAllocatableResources = None()) = 0; virtual void recover( @@ -364,7 +364,8 @@ inline void MesosAllocator<AllocatorProcess>::initialize( const Option<std::set<std::string>>& fairnessExcludeResourceNames, bool filterGpuResources, const Option<DomainInfo>& domain, - const Option<std::vector<Resources>>& minAllocatableResources) + const Option<std::vector<mesos::internal::ResourceQuantities>>& + minAllocatableResources) { process::dispatch( process, diff --git a/src/master/allocator/mesos/hierarchical.cpp b/src/master/allocator/mesos/hierarchical.cpp index f6aaa45..e63f886 100644 --- a/src/master/allocator/mesos/hierarchical.cpp +++ b/src/master/allocator/mesos/hierarchical.cpp @@ -41,6 +41,7 @@ #include <stout/stringify.hpp> #include "common/protobuf_utils.hpp" +#include "common/resource_quantities.hpp" using std::set; using std::string; @@ -156,7 +157,8 @@ void HierarchicalAllocatorProcess::initialize( const Option<set<string>>& _fairnessExcludeResourceNames, bool _filterGpuResources, const Option<DomainInfo>& _domain, - const Option<std::vector<Resources>>& _minAllocatableResources) + const Option<std::vector<mesos::internal::ResourceQuantities>>& + _minAllocatableResources) { allocationInterval = _allocationInterval; offerCallback = _offerCallback; @@ -2457,11 +2459,10 @@ bool HierarchicalAllocatorProcess::allocatable(const Resources& resources) return true; } - // We remove the static reservation metadata here via `toUnreserved()`. - Resources quantity = resources.createStrippedScalarQuantity().toUnreserved(); foreach ( - const Resources& minResources, CHECK_NOTNONE(minAllocatableResources)) { - if (quantity.contains(minResources)) { + const ResourceQuantities& resourceQuantities, + CHECK_NOTNONE(minAllocatableResources)) { + if (resources.contains(resourceQuantities)) { return true; } } diff --git a/src/master/allocator/mesos/hierarchical.hpp b/src/master/allocator/mesos/hierarchical.hpp index 0f6c0e9..396f201 100644 --- a/src/master/allocator/mesos/hierarchical.hpp +++ b/src/master/allocator/mesos/hierarchical.hpp @@ -112,7 +112,7 @@ public: fairnessExcludeResourceNames = None(), bool filterGpuResources = true, const Option<DomainInfo>& domain = None(), - const Option<std::vector<Resources>>& + const Option<std::vector<mesos::internal::ResourceQuantities>>& minAllocatableResources = None()); void recover( @@ -526,7 +526,8 @@ protected: Option<DomainInfo> domain; // The minimum allocatable resources, if any. - Option<std::vector<Resources>> minAllocatableResources; + Option<std::vector<mesos::internal::ResourceQuantities>> + minAllocatableResources; // There are two stages of allocation: // diff --git a/src/master/flags.cpp b/src/master/flags.cpp index 6ad53ed..38ba888 100644 --- a/src/master/flags.cpp +++ b/src/master/flags.cpp @@ -488,14 +488,20 @@ mesos::internal::master::Flags::Flags() add(&Flags::min_allocatable_resources, "min_allocatable_resources", - "One or more sets of resources that define the minimum allocatable\n" - "resources for the allocator. The allocator will only offer resources\n" - "that contain at least one of the specified sets. The resources in\n" - "each set should be delimited by semicolons, and the sets should be\n" - "delimited by the pipe character.\n" - "(Example: `disk:1|cpu:1;mem:32` configures the allocator to only offer\n" - "resources if they contain a disk resource of at least 1 megabyte, or\n" - "if they contain both 1 cpu and 32 megabytes of memory.)\n", + "One or more sets of resource quantities that define the minimum\n" + "allocatable resources for the allocator. The allocator will only offer\n" + "resources that meets the quantity requirement of at least one of the\n" + "specified sets. For `SCALAR` type resources, its quantity is its\n" + "scalar value. For `RANGES` and `SET` type, their quantities are the\n" + "number of different instances in the range or set. For example,\n" + "`range:[1-5]` has a quantity of 5 and `set:{a,b}` has a quantity of 2.\n" + "The resources in each set should be delimited by semicolons (acting as\n" + "logical AND), and each set should be delimited by the pipe character\n" + "(acting as logical OR).\n" + "(Example: `disk:1|cpu:1;mem:32;port:1` configures the allocator to\n" + "only offer resources if they contain a disk resource of at least\n" + "1 megabyte, or if they at least contain 1 cpu, 32 megabytes of memory\n" + "and 1 port.)\n", "cpus:" + stringify(MIN_CPUS) + "|mem:" + stringify((double)MIN_MEM.bytes() / Bytes::MEGABYTES)); diff --git a/src/master/master.cpp b/src/master/master.cpp index 2aaff39..8776663 100644 --- a/src/master/master.cpp +++ b/src/master/master.cpp @@ -78,6 +78,7 @@ #include "common/build.hpp" #include "common/http.hpp" #include "common/protobuf_utils.hpp" +#include "common/resource_quantities.hpp" #include "common/status_utils.hpp" #include "credentials/credentials.hpp" @@ -148,6 +149,7 @@ using mesos::master::contender::MasterContender; using mesos::master::detector::MasterDetector; +using mesos::internal::ResourceQuantities; static bool isValidFailoverTimeout(const FrameworkInfo& frameworkInfo); @@ -769,39 +771,31 @@ void Master::initialize() } // Parse min_allocatable_resources. - Try<vector<Resources>> minAllocatableResources = - [](const string& resourceString) -> Try<vector<Resources>> { - vector<Resources> result; - - foreach (const string& token, strings::tokenize(resourceString, "|")) { - Try<vector<Resource>> resourceVector = - Resources::fromSimpleString(token); + vector<ResourceQuantities> minAllocatableResources; + foreach ( + const string& token, + strings::tokenize(flags.min_allocatable_resources, "|")) { + Try<ResourceQuantities> resourceQuantities = + ResourceQuantities::fromString(token); - if (resourceVector.isError()) { - return Error(resourceVector.error()); - } + if (resourceQuantities.isError()) { + EXIT(EXIT_FAILURE) << "Error parsing min_allocatable_resources '" + << flags.min_allocatable_resources + << "': " << resourceQuantities.error(); + } - result.push_back(Resources(CHECK_NOTERROR(resourceVector))); + // We check the configuration against first-class resources and warn + // against possible mis-configuration (e.g. typo). + set<string> firstClassResources = {"cpus", "mem", "disk", "ports", "gpus"}; + for (auto it = resourceQuantities->begin(); it != resourceQuantities->end(); + ++it) { + if (firstClassResources.count(it->first) == 0) { + LOG(WARNING) << "Non-first-class resource '" << it->first + << "' is configured as part of min_allocatable_resources"; } - - return result; - }(flags.min_allocatable_resources); - - if (minAllocatableResources.isError()) { - EXIT(EXIT_FAILURE) << "Error parsing min_allocatable_resources: '" - << flags.min_allocatable_resources - << "': " << minAllocatableResources.error(); - } - - // Validate that configured minimum resources are "pure" scalar quantities. - foreach ( - const Resources& resources, CHECK_NOTERROR(minAllocatableResources)) { - if (!Resources::isScalarQuantity(resources)) { - EXIT(EXIT_FAILURE) << "Invalid min_allocatable_resources: '" - << flags.min_allocatable_resources << "': " - << "minimum allocatable resources should only" - << "have name, type (scalar) and value set"; } + + minAllocatableResources.push_back(resourceQuantities.get()); } // Initialize the allocator. @@ -812,7 +806,7 @@ void Master::initialize() flags.fair_sharing_excluded_resource_names, flags.filter_gpu_resources, flags.domain, - CHECK_NOTERROR(minAllocatableResources)); + minAllocatableResources); // Parse the whitelist. Passing Allocator::updateWhitelist() // callback is safe because we shut down the whitelistWatcher in diff --git a/src/tests/allocator.hpp b/src/tests/allocator.hpp index 73fc060..c57d208 100644 --- a/src/tests/allocator.hpp +++ b/src/tests/allocator.hpp @@ -379,7 +379,7 @@ public: const Option<std::set<std::string>>&, bool, const Option<DomainInfo>&, - const Option<std::vector<Resources>>&)); + const Option<std::vector<mesos::internal::ResourceQuantities>>&)); MOCK_METHOD2(recover, void( const int expectedAgentCount, diff --git a/src/tests/hierarchical_allocator_tests.cpp b/src/tests/hierarchical_allocator_tests.cpp index 597469e..5dc7815 100644 --- a/src/tests/hierarchical_allocator_tests.cpp +++ b/src/tests/hierarchical_allocator_tests.cpp @@ -40,6 +40,8 @@ #include <stout/stopwatch.hpp> #include <stout/utils.hpp> +#include "common/resource_quantities.hpp" + #include "master/constants.hpp" #include "master/flags.hpp" @@ -61,6 +63,8 @@ using mesos::internal::protobuf::createLabel; using mesos::internal::slave::AGENT_CAPABILITIES; +using mesos::internal::ResourceQuantities; + using mesos::allocator::Allocator; using process::Clock; @@ -176,11 +180,12 @@ protected: }; } - vector<Resources> minAllocatableResources; + vector<ResourceQuantities> minAllocatableResources; + minAllocatableResources.push_back(CHECK_NOTERROR( + ResourceQuantities::fromString("cpus:" + stringify(MIN_CPUS)))); minAllocatableResources.push_back( - CHECK_NOTERROR(Resources::parse("cpus:" + stringify(MIN_CPUS)))); - minAllocatableResources.push_back(CHECK_NOTERROR(Resources::parse( - "mem:" + stringify((double)MIN_MEM.bytes() / Bytes::MEGABYTES)))); + CHECK_NOTERROR(ResourceQuantities::fromString( + "mem:" + stringify((double)MIN_MEM.bytes() / Bytes::MEGABYTES)))); allocator->initialize( flags.allocation_interval, diff --git a/src/tests/master_allocator_tests.cpp b/src/tests/master_allocator_tests.cpp index 58821a4..72c549f 100644 --- a/src/tests/master_allocator_tests.cpp +++ b/src/tests/master_allocator_tests.cpp @@ -1235,6 +1235,112 @@ TYPED_TEST(MasterAllocatorTest, MemoryOnlyOfferedAndTaskLaunched) } +// This test verifies that the allocator honors the `min_allocatable_resources` +// flag and only offers resources that are more than at least one of the +// specified resources quantities. +TYPED_TEST(MasterAllocatorTest, MinAllocatableResources) +{ + Clock::pause(); + + master::Flags masterFlags = this->CreateMasterFlags(); + masterFlags.min_allocatable_resources = + "cpus:1;mem:100|disk:100|cpus:1;ports:10"; + + TestAllocator<TypeParam> allocator; + + Try<Owned<cluster::Master>> master = + this->StartMaster(&allocator, masterFlags); + ASSERT_SOME(master); + + Owned<MasterDetector> detector = master.get()->createDetector(); + + // Prevent agents from destruction until end of test. + vector<Owned<cluster::Slave>> slaves; + + // Add agents with non-allocatable resources. + vector<string> nonAllocatableAgentResources = { + "cpus:1;mem:10;disk:10;ports:[1-5]", + "cpus:0.5;mem:100;disk:10;ports:[6-10]", + "cpus:0.5;mem:200;disk:10;ports:[11-15]", + "cpus:0.5;mem:10;disk:50;ports:[16-20]"}; + + foreach (const string& resourcesString, nonAllocatableAgentResources) { + Future<SlaveRegisteredMessage> slaveRegisteredMessage = + FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _); + + slave::Flags flags = this->CreateSlaveFlags(); + flags.resources = resourcesString; + + Try<Owned<cluster::Slave>> slave = this->StartSlave(detector.get(), flags); + ASSERT_SOME(slave); + + // Advance the clock to trigger agent registration. + Clock::advance(flags.registration_backoff_factor); + Clock::settle(); + + AWAIT_READY(slaveRegisteredMessage); + + slaves.push_back(slave.get()); + } + + // Add agents with allocatable resources. + vector<string> allocatableAgentResources = { + "cpus:1;mem:100;disk:0;ports:[1-5]", // contains `cpus:1;mem:100` + "cpus:0.5;mem:10;disk:100;ports:[1-5]", // contains `disk:100` + "cpus:1;mem:100;disk:50", // contains `cpus:1;mem:100` + "cpus:1;mem:10;disk:10;ports:[1-10]", // contains `cpus:1;ports:10` + }; + + hashset<SlaveID> allocatableAgents; + foreach (const string& resourcesString, allocatableAgentResources) { + Future<SlaveRegisteredMessage> slaveRegisteredMessage = + FUTURE_PROTOBUF(SlaveRegisteredMessage(), _, _); + + slave::Flags flags = this->CreateSlaveFlags(); + flags.resources = resourcesString; + + Try<Owned<cluster::Slave>> slave = this->StartSlave(detector.get(), flags); + ASSERT_SOME(slave); + + // Advance the clock to trigger agent registration. + Clock::advance(flags.registration_backoff_factor); + Clock::settle(); + + AWAIT_READY(slaveRegisteredMessage); + + slaves.push_back(slave.get()); + allocatableAgents.insert(slaveRegisteredMessage->slave_id()); + } + + // Add a framework. This will trigger an event-driven allocation. + // Since it is the only framework, it will get all offers. + MockScheduler sched; + MesosSchedulerDriver driver( + &sched, DEFAULT_FRAMEWORK_INFO, master.get()->pid, DEFAULT_CREDENTIAL); + + EXPECT_CALL(sched, registered(_, _, _)); + + Future<vector<Offer>> offers; + EXPECT_CALL(sched, resourceOffers(&driver, _)) + .WillOnce(FutureArg<1>(&offers)); + + driver.start(); + + AWAIT_READY(offers); + + // All allocatable agents are offered. None of the non-allocatable + // agents should be offered. + EXPECT_EQ(offers->size(), allocatableAgents.size()); + + foreach (const Offer& offer, offers.get()) { + EXPECT_TRUE(allocatableAgents.count(offer.slave_id()) != 0); + } + + driver.stop(); + driver.join(); +} + + // Checks that changes to the whitelist are sent to the allocator. // The allocator whitelisting is tested in the allocator unit tests. // TODO(bmahler): Move this to a whitelist unit test.
