Auto push metrics (#62)

Author: awegrzyn

CMakeLists.txt

@@ -171,6 +171,7 @@ set(EXAMPLES
         examples/8-Multiple.cxx
         examples/9-Timer.cxx
         examples/10-Buffering.cxx
+	examples/11-AutoUpdate.cxx
         )
 
 foreach (example ${EXAMPLES})

examples/10-Buffering.cxx

@@ -10,7 +10,7 @@ using Monitoring = o2::monitoring::MonitoringFactory;
 int main() {
   // Configure monitoring
   // Pass string with list of URLs as parameter
-  auto monitoring = Monitoring::Get("infologger://,flume://localhost:1234");
+  auto monitoring = Monitoring::Get("infologger://");
   monitoring->enableBuffering(10);
 
   // now send an application specific metric

examples/11-AutoUpdate.cxx

@@ -0,0 +1,19 @@
+///
+/// \file 1-Basic.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "Monitoring/MonitoringFactory.h"
+
+using namespace o2::monitoring;
+
+int main() {
+  auto monitoring = MonitoringFactory::Get("infologger://");
+  monitoring->enableAutoPush();
+  auto& qcMetric = monitoring->getAutoPushMetric("qcMetric"); 
+  auto& qcMetric2 = monitoring->getAutoPushMetric("qcMetric2");
+  std::this_thread::sleep_for (std::chrono::milliseconds(2000));
+  qcMetric = 123;
+  qcMetric2 = 321;
+  std::this_thread::sleep_for (std::chrono::milliseconds(2000));
+}	

include/Monitoring/Metric.h

@@ -8,6 +8,7 @@
 
 #include <string>
 #include <chrono>
+#include <mutex>
 #include <vector>
 #include <boost/variant.hpp>
 #include "Tag.h"
@@ -56,6 +57,15 @@ class Metric
 
     /// Default destructor
     ~Metric() = default;
+
+    /// Copy initialization
+    Metric(const Metric& other);
+
+    /// Copy assignment
+    Metric& operator=(Metric const& other);
+
+    /// Assign operator overload, assignes new values to the metric object
+    Metric& operator=(const boost::variant< int, std::string, double, uint64_t >& value);
 
     /// Name getter
     /// \return	metric name
@@ -107,6 +117,9 @@ class Metric
 
     /// Metric tags
     std::vector<Tag> tagSet;
+
+    /// Mutex for accesing metric value
+    mutable std::mutex mValueMutex;
 };
 
 } // namespace monitoring

include/Monitoring/Monitoring.h

@@ -15,6 +15,7 @@
 #include <tuple>
 #include <unordered_map>
 #include <vector>
+#include <deque>
 
 #include "Monitoring/Backend.h"
 #include "Monitoring/DerivedMetrics.h"
@@ -96,6 +97,16 @@ class Monitoring
     /// \param name 		tag name
     /// \param value 		tag value
     void addGlobalTag(std::string name, std::string value);
+
+    /// Returns a metric which will be periodically sent to backends
+    /// \param name 		metric name
+    /// \return 		periodically send metric
+    Metric& getAutoPushMetric(std::string name);
+
+    /// Enables periodical push interval
+    /// \param interval 	interval in seconds
+    void enableAutoPush(const unsigned int interval = 1);
+
   private:
     /// Derived metrics handler
     /// \see class DerivedMetrics
@@ -119,9 +130,8 @@ class Monitoring
     /// Process Monitor object that sends updates about the process itself
     std::unique_ptr<ProcessMonitor> mProcessMonitor;
 
-    /// Process Monitor thread loop
-    /// \param interval 	sleep time in seconds
-    void processMonitorLoop(int interval);
+    /// Push metric loop
+    void pushLoop();
 
     /// Metric buffer
     std::vector<Metric> mStorage;
@@ -131,6 +141,15 @@ class Monitoring
 
     /// Size of buffer
     unsigned int mBufferSize;
+
+    /// Store for automatically pushed metrics
+    std::deque<Metric> mPushStore;
+
+    /// Process monitor interval
+    std::atomic<unsigned int> mProcessMonitoringInterval;
+
+    /// Automatic metric push interval
+    std::atomic<unsigned int> mAutoPushInterval;
 };
 
 } // namespace monitoring

src/Metric.cxx

@@ -55,11 +55,40 @@ Metric::Metric(boost::variant< int, std::string, double, uint64_t > value, const
   mValue(value), mName(name), mTimestamp(timestamp)
 {}
 
+Metric::Metric(const Metric& other)
+{
+  std::lock_guard<std::mutex> lock(other.mValueMutex);
+  mName = other.mName;
+  mValue = other.mValue;
+  mTimestamp = other.mTimestamp;
+  tagSet = other.tagSet;
+}
+
+Metric& Metric::operator=(Metric const& other)
+{
+  if (&other != this) {
+    std::unique_lock<std::mutex> lockThis(mValueMutex, std::defer_lock);
+    std::unique_lock<std::mutex> lockOther(other.mValueMutex, std::defer_lock);
+    std::lock(lockThis, lockOther);
+
+    mName = other.mName;
+    mValue = other.mValue;
+    mTimestamp = other.mTimestamp;
+    tagSet = other.tagSet;
+  }
+  return *this;
+}
+
 boost::variant< int, std::string, double, uint64_t > Metric::getValue() const
 {
   return mValue;
 }
 
+Metric& Metric::operator=(const boost::variant< int, std::string, double, uint64_t >& value) {
+  mValue = value;
+  return *this;
+}
+
 Metric&& Metric::addTags(std::vector<Tag>&& tags)
 {
   tagSet = std::move(tags);

src/Monitoring.cxx

@@ -39,6 +39,9 @@ Monitoring::Monitoring()
   mProcessMonitor = std::make_unique<ProcessMonitor>();
   mDerivedHandler = std::make_unique<DerivedMetrics>();
   mBuffering = false;
+  mProcessMonitoringInterval = 0;
+  mAutoPushInterval = 0;
+  mMonitorRunning = false;
 }
 
 void Monitoring::enableBuffering(const unsigned int size)
@@ -59,13 +62,16 @@ void Monitoring::flushBuffer() {
 }
 
 void Monitoring::enableProcessMonitoring(const unsigned int interval) {
-  mMonitorRunning = true;
-  mMonitorThread = std::thread(&Monitoring::processMonitorLoop, this, interval);
-#ifdef _OS_LINUX
+  mProcessMonitoringInterval = interval;
+  if (!mMonitorRunning) {
+    mMonitorRunning = true;
+    mMonitorThread = std::thread(&Monitoring::pushLoop, this);
+  }
+  #ifdef _OS_LINUX
   MonLogger::Get() << "Process Monitor : Automatic updates enabled" << MonLogger::End();
-#else
+  #else
   MonLogger::Get() << "!! Process Monitor : Limited metrics available" << MonLogger::End();
-#endif
+  #endif
 }
 
 void Monitoring::startTimer(std::string name) {
@@ -115,21 +121,39 @@ Monitoring::~Monitoring()
   }
 }
 
-void Monitoring::processMonitorLoop(int interval)
+void Monitoring::pushLoop()
 {
-  // loopCount - no need to wait full sleep time to terminame the thread
-  int loopCount = 0;
+  unsigned int loopCount = 0;
+  std::this_thread::sleep_for (std::chrono::milliseconds(100));
   while (mMonitorRunning) {
-    std::this_thread::sleep_for (std::chrono::milliseconds(interval*10));
-    if ((++loopCount % 100) != 0) continue;
-    send(mProcessMonitor->getCpuAndContexts());
-#ifdef _OS_LINUX
-    send(mProcessMonitor->getMemoryUsage());
-#endif
-    loopCount = 0;
+    if (mProcessMonitoringInterval != 0 && (loopCount % (mProcessMonitoringInterval*10)) == 0) {
+      send(mProcessMonitor->getCpuAndContexts());
+      #ifdef _OS_LINUX
+      send(mProcessMonitor->getMemoryUsage());
+      #endif
+    }
+
+    if (mAutoPushInterval != 0 && (loopCount % (mAutoPushInterval*10)) == 0) {
+      std::vector<Metric> metrics;
+      for (auto& metric : mPushStore) {
+        metrics.push_back(metric);
+      }
+      send(std::move(metrics));
+    }
+    std::this_thread::sleep_for (std::chrono::milliseconds(100));
+    (loopCount >= 600) ? loopCount = 0 : loopCount++;
   }
 }
 
+Metric& Monitoring::getAutoPushMetric(std::string name)
+{
+  if (mAutoPushInterval == 0) {
+    MonLogger::Get() << "[WARN] AutoPush is not enabled" << MonLogger::End();
+  }
+  mPushStore.emplace_back(boost::variant< int, std::string, double, uint64_t > {}, name);
+  return mPushStore.back();
+}
+
 void Monitoring::sendGrouped(std::string measurement, std::vector<Metric>&& metrics)
 {
   for (auto& b: mBackends) {
@@ -153,6 +177,15 @@ void Monitoring::debug(Metric&& metric)
   }
 }
 
+void Monitoring::enableAutoPush(unsigned int interval)
+{
+  if (!mMonitorRunning) {
+    mMonitorRunning = true;
+    mMonitorThread = std::thread(&Monitoring::pushLoop, this);
+  }
+  mAutoPushInterval = interval;
+}
+
 void Monitoring::pushToBackends(Metric&& metric)
 {
   if (mBuffering) {

src/ProcessMonitor.cxx

@@ -24,9 +24,9 @@ ProcessMonitor::ProcessMonitor()
   mPid = static_cast<unsigned int>(::getpid());
   getrusage(RUSAGE_SELF, &mPreviousGetrUsage);
   mTimeLastRun = std::chrono::high_resolution_clock::now();
-#ifdef _OS_LINUX
+  #ifdef _OS_LINUX
   setTotalMemory();
-#endif
+  #endif
 }
 
 void ProcessMonitor::setTotalMemory()
@@ -62,8 +62,9 @@ std::vector<Metric> ProcessMonitor::getCpuAndContexts() {
   getrusage(RUSAGE_SELF, &currentUsage);
   auto timeNow = std::chrono::high_resolution_clock::now();
   double timePassed = std::chrono::duration_cast<std::chrono::microseconds>(timeNow - mTimeLastRun).count();
-  if (timePassed < 950) { // do not run too often
-    throw MonitoringInternalException("Process Monitor getrusage", "Do not invoke more often then 1ms");
+  if (timePassed < 950) {
+     MonLogger::Get() << "[WARN] Do not invoke Process Monitor more frequent then every 1s" << MonLogger::End();
+     return {};
   }
   double fractionCpuUsed = (
       currentUsage.ru_utime.tv_sec*1000000.0 + currentUsage.ru_utime.tv_usec - (mPreviousGetrUsage.ru_utime.tv_sec*1000000.0 + mPreviousGetrUsage.ru_utime.tv_usec)

test/testMetric.cxx

@@ -76,6 +76,19 @@ BOOST_AUTO_TEST_CASE(tags) {
   }
 }
 
+BOOST_AUTO_TEST_CASE(customCopyConstructor) {
+  o2::monitoring::Metric metric = o2::monitoring::Metric{10, "myMetric"}.addTags({{"tag1", "value1"}, {"tag2", "value2"}});
+  auto copied = metric;
+  BOOST_CHECK_EQUAL(boost::get<int>(copied.getValue()), 10);
+  BOOST_CHECK_EQUAL(copied.getName(), "myMetric");
+
+  std::vector<Tag> tags = copied.getTags();
+  for (auto const& tag: tags) {
+    BOOST_TEST(tag.name.find("tag") != std::string::npos);
+    BOOST_TEST(tag.value.find("value") != std::string::npos);
+  }
+}
+
 
 
 } // namespace Test

test/testMonitoring.cxx

@@ -50,6 +50,18 @@ BOOST_AUTO_TEST_CASE(testTimer)
   monitoring->stopAndSendTimer("timer");
 }
 
+BOOST_AUTO_TEST_CASE(testPush)
+{
+  auto monitoring = Monitoring::Get("infologger://");
+  monitoring->enableAutoPush();
+  auto& qcMetric = monitoring->getAutoPushMetric("qcMetric");
+  auto& qcMetric2 = monitoring->getAutoPushMetric("qcMetric2");
+  std::this_thread::sleep_for (std::chrono::milliseconds(1500));
+  qcMetric = 133 + 11 + 2.2;
+  qcMetric2 = 133 - 11 - 2.2;
+  std::this_thread::sleep_for (std::chrono::milliseconds(1500));
+}
+
 BOOST_AUTO_TEST_CASE(testSymbols)
 {
   BOOST_WARN_MESSAGE(!BOOST_IS_DEFINED( _WITH_APPMON ), "ApMon Backend disabled");