[OMON-584] Modify example to write SOR and EOR times for each run (#293)

Author: awegrzyn

examples/12-KafkaToInfluxDb.cxx

@@ -2,7 +2,6 @@
 /// \file 12-KafkaToInfluxDb.cxx
 /// \author Adam Wegrzynek <[email protected]>
 ///
-
 #include "../src/Transports/KafkaConsumer.h"
 #include "../src/Backends/InfluxDB.h"
 #include "../src/Transports/HTTP.h"
@@ -23,7 +22,6 @@ 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-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")
@@ -32,8 +30,8 @@ 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 = vm["kafka-topics"].as<std::vector<std::string>>();
-  auto kafkaConsumer = std::make_unique<transports::KafkaConsumer>(vm["kafka-host"].as<std::string>() + ":9092", topics, "kafka-to-influxdb");
+  std::vector<std::string> topics = {"aliecs.env_leave_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>(
     "http://" + vm["influxdb-host"].as<std::string>() + ":8086/api/v2/write?" +
     "org=" + vm["influxdb-org"].as<std::string>() + "&" +
@@ -50,19 +48,14 @@ int main(int argc, char* argv[])
         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");
-        }
+        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");
         int run = stateChange.envinfo().runnumber();
         if (run > 1) {
-          metric.addValue(run, "run");
+          influxdbBackend->sendWithRun(metric, stateChange.envinfo().environmentid(), std::to_string(run));
         }
-        influxdbBackend->sendWithId(metric, stateChange.envinfo().environmentid());
       }
     }
     std::this_thread::sleep_for(std::chrono::milliseconds(1000));

src/Backends/InfluxDB.cxx

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

src/Backends/InfluxDB.h

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