Allow sending vector of metrics (#40)

Author: awegrzyn

examples/5-Benchmark.cxx

@@ -57,7 +57,7 @@ int main(int argc, char *argv[]) {
     }
   } else {
     for (int i = 0; i <= count; i += add) {
-      monitoring->send("benchmarkMeasurement",{
+      monitoring->sendGrouped("benchmarkMeasurement",{
         {"string" + std::to_string(intDist(mt)), "stringMetric"},
         {doubleDist(mt), "doubleMetric"},
         {intDist(mt), "intMetric"}

examples/8-Multiple.cxx

@@ -10,7 +10,8 @@ using Monitoring = o2::monitoring::MonitoringFactory;
 int main() {
   // Configure monitoring
   // Pass string with list of URLs as parameter
-  auto monitoring = Monitoring::Get("infologger://");
+  auto monitoring = Monitoring::Get("flume://pcald03.cern.ch:8092");
 
-  monitoring->send("measurementName", {{20, "myMetricIntMultiple"}, {20.30, "myMetricFloatMultple"}});
+  //monitoring->sendGrouped("measurementName", {{20, "myMetricIntMultiple"}, {20.30, "myMetricFloatMultple"}});
+  monitoring->send({{201, "myMetricIntMultiple"}, {2.34, "myMetricFloatMultple"}});
 }

include/Monitoring/Backend.h

@@ -32,7 +32,11 @@ class Backend
     /// Sends metric via backend
     virtual void send(const Metric& metric) = 0;
 
-    /// Sends multiple metrics (if supported), otherwise falls back into sending single metrics
+    /// Sends multiple metrics not related to each other
+    virtual void send(std::vector<Metric>&& metrics) = 0;
+
+    /// Sends multiple related to each other metrics under a common name
+    /// If not supported by backend, falls back into sending single metrics
     virtual void sendMultiple(std::string measurement, std::vector<Metric>&& metrics) = 0;
 
     /// Sets a tag

include/Monitoring/Monitoring.h

@@ -54,11 +54,16 @@ class Monitoring
     /// \param metric            r-value to metric object
     void send(Metric&& metric, DerivedMetricMode mode = DerivedMetricMode::NONE);
 
-    /// Sends multiple metrics to as a single measurement
-    /// If it's not supported by backend it fallbacks into sending multiple metrics
+    /// Sends multiple not related to each other metrics
+    /// \@param metrics  vector of metrics
+    void send(std::vector<Metric>&& metrics);
+
+    /// Sends multiple realated to each other metrics under a common  measurement name
+    /// You can imagine it as a data point with multiple values
+    /// If it's not supported by a backend it fallbacks into sending one by one
     /// \param name		measurement name
     /// \param metrics		list of metrics
-    void send(std::string name, std::vector<Metric>&& metrics);
+    void sendGrouped(std::string name, std::vector<Metric>&& metrics);
 
     /// Enables process monitoring
     /// \param interval		refresh interval

src/Backends/ApMonBackend.cxx

@@ -40,6 +40,56 @@ void ApMonBackend::addGlobalTag(std::string, std::string value)
   entity += value;
 }
 
+void ApMonBackend::send(std::vector<Metric>&& metrics)
+{
+  int noMetrics = metrics.size();
+  char **paramNames, **paramValues;
+  int *valueTypes;
+  paramNames = (char **)std::malloc(noMetrics * sizeof(char *));
+  paramValues = (char **)std::malloc(noMetrics * sizeof(char *));
+  valueTypes = (int *)std::malloc(noMetrics * sizeof(int));
+  // the scope of values must be the same as sendTimedParameters method
+  int intValue;
+  double doubleValue;
+  std::string stringValue;
+
+  for (int i = 0; i < noMetrics; i++) {
+    paramNames[i] = const_cast<char*>(metrics[i].getName().c_str());
+    switch(metrics[i].getType()) {
+      case MetricType::INT :
+      {
+        valueTypes[i] = XDR_INT32;
+        intValue = boost::get<int>(metrics[i].getValue());
+        paramValues[i] = reinterpret_cast<char*>(&intValue);
+      }
+      break;
+
+      case MetricType::STRING :
+      {
+        valueTypes[i] = XDR_STRING;
+        stringValue = boost::get<std::string>(metrics[i].getValue());
+        paramValues[i] = const_cast<char*>(stringValue.c_str());
+      }
+      break;
+
+      case MetricType::DOUBLE :
+      {
+        valueTypes[i] = XDR_REAL64;
+        doubleValue = boost::get<double>(metrics[i].getValue());
+        paramValues[i] = reinterpret_cast<char*>(&doubleValue);
+      }
+      break;
+    }
+  }
+
+  mApMon->sendTimedParameters(const_cast<char*>(entity.c_str()), const_cast<char*>(entity.c_str()),
+    noMetrics, paramNames, valueTypes, paramValues, convertTimestamp(metrics[0].getTimestamp()));
+
+  std::free(paramNames);
+  std::free(paramValues);
+  std::free(valueTypes); 
+}
+
 void ApMonBackend::sendMultiple(std::string, std::vector<Metric>&& metrics)
 {
   int noMetrics = metrics.size();

src/Backends/ApMonBackend.h

@@ -37,12 +37,16 @@ class ApMonBackend final : public Backend
     /// Default destructor
     ~ApMonBackend() = default;
 
+    /// Sends multiple metrics not related to each other
+    /// \@param metrics  vector of metrics
+    void send(std::vector<Metric>&& metrics) override;
+
     /// Sends metric via MonALISA
     /// ApMonBackend's intances is type-aware therefore cast of metric value is needed
     /// \param metric           reference to metric object:
     void send(const Metric& metric) override;
 
-    /// Sends multiple metric in single packet
+    /// Sends grouped metrics  under common measuremet name
     /// \param name     measurement name
     /// \param metrics  list of metrics
     void sendMultiple(std::string measurement, std::vector<Metric>&& metrics) override;

src/Backends/Flume.cxx

@@ -31,12 +31,12 @@ std::string Flume::metricToJson(const Metric& metric)
     boost::property_tree::ptree header = globalHeader;
     header.put<std::string>("timestamp", std::to_string(convertTimestamp(metric.getTimestamp())));
     header.put<std::string>("name", metric.getName());
-    header.put<std::string>("value", boost::lexical_cast<std::string>(metric.getValue()));
+    header.put<std::string>("value_value", boost::lexical_cast<std::string>(metric.getValue()));
     for (const auto& tag : metric.getTags()) {
-      header.put<std::string>(tag.name, tag.value);
+      header.put<std::string>("tag_" + tag.name, tag.value);
     }   
     event.push_back(std::make_pair("headers", header));
-    event.put<std::string>("body", boost::lexical_cast<std::string>(metric.getValue()));
+    event.put<std::string>("body", "");
     std::stringstream ss; 
     write_json(ss, event);
     std::string s = ss.str();
@@ -45,14 +45,41 @@ std::string Flume::metricToJson(const Metric& metric)
     return s;
 }
 
+void Flume::send(std::vector<Metric>&& metrics)
+{
+  std::string flumeMetrics = "";
+  for (auto& metric : metrics) {
+    flumeMetrics += metricToJson(metric);
+    flumeMetrics += "\n";
+  }
+  mTransport->send(std::move(flumeMetrics));
+}
+
 void Flume::sendMultiple(std::string measurement, std::vector<Metric>&& metrics)
 {
-  for (auto& m : metrics) {
-    std::string tempName = m.getName();
-    m.setName(measurement + "-" + m.getName());
-    send(m);
-    m.setName(tempName);
+  mTransport->send(metricsToJson(measurement, std::move(metrics)));
+}
+
+std::string Flume::metricsToJson(std::string measurement, std::vector<Metric>&& metrics)
+{
+  boost::property_tree::ptree event;
+  boost::property_tree::ptree header = globalHeader;
+  header.put<std::string>("timestamp", std::to_string(convertTimestamp(metrics.front().getTimestamp())));
+  header.put<std::string>("name", measurement);
+  for (const auto& tag : metrics.front().getTags()) {
+    header.put<std::string>("tag_" + tag.name, tag.value);
+  }
+  for (auto& metric : metrics) {
+    header.put<std::string>("value_" + metric.getName(), boost::lexical_cast<std::string>(metric.getValue()));
   }
+  event.push_back(std::make_pair("headers", header));
+    event.put<std::string>("body", "");
+    std::stringstream ss;
+    write_json(ss, event);
+    std::string s = ss.str();
+    s.erase(std::remove_if( s.begin(), s.end(),
+      [](char c){ return (c =='\r' || c =='\t' || c == ' ' || c == '\n');}), s.end() );
+    return s;
 }
 
 void Flume::send(const Metric& metric)

src/Backends/Flume.h

@@ -46,6 +46,10 @@ class Flume final : public Backend
     /// \param metric    reference to metric object
     void send(const Metric& metric) override;
 
+    /// Sends multiple metrics not related to each other
+    /// \@param metrics  vector of metrics
+    void send(std::vector<Metric>&& metrics) override;
+
     /// Sends multiple metric in single packet
     /// Not supported by the backend therefore it falls back to sending metric one by one
     /// TODO: changed required in Flume Source
@@ -68,7 +72,8 @@ class Flume final : public Backend
     /// Serializes metric object to JSON
     /// \param metric
     /// \return JSON serializes metric
-    std::string metricToJson(const Metric& metric); 
+    std::string metricToJson(const Metric& metric);
+    std::string metricsToJson(std::string measurement, std::vector<Metric>&& metrics);
 };
 
 } // namespace backends

src/Backends/InfluxDB.cxx

@@ -69,8 +69,29 @@ void InfluxDB::sendMultiple(std::string measurement, std::vector<Metric>&& metri
   }
 }
 
+void InfluxDB::send(std::vector<Metric>&& metrics) {
+  std::string influxMetrics = "";
+  for (const auto& metric : metrics) {
+    influxMetrics += toInfluxLineProtocol(metric);
+    influxMetrics += "\n";
+  }
+
+  try {
+    transport->send(std::move(influxMetrics));
+  } catch (MonitoringInternalException&) {
+  }
+
+}
+
 void InfluxDB::send(const Metric& metric)
 {
+  try {
+    transport->send(toInfluxLineProtocol(metric));
+  } catch (MonitoringInternalException&) {
+  }
+}
+
+std::string InfluxDB::toInfluxLineProtocol(const Metric& metric) {
   std::string metricTags{};
   for (const auto& tag : metric.getTags()) {
     metricTags += "," + tag.name + "=" + tag.value;
@@ -83,11 +104,7 @@ void InfluxDB::send(const Metric& metric)
 
   std::stringstream convert;
   convert << name << "," << tagSet << metricTags << " value=" << value << " " << convertTimestamp(metric.getTimestamp());
-
-  try {
-    transport->send(convert.str());
-  } catch (MonitoringInternalException&) {
-  }
+  return convert.str();
 }
 
 void InfluxDB::prepareValue(std::string& value, int type)

src/Backends/InfluxDB.h

@@ -50,6 +50,10 @@ class InfluxDB final : public Backend
     /// \param metric    reference to metric object
     void send(const Metric& metric) override;
 
+    /// Sends multiple metrics not related to each other
+    /// \@param metrics  vector of metrics
+    void send(std::vector<Metric>&& metrics) override;
+
     /// Sends multiple values in single measurement
     /// \param name 	measurement name
     /// \param metrics 	list of metrics
@@ -61,7 +65,7 @@ class InfluxDB final : public Backend
     void addGlobalTag(std::string name, std::string value) override;
 
   private:
-    std::unique_ptr<transports::TransportInterface> transport;
+    std::unique_ptr<transports::TransportInterface> transport; ///< InfluxDB transport
     std::string tagSet; ///< Global tagset (common for each metric)
 
     /// Escapes " ", "," and "=" characters
@@ -72,6 +76,7 @@ class InfluxDB final : public Backend
     /// \param value 	reference to value
     /// \param type	type of the metric
     void prepareValue(std::string& value, int type);
+    std::string toInfluxLineProtocol(const Metric& metric);
 };
 
 } // namespace backends