Refactored SerenityConfig

src/contention_detectors/overload.cpp

@@ -5,39 +5,38 @@
 
 #include "mesos/resources.hpp"
 
+#include "serenity/resource_helper.hpp"
+
 namespace mesos {
 namespace serenity {
 
-Try<Nothing> OverloadDetector::consume(const ResourceUsage& in) {
+const constexpr char* OverloadDetector::UTILIZATION_THRESHOLD_KEY;
+
+void OverloadDetector::allProductsReady() {
   Contentions product;
+  ResourceUsage usage = getConsumable().get();
 
-  if (in.total_size() == 0) {
-    return Error(std::string(NAME) + " No total in ResourceUsage");
+  if (usage.total_size() == 0) {
+    SERENITY_LOG(ERROR) << std::string(NAME) << " No total in ResourceUsage";
+    produce(Contentions());
   }
 
-  Resources totalAgentResources(in.total());
+  Resources totalAgentResources(usage.total());
   Option<double_t> totalAgentCpus = totalAgentResources.cpus();
 
   if (totalAgentCpus.isNone()) {
-    return Error(std::string(NAME) + " No total cpus in ResourceUsage");
+    SERENITY_LOG(ERROR) << std::string(NAME)
+                        << " No total cpus in ResourceUsage";
+    produce(product);
   }
 
   double_t thresholdCpus = this->cfgUtilizationThreshold * totalAgentCpus.get();
   double_t agentSumCpus = 0;
   uint64_t beExecutors = 0;
 
-  for (const ResourceUsage_Executor& inExec : in.executors()) {
-    if (!inExec.has_executor_info()) {
-      SERENITY_LOG(ERROR) << "Executor <unknown>"
-      << " does not include executor_info";
-      // Filter out these executors.
-      continue;
-    }
-    if (!inExec.has_statistics()) {
-      SERENITY_LOG(ERROR) << "Executor "
-      << inExec.executor_info().executor_id().value()
-      << " does not include statistics.";
-      // Filter out these executors.
+  for (const ResourceUsage_Executor& inExec : usage.executors()) {
+    // Validate for statistics and executor info.
+    if (!hasRequiredFields(inExec)) {
       continue;
     }
 
@@ -49,13 +48,14 @@ Try<Nothing> OverloadDetector::consume(const ResourceUsage& in) {
 
     agentSumCpus += value.get();
 
-    if (!Resources(inExec.allocated()).revocable().empty()) {
+    if (ResourceUsageHelper::isRevocableExecutor(inExec)) {
       beExecutors++;
     }
   }
 
   SERENITY_LOG(INFO) << "Sum = " << agentSumCpus << " vs total = "
-    << totalAgentCpus.get() << " [threshold = " << thresholdCpus << "]";
+                     << totalAgentCpus.get() << " [threshold = "
+                     << thresholdCpus << "]";
 
   if (agentSumCpus > thresholdCpus) {
     if (beExecutors == 0) {
@@ -70,12 +70,25 @@ Try<Nothing> OverloadDetector::consume(const ResourceUsage& in) {
     }
   }
 
-  // Continue pipeline.
-  this->produce(product);
-
-  return Nothing();
+  produce(product);
 }
 
+bool OverloadDetector::hasRequiredFields(const ResourceUsage_Executor& inExec) {
+  if (!inExec.has_executor_info()) {
+    SERENITY_LOG(ERROR) << "Executor <unknown>"
+    << " does not include executor_info";
+    return false;
+  }
+
+  if (!inExec.has_statistics()) {
+    SERENITY_LOG(ERROR) << "Executor "
+    << inExec.executor_info().executor_id().value()
+    << " does not include statistics.";
+    return false;
+  }
+
+  return true;
+}
 
 }  // namespace serenity
 }  // namespace mesos

src/contention_detectors/overload.hpp

@@ -20,22 +20,6 @@
 namespace mesos {
 namespace serenity {
 
-class OverloadDetectorConfig : public SerenityConfig {
- public:
-  OverloadDetectorConfig() { }
-
-  explicit OverloadDetectorConfig(const SerenityConfig& customCfg) {
-    this->initDefaults();
-    this->applyConfig(customCfg);
-  }
-
-  void initDefaults() {
-    //! double_t
-    //! Detector threshold.
-    this->fields[detector::THRESHOLD] =
-      detector::DEFAULT_UTILIZATION_THRESHOLD;
-  }
-};
 
 /**
  * OverloadDetector is able to create contention if utilization is above
@@ -48,28 +32,40 @@ class OverloadDetector :
   OverloadDetector(
       Consumer<Contentions>* _consumer,
       const lambda::function<usage::GetterFunction>& _cpuUsageGetFunction,
-      SerenityConfig _conf,
+      const Config& _conf,
       const Tag& _tag = Tag(QOS_CONTROLLER, NAME))
     : tag(_tag),
       cpuUsageGetFunction(_cpuUsageGetFunction),
       Producer<Contentions>(_consumer) {
-    SerenityConfig config = OverloadDetectorConfig(_conf);
-    this->cfgUtilizationThreshold =
-      config.getD(detector::THRESHOLD);
+    // Parse config values.
+    setUtilizationThreshold(
+        _conf.getValue<double_t>(UTILIZATION_THRESHOLD_KEY));
   }
 
   ~OverloadDetector() {}
 
-  Try<Nothing> consume(const ResourceUsage& in) override;
-
   static const constexpr char* NAME = "OverloadDetector";
 
+  void setUtilizationThreshold(const Result<double_t>& value) {
+    cfgUtilizationThreshold =
+      ConfigValidator<double_t>(value, UTILIZATION_THRESHOLD_KEY)
+        .validateValueIsPositive()
+        .getOrElse(DEFAULT_UTILIZATION_THRESHOLD);
+  }
+
+  static const constexpr char* UTILIZATION_THRESHOLD_KEY = "THRESHOLD";
+
  protected:
+  void allProductsReady() override;
+  bool hasRequiredFields(const ResourceUsage_Executor& inExec);
+
   const Tag tag;
   const lambda::function<usage::GetterFunction> cpuUsageGetFunction;
 
   // cfg parameters.
   double_t cfgUtilizationThreshold;
+
+  static const constexpr double_t DEFAULT_UTILIZATION_THRESHOLD = 0.72;
 };
 
 }  // namespace serenity

src/contention_detectors/signal_analyzers/drop.cpp

@@ -1,3 +1,4 @@
+#include <algorithm>
 #include <list>
 #include <utility>
 
@@ -10,6 +11,15 @@
 namespace mesos {
 namespace serenity {
 
+
+const constexpr char* SignalDropAnalyzer::WINDOW_SIZE_KEY;
+const constexpr char* SignalDropAnalyzer::FRACTIONAL_THRESHOLD_KEY;
+const constexpr char* SignalDropAnalyzer::SEVERITY_FRACTION_KEY;
+const constexpr char* SignalDropAnalyzer::NEAR_FRACTION_KEY;
+const constexpr char* SignalDropAnalyzer::MAX_CHECKPOINTS_KEY;
+const constexpr char* SignalDropAnalyzer::QUORUM_FRACTION_KEY;
+const constexpr double_t SignalDropAnalyzer::START_VALUE_DEFAULT;
+
 void SignalDropAnalyzer::shiftBasePoints() {
   for (std::list<double_t>::iterator& basePoint : this->basePoints) {
     basePoint++;
@@ -22,43 +32,39 @@ void SignalDropAnalyzer::recalculateParams() {
   this->window.clear();
   this->basePoints.clear();
 
-  uint64_t windowSize = this->cfgWindowSize;
-
+  uint64_t windowSize  = cfgWindowSize;
   // Find the biggest n in the T-2^n which fits within window length.
   uint64_t checkpoints = 0;
   while (windowSize > 0) {
     windowSize = windowSize >> 1;
     checkpoints++;
   }
 
-  // Make sure it does not exceed MAX_CHECKPOINSs option.
-  if (checkpoints > this->cfgMaxCheckpoints) {
-    checkpoints = this->cfgMaxCheckpoints;
-  }
+  // Make sure it does not exceed MAX_CHECKPOINTS option.
+  checkpoints = std::min((int64_t)checkpoints, cfgMaxCheckpoints);
 
   // Get the Quorum number from QUORUM fraction parameter.
-  this->quorumNum = this->cfgQuroum * checkpoints;
-  if (this->quorumNum == 0 || this->quorumNum > checkpoints) {
+  quorumNum = cfgQuroumFraction * checkpoints;
+  if (quorumNum == 0 || quorumNum > checkpoints) {
     SERENITY_LOG(WARNING) << "Bad value for Quorum parameter. Creating 100%"
                           << " quorum.";
-    this->quorumNum = checkpoints;
+    quorumNum = checkpoints;
   }
 
   std::stringstream checkpointLog;
-  checkpointLog << "Assurance Parameters: Quorum = "
-                << this->quorumNum << "/"
+  checkpointLog << "Assurance Parameters: Quorum = " << quorumNum << "/"
                 << checkpoints << " Checkpoints [ ";
   // Iterate over window and initialize it. Choose proper base points starting
   // from the end of window.
   uint64_t choosenNum = pow(2, (--checkpoints));
-  for (uint64_t i = this->cfgWindowSize; i > 0 ; i--) {
-    this->window.push_back(detector::DEFAULT_START_VALUE);
+  for (uint64_t i = cfgWindowSize; i > 0 ; i--) {
+    window.push_back(START_VALUE_DEFAULT);
 
     if (choosenNum == i) {
       checkpointLog << "T-" << choosenNum << " ";
 
       choosenNum /= 2;
-      basePoints.push_back(--this->window.end());
+      basePoints.push_back(--window.end());
     }
   }
   checkpointLog << "]";
@@ -68,6 +74,12 @@ void SignalDropAnalyzer::recalculateParams() {
 
 
 Result<Detection> SignalDropAnalyzer::processSample(double_t in) {
+  if (paramsChanged) {
+    recalculateParams();
+    paramsChanged = false;
+  }
+
+
   // Fill window.
   if (in < 0.1)
     in = 0.1;