Refactored SerenityConfig

src/contention_detectors/overload.cpp

@@ -5,39 +5,36 @@
 
 #include "mesos/resources.hpp"
 
+#include "serenity/resource_helper.hpp"
+
 namespace mesos {
 namespace serenity {
 
-Try<Nothing> OverloadDetector::consume(const ResourceUsage& in) {
+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 +46,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 +68,27 @@ 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";
+    // Filter out these executors.
+    return false;
+  }
+
+  if (!inExec.has_statistics()) {
+    SERENITY_LOG(ERROR) << "Executor "
+    << inExec.executor_info().executor_id().value()
+    << " does not include statistics.";
+    // Filter out these executors.
+    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,23 +32,29 @@ class OverloadDetector :
   OverloadDetector(
       Consumer<Contentions>* _consumer,
       const lambda::function<usage::GetterFunction>& _cpuUsageGetFunction,
-      SerenityConfig _conf,
+      const SerenityConfig& _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.
+    setCfgUtilizationThreshold(
+      _conf.item<double_t>(detector::THRESHOLD,
+                           detector::DEFAULT_UTILIZATION_THRESHOLD));
   }
 
   ~OverloadDetector() {}
 
-  Try<Nothing> consume(const ResourceUsage& in) override;
-
   static const constexpr char* NAME = "OverloadDetector";
 
+  void setCfgUtilizationThreshold(double_t cfgUtilizationThreshold) {
+    OverloadDetector::cfgUtilizationThreshold = cfgUtilizationThreshold;
+  }
+
  protected:
+  void allProductsReady() override;
+  bool hasRequiredFields(const ResourceUsage_Executor& inExec);
+
   const Tag tag;
   const lambda::function<usage::GetterFunction> cpuUsageGetFunction;
 

src/contention_detectors/signal_analyzers/drop.hpp

@@ -30,56 +30,6 @@ namespace serenity {
 
 #define SIGNAL_DROP_ANALYZER_NAME "AssuranceDropAnalyzer"
 
-class SignalDropAnalyzerConfig : public SerenityConfig {
- public:
-  SignalDropAnalyzerConfig() {}
-
-  explicit SignalDropAnalyzerConfig(const SerenityConfig& customCfg) {
-    this->initDefaults();
-    this->applyConfig(customCfg);
-  }
-
-  void initDefaults() {
-    this->fields[detector::ANALYZER_TYPE] = SIGNAL_DROP_ANALYZER_NAME;
-    //! uint64_t
-    //! How far in the past we look.
-    this->fields[detector::WINDOW_SIZE] =
-      detector::DEFAULT_WINDOW_SIZE;
-
-    //! double_t
-    //! Defines how much (relatively to base point) value must drop to trigger
-    //! contention.
-    //! Most signal_analyzer will use that.
-    this->fields[detector::FRACTIONAL_THRESHOLD] =
-      detector::DEFAULT_FRACTIONAL_THRESHOLD;
-
-    //! double_t
-    //! You can adjust how big severity is created for a defined drop.
-    //! if -1 then unknown severity will be reported.
-    this->fields[detector::SEVERITY_FRACTION] = (double_t) -1;
-
-    //! double_t
-    //! Tolerance fraction of threshold if signal is accepted as returned to
-    //! previous state after drop.
-    this->fields[detector::NEAR_FRACTION] =
-      detector::DEFAULT_NEAR_FRACTION;
-
-    //! uint64_t
-    //! Maximum number of checkpoints we will have in our assurance detector.
-    //! Checkpoints are the reference assurance_test(base) points which we refer
-    //! to in the past when detecting drop or not.
-    //! It needs to be 0 < < WINDOW_SIZE
-    this->fields[detector::MAX_CHECKPOINTS] =
-      detector::DEFAULT_MAX_CHECKPOINTS;
-
-    //! double_t
-    //! Fraction of checkpoints' votes that important decision needs to obtain.
-    this->fields[detector::QUORUM] =
-      detector::DEFAULT_QUORUM;
-  }
-};
-
-
 /**
  * Dynamic implementation of sequential change point detection.
  *
@@ -109,13 +59,29 @@ class SignalDropAnalyzer : public SignalAnalyzer {
     : SignalAnalyzer(_tag),
       valueBeforeDrop(None()),
       quorumNum(0) {
-    SerenityConfig config = SignalDropAnalyzerConfig(_config);
-    this->cfgWindowSize = config.getU64(detector::WINDOW_SIZE);
-    this->cfgMaxCheckpoints = config.getU64(detector::MAX_CHECKPOINTS);
-    this->cfgQuroum = config.getD(detector::QUORUM);
-    this->cfgFractionalThreshold = config.getD(detector::FRACTIONAL_THRESHOLD);
-    this->cfgNearFraction = config.getD(detector::NEAR_FRACTION);
-    this->cfgSeverityFraction = config.getD(detector::SEVERITY_FRACTION);
+    setCfgWindowSize(_config.item<int64_t>(
+        detector::WINDOW_SIZE,
+        detector::DEFAULT_WINDOW_SIZE));
+
+    setCfgMaxCheckpoints(_config.item<int64_t>(
+        detector::MAX_CHECKPOINTS,
+        detector::DEFAULT_MAX_CHECKPOINTS));
+
+    setCfgQuroum(_config.item<double_t>(
+        detector::QUORUM,
+        detector::DEFAULT_QUORUM));
+
+    setCfgFractionalThreshold(_config.item<double_t>(
+        detector::FRACTIONAL_THRESHOLD,
+        detector::DEFAULT_FRACTIONAL_THRESHOLD));
+
+    setCfgNearFraction(_config.item<double_t>(
+        detector::NEAR_FRACTION,
+        detector::DEFAULT_NEAR_FRACTION));
+
+    setCfgSeverityFraction(_config.item<double_t>(
+        detector::SEVERITY_FRACTION,
+        detector::DEFAULT_SEVERITY_FRACTION));
 
     this->recalculateParams();
   }
@@ -131,6 +97,49 @@ class SignalDropAnalyzer : public SignalAnalyzer {
    */
   void shiftBasePoints();
 
+  //! int64_t
+  //! How far in the past we look.
+  void setCfgWindowSize(int64_t cfgWindowSize) {
+    SignalDropAnalyzer::cfgWindowSize = cfgWindowSize;
+  }
+
+  //! int64_t
+  //! Maximum number of checkpoints we will have in our assurance detector.
+  //! Checkpoints are the reference assurance_test(base) points which we refer
+  //! to in the past when detecting drop or not.
+  //! It needs to be 0 < < WINDOW_SIZE
+  void setCfgMaxCheckpoints(int64_t cfgMaxCheckpoints) {
+    SignalDropAnalyzer::cfgMaxCheckpoints = cfgMaxCheckpoints;
+  }
+
+  //! double_t
+  //! Fraction of checkpoints' votes that important decision needs to obtain.
+  void setCfgQuroum(double_t cfgQuroum) {
+    SignalDropAnalyzer::cfgQuroum = cfgQuroum;
+  }
+
+  //! double_t
+  //! Defines how much (relatively to base point) value must drop to trigger
+  //! contention.
+  //! Most signal_analyzer will use that.
+  void setCfgFractionalThreshold(double_t cfgFractionalThreshold) {
+    SignalDropAnalyzer::cfgFractionalThreshold = cfgFractionalThreshold;
+  }
+
+  //! double_t
+  //! You can adjust how big severity is created for a defined drop.
+  //! if -1 then unknown severity will be reported.
+  void setCfgSeverityFraction(double_t cfgSeverityFraction) {
+    SignalDropAnalyzer::cfgSeverityFraction = cfgSeverityFraction;
+  }
+
+  //! double_t
+  //! Tolerance fraction of threshold if signal is accepted as returned to
+  //! previous state after drop.
+  void setCfgNearFraction(double_t cfgNearFraction) {
+    SignalDropAnalyzer::cfgNearFraction = cfgNearFraction;
+  }
+
  protected:
   std::list<double_t> window;
   std::list<std::list<double_t>::iterator> basePoints;
@@ -142,16 +151,16 @@ class SignalDropAnalyzer : public SignalAnalyzer {
   uint32_t quorumNum;
 
   // cfg parameters.
-  uint64_t cfgWindowSize;
-  uint64_t cfgMaxCheckpoints;
+  int64_t cfgWindowSize;
+  int64_t cfgMaxCheckpoints;
   double_t cfgQuroum;
   double_t cfgFractionalThreshold;
   double_t cfgSeverityFraction;
   double_t cfgNearFraction;
 
   /**
-  * It is possible to dynamically change analyzer configuration.
-  */
+   * It is possible to dynamically change analyzer configuration.
+   */
   void recalculateParams();
 };
 

src/filters/too_low_usage.hpp

@@ -14,24 +14,6 @@
 namespace mesos {
 namespace serenity {
 
-class TooLowUsageFilterConfig : public SerenityConfig {
- public:
-  TooLowUsageFilterConfig() { }
-
-  explicit TooLowUsageFilterConfig(const SerenityConfig& customCfg) {
-    this->initDefaults();
-    this->applyConfig(customCfg);
-  }
-
-  void initDefaults() {
-    //! double_t
-    //! Minimal cpu usage
-    this->fields[too_low_usage::MINIMAL_CPU_USAGE] =
-      too_low_usage::DEFAULT_MINIMAL_CPU_USAGE;
-  }
-};
-
-
 /**
  * Filter out PR executors with too low metrics.
  * Currently we filter out when CPU Usage is below specified threshold.
@@ -44,11 +26,12 @@ class TooLowUsageFilter :
 
   explicit TooLowUsageFilter(
       Consumer<ResourceUsage>* _consumer,
-      SerenityConfig _conf,
+      const SerenityConfig& _conf,
       const Tag& _tag = Tag(QOS_CONTROLLER, NAME))
       : Producer<ResourceUsage>(_consumer), tag(_tag) {
-    SerenityConfig config = TooLowUsageFilterConfig(_conf);
-    this->cfgMinimalCpuUsage = config.getD(too_low_usage::MINIMAL_CPU_USAGE);
+    setCfgMinimalCpuUsage(_conf.item<double_t>(
+        too_low_usage::MINIMAL_CPU_USAGE,
+        too_low_usage::DEFAULT_MINIMAL_CPU_USAGE));
   }
 
   ~TooLowUsageFilter();
@@ -57,7 +40,11 @@ class TooLowUsageFilter :
 
   Try<Nothing> consume(const ResourceUsage& in);
 
- public:
+  void setCfgMinimalCpuUsage(double_t cfgMinimalCpuUsage) {
+    TooLowUsageFilter::cfgMinimalCpuUsage = cfgMinimalCpuUsage;
+  }
+
+ protected:
   const Tag tag;
 
   double_t cfgMinimalCpuUsage;