[OMON-532] Document connecting to metric stream using library (#290)

Author: awegrzyn

README.md

@@ -190,3 +190,50 @@ Metric::setVerbosityPolicy(Verbosity verbosity, const std::regex& regex)
 
 ## System monitoring, server-side backends installation and configuration
 This guide explains manual installation. For `ansible` deployment see [AliceO2Group/system-configuration](https://gitlab.cern.ch/AliceO2Group/system-configuration/tree/master/ansible) gitlab repo.
+
+---
+
+## Receiving metrics from Monitoring system (development instructions)
+
+### Compile Monitoring library with Kafka backend
+- Install `boost`
+- Compile `librdkafka`
+  ```bash
+  git clone -b v1.8.2 https://github.com/edenhill/librdkafka && cd librdkafka
+  cmake -H. -B./_cmake_build -DENABLE_LZ4_EXT=OFF -DCMAKE_INSTALL_LIBDIR=lib -DRDKAFKA_BUILD_TESTS=OFF -DRDKAFKA_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=~/librdkafka_install
+  cmake --build ./_cmake_build --target install -j
+  ```
+- Compile Monitoring library, make sure to define `RdKafka_DIR` and point to CMake config directory:
+  ```bash
+  git clone https://github.com/AliceO2Group/Monitoring && cd Monitoring
+  cmake -H. -B./_cmake_build -DRdKafka_DIR=~/librdkafka_install/lib/cmake/RdKafka/ -DCMAKE_INSTALL_PREFIX=~/Monitoring_install
+  cmake --build ./_cmake_build --target install -j
+  ```
+
+### Look for Monitoring library in your CMake
+As `librdkafka` is optional dependency of Monitoring it is not handled by CMakeConfig, therefore you need:
+```cmake
+find_package(RdKafka CONFIG REQUIRED)
+find_package(Monitoring CONFIG REQUIRED)
+```
+
+And then, link against `AliceO2::Monitoring` target.
+
+### Connect to Monitoring server
+```cpp
+#include "Monitoring/MonitoringFactory.h"
+...
+
+std::vector<std::string> topics = {"topic-to-subscribe"};
+auto client = MonitoringFactory::GetPullClient("kafka-server:9092", topics);
+for (;;) {
+  auto metrics = client->pull();
+  if (!metrics.empty()) {
+    //  DO SOMETHING !
+  }
+  std::this_thread::sleep_for(std::chrono::milliseconds(100));
+}
+```
+
+### Data format
+Native data format is [Influx Line Protocol](https://docs.influxdata.com/influxdb/latest/reference/syntax/line-protocol/) but metrics can be converted into any format listed in here: https://docs.influxdata.com/telegraf/latest/data_formats/output/

doc/DPL.md

@@ -3,7 +3,7 @@
 /// \author Adam Wegrzynek <[email protected]>
 ///
 
-#include "../src/Transports/KafkaConsumer.h"
+#include "Monitoring/MonitoringFactory.h"
 #include "../src/Transports/HTTP.h"
 #include "../src/Transports/WebSocket.h"
 
@@ -27,7 +27,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);
 
-  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 kafkaConsumer = MonitoringFactory::GetPullClient(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 (;;) {
@@ -36,7 +37,7 @@ int main(int argc, char* argv[])
     }
   });
   for (;;) {
-    auto metrics = kafkaConsumer->receive();
+    auto metrics = kafkaConsumer->pull();
     if (!metrics.empty()) {
       outTransport->send(boost::algorithm::join(metrics, "\n"));
     }

examples/11-KafkaToWebsocket.cxx

@@ -33,7 +33,7 @@ int main(int argc, char* argv[])
   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);
+  auto kafkaConsumer = std::make_unique<transports::KafkaConsumer>(vm["kafka-host"].as<std::string>() + ":9092", topics, "kafka-to-influxdb");
   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>() + "&" +
@@ -42,7 +42,7 @@ int main(int argc, char* argv[])
   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();
+    auto changes = kafkaConsumer->pull();
     if (!changes.empty()) {
       for (auto& change : changes) {
         aliceo2::envs::NewStateNotification stateChange;

examples/12-KafkaToInfluxDb.cxx

@@ -231,10 +231,10 @@ 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>>()}
+    vm["kafka-host"].as<std::string>() + ":9092", std::vector<std::string>{vm["kafka-topics"].as<std::vector<std::string>>()}, "kafka-aliecs"
   );
   for (;;) {
-    auto serializedRuns = kafkaConsumer->receive();
+    auto serializedRuns = kafkaConsumer->pull();
     if (!serializedRuns.empty()) {
       deserializeActiveRuns(serializedRuns.back());
     }

examples/8-KafkaToHttpServer.cxx

@@ -18,6 +18,7 @@
 #define ALICEO2_MONITORING_FACTORY_H
 
 #include "Monitoring/Monitoring.h"
+#include "Monitoring/PullClient.h"
 
 namespace o2
 {
@@ -42,7 +43,7 @@ class MonitoringFactory
   /// \return		monitoring backend
   /// \throw MonitoringException when backend initialisation failed
   static std::unique_ptr<Backend> GetBackend(std::string& url);
-
+  static std::unique_ptr<PullClient> GetPullClient(const std::string &url, const std::vector<std::string>& topics, const std::string &label = "o2-monitoring-group");
  private:
   /// Private constructor disallows to create instance of Factory
   MonitoringFactory() = default;

include/Monitoring/MonitoringFactory.h

@@ -0,0 +1,44 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+///
+/// \file PullClient.h
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#ifndef ALICEO2_MONITORING_CORE_PULLCLIENT_H
+#define ALICEO2_MONITORING_CORE_PULLCLIENT_H
+
+#include <vector>
+#include <string>
+
+namespace o2
+{
+/// ALICE O2 Monitoring system
+namespace monitoring
+{
+
+/// \brief PullClient pure virtual interface
+///
+/// Interface that allows to send a metric to remote backend.
+/// In addition, default tagset (for all handled metrics) can be created.
+class PullClient
+{
+ public:
+  /// Default destructor
+  virtual ~PullClient() = default;
+  virtual std::vector<std::string> pull() = 0;
+};
+
+} // namespace monitoring
+} // namespace o2
+
+#endif // ALICEO2_MONITORING_CORE_PULLCLIENT_H

include/Monitoring/PullClient.h

@@ -219,6 +219,15 @@ std::unique_ptr<Monitoring> MonitoringFactory::Get(std::string urlsString)
   }
   return monitoring;
 }
+#ifdef O2_MONITORING_WITH_KAFKA
+std::unique_ptr<PullClient> MonitoringFactory::GetPullClient(const std::string &url, const std::vector<std::string>& topics, const std::string &label) {
+  auto client = std::make_unique<transports::KafkaConsumer>(url, topics, label);
+  return client;
+#else
+std::unique_ptr<PullClient> MonitoringFactory::GetPullClient(const std::string&, const std::vector<std::string>&, const std::string&) {
+  throw MonitoringException("Factory", "Compile library with Kafka");
+#endif
+}
 
 } // namespace monitoring
 } // namespace o2

src/MonitoringFactory.cxx

@@ -17,6 +17,7 @@
 #include "KafkaConsumer.h"
 #include <string>
 #include "../MonLogger.h"
+#include "../Exceptions/MonitoringException.h"
 
 namespace o2
 {
@@ -27,13 +28,13 @@ namespace monitoring
 namespace transports
 {
 
-KafkaConsumer::KafkaConsumer(const std::string& host, unsigned int port, const std::vector<std::string>& topics) : mTopics(topics)
+KafkaConsumer::KafkaConsumer(const std::string& url, const std::vector<std::string>& topics, const std::string& groupId) : mTopics(topics)
 {
   std::string errstr;
   std::unique_ptr<RdKafka::Conf> conf{RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL)};
-  if (conf->set("bootstrap.servers", host + ":" + std::to_string(port), errstr) != RdKafka::Conf::CONF_OK
+  if (conf->set("bootstrap.servers", url, errstr) != RdKafka::Conf::CONF_OK
    || conf->set("enable.partition.eof", "false", errstr) != RdKafka::Conf::CONF_OK
-   || conf->set("group.id", "o2-monitoring-group", errstr) != RdKafka::Conf::CONF_OK
+   || conf->set("group.id", groupId, errstr) != RdKafka::Conf::CONF_OK
    || conf->set("auto.offset.reset", "latest", errstr) != RdKafka::Conf::CONF_OK
    || conf->set("heartbeat.interval.ms", "2000", errstr) != RdKafka::Conf::CONF_OK
    || conf->set("session.timeout.ms", "6000", errstr) != RdKafka::Conf::CONF_OK
@@ -50,7 +51,7 @@ KafkaConsumer::KafkaConsumer(const std::string& host, unsigned int port, const s
   }
 }
 
-std::vector<std::string> KafkaConsumer::receive()
+std::vector<std::string> KafkaConsumer::pull()
 {
   std::vector<std::string> received;
   size_t batch_size = 5;

src/Transports/KafkaConsumer.cxx

@@ -14,10 +14,10 @@
 /// \author Adam Wegrzynek <[email protected]>
 ///
 
-#ifndef ALICEO2_MONITORING_TRANSPORTS_KAFKA_H
-#define ALICEO2_MONITORING_TRANSPORTS_KAFKA_H
+#ifndef ALICEO2_MONITORING_TRANSPORTS_KAFKACONSUMER_H
+#define ALICEO2_MONITORING_TRANSPORTS_KAFKACONSUMER_H
 
-#include "TransportInterface.h"
+#include "Monitoring/PullClient.h"
 
 #include <chrono>
 #include <string>
@@ -33,25 +33,21 @@ namespace transports
 {
 
 /// \brief Transport that sends string formatted metrics via Kafka
-class KafkaConsumer : public TransportInterface
+class KafkaConsumer : public PullClient
 {
  public:
   /// Creates producer
-  /// \param hostname      Hostname
-  /// \param port          Port number
-  /// \param topic 	   Kafka topic
-  KafkaConsumer(const std::string& host, unsigned int port, const std::vector<std::string>& topic);
+  /// \param url  Broker URL (host:port)
+  /// \param topics 	   Kafka topics to subscribe to
+  /// \param groupId     Kafka consumer group id
+  KafkaConsumer(const std::string& url, const std::vector<std::string>& topics, const std::string& groupId);
 
   /// Deletes producer
   ~KafkaConsumer();
 
-  void send(std::string&&/* message*/) override {
-    throw MonitoringException("Transport", "This transport does not implement sending");
-  }
-
   /// Sends metric via Kafka
   /// \param message   r-value string formated
-  std::vector<std::string> receive() override;
+  std::vector<std::string> pull() override;
  private:
   /// Kafka producer instance
   RdKafka::KafkaConsumer* mConsumer;
@@ -64,4 +60,4 @@ class KafkaConsumer : public TransportInterface
 } // namespace monitoring
 } // namespace o2
 
-#endif // ALICEO2_MONITORING_TRANSPORTS_KAFKA_H
+#endif // ALICEO2_MONITORING_TRANSPORTS_KAFKACONSUMER_H