Add Kafka backend (#97)

Author: awegrzyn

CMakeLists.txt

@@ -53,7 +53,7 @@ list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_LIST_DIR}/cmake")
 find_package(Boost REQUIRED COMPONENTS unit_test_framework program_options system filesystem)
 find_package(Git QUIET)
 find_package(ApMon MODULE)
-
+find_package(RdKafka CONFIG)
 
 ####################################
 # Library
@@ -86,6 +86,7 @@ add_library(Monitoring SHARED
   src/Transports/Unix.cxx
   src/Exceptions/MonitoringException.cxx
   $<$<BOOL:${ApMon_FOUND}>:src/Backends/ApMonBackend.cxx>
+  $<$<BOOL:${RdKafka_FOUND}>:src/Backends/Kafka.cxx>
 )
 
 target_include_directories(Monitoring
@@ -103,13 +104,18 @@ target_link_libraries(Monitoring
   PRIVATE
     Boost::system
     $<$<BOOL:${ApMon_FOUND}>:ApMon::ApMon>
+    $<$<BOOL:${RdKafka_FOUND}>:RdKafka::rdkafka++>
 )
 
 # Handle ApMon optional dependency
 if(ApMon_FOUND)
   message(STATUS "  Compiling ApMon backend")
 endif()
 
+if(RdKafka_FOUND)
+   message(STATUS "  Compiling Kafka backend")
+endif()
+
 # Detect operating system
 if (UNIX AND NOT APPLE)
   message(STATUS "Detected Linux: Process monitor enabled")
@@ -126,6 +132,7 @@ target_compile_definitions(Monitoring
     $<$<BOOL:${APPLE}>:O2_MONITORING_OS_MAC>
     $<$<BOOL:${LINUX}>:O2_MONITORING_OS_LINUX>
     $<$<BOOL:${ApMon_FOUND}>:O2_MONITORING_WITH_APPMON>
+    $<$<BOOL:${RdKafka_FOUND}>:O2_MONITORING_WITH__WITH_KAFKA>
   )
 
 # Use C++17

README.md

@@ -49,6 +49,7 @@ See the table below to find `URI`s for supported backends:
 | ApMon        | UDP         | `apmon`                | -          | `info`            |
 | StdOut       | -           | `stdout`, `infologger` | [Prefix]   | `debug`           |
 | Flume        | UDP         | `flume`                | -          | `info`            |
+| Kafka        | TCP         | `kafka`                | -          | `info`            |
 
 ##### StdCout output format
 ```

src/Backends/InfluxDB.cxx

@@ -39,6 +39,8 @@ InfluxDB::InfluxDB(const std::string& host, unsigned int port) :
                    << " ("<< host << ":" << port << ")" << MonLogger::End();
 }
 
+InfluxDB::InfluxDB() {}
+
 InfluxDB::InfluxDB(const std::string& socketPath) :
   mTransport(std::make_unique<transports::Unix>(socketPath))
 {

src/Backends/InfluxDB.h

@@ -42,9 +42,13 @@ class InfluxDB final : public Backend
     /// \param port      InfluxDB UDP endpoint port number
     InfluxDB(const std::string& host, unsigned int port);
 
+    /// Constructor for other backends
+    InfluxDB();
+
     /// Constructor for Unix socket transport
     InfluxDB(const std::string& socketPath);
 
+
     /// Default destructor
     ~InfluxDB() = default;
 
@@ -70,18 +74,17 @@ class InfluxDB final : public Backend
     /// \param name      tag name
     /// \param value     tag value
     void addGlobalTag(std::string_view name, std::string_view value) override;
-  
+
+    /// Converts metric to Influx Line Protocol format
+    /// \param metric
+    std::string toInfluxLineProtocol(const Metric& metric);
   private:
     std::unique_ptr<transports::TransportInterface> mTransport; ///< InfluxDB transport
     std::string tagSet; ///< Global tagset (common for each metric)
 
     /// Escapes " ", "," and "=" characters
     /// \param escaped   string rerference to escape characters from
     void escape(std::string& escaped);
-
-    /// Converts metric to Influx Line Protocol format
-    /// \param metric
-    std::string toInfluxLineProtocol(const Metric& metric);
 };
 
 } // namespace backends

src/Backends/Kafka.cxx

@@ -0,0 +1,90 @@
+///
+/// \file Kafka.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "Kafka.h"
+#include <string>
+#include <boost/lexical_cast.hpp>
+
+namespace o2
+{
+/// ALICE O2 Monitoring system
+namespace monitoring
+{
+/// Monitoring backends
+namespace backends
+{
+
+Kafka::Kafka(const std::string& host, unsigned int port, const std::string& topic) :
+  mInfluxDB(), mTopic(topic)
+{
+  std::string errstr;
+  RdKafka::Conf *conf = RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL);
+  conf->set("bootstrap.servers", host + ":" + std::to_string(port), errstr);
+  conf->set("request.required.acks", "0", errstr);
+  conf->set("message.send.max.retries", "0", errstr);
+  conf->set("queue.buffering.max.ms", "10", errstr);
+  conf->set("batch.num.messages", "1000", errstr);
+
+  producer = RdKafka::Producer::create(conf, errstr);
+    if (!producer) {
+      MonLogger::Get() << "Failed to create producer: " << errstr << MonLogger::End();
+      exit(1);
+    }
+
+  MonLogger::Get() << "Kafka backend initialized"
+                   << " ("<< host << ":" << port << ")" << MonLogger::End();
+}
+
+Kafka::~Kafka()
+{
+  delete producer;
+}
+
+inline std::string Kafka::convertTimestamp(const std::chrono::time_point<std::chrono::system_clock>& timestamp)
+{
+  return std::to_string(std::chrono::duration_cast <std::chrono::nanoseconds>(
+    timestamp.time_since_epoch()
+  ).count());
+}
+
+void Kafka::sendMultiple(std::string /*measurement*/, std::vector<Metric>&& /*metrics*/)
+{
+}
+
+void Kafka::send(std::vector<Metric>&& /*metrics*/)
+{
+}
+
+void Kafka::send(const Metric& metric)
+{
+  std::string influxLine = mInfluxDB.toInfluxLineProtocol(metric);
+  int32_t partition = RdKafka::Topic::PARTITION_UA;
+
+  RdKafka::ErrorCode resp = producer->produce(
+    mTopic, partition,
+    RdKafka::Producer::RK_MSG_COPY,
+    const_cast<char*>(influxLine.c_str()), influxLine.size(),
+    NULL, 0,
+    0,  
+    NULL,
+    NULL
+  );  
+  if (resp != RdKafka::ERR_NO_ERROR) {
+    MonLogger::Get() << "% Produce failed: " << RdKafka::err2str(resp) << MonLogger::End();
+  }
+  producer->poll(0);
+}
+
+void Kafka::addGlobalTag(std::string_view name, std::string_view value)
+{
+  std::string sName = name.data();
+  std::string sValue = value.data();
+  if (!tagSet.empty()) tagSet += ",";
+  tagSet += sName + "=" + sValue;
+}
+
+} // namespace backends
+} // namespace monitoring
+} // namespace o2

src/Backends/Kafka.h

@@ -0,0 +1,71 @@
+///
+/// \file Kafka.h
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#ifndef ALICEO2_MONITORING_BACKENDS_KAFKA_H
+#define ALICEO2_MONITORING_BACKENDS_KAFKA_H
+
+#include "Monitoring/Backend.h"
+#include "InfluxDB.h"
+#include "../MonLogger.h"
+#include <chrono>
+#include <string>
+#include <unordered_map>
+#include <librdkafka/rdkafkacpp.h>
+
+namespace o2
+{
+/// ALICE O2 Monitoring system
+namespace monitoring
+{
+/// Monitoring backends
+namespace backends
+{
+
+/// \brief Backend that sends metrics to Kafka over Influx Line protocol
+class Kafka final : public Backend
+{
+  public:
+    /// \param host      Kafka UDP endpoint hostname
+    /// \param port      Kafka UDP endpoint port number
+    Kafka(const std::string& host, unsigned int port, const std::string& topic = "test");
+
+    /// Default destructor
+    ~Kafka();
+
+    /// Convert timestamp to unsigned long as required by Kafka
+    /// \param timestamp  chrono time_point timestamp
+    /// \return  	 timestamp in nanoseconds
+    inline std::string convertTimestamp(const std::chrono::time_point<std::chrono::system_clock>& timestamp);
+
+    /// Sends metric to Kafka instance via one transport
+    /// \param metric    reference to metric object
+    void send(const Metric& metric) override;
+
+    /// Sends multiple metrics not related to each other - NOT SUPPORTED
+    /// \@param metrics  vector of metrics
+    void send(std::vector<Metric>&& metrics) override;
+
+    /// Sends multiple values in single measurement - NOT SUPPORTED
+    /// \param measurement measurement name
+    /// \param metrics 	list of metrics
+    void sendMultiple(std::string measurement, std::vector<Metric>&& metrics) override;
+
+    /// Adds tag
+    /// \param name      tag name
+    /// \param value     tag value
+    void addGlobalTag(std::string_view name, std::string_view value) override;
+  
+  private:
+    RdKafka::Producer *producer; ///< Kafka producer instance
+    std::string tagSet; ///< Global tagset (common for each metric)
+    InfluxDB mInfluxDB; ///< InfluxDB instance
+    std::string mTopic; ///< Kafka topic
+};
+
+} // namespace backends
+} // namespace monitoring
+} // namespace o2
+
+#endif // ALICEO2_MONITORING_BACKENDS_KAFKA_H

src/MonitoringFactory.cxx

@@ -30,11 +30,28 @@
 #include "Backends/ApMonBackend.h"
 #endif
 
+#ifdef O2_MONITORING__WITH_KAFKA
+#include "Backends/Kafka.h"
+#endif
+
 namespace o2 
 {
 /// ALICE O2 Monitoring system
 namespace monitoring 
 {
+#ifdef _WITH_KAFKA
+std::unique_ptr<Backend> getKafka(http::url uri) {
+  if (uri.search.size() > 0) {
+    return std::make_unique<backends::Kafka>(uri.host, uri.port, uri.search);
+  } else {
+    return std::make_unique<backends::Kafka>(uri.host, uri.port);
+  }
+}
+#else
+std::unique_ptr<Backend> getKafka(http::url /*uri*/) {
+  throw std::runtime_error("Kafka backend is not enabled");
+}
+#endif
 
 static const std::map<std::string, Verbosity> verbosities = {
   {"/prod", Verbosity::Prod},
@@ -108,7 +125,8 @@ std::unique_ptr<Backend> MonitoringFactory::GetBackend(std::string& url) {
     {"influxdb-unix", getInfluxDb},
     {"apmon", getApMon},
     {"flume", getFlume},
-    {"no-op", getNoop}
+    {"no-op", getNoop},
+    {"kafka", getKafka}
   };
 
   http::url parsedUrl = http::ParseHttpUrl(url);