Use Kafka as Influx Line Protocol transport (#187)

Author: awegrzyn

CMakeLists.txt

@@ -105,7 +105,7 @@ add_library(Monitoring SHARED
   src/Transports/StdOut.cxx
   src/Exceptions/MonitoringException.cxx
   $<$<BOOL:${ApMon_FOUND}>:src/Backends/ApMonBackend.cxx>
-  $<$<BOOL:${RdKafka_FOUND}>:src/Backends/Kafka.cxx>
+  $<$<BOOL:${RdKafka_FOUND}>:src/Transports/Kafka.cxx>
 )
 
 target_include_directories(Monitoring
@@ -136,7 +136,7 @@ if(ApMon_FOUND)
 endif()
 
 if(RdKafka_FOUND)
-   message(STATUS "  Compiling Kafka backend")
+   message(STATUS "  Compiling Kafka transport")
 endif()
 
 # Detect operating system

README.md

@@ -41,14 +41,15 @@ std::unique_ptr<Monitoring> monitoring = MonitoringFactory::Get("backend[-protoc
 
 See the table below to find `URI`s for supported backends:
 
-| Backend name | Transport   | URI backend[-protocol] | URI query  | Default verbosity |
-| ------------ |:-----------:|:----------------------:|:----------:| -----------------:|
-| No-op        | -           | `no-op`                |            | -                 |
-| InfluxDB     | UDP         | `influxdb-udp`         | -          | `info`            |
-| InfluxDB     | Unix socket | `influxdb-unix`        | -          | `info`            |
-| ApMon        | UDP         | `apmon`                | -          | `info`            |
-| StdOut       | -           | `stdout`, `infologger` | [Prefix]   | `debug`           |
-| Kafka        | TCP         | `kafka`                | -          | `info`            |
+| Backend name | Transport   | URI backend[-protocol] | URI query   | Default verbosity |
+| ------------ |:-----------:|:----------------------:|:-----------:| -----------------:|
+| No-op        | -           | `no-op`                |             | -                 |
+| InfluxDB     | UDP         | `influxdb-udp`         | -           | `info`            |
+| InfluxDB     | Unix socket | `influxdb-unix`        | -           | `info`            |
+| InfluxDB     | StdOut      | `influxdb-stdout`      | -           | `info`            |
+| InfluxDB     | Kafka       | `influxdb-kafka`       | Kafka topic | `info`            |
+| ApMon        | UDP         | `apmon`                | -           | `info`            |
+| StdOut       | -           | `stdout`, `infologger` | [Prefix]    | `debug`           |
 
 ##### StdCout output format
 ```

src/Backends/InfluxDB.cxx

@@ -42,8 +42,6 @@ InfluxDB::InfluxDB(std::unique_ptr<transports::TransportInterface> transport)
   MonLogger::Get() << "InfluxDB backend initialized" << MonLogger::End();
 }
 
-InfluxDB::InfluxDB() {}
-
 inline unsigned long InfluxDB::convertTimestamp(const std::chrono::time_point<std::chrono::system_clock>& timestamp)
 {
   return std::chrono::duration_cast<std::chrono::nanoseconds>(

src/Backends/InfluxDB.h

@@ -41,9 +41,6 @@ class InfluxDB final : public Backend
   /// \param transport 	Any available transport (udp, unix, kafka)
   InfluxDB(std::unique_ptr<transports::TransportInterface> transport);
 
-  /// Constructor for other backends
-  InfluxDB();
-
   /// Default destructor
   ~InfluxDB() = default;
 

src/Backends/Kafka.cxx

@@ -33,29 +33,14 @@
 #endif
 
 #ifdef O2_MONITORING_WITH_KAFKA
-#include "Backends/Kafka.h"
+#include "Transports/Kafka.h"
 #endif
 
 namespace o2
 {
 /// ALICE O2 Monitoring system
 namespace monitoring
 {
-#ifdef O2_MONITORING_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},
@@ -95,7 +80,15 @@ std::unique_ptr<Backend> getInfluxDb(http::url uri)
     auto transport = std::make_unique<transports::StdOut>();
     return std::make_unique<backends::InfluxDB>(std::move(transport));
   }
-  throw std::runtime_error("InfluxDB transport protocol not supported");
+  if (uri.protocol == "kafka") {
+#ifdef O2_MONITORING_WITH_KAFKA
+    auto transport = std::make_unique<transports::Kafka>(uri.host, uri.port, uri.search);
+    return std::make_unique<backends::InfluxDB>(std::move(transport));
+#else
+    throw std::runtime_error("Kafka transport is not enabled");
+#endif
+  }
+  throw std::runtime_error("InfluxDB transport not supported: " + uri.protocol);
 }
 
 #ifdef O2_MONITORING_WITH_APPMON
@@ -135,9 +128,10 @@ std::unique_ptr<Backend> MonitoringFactory::GetBackend(std::string& url)
     {"influxdb-udp", getInfluxDb},
     {"influxdb-unix", getInfluxDb},
     {"influxdb-stdout", getInfluxDb},
+    {"influxdb-kafka", getInfluxDb},
     {"apmon", getApMon},
-    {"no-op", getNoop},
-    {"kafka", getKafka}};
+    {"no-op", getNoop}
+  };
 
   http::url parsedUrl = http::ParseHttpUrl(url);
   if (parsedUrl.protocol.empty()) {

src/Backends/Kafka.h

@@ -0,0 +1,73 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// 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 Kafka.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "Kafka.h"
+#include <string>
+#include "../MonLogger.h"
+
+namespace o2
+{
+/// ALICE O2 Monitoring system
+namespace monitoring
+{
+/// Monitoring transports
+namespace transports
+{
+
+Kafka::Kafka(const std::string& host, unsigned int port, const std::string& topic)
+{
+  topic.length() > 0 ? mTopic = topic : mTopic = "test";
+  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", "100", errstr);
+  conf->set("batch.num.messages", "1000", errstr);
+
+  producer = RdKafka::Producer::create(conf, errstr);
+  if (!producer) {
+    throw std::runtime_error("Failed to create Kafka producer: " + errstr);
+  }
+
+  MonLogger::Get() << "Kafka transport initialized (" << host << ":" << port << "/" << mTopic << ")" << MonLogger::End();
+}
+
+Kafka::~Kafka()
+{
+  delete producer;
+}
+
+void Kafka::send(std::string&& message)
+{
+  int32_t partition = RdKafka::Topic::PARTITION_UA;
+
+  RdKafka::ErrorCode resp = producer->produce(
+    mTopic, partition,
+    RdKafka::Producer::RK_MSG_COPY,
+    const_cast<char*>(message.c_str()), message.size(),
+    NULL, 0,
+    0,
+    NULL,
+    NULL);
+  if (resp != RdKafka::ERR_NO_ERROR) {
+    MonLogger::Get() << "Kafka send failed: " << RdKafka::err2str(resp) << MonLogger::End();
+  }
+  producer->poll(0);
+}
+
+} // namespace transports
+} // namespace monitoring
+} // namespace o2