Added a metric for the number of per-role active offer filters.

Review: https://reviews.apache.org/r/43883/

docs/monitoring.md

@@ -950,6 +950,13 @@ and resource allocations in the allocator.
   <td>Number of dispatch events in the event queue</td>
   <td>Gauge</td>
 </tr>
+<tr>
+  <td>
+  <code>allocator/mesos/offer_filters/roles/&lt;role&gt;/active</code>
+  </td>
+  <td>Number of active offer filters for all frameworks within the role</td>
+  <td>Gauge</td>
+</tr>
 <tr>
   <td>
   <code>allocator/mesos/quota/roles/&lt;role&gt;/resources/&lt;resource&gt;/offered_or_allocated</code>

src/master/allocator/mesos/hierarchical.cpp

@@ -226,6 +226,7 @@ void HierarchicalAllocatorProcess::addFramework(
     activeRoles[role] = 1;
     roleSorter->add(role, roleWeight(role));
     frameworkSorters[role] = frameworkSorterFactory();
+    metrics.addRole(role);
   } else {
     activeRoles[role]++;
   }
@@ -315,6 +316,8 @@ void HierarchicalAllocatorProcess::removeFramework(
     CHECK(frameworkSorters.contains(role));
     delete frameworkSorters[role];
     frameworkSorters.erase(role);
+
+    metrics.removeRole(role);
   }
 
   // Do not delete the filters contained in this
@@ -1758,6 +1761,25 @@ double HierarchicalAllocatorProcess::_quota_allocated(
   return used.isSome() ? used->value() : 0;
 }
 
+
+double HierarchicalAllocatorProcess::_offer_filters_active(
+    const string& role)
+{
+  double result = 0;
+
+  foreachvalue (const Framework& framework, frameworks) {
+    if (framework.role != role) {
+      continue;
+    }
+
+    foreachkey (const SlaveID& slaveId, framework.offerFilters) {
+      result += framework.offerFilters.get(slaveId)->size();
+    }
+  }
+
+  return result;
+}
+
 } // namespace internal {
 } // namespace allocator {
 } // namespace master {

src/master/allocator/mesos/hierarchical.hpp

@@ -295,6 +295,9 @@ class HierarchicalAllocatorProcess : public MesosAllocatorProcess
       const std::string& role,
       const std::string& resource);
 
+  double _offer_filters_active(
+      const std::string& role);
+
   hashmap<FrameworkID, Framework> frameworks;
 
   struct Slave

src/master/allocator/mesos/metrics.cpp

@@ -109,6 +109,10 @@ Metrics::~Metrics()
       process::metrics::remove(gauge);
     }
   }
+
+  foreachvalue (const Gauge& gauge, offer_filters_active) {
+    process::metrics::remove(gauge);
+  }
 }
 
 
@@ -163,6 +167,34 @@ void Metrics::removeQuota(const string& role)
   quota_allocated.erase(role);
 }
 
+
+void Metrics::addRole(const string& role)
+{
+  CHECK(!offer_filters_active.contains(role));
+
+  Gauge gauge(
+      "allocator/mesos/offer_filters/roles/" + role + "/active",
+      defer(allocator,
+            &HierarchicalAllocatorProcess::_offer_filters_active,
+            role));
+
+  offer_filters_active.put(role, gauge);
+
+  process::metrics::add(gauge);
+}
+
+
+void Metrics::removeRole(const string& role)
+{
+  Option<Gauge> gauge = offer_filters_active.get(role);
+
+  CHECK_SOME(gauge);
+
+  offer_filters_active.erase(role);
+
+  process::metrics::remove(gauge.get());
+}
+
 } // namespace internal {
 } // namespace allocator {
 } // namespace master {

src/master/allocator/mesos/metrics.hpp

@@ -48,6 +48,9 @@ struct Metrics
   void setQuota(const std::string& role, const Quota& quota);
   void removeQuota(const std::string& role);
 
+  void addRole(const std::string& role);
+  void removeRole(const std::string& role);
+
   const process::PID<HierarchicalAllocatorProcess> allocator;
 
   // Number of dispatch events currently waiting in the allocator process.
@@ -76,6 +79,9 @@ struct Metrics
   // Gauges for the per-role quota guarantee for each resource.
   hashmap<std::string, hashmap<std::string, process::metrics::Gauge>>
     quota_guarantee;
+
+  // Gauges for the per-role count of active offer filters.
+  hashmap<std::string, process::metrics::Gauge> offer_filters_active;
 };
 
 } // namespace internal {

src/tests/hierarchical_allocator_tests.cpp

@@ -540,6 +540,12 @@ TEST_F(HierarchicalAllocatorTest, OfferFilter)
   // Ensure the offer filter timeout is set before advancing the clock.
   Clock::settle();
 
+  JSON::Object metrics = Metrics();
+
+  string activeOfferFilters =
+    "allocator/mesos/offer_filters/roles/" + ROLE + "/active";
+  EXPECT_EQ(1, metrics.values[activeOfferFilters]);
+
   // Trigger a batch allocation.
   Clock::advance(flags.allocation_interval);
   Clock::settle();
@@ -557,6 +563,10 @@ TEST_F(HierarchicalAllocatorTest, OfferFilter)
   AWAIT_READY(allocation);
   EXPECT_EQ(framework1.id(), allocation.get().frameworkId);
   EXPECT_EQ(agent1.resources(), Resources::sum(allocation.get().resources));
+
+  metrics = Metrics();
+
+  EXPECT_EQ(0, metrics.values[activeOfferFilters]);
 }
 
 
@@ -2643,6 +2653,98 @@ TEST_F(HierarchicalAllocatorTest, AllocationRunTimerMetrics)
 }
 
 
+// This test checks that per-role active offer filter metrics
+// are correctly reported in the metrics endpoint.
+TEST_F(HierarchicalAllocatorTest, ActiveOfferFiltersMetrics)
+{
+  // Pausing the clock is not necessary, but ensures that the test
+  // doesn't rely on the batch allocation in the allocator, which
+  // would slow down the test.
+  Clock::pause();
+
+  initialize();
+
+  SlaveInfo agent = createSlaveInfo("cpus:2;mem:1024;disk:0");
+  allocator->addSlave(agent.id(), agent, None(), agent.resources(), {});
+
+  // Register three frameworks, two of which are in the same role.
+  // For every offer the frameworks install practically indefinite
+  // offer filters.
+  Duration filterTimeout = flags.allocation_interval * 100;
+  Filters offerFilter;
+  offerFilter.set_refuse_seconds(filterTimeout.secs());
+
+  FrameworkInfo framework1 = createFrameworkInfo("roleA");
+  allocator->addFramework(framework1.id(), framework1, {});
+
+  Future<Allocation> allocation = allocations.get();
+
+  AWAIT_READY(allocation);
+  ASSERT_EQ(framework1.id(), allocation->frameworkId);
+
+  allocator->recoverResources(
+      allocation->frameworkId,
+      agent.id(),
+      allocation->resources.get(agent.id()).get(),
+      offerFilter);
+
+  JSON::Object expected;
+  expected.values = {
+      {"allocator/mesos/offer_filters/roles/roleA/active", 1},
+  };
+
+  JSON::Value metrics = Metrics();
+
+  EXPECT_TRUE(metrics.contains(expected));
+
+  FrameworkInfo framework2 = createFrameworkInfo("roleB");
+  allocator->addFramework(framework2.id(), framework2, {});
+
+  allocation = allocations.get();
+
+  AWAIT_READY(allocation);
+  ASSERT_EQ(framework2.id(), allocation->frameworkId);
+
+  allocator->recoverResources(
+      allocation->frameworkId,
+      agent.id(),
+      allocation->resources.get(agent.id()).get(),
+      offerFilter);
+
+  expected.values = {
+      {"allocator/mesos/offer_filters/roles/roleA/active", 1},
+      {"allocator/mesos/offer_filters/roles/roleB/active", 1},
+  };
+
+  metrics = Metrics();
+
+  EXPECT_TRUE(metrics.contains(expected));
+
+  FrameworkInfo framework3 = createFrameworkInfo("roleA");
+  allocator->addFramework(framework3.id(), framework3, {});
+
+  allocation = allocations.get();
+
+  AWAIT_READY(allocation);
+  ASSERT_EQ(framework3.id(), allocation->frameworkId);
+
+  allocator->recoverResources(
+      allocation->frameworkId,
+      agent.id(),
+      allocation->resources.get(agent.id()).get(),
+      offerFilter);
+
+  expected.values = {
+      {"allocator/mesos/offer_filters/roles/roleA/active", 2},
+      {"allocator/mesos/offer_filters/roles/roleB/active", 1},
+  };
+
+  metrics = Metrics();
+
+  EXPECT_TRUE(metrics.contains(expected));
+}
+
+
 // This test ensures that resource allocation is done according to each role's
 // weight. This is done by having six agents and three frameworks and making
 // sure each framework gets the appropriate number of resources.