[OMON-539] Send AliECS state transitions to InfluxDB in example (#284)

Author: awegrzyn

CMakeLists.txt

@@ -242,7 +242,7 @@ endif()
 ####################################
 # Kafka protobuf deserializer
 ####################################
-if(RdKafka_FOUND AND Protobuf_FOUND)
+if(RdKafka_FOUND AND Protobuf_FOUND AND CURL_FOUND)
   set(PROTO_FILE ${CMAKE_CURRENT_SOURCE_DIR}/proto/envs.proto)
   get_filename_component(PROTO_OUTPUT_NAME ${PROTO_FILE} NAME_WE)
   get_filename_component(PROTO_FILE_PREFIX ${PROTO_FILE} PATH)

examples/11-KafkaToWebsocket.cxx

@@ -20,14 +20,14 @@ int main(int argc, char* argv[])
   boost::program_options::options_description desc("Program options");
   desc.add_options()
     ("kafka-host", boost::program_options::value<std::string>()->required(), "Kafka broker hostname")
-    ("kafka-topic", boost::program_options::value<std::string>()->required(), "Kafka topic")
+    ("kafka-topics", boost::program_options::value<std::vector<std::string>>()->multitoken()->required(), "Kafka topics")
     ("grafana-host", boost::program_options::value<std::string>()->required(), "Grafana hostname")
     ("grafana-key", boost::program_options::value<std::string>()->required(), "Grafana API key");
   boost::program_options::variables_map vm;
   boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);
   boost::program_options::notify(vm);
 
-  auto kafkaConsumer = std::make_unique<transports::KafkaConsumer>(vm["kafka-host"].as<std::string>(), 9092, vm["kafka-topic"].as<std::string>());
+  auto kafkaConsumer = std::make_unique<transports::KafkaConsumer>(vm["kafka-host"].as<std::string>(), 9092, std::vector<std::string>{vm["kafka-topics"].as<std::vector<std::string>>()});
   auto outTransport = std::make_unique<transports::WebSocket>(vm["grafana-host"].as<std::string>(), 3000, vm["grafana-key"].as<std::string>(), "alice_o2");
   std::thread readThread([&outTransport](){
     for (;;) {

examples/12-KafkaToInfluxDb.cxx

@@ -4,14 +4,15 @@
 ///
 
 #include "../src/Transports/KafkaConsumer.h"
+#include "../src/Backends/InfluxDB.h"
 #include "../src/Transports/HTTP.h"
-#include "../src/Transports/WebSocket.h"
 
 #include <iostream>
 #include <memory>
 #include <thread>
-#include <boost/algorithm/string/join.hpp>
 #include <boost/program_options.hpp>
+#include <boost/algorithm/string/join.hpp>
+#include <boost/algorithm/string.hpp>
 
 #include "envs.pb.h"
 
@@ -22,21 +23,46 @@ int main(int argc, char* argv[])
   boost::program_options::options_description desc("Program options");
   desc.add_options()
     ("kafka-host", boost::program_options::value<std::string>()->required(), "Kafka broker hostname")
-    ("kafka-topic", boost::program_options::value<std::string>()->required(), "Kafka topic");
+    ("kafka-topics", boost::program_options::value<std::vector<std::string>>()->multitoken()->required(), "Kafka topics")
+    ("influxdb-host", boost::program_options::value<std::string>()->required(), "InfluxDB hostname")
+    ("influxdb-token", boost::program_options::value<std::string>()->required(), "InfluxDB token")
+    ("influxdb-org", boost::program_options::value<std::string>()->default_value("cern"), "InfluxDB organisation")
+    ("influxdb-bucket", boost::program_options::value<std::string>()->default_value("aliecs"), "InfluxDB bucket");
   boost::program_options::variables_map vm;
   boost::program_options::store(boost::program_options::parse_command_line(argc, argv, desc), vm);
   boost::program_options::notify(vm);
 
-  auto kafkaConsumer = std::make_unique<transports::KafkaConsumer>(vm["kafka-host"].as<std::string>(), 9092, vm["kafka-topic"].as<std::string>());
+  std::vector<std::string> topics = vm["kafka-topics"].as<std::vector<std::string>>();
+  auto kafkaConsumer = std::make_unique<transports::KafkaConsumer>(vm["kafka-host"].as<std::string>(), 9092, topics);
+  auto httpTransport = std::make_unique<transports::HTTP>(
+    "http://" + vm["influxdb-host"].as<std::string>() + ":8086/api/v2/write?" +
+    "org=" + vm["influxdb-org"].as<std::string>() + "&" +
+    "bucket=" + vm["influxdb-bucket"].as<std::string>()
+  );
+  httpTransport->addHeader("Authorization: Token " + vm["influxdb-token"].as<std::string>());
+  auto influxdbBackend = std::make_unique<backends::InfluxDB>(std::move(httpTransport));
   for (;;) {
     auto changes = kafkaConsumer->receive();
     if (!changes.empty()) {
       for (auto& change : changes) {
         aliceo2::envs::NewStateNotification stateChange;
         stateChange.ParseFromString(change);
-        std::cout << "EnvID: " << stateChange.envinfo().environmentid() << std::endl
-                  << "State: " << stateChange.envinfo().state() << std::endl
-                  << "First detector: " << stateChange.envinfo().detectors().Get(0) << std::endl;
+        if (stateChange.envinfo().state().empty()) {
+          continue;
+        }
+        std::cout << stateChange.envinfo().environmentid() << " => " << stateChange.envinfo().state()
+                  << " (" << stateChange.envinfo().runnumber() << ")" << std::endl;
+        auto metric = Metric{"env_info"}.addValue(stateChange.envinfo().state(), "state");
+        auto detectorsProto = stateChange.envinfo().detectors();
+        std::vector<std::string> detectors(detectorsProto.begin(), detectorsProto.end());
+        if (detectors.size() > 0) {
+          metric.addValue(boost::algorithm::join(detectors, " "), "detectors");
+        }
+        int run = stateChange.envinfo().runnumber();
+        if (run > 1) {
+          metric.addValue(run, "run");
+        }
+        influxdbBackend->sendWithId(metric, stateChange.envinfo().environmentid());
       }
     }
     std::this_thread::sleep_for(std::chrono::milliseconds(1000));

src/Backends/InfluxDB.cxx

@@ -78,6 +78,13 @@ void InfluxDB::send(const Metric& metric)
   mTransport->send(toInfluxLineProtocol(metric));
 }
 
+void InfluxDB::sendWithId(const Metric& metric, const std::string& id)
+{
+  auto serialized = toInfluxLineProtocol(metric);
+  serialized.insert(serialized.find(',') + 1, "id=" + id);
+  mTransport->send(std::move(serialized));
+}
+
 std::string InfluxDB::toInfluxLineProtocol(const Metric& metric)
 {
   std::stringstream convert;

src/Backends/InfluxDB.h

@@ -54,6 +54,11 @@ class InfluxDB final : public Backend
   /// \param metric    reference to metric object
   void send(const Metric& metric) override;
 
+  /// Sends metric with ID tag
+  /// \param metric    reference to metric object
+  /// \param id        ID tag value
+  void sendWithId(const Metric& metric, const std::string& id);
+
   /// Sends multiple metrics not related to each other
   /// \@param metrics  vector of metrics
   void send(std::vector<Metric>&& metrics) override;

src/Transports/KafkaConsumer.cxx

@@ -27,7 +27,7 @@ namespace monitoring
 namespace transports
 {
 
-KafkaConsumer::KafkaConsumer(const std::string& host, unsigned int port, const std::string& topic) : mTopic(topic)
+KafkaConsumer::KafkaConsumer(const std::string& host, unsigned int port, const std::vector<std::string>& topics) : mTopics(topics)
 {
   std::string errstr;
   std::unique_ptr<RdKafka::Conf> conf{RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL)};
@@ -45,7 +45,7 @@ KafkaConsumer::KafkaConsumer(const std::string& host, unsigned int port, const s
   if (!mConsumer) {
     MonLogger::Get(Severity::Error) << "Could not initialize Kafka consumer" << MonLogger::End();
   }
-  if (mConsumer->subscribe({mTopic})) {
+  if (mConsumer->subscribe(mTopics)) {
     MonLogger::Get(Severity::Warn) << "Failed to subscribe to topic" << MonLogger::End();
   }
 }

src/Transports/KafkaConsumer.h

@@ -40,7 +40,7 @@ class KafkaConsumer : public TransportInterface
   /// \param hostname      Hostname
   /// \param port          Port number
   /// \param topic 	   Kafka topic
-  KafkaConsumer(const std::string& host, unsigned int port, const std::string& topic);
+  KafkaConsumer(const std::string& host, unsigned int port, const std::vector<std::string>& topic);
 
   /// Deletes producer
   ~KafkaConsumer();
@@ -57,7 +57,7 @@ class KafkaConsumer : public TransportInterface
   RdKafka::KafkaConsumer* mConsumer;
 
   /// Kafka topic
-  std::string mTopic;
+  std::vector<std::string> mTopics;
 };
 
 } // namespace transports