[OMON-520] Add Kafka bridge example (#277)

Author: awegrzyn

CMakeLists.txt

@@ -216,7 +216,7 @@ set(EXAMPLES
 )
 
 if(RdKafka_FOUND)
-  list(APPEND EXAMPLES "examples/11-KafkaToHttp.cxx")
+  list(APPEND EXAMPLES "examples/11-KafkaToWebsocket.cxx")
 endif()
 
 foreach (example ${EXAMPLES})
@@ -231,6 +231,9 @@ endforeach()
 
 set_target_properties(5-Benchmark PROPERTIES OUTPUT_NAME "o2-monitoring-benchmark")
 set_target_properties(8-DbFiller PROPERTIES OUTPUT_NAME "o2-monitoring-dbfiller")
+if(RdKafka_FOUND)
+  set_target_properties(11-KafkaToWebsocket PROPERTIES OUTPUT_NAME "o2-monitoring-kafka-to-grafana")
+endif()
 
 ####################################
 # Tests
@@ -278,6 +281,9 @@ install(TARGETS Monitoring 5-Benchmark 8-DbFiller
   ARCHIVE DESTINATION ${CMAKE_INSTALL_LIBDIR}
   RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR}
 )
+if(RdKafka_FOUND)
+  install(TARGETS 11-KafkaToWebsocket)
+endif()
 
 # Create version file
 include(CMakePackageConfigHelpers)

examples/11-KafkaToWebsocket.cxx

@@ -0,0 +1,45 @@
+///
+/// \file 11-KafkaToHttp.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "../src/Transports/KafkaConsumer.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>
+
+using namespace o2::monitoring;
+
+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")
+    ("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 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 (;;) {
+      outTransport->read();
+      std::this_thread::sleep_for(std::chrono::milliseconds(500));
+    }
+  });
+  for (;;) {
+    auto metrics = kafkaConsumer->receive();
+    if (!metrics.empty()) {
+      outTransport->send(boost::algorithm::join(metrics, "\n"));
+    }
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+  }
+}

src/Transports/WebSocket.cxx

@@ -40,6 +40,11 @@ WebSocket::WebSocket(const std::string& hostname, int port, const std::string& t
   mWebSocket.handshake(hostname, "/api/live/push/" + stream);
 }
 
+void WebSocket::read() {
+  beast::flat_buffer buffer;
+  mWebSocket.read(buffer);
+}
+
 WebSocket::~WebSocket()
 {
   mWebSocket.close(beast::websocket::close_code::normal);

src/Transports/WebSocket.h

@@ -54,6 +54,8 @@ class WebSocket : public TransportInterface
   /// \param message   r-value string formated
   void send(std::string&& message) override;
 
+  /// Read control frames from socket
+  void read();
  private:
   /// IO context
   boost::asio::io_context mIoContext;