Metric buffering (#41)

Author: awegrzyn

CMakeLists.txt

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

README.md

@@ -49,45 +49,7 @@ eval `modulecmd bash load Monitoring/v1.5.0-1`
 The installation directory is: `/opt/alisw/el7/Monitoring/v1.5.0-1`
 
 ### aliBuild
-<details>
- <summary><strong>Click here if you don't have <i>aliBuild</i> installed</strong></summary>
-<br>
-
-+ Enable Software Collections in order to install `devtoolset-6` which includes `gcc 6.2.0` **(as root)**
-~~~
-yum install -y centos-release-scl
-yum-config-manager --enable rhel-server-rhscl-7-rpms
-~~~
-
-+ Install packages required by aliBuild **(as root)**
-~~~
-yum install -y curl curl-devel python python-devel devtoolset-6           \
-               python-pip bzip2-devel autoconf automake texinfo git       \
-               gettext-devel libtool freetype freetype-devel libpng       \
-               libpng-devel sqlite sqlite-devel ncurses-devel gettext     \
-               mesa-libGLU-devel libX11-devel libXpm-devel libXext-devel  \
-               libXft-devel libxml2 libxml2-devel environment-modules
-~~~
-
-+ Upgrade `pip` and `setuptools`, install [aliBuild](https://alisw.github.io/alibuild/) and missing python packags **(as root)**
-~~~
-pip install --upgrade pip
-pip install -U setuptools
-pip install alibuild matplotlib numpy certifi ipython==5.4.1 ipywidgets   \
-            ipykernel notebook metakernel pyyaml
-~~~
-
-+ Enable `devtoolset-6`. It's recommened to add it to `~/.basrc` or similar file
-~~~
-source /opt/rh/devtoolset-6/enable
-~~~
-
-+ Set `alienv` root directory and load the environment
-~~~
-mkdir alice; cd alice
-ALICE_WORK_DIR=`pwd`/sw; eval "`alienv shell-helper`"
-~~~
-</details>
+<strong>[Click here if you don't have aliBuild installed](https://alice-doc.github.io/alice-analysis-tutorial/building/)</strong>
 <br>
 
 + Compile `Monitoring` and its dependecies via `aliBuild`
@@ -140,19 +102,16 @@ See table below to find out how to create `URI` for each backend:
 | Flume        | UDP       | `flume`                | -                |
 
 ### Sending metric
-Simplified `send` method:
-```cpp
-send(T value, std::string name)
-```
-
-Or more advanced overload of `send`:
 ```cpp
 send(Metric&& metric)
 ```
+Where metric constructor receives following parameters:
+  - `T value`
+  - `std::string& name`
+  - `[time_point<system_clock> timestamp]`
 
 For example:
 ```cpp
-monitoring->send(10, "myMetricInt");
 monitoring->send({10, "myMetricInt"});
 ```
 

examples/10-Buffering.cxx

@@ -0,0 +1,24 @@
+///
+/// \file 10-Buffering.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "Monitoring/MonitoringFactory.h"
+
+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");
+  monitoring->enableBuffering(10);
+
+  // now send an application specific metric
+  // 10 is the value
+  // myMetric is the name of the metric by creating and moving Metric object
+  for (int i = 0; i < 25; i++) {
+    monitoring->send({10, "myMetricInt"});
+  }
+  monitoring->send({10.10, "myMetricFloat"});
+  monitoring->flushBuffer();
+}	

include/Monitoring/Monitoring.h

@@ -55,7 +55,7 @@ class Monitoring
     void send(Metric&& metric, DerivedMetricMode mode = DerivedMetricMode::NONE);
 
     /// Sends multiple not related to each other metrics
-    /// \@param metrics  vector of metrics
+    /// \param metrics  vector of metrics
     void send(std::vector<Metric>&& metrics);
 
     /// Sends multiple realated to each other metrics under a common  measurement name
@@ -67,18 +67,25 @@ class Monitoring
 
     /// Enables process monitoring
     /// \param interval		refresh interval
-    void enableProcessMonitoring(int interval = 5);
+    void enableProcessMonitoring(const unsigned int interval = 5);
 
     /// Starts timing
     /// Sets a start timestamp and timeout
-    /// \@param name 		metric name
-    /// \@param timeout		timeout
+    /// \param name 		metric name
+    /// \param timeout		timeout
     void startTimer(std::string name);
 
     /// Stops timing
     /// Sets stop timestamp, calculates delta and sends value
-    /// \@param name 		metric name
+    /// \param name 		metric name
     void stopAndSendTimer(std::string name);
+
+    /// Flushes metric buffer (this can also happen when buffer is full)
+    void flushBuffer();
+
+    /// Enables metric buffering
+    /// \param size 		buffer size
+    void enableBuffering(const unsigned int size = 20);
   private:
     /// Derived metrics handler
     /// \see class DerivedMetrics
@@ -102,6 +109,15 @@ class Monitoring
     /// Process Monitor thread loop
     /// \param interval 	sleep time in seconds
     void processMonitorLoop(int interval);
+
+    /// Metric buffer
+    std::vector<Metric> mStorage;
+
+    /// Flag stating whether metric buffering is enabled
+    bool mBuffering;
+
+    /// Size of buffer
+    unsigned int mBufferSize;
 };
 
 } // namespace monitoring

src/Monitoring.cxx

@@ -34,12 +34,33 @@ namespace o2
 namespace monitoring 
 {
 
-Monitoring::Monitoring() {
+Monitoring::Monitoring()
+{
   mProcessMonitor = std::make_unique<ProcessMonitor>();
   mDerivedHandler = std::make_unique<DerivedMetrics>();
+  mBuffering = false;
+}
+
+void Monitoring::enableBuffering(const unsigned int size)
+{
+  mBufferSize = size;
+  mBuffering = true;
+  mStorage.reserve(size);
+  MonLogger::Get() << "Buffering enabled (" << mStorage.capacity() << ")" << MonLogger::End();
+}
+
+void Monitoring::flushBuffer() {
+  if (!mBuffering) {
+    MonLogger::Get() << "Cannot flush as buffering is disabled" << MonLogger::End();
+    return;
+  }
+  auto capacity = mStorage.capacity();
+  auto size = mStorage.size();
+  send(std::move(mStorage));
+  mStorage.clear();
 }
 
-void Monitoring::enableProcessMonitoring(int interval) {
+void Monitoring::enableProcessMonitoring(const unsigned int interval) {
 #ifdef _OS_LINUX
   mMonitorRunning = true;
   mMonitorThread = std::thread(&Monitoring::processMonitorLoop, this, interval);
@@ -84,6 +105,9 @@ Monitoring::~Monitoring()
   if (mMonitorThread.joinable()) {
     mMonitorThread.join();
   }
+  if (mBuffering) {
+    flushBuffer();
+  }
 }
 
 void Monitoring::processMonitorLoop(int interval)
@@ -128,9 +152,15 @@ void Monitoring::send(Metric&& metric, DerivedMetricMode mode)
   if (mode == DerivedMetricMode::INCREMENT) {
     metric = mDerivedHandler->increment(metric);
   }
-
-  for (auto& b: mBackends) {
-    b->send(metric);
+  if (mBuffering) {
+    mStorage.push_back(std::move(metric));
+    if (mStorage.size() >= mBufferSize) {
+      flushBuffer();
+    }
+  } else {
+    for (auto& b: mBackends) {
+      b->send(metric);
+    }
   }
 }
 

test/testMonitoring.cxx

@@ -27,6 +27,16 @@ BOOST_AUTO_TEST_CASE(createMonitoring)
   monitoring->send({doubleMetric, "myCrazyMetricD"});
 }
 
+BOOST_AUTO_TEST_CASE(buffering)
+{
+  auto monitoring = Monitoring::Get("infologger://,flume://localhost:1234");
+  monitoring->enableBuffering(10);
+  for (int i = 0; i < 25; i++) {
+    monitoring->send({10, "myMetricInt"});
+  }
+  monitoring->flushBuffer();
+}
+
 BOOST_AUTO_TEST_CASE(testTimer)
 {
   auto monitoring = Monitoring::Get("infologger://");