[OMON-584] Pass topic along with the message from Kafka consumer (#297)

Author: awegrzyn

CMakeLists.txt

@@ -31,7 +31,7 @@ endif()
 
 # Define project
 project(Monitoring
-  VERSION 3.12.10
+  VERSION 3.12.11
   DESCRIPTION "O2 Monitoring library"
   LANGUAGES CXX
 )

README.md

@@ -232,7 +232,7 @@ auto client = MonitoringFactory::GetPullClient("kafka-server:9092", topics);
 for (;;) {
   auto metrics = client->pull();
   if (!metrics.empty()) {
-    //  DO SOMETHING !
+    /// metric.first => topic name; metric.second => metric itself
   }
   std::this_thread::sleep_for(std::chrono::milliseconds(100));
 }

examples/11-KafkaToWebsocket.cxx

@@ -10,7 +10,6 @@
 #include <iostream>
 #include <memory>
 #include <thread>
-#include <boost/algorithm/string/join.hpp>
 #include <boost/program_options.hpp>
 
 using namespace o2::monitoring;
@@ -38,8 +37,8 @@ int main(int argc, char* argv[])
   });
   for (;;) {
     auto metrics = kafkaConsumer->pull();
-    if (!metrics.empty()) {
-      outTransport->send(boost::algorithm::join(metrics, "\n"));
+    for (auto& metric : metrics) {
+      outTransport->send(std::move(metric.second));
     }
     std::this_thread::sleep_for(std::chrono::milliseconds(100));
   }

examples/12-KafkaToInfluxDb.cxx

@@ -30,7 +30,7 @@ int main(int argc, char* argv[])
   boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);
   boost::program_options::notify(vm);
 
-  std::vector<std::string> topics = {"aliecs.env_leave_state.RUNNING"};
+  std::vector<std::string> topics = {"aliecs.env_leave_state.RUNNING", "aliecs.env_state.RUNNING"};
   auto kafkaConsumer = std::make_unique<transports::KafkaConsumer>(vm["kafka-host"].as<std::string>() + ":9092", topics, "aliecs-run-times");
   auto httpTransport = std::make_unique<transports::HTTP>(
     vm["influxdb-url"].as<std::string>() + "/api/v2/write?" +
@@ -44,14 +44,17 @@ int main(int argc, char* argv[])
     if (!changes.empty()) {
       for (auto& change : changes) {
         aliceo2::envs::NewStateNotification stateChange;
-        stateChange.ParseFromString(change);
+        stateChange.ParseFromString(change.second);
         if (stateChange.envinfo().state().empty()) {
           continue;
         }
-        std::cout << stateChange.envinfo().environmentid() << " (" << stateChange.envinfo().runnumber() << ") EOR: from " <<stateChange.envinfo().enterstatetimestamp() << " to " << stateChange.timestamp() << std::endl;
-        auto metric = Metric{"run_times"}
-          .addValue(stateChange.envinfo().enterstatetimestamp(), "sor")
-          .addValue(stateChange.timestamp(), "eor");
+        std::cout << stateChange.envinfo().environmentid() << "/" << stateChange.envinfo().runnumber() << "  " << change.first << " SOR: " <<stateChange.envinfo().enterstatetimestamp() << " EOR: " << stateChange.timestamp() << std::endl;
+        auto metric = Metric{"run_times"};
+        if (change.first.find("leave") != std::string::npos) {
+          metric.addValue(stateChange.envinfo().enterstatetimestamp(), "sor").addValue(stateChange.timestamp(), "eor");
+        } else {
+          metric.addValue(stateChange.envinfo().runtype(), "type");
+        }
         int run = stateChange.envinfo().runnumber();
         if (run > 1) {
           influxdbBackend->sendWithRun(metric, stateChange.envinfo().environmentid(), std::to_string(run));

examples/8-KafkaToHttpServer.cxx

@@ -231,12 +231,12 @@ int main(int argc, char* argv[]) {
   });
 
   auto kafkaConsumer = std::make_unique<o2::monitoring::transports::KafkaConsumer>(
-    vm["kafka-host"].as<std::string>() + ":9092", std::vector<std::string>{vm["kafka-topics"].as<std::vector<std::string>>()}, "kafka-aliecs"
+    vm["kafka-host"].as<std::string>() + ":9092", std::vector<std::string>{vm["kafka-topics"].as<std::vector<std::string>>()}, "kafka-aliecs-active-envs"
   );
   for (;;) {
     auto serializedRuns = kafkaConsumer->pull();
     if (!serializedRuns.empty()) {
-      deserializeActiveRuns(serializedRuns.back());
+      deserializeActiveRuns(serializedRuns.back().second);
     }
     std::this_thread::sleep_for(std::chrono::milliseconds(500));
   }

include/Monitoring/PullClient.h

@@ -35,7 +35,7 @@ class PullClient
  public:
   /// Default destructor
   virtual ~PullClient() = default;
-  virtual std::vector<std::string> pull() = 0;
+  virtual std::vector<std::pair<std::string, std::string>> pull() = 0;
 };
 
 } // namespace monitoring

src/Transports/KafkaConsumer.cxx

@@ -51,9 +51,9 @@ KafkaConsumer::KafkaConsumer(const std::string& url, const std::vector<std::stri
   }
 }
 
-std::vector<std::string> KafkaConsumer::pull()
+std::vector<std::pair<std::string, std::string>> KafkaConsumer::pull()
 {
-  std::vector<std::string> received;
+  std::vector<std::pair<std::string, std::string>> received;
   size_t batch_size = 5;
   int remaining_timeout = 1000;
   auto start = std::chrono::high_resolution_clock::now();
@@ -64,7 +64,7 @@ std::vector<std::string> KafkaConsumer::pull()
     case RdKafka::ERR__TIMED_OUT:
       break;
     case RdKafka::ERR_NO_ERROR:
-      received.push_back(std::string(static_cast<char*>(message->payload()), message->len()));
+      received.push_back({message->topic_name(), std::string(static_cast<char*>(message->payload()), message->len())});
       break;
     default:
       std::cerr << "%% Consumer error: " << message->errstr() << std::endl;

src/Transports/KafkaConsumer.h

@@ -45,9 +45,10 @@ class KafkaConsumer : public PullClient
   /// Deletes producer
   ~KafkaConsumer();
 
-  /// Sends metric via Kafka
+  /// Pulls metrics from Kafka
   /// \param message   r-value string formated
-  std::vector<std::string> pull() override;
+  /// \returns Key-Value pair list as topic : metric
+  std::vector<std::pair<std::string, std::string>> pull() override;
  private:
   /// Kafka producer instance
   RdKafka::KafkaConsumer* mConsumer;