Clean up derived metrics (#39)

Author: awegrzyn

examples/4-RateDerivedMetric.cxx

@@ -6,18 +6,16 @@
 #include "Monitoring/MonitoringFactory.h"
 
 using Monitoring = o2::monitoring::MonitoringFactory;
+using DerivedMetricMode =  o2::monitoring::DerivedMetricMode;
 
 int main() {
   // Configure monitoring
   // Pass string with list of URLs as parameter
   auto monitoring = Monitoring::Get("infologger://");
 
-  // derived metric :  rate
-  monitoring->addDerivedMetric("myMetric", o2::monitoring::DerivedMetricMode::RATE);
-
   // now send at least two metrics to see the result
   for (int i = 0; i < 101; i += 10) {
-    monitoring->send({i, "myMetric"});
+    monitoring->send({i, "myMetric"}, DerivedMetricMode::RATE);
     std::this_thread::sleep_for(std::chrono::milliseconds(250));
   }
 }

examples/6-Increment.cxx

@@ -1,23 +1,19 @@
 ///
-/// \file 1-Basic.cxx
+/// \file 6-Increment.cxx
 /// \author Adam Wegrzynek <[email protected]>
 ///
 
 #include "Monitoring/MonitoringFactory.h"
 
 using Monitoring = o2::monitoring::MonitoringFactory;
+using DerivedMetricMode = o2::monitoring::DerivedMetricMode;
 
 int main() {
   // Configure monitoring
   // Pass string with list of URLs as parameter
   auto monitoring = Monitoring::Get("infologger://");
 
-  // now send an application specific metric
-  // 10 is the value
-  // myMetric is the name of the metric
-  //  
-  // 1. by copying values
-  monitoring->increment(10, "myIncrementMetric");
-  monitoring->increment(5, "myIncrementMetric");
-  monitoring->increment(15, "myIncrementMetric");
+  monitoring->send({10, "myIncrementMetric"}, DerivedMetricMode::INCREMENT);
+  monitoring->send({5, "myIncrementMetric"}, DerivedMetricMode::INCREMENT);
+  monitoring->send({15, "myIncrementMetric"}, DerivedMetricMode::INCREMENT);
 }	

include/Monitoring/DerivedMetrics.h

@@ -6,7 +6,7 @@
 #ifndef ALICEO2_MONITORING_CORE_DERIVED_METRICS_H
 #define ALICEO2_MONITORING_CORE_DERIVED_METRICS_H
 
-#include <map>
+#include <unordered_map>
 #include <memory>
 #include <string>
 #include <vector>
@@ -20,7 +20,7 @@ namespace monitoring
 {
 
 /// Available derived metric modes : RATE and AVERAGE values
-enum class DerivedMetricMode { RATE, AVERAGE, NONE };
+enum class DerivedMetricMode { RATE, INCREMENT, NONE };
 
 /// Calculates derived metrics
 ///
@@ -31,47 +31,19 @@ class DerivedMetrics
 {
   public:
     /// Default constructor
-    DerivedMetrics(const unsigned int cacheSize);
+    DerivedMetrics() = default;
 
     /// Default destructor
     ~DerivedMetrics() = default;
 
-    /// States whether metric has been registered or not
-    /// \param name     metric name
-    /// \return 	true when metric is present in mRegistered, false otherwise
-    bool isRegistered(std::string name);
-
-    /// Registers metric to be processed (adds its name to mRegistered map)
-    /// Processing modes are enumerated in DerivedMetricMode class
-    /// \param name      name, metrics name
-    /// \param mode      mode, see DerivedMetricMode
-    void registerMetric(std::string name, DerivedMetricMode mode);
-
-    /// Handles actual metric processing; finds out whether metric needs to be processed or not
-    /// If yes, passing it to one of methods that handles calculation of derived metric
-    /// \param metric    reference to metric instance
-    /// \return          metric object with calculated derived metric in it
-    Metric processMetric(Metric& metric);
-  
-  private:
     /// Calculates rate value based on metrics stored in mCache map
     /// \param name 	metric name
     /// \return 	metric with calculated rate value
-    Metric calculateRate(std::string name);
-
-    /// Calculates average value based on metrics stored in mCache map
-    /// \param name 	metric name
-    /// \return		metric with calculated average value
-    Metric calculateAverage(std::string name);
-
-    /// maximum size of cache map
-    const unsigned int mMaxVectorSize;
+    Metric rate(Metric& metric);
+    Metric increment(Metric& metric);
 
     /// Cache of registered metrics (metric name / vector of metric pointers).
-    std::map <std::string, std::vector<Metric>> mCache;
-
-    /// Registered metrics with their modes (metric name, registered mode).
-    std::map <std::string, DerivedMetricMode> mRegistered;
+    std::unordered_map <std::string, Metric> mStorage;
 };
 
 } // namespace monitoring

include/Monitoring/Monitoring.h

@@ -60,22 +60,10 @@ class Monitoring
     /// \param metrics		list of metrics
     void send(std::string name, std::vector<Metric>&& metrics);
 
-    /// Adds metric to derived metric list - each time the metric arrives the derived metric is calculated and pushed to all backends
-    /// Following processing modes are supported: DerivedMetricMode::RATE, DerivedMetricMode::AVERAGE
-    /// \param name             metric name
-    /// \param mode             mode
-    void addDerivedMetric(std::string name, DerivedMetricMode mode);
-
     /// Enables process monitoring
     /// \param interval		refresh interval
     void enableProcessMonitoring(int interval = 5);
 
-    /// Increment value of a metric (or initialize if not exists)
-    /// \@param value		incremental value
-    /// \@param name		name of the metric
-    template<typename T>
-    void increment(T value, std::string name);
-
     /// Starts timing
     /// Sets a start timestamp and timeout
     /// \@param name 		metric name
@@ -94,9 +82,6 @@ class Monitoring
     /// Vector of backends (where metrics are passed to)
     std::vector <std::unique_ptr<Backend>> mBackends;
 
-    /// Cache for the metric increment feature
-    std::unordered_map <std::string, Metric> mIncrementCache;
-
     /// List of timers
     std::unordered_map <std::string, std::chrono::time_point<std::chrono::steady_clock>> mTimers;
 
@@ -112,10 +97,6 @@ class Monitoring
     /// Process Monitor thread loop
     /// \param interval 	sleep time in seconds
     void processMonitorLoop(int interval);
-
-    /// Increments metrics, stores calculated value in cache
-    template<typename T>
-    Metric incrementMetric(T value, std::string name);
 };
 
 } // namespace monitoring

src/DerivedMetrics.cxx

@@ -15,91 +15,64 @@
 #include <vector>
 #include "MonLogger.h"
 #include "VariantVisitorRate.h"
+#include "VariantVisitorAdd.h"
 
 namespace o2
 {
 /// ALICE O2 Monitoring system
 namespace monitoring
 {
 
-DerivedMetrics::DerivedMetrics(const unsigned int cacheSize) : mMaxVectorSize(cacheSize) 
+Metric DerivedMetrics::rate(Metric& metric)
 {
-}
-
-void DerivedMetrics::registerMetric(std::string name, DerivedMetricMode mode)
-{
-  mRegistered.emplace(std::pair<std::string, DerivedMetricMode>(name, mode));
-  MonLogger::Get() << "Monitoring : Metric " << name << " added to derived metrics" 
-                       << MonLogger::End();
-}
-
-bool DerivedMetrics::isRegistered(std::string name)
-{
-  auto search = mRegistered.find(name);
-  return (search != mRegistered.end());
-}
+  // disallow string
+  std::string name = metric.getName();
+  if (metric.getType() == MetricType::STRING) {
+    throw MonitoringInternalException("DerivedMetrics/ProcessMetric", "Not able to process string values");
+  }
 
-Metric DerivedMetrics::calculateRate(std::string name)
-{
-  auto search = mCache.find(name);
-  int size = search->second.size();
-  if (search == mCache.end() || size < 2) {
-    throw MonitoringInternalException("DerivedMetrics/Calculate rate", "Not enough values");
+  // search for previous value
+  auto search = mStorage.find(name);
+  if (search == mStorage.end()) {
+    mStorage.insert(std::make_pair(name, metric));
+    return Metric{(double) 0.0, name + "Rate"};
   }
 
   auto timestampDifference = std::chrono::duration_cast<std::chrono::milliseconds>(
-    search->second.at(size - 1).getTimestamp()
-    - search->second.at(size - 2).getTimestamp()
+    metric.getTimestamp()
+    - search->second.getTimestamp()
   );
-  boost::variant< int, std::string, double, uint64_t > current = search->second.at(size - 1).getValue();
-  boost::variant< int, std::string, double, uint64_t > previous = search->second.at(size - 2).getValue();
   int timestampCount = timestampDifference.count();
-
   // disallow dividing by 0
   if (timestampCount == 0) {
     throw MonitoringInternalException("DerivedMetrics/Calculate rate", "Division by 0");
   }
 
+  auto current = metric.getValue();
+  auto previous = search->second.getValue();
   auto rate =  boost::apply_visitor(VariantVisitorRate(timestampCount), current, previous);
-  return Metric{rate, name + "Rate"};
-}
 
-Metric DerivedMetrics::calculateAverage(std::string name)
-{
-  double total = 0;
-  for (auto& m : mCache.at(name)) {
-    total += boost::lexical_cast<double>(m.getValue());
-  }
-  double average = total / mCache.at(name).size();
-  return Metric{average, name + "Average"};
+  // swap metrics
+  mStorage.erase(name);
+  mStorage.insert(std::make_pair(name, metric));
+
+  return Metric{rate, name + "Rate"};
 }
 
-Metric DerivedMetrics::processMetric(Metric& metric)
-{
+Metric DerivedMetrics::increment(Metric& metric) {
   std::string name = metric.getName();
-  if (metric.getType() == MetricType::STRING) {
-    throw MonitoringInternalException("DerivedMetrics/ProcessMetric", "Not able to process string values");
-  }
-  auto search = mCache.find(name);
-  // create vector if this is first metric of this kind
-  if (search == mCache.end()) {
-    mCache.emplace(std::make_pair(name, std::vector<Metric>()));
-  }
-  // remove first value if vector too large
-  if (mCache.at(name).size() > mMaxVectorSize) {
-    mCache.at(name).erase( mCache.at(name).begin() );
-  }
-  mCache[name].push_back(metric);
-
-  auto derived = mRegistered.find(name);
-  if (derived->second == DerivedMetricMode::RATE) {
-    return calculateRate(name);
-  }
-  else if (derived->second == DerivedMetricMode::AVERAGE)  {
-    return calculateAverage(name);
-  } else {
-    throw MonitoringInternalException("DerivedMetrics/ProcessMetric", "Processing mode not supported");
+  auto search = mStorage.find(name);
+  if (search != mStorage.end()) {
+    auto currentValue = metric.getValue();
+    auto storedValue = search->second.getValue();
+    auto value = boost::apply_visitor(VariantVisitorAdd(), currentValue, storedValue);
+    mStorage.erase(search);
+    Metric result = Metric{value, name};
+    mStorage.insert(std::make_pair(name, result));
+    return result;
   }
+  mStorage.insert(std::make_pair(name, metric));
+  return metric;
 }
 
 } // namespace monitoring

src/Monitoring.cxx

@@ -36,7 +36,7 @@ namespace monitoring
 
 Monitoring::Monitoring() {
   mProcessMonitor = std::make_unique<ProcessMonitor>();
-  mDerivedHandler = std::make_unique<DerivedMetrics>(1000);
+  mDerivedHandler = std::make_unique<DerivedMetrics>();
 }
 
 void Monitoring::enableProcessMonitoring(int interval) {
@@ -71,24 +71,6 @@ void Monitoring::stopAndSendTimer(std::string name) {
   }
 }
 
-template<typename T>
-Metric Monitoring::incrementMetric(T value, std::string name) {
-  auto search = mIncrementCache.find(name);
-  if (search != mIncrementCache.end()) {
-    T current = boost::lexical_cast<T>(search->second.getValue());
-    value += current;
-    mIncrementCache.erase(search);
-  }
-  Metric result = Metric{value, name};
-  mIncrementCache.insert(std::make_pair(name, result));
-  return result;
-}
-
-template<typename T>
-void Monitoring::increment(T value, std::string name) {
-  send(incrementMetric(value, name));
-}
-
 void Monitoring::addBackend(std::unique_ptr<Backend> backend) {
    ProcessDetails processDetails{};
    backend->addGlobalTag("hostname", processDetails.getHostname());
@@ -124,10 +106,6 @@ void Monitoring::processMonitorLoop(int interval)
   }
 }
 
-void Monitoring::addDerivedMetric(std::string name, DerivedMetricMode mode) {
-  mDerivedHandler->registerMetric(name, mode);
-}
-
 void Monitoring::send(std::string measurement, std::vector<Metric>&& metrics)
 {
   for (auto& b: mBackends) {
@@ -137,21 +115,18 @@ void Monitoring::send(std::string measurement, std::vector<Metric>&& metrics)
 
 void Monitoring::send(Metric&& metric, DerivedMetricMode mode)
 {
+  if (mode == DerivedMetricMode::RATE) {
+    metric = mDerivedHandler->rate(metric);
+  }
+
+  if (mode == DerivedMetricMode::INCREMENT) {
+    metric = mDerivedHandler->increment(metric);
+  }
+
   for (auto& b: mBackends) {
     b->send(metric);
   }
-  if (mDerivedHandler->isRegistered(metric.getName())) {
-    try {
-      Metric&& derived = mDerivedHandler->processMetric(metric);
-      for (auto& b: mBackends) {
-        b->send(derived);
-      }   
-    } catch (MonitoringInternalException&) { }
-  }
 }
 
-template void Monitoring::increment(int, std::string);
-template void Monitoring::increment(double, std::string);
-template void Monitoring::increment(uint64_t, std::string);
 } // namespace monitoring
 } // namespace o2