[mesos] 05/07: Wired up the `OfferConstraintsFilter` into the hierarchical allocator.

[asekretenko]

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

asekretenko pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git

commit e73306e3e9a6759e1fc17e1aefc9ae17b4848f2b
Author: Andrei Sekretenko <asekrete...@apache.org>
AuthorDate: Thu Jul 16 20:45:43 2020 +0200

    Wired up the `OfferConstraintsFilter` into the hierarchical allocator.
    
    Review: https://reviews.apache.org/r/72774
---
 src/master/allocator/mesos/hierarchical.cpp | 48 +++++++++++++++++++++--------
 src/master/allocator/mesos/hierarchical.hpp |  4 +++
 2 files changed, 40 insertions(+), 12 deletions(-)

diff --git a/src/master/allocator/mesos/hierarchical.cpp 
b/src/master/allocator/mesos/hierarchical.cpp
index b061a05..47f8a2b 100644
--- a/src/master/allocator/mesos/hierarchical.cpp
+++ b/src/master/allocator/mesos/hierarchical.cpp
@@ -606,7 +606,9 @@ Framework::Framework(
     active(_active),
     metrics(new FrameworkMetrics(frameworkInfo, publishPerFrameworkMetrics)),
     minAllocatableResources(
-        unpackFrameworkOfferFilters(frameworkInfo.offer_filters())) {}
+        unpackFrameworkOfferFilters(frameworkInfo.offer_filters())),
+    offerConstraintsFilter(std::move(options.offerConstraintsFilter))
+{}
 
 
 void HierarchicalAllocatorProcess::initialize(
@@ -726,12 +728,16 @@ void HierarchicalAllocatorProcess::addFramework(
   // TODO(mzhu): remove the `frameworkId` parameter.
   CHECK_EQ(frameworkId, frameworkInfo.id());
 
-  frameworks.insert({frameworkId,
+  // NOTE: We don't use insert({frameworkId, Framework()}) here because with
+  // older versions of libstdc++ the argument type of
+  // std::unordered_map::insert(P&&)is deduced to be a non-copyable non-movable
+  // std::pair<const FrameworkId, Framework>.
+  frameworks.emplace(frameworkId,
                      Framework(
                          frameworkInfo,
                          std::move(frameworkOptions),
                          active,
-                         options.publishPerFrameworkMetrics)});
+                         options.publishPerFrameworkMetrics));
 
   const Framework& framework = *CHECK_NOTNONE(getFramework(frameworkId));
 
@@ -929,6 +935,9 @@ void HierarchicalAllocatorProcess::updateFramework(
   framework.minAllocatableResources =
     unpackFrameworkOfferFilters(frameworkInfo.offer_filters());
 
+  framework.offerConstraintsFilter =
+    std::move(frameworkOptions.offerConstraintsFilter);
+
   suppressRoles(
       framework,
       frameworkOptions.suppressedRoles - oldSuppressedRoles);
@@ -2145,6 +2154,18 @@ void HierarchicalAllocatorProcess::__generateOffers()
         if (unsatisfiedQuotaGuarantees.empty()) {
           break;
         }
+        FrameworkID frameworkId;
+        frameworkId.set_value(frameworkId_);
+
+        const Framework& framework = *CHECK_NOTNONE(getFramework(frameworkId));
+        CHECK(framework.active) << frameworkId;
+
+        if (framework.offerConstraintsFilter.isSome() &&
+            framework.offerConstraintsFilter->isAgentExcluded(
+                role, slave.info)) {
+          // Framework filters the agent regardless of remaining resources.
+          continue;
+        }
 
         // Offer a shared resource only if it has not been offered in this
         // offer cycle to a framework.
@@ -2156,11 +2177,6 @@ void HierarchicalAllocatorProcess::__generateOffers()
           break; // Nothing left for the role.
         }
 
-        FrameworkID frameworkId;
-        frameworkId.set_value(frameworkId_);
-
-        const Framework& framework = *CHECK_NOTNONE(getFramework(frameworkId));
-        CHECK(framework.active) << frameworkId;
 
         // An early `continue` optimization.
         if (!allocatable(available, role, framework)) {
@@ -2375,6 +2391,18 @@ void HierarchicalAllocatorProcess::__generateOffers()
       Sorter* frameworkSorter = CHECK_NOTNONE(getFrameworkSorter(role));
 
       foreach (const string& frameworkId_, frameworkSorter->sort()) {
+        FrameworkID frameworkId;
+        frameworkId.set_value(frameworkId_);
+
+        const Framework& framework = *CHECK_NOTNONE(getFramework(frameworkId));
+
+        if (framework.offerConstraintsFilter.isSome() &&
+            framework.offerConstraintsFilter->isAgentExcluded(
+                role, slave.info)) {
+          // Framework filters the agent regardless of remaining resources.
+          continue;
+        }
+
         // Offer a shared resource only if it has not been offered in this
         // offer cycle to a framework.
         Resources available =
@@ -2385,10 +2413,6 @@ void HierarchicalAllocatorProcess::__generateOffers()
           break; // Nothing left for the role.
         }
 
-        FrameworkID frameworkId;
-        frameworkId.set_value(frameworkId_);
-
-        const Framework& framework = *CHECK_NOTNONE(getFramework(frameworkId));
 
         // An early `continue` optimization.
         if (!allocatable(available, role, framework)) {
diff --git a/src/master/allocator/mesos/hierarchical.hpp 
b/src/master/allocator/mesos/hierarchical.hpp
index 9c1932d..7e1980e 100644
--- a/src/master/allocator/mesos/hierarchical.hpp
+++ b/src/master/allocator/mesos/hierarchical.hpp
@@ -108,7 +108,11 @@ struct Framework
   process::Owned<FrameworkMetrics> metrics;
 
   // TODO(bbannier): Consider documenting examples on how to use this setting.
+  // TODO(asekretenko): reimplement minAllocatableResources in terms of
+  // offer constraints.
   hashmap<std::string, std::vector<ResourceQuantities>> 
minAllocatableResources;
+
+  Option<::mesos::allocator::OfferConstraintsFilter> offerConstraintsFilter;
 };