[OMON-595] Use run number as record key in Kafka transport (#304)

Author: awegrzyn

examples/5-Benchmark.cxx

@@ -66,6 +66,7 @@ int main(int argc, char* argv[])
   }
 
   auto monitoring = MonitoringFactory::Get(vm["url"].as<std::string>());
+  monitoring->setRunNumber(intDist(mt));
   if (vm["monitor"].as<bool>()) {
     monitoring->enableProcessMonitoring(1, {PmMeasurement::Cpu, PmMeasurement::Mem, PmMeasurement::Smaps});
   }
@@ -102,7 +103,7 @@ int main(int argc, char* argv[])
       for (int i = 1; i <= measurements; i++) {
         for (int k = 1; k <= flps; k++) {
           monitoring->send(Metric{doubleDist(mt), "doubleMetric" + std::to_string(i)}.addTag(tags::Key::FLP, k));
-          monitoring->send(Metric{intDist(mt), "intMetric" + std::to_string(i)}.addTag(tags::Key::FLP, k));
+          monitoring->send(Metric{intDist(mt), "intMetric" + std::to_string(i)}.addTag(tags::Key::FLP, k).addTag(tags::Key::Subsystem, tags::Value::Readout));
           monitoring->send(Metric{std::rand() % 2, "onOffMetric" + std::to_string(i)}.addTag(tags::Key::FLP, k));
           std::this_thread::sleep_for(std::chrono::microseconds(10));
         }

include/Monitoring/Backend.h

@@ -49,7 +49,7 @@ class Backend
   }
 
   /// Run number setter
-  void setRunNumber(uint32_t runNumber) {
+  virtual void setRunNumber(uint32_t runNumber) {
     mRunNumber = runNumber;
   }
 

src/Backends/InfluxDB.cxx

@@ -73,6 +73,11 @@ void InfluxDB::send(std::vector<Metric>&& metrics)
   mTransport->send(std::move(influxMetrics));
 }
 
+void InfluxDB::setRunNumber(uint32_t runNumber) {
+  mRunNumber = runNumber;
+  mTransport->setKey(std::to_string(runNumber));
+}
+
 void InfluxDB::send(const Metric& metric)
 {
   mTransport->send(toInfluxLineProtocol(metric));

src/Backends/InfluxDB.h

@@ -50,6 +50,10 @@ class InfluxDB final : public Backend
   /// \return  	 timestamp in nanoseconds
   inline unsigned long convertTimestamp(const std::chrono::time_point<std::chrono::system_clock>& timestamp);
 
+  /// Reimplements setting run number and some transport needs to be notified about that as well
+  /// \param runNumber run number to be set
+  virtual void setRunNumber(uint32_t runNumber) override;
+
   /// Sends metric to InfluxDB instance via one transport
   /// \param metric    reference to metric object
   void send(const Metric& metric) override;

src/Transports/KafkaProducer.cxx

@@ -60,7 +60,7 @@ void KafkaProducer::send(std::string&& message)
     mTopic, partition,
     RdKafka::Producer::RK_MSG_COPY,
     const_cast<char*>(message.c_str()), message.size(),
-    NULL, 0,
+    const_cast<char*>(mKey.c_str()), mKey.size(),
     0,
     NULL,
     NULL);

src/Transports/TransportInterface.h

@@ -17,6 +17,7 @@
 #ifndef ALICEO2_MONITORING_TRANSPORTS_TRANSPORTINTERFACE_H
 #define ALICEO2_MONITORING_TRANSPORTS_TRANSPORTINTERFACE_H
 
+#include <atomic>
 #include <string>
 #include <vector>
 
@@ -32,6 +33,12 @@ namespace transports
 /// \brief Transport interface for backends
 class TransportInterface
 {
+ protected:
+  /// Transport record key
+  //  This is needed by some transports to route metrics, eg. Kafka
+  //  This can be set to run number or measurement name and run number
+  std::string mKey;
+
  public:
   TransportInterface() = default;
 
@@ -40,6 +47,11 @@ class TransportInterface
   /// Sends metric via given transport
   /// \param message   r-value to string formatted metric
   virtual void send(std::string&& message) = 0;
+
+  /// Transport record key setter
+  void setKey(const std::string& key) {
+    mKey = key;
+  }
 };
 
 } // namespace transports