[OMON-501] Add suppress mode to avoid sending same value over again (#270)

awegrzyn · web-flow · commit 47907b306718 · 2021-10-29T18:09:25.000+02:00
diff --git a/README.md b/README.md
@@ -142,6 +142,7 @@ send(Metric&& metric, [DerivedMetricMode mode])
 Two modes are available:
   + `DerivedMetricMode::RATE` - rate between two following values,
   + `DerivedMetricMode::INCREMENT` - sum of all passed values.
+  + `DerivedMetricMode::SUPPRESS` - suppresses forthcoming metric with same value, this happens until timeout is reached (configurable using `DerivedMetrics::mSuppressTimeout`)
 
 The derived value is generated only from the first value of the metric and it is added to the same metric with the value name suffixed with `_rate`, `_increment` accordingly.
 
diff --git a/examples/4-RateDerivedMetric.cxx b/examples/4-RateDerivedMetric.cxx
@@ -14,8 +14,17 @@ int main()
   auto monitoring = MonitoringFactory::Get("stdout://");
 
   // now send at least two metrics to see the result
-  for (int i = 0; i < 101; i += 10) {
+  for (int i = 0; i < 41; i += 10) {
     monitoring->send(Metric{i, "myMetric"}.addTag(tags::Key::Subsystem, tags::Value::Readout), DerivedMetricMode::RATE);
     std::this_thread::sleep_for(std::chrono::milliseconds(100));
   }
+
+  // present "suppress" mode
+  for (int i = 0; i < 10; i++) {
+    monitoring->send(Metric{1, "slowMetric"}, DerivedMetricMode::SUPPRESS); // only 1 should be displayed
+  }
+  DerivedMetrics::mSuppressTimeout = std::chrono::seconds(1); // change timeout to 1 s
+  monitoring->send(Metric{2, "slowMetric"}, DerivedMetricMode::SUPPRESS); // value changes
+  std::this_thread::sleep_for(std::chrono::seconds(1));
+  monitoring->send(Metric{2, "slowMetric"}, DerivedMetricMode::SUPPRESS); // timeout reached !
 }
diff --git a/include/Monitoring/DerivedMetrics.h b/include/Monitoring/DerivedMetrics.h
@@ -17,6 +17,7 @@
 #ifndef ALICEO2_MONITORING_CORE_DERIVED_METRICS_H
 #define ALICEO2_MONITORING_CORE_DERIVED_METRICS_H
 
+#include <chrono>
 #include <unordered_map>
 #include <memory>
 #include <string>
@@ -33,7 +34,8 @@ namespace monitoring
 /// Available derived metric modes
 enum class DerivedMetricMode { RATE,
                                INCREMENT,
-                               NONE };
+                               NONE,
+                               SUPPRESS };
 
 /// \brief Enables Calculation of derived metrics
 ///
@@ -48,12 +50,14 @@ class DerivedMetrics
   /// Default destructor
   ~DerivedMetrics() = default;
 
-  /// Metrics store necessary for derived metrics
-  std::unordered_map<std::string, Metric> mStorage;
-
   /// Entry method to DerivedMetrics
   /// Switches over processing modes: rate and increment
-  void process(Metric& metric, DerivedMetricMode mode);
+  bool process(Metric& metric, DerivedMetricMode mode);
+
+  static std::chrono::seconds mSuppressTimeout;
+ private:
+  /// Metrics store necessary for derived metrics
+  std::unordered_map<std::string, Metric> mStorage;
 };
 
 } // namespace monitoring
diff --git a/src/DerivedMetrics.cxx b/src/DerivedMetrics.cxx
@@ -37,9 +37,9 @@ namespace o2
 namespace monitoring
 {
 
-void DerivedMetrics::process(Metric& metric, DerivedMetricMode mode)
+bool DerivedMetrics::process(Metric& metric, DerivedMetricMode mode)
 {
-  const std::map<DerivedMetricMode, std::function<void(Metric&)>> map = {
+  const std::map<DerivedMetricMode, std::function<bool(Metric&)>> map = {
     {DerivedMetricMode::INCREMENT, [this](Metric& metric) {
        auto tags = metric.getTags();
        std::string key = metric.getName();
@@ -57,6 +57,7 @@ void DerivedMetrics::process(Metric& metric, DerivedMetricMode mode)
          metric.addValue(metric.getFirstValue().second, metric.getFirstValue().first + "_increment");
        }
        mStorage.insert(std::make_pair(key, metric));
+       return true;
      }},
     {DerivedMetricMode::RATE, [this](Metric& metric) {
        /// create pseudo unique key
@@ -71,7 +72,7 @@ void DerivedMetrics::process(Metric& metric, DerivedMetricMode mode)
        if (search == mStorage.end()) {
          mStorage.insert(std::make_pair(key, metric));
          metric.addValue((double)0.0, metric.getFirstValue().first + "_rate");
-         return;
+         return true;
        }
 
        auto timestampDifference = std::chrono::duration_cast<std::chrono::milliseconds>(
@@ -100,13 +101,41 @@ void DerivedMetrics::process(Metric& metric, DerivedMetricMode mode)
        mStorage.insert(std::make_pair(key, metric));
        // add rate field
        metric.addValue(rate, metric.getFirstValue().first + "_rate");
-     }}};
+       return true;
+    }},
+    {DerivedMetricMode::SUPPRESS, [this](Metric& metric) {
+       auto tags = metric.getTags();
+       std::string key = metric.getName();
+       std::for_each(tags.begin(), tags.end(), [&key](auto const& pair) {
+         key += pair.second;
+       });
+
+       // if no previous values -> store and send
+       auto search = mStorage.find(key);
+       if (search == mStorage.end()) {
+         mStorage.insert(std::make_pair(key, metric));
+         return true;
+       }
+       // if same values and timeout not reached -> skip
+       if (metric.getFirstValue().second == search->second.getFirstValue().second
+           && std::chrono::duration_cast<std::chrono::seconds>(
+             metric.getTimestamp() - search->second.getTimestamp()
+           ) < std::chrono::seconds(DerivedMetrics::mSuppressTimeout)) {
+         return false;
+       }
+       mStorage.erase(key);
+       mStorage.insert(std::make_pair(key, metric));
+       return true;
+     }}
+  };
   auto iterator = map.find(mode);
   if (iterator == map.end()) {
     throw MonitoringException("DerivedMetrics", "Unknown mode");
   }
-  iterator->second(metric);
+  return iterator->second(metric);
 }
 
+std::chrono::seconds DerivedMetrics::mSuppressTimeout{300};
+
 } // namespace monitoring
 } // namespace o2
diff --git a/src/Monitoring.cxx b/src/Monitoring.cxx
@@ -180,7 +180,9 @@ void Monitoring::send(Metric&& metric, DerivedMetricMode mode)
 {
   if (mode != DerivedMetricMode::NONE) {
     try {
-      mDerivedHandler->process(metric, mode);
+      if (!mDerivedHandler->process(metric, mode)) {
+        return;
+      }
     } catch (MonitoringException& e) {
       MonLogger::Get() << e.what() << MonLogger::End();
     }
diff --git a/test/testDerived.cxx b/test/testDerived.cxx
@@ -220,6 +220,61 @@ BOOST_AUTO_TEST_CASE(derivedIncrementDouble)
   }
 }
 
+BOOST_AUTO_TEST_CASE(suppressSimple)
+{
+  DerivedMetrics derivedHandler;
+  Metric m1{3, "metricInt"};
+  BOOST_CHECK(derivedHandler.process(m1, DerivedMetricMode::SUPPRESS));
+  Metric m2{3, "metricInt"};
+  BOOST_CHECK(!derivedHandler.process(m2, DerivedMetricMode::SUPPRESS));
+  Metric m3{4, "metricInt"};
+  BOOST_CHECK(derivedHandler.process(m3, DerivedMetricMode::SUPPRESS));
+  Metric m4{4, "metricInt"};
+  BOOST_CHECK(!derivedHandler.process(m4, DerivedMetricMode::SUPPRESS));
+}
+
+BOOST_AUTO_TEST_CASE(suppressTag)
+{
+  DerivedMetrics derivedHandler;
+  Metric metric{3, "metricInt"};
+  metric.addTag(o2::monitoring::tags::Key::Subsystem, o2::monitoring::tags::Value::Readout);
+  Metric metric2{3, "metricInt"};
+  metric2.addTag(o2::monitoring::tags::Key::Subsystem, o2::monitoring::tags::Value::DPL);
+  BOOST_CHECK(derivedHandler.process(metric, DerivedMetricMode::SUPPRESS));
+  BOOST_CHECK(derivedHandler.process(metric2, DerivedMetricMode::SUPPRESS));
+  BOOST_CHECK(!derivedHandler.process(metric, DerivedMetricMode::SUPPRESS));
+  BOOST_CHECK(!derivedHandler.process(metric, DerivedMetricMode::SUPPRESS));
+}
+
+BOOST_AUTO_TEST_CASE(suppressTimeout)
+{
+
+  DerivedMetrics::mSuppressTimeout = std::chrono::seconds(200);
+  DerivedMetrics derivedHandler;
+  Metric m1{3, "metricInt"};
+  BOOST_CHECK(derivedHandler.process(m1, DerivedMetricMode::SUPPRESS));
+
+  auto timestamp = std::chrono::system_clock::now();
+  timestamp += std::chrono::seconds(150);
+  Metric m2("metricInt", Metric::DefaultVerbosity, timestamp);
+  m2.addValue(3, "value");
+  BOOST_CHECK(!derivedHandler.process(m2, DerivedMetricMode::SUPPRESS));
+
+  timestamp += std::chrono::seconds(60);
+  Metric m3("metricInt", Metric::DefaultVerbosity, timestamp);
+  m3.addValue(3, "value");
+  BOOST_CHECK(derivedHandler.process(m3, DerivedMetricMode::SUPPRESS));
+
+  timestamp += std::chrono::seconds(60);
+  Metric m4("metricInt", Metric::DefaultVerbosity, timestamp);
+  m4.addValue(3, "value");
+  BOOST_CHECK(!derivedHandler.process(m4, DerivedMetricMode::SUPPRESS));
+
+  Metric m5("metricInt", Metric::DefaultVerbosity, timestamp);
+  m5.addValue(4, "value");
+  BOOST_CHECK(derivedHandler.process(m5, DerivedMetricMode::SUPPRESS));
+}
+
 BOOST_AUTO_TEST_CASE(testBoostVisitor)
 {
   {

Original file line number	Diff line number	Diff line change
`@@ -180,7 +180,9 @@ void Monitoring::send(Metric&& metric, DerivedMetricMode mode)`
`180`	`180`	`{`
`181`	`181`	`if (mode != DerivedMetricMode::NONE) {`
`182`	`182`	`try {`
`183`		`- mDerivedHandler->process(metric, mode);`
	`183`	`+ if (!mDerivedHandler->process(metric, mode)) {`
	`184`	`+ return;`
	`185`	`+ }`
`184`	`186`	`} catch (MonitoringException& e) {`
`185`	`187`	`MonLogger::Get() << e.what() << MonLogger::End();`
`186`	`188`	`}`