[OMON-517] Add WebSocket backend to the library (#273)

Author: awegrzyn

CMakeLists.txt

@@ -31,7 +31,7 @@ endif()
 
 # Define project
 project(Monitoring
-  VERSION 3.9.0
+  VERSION 3.10.0
   DESCRIPTION "O2 Monitoring library"
   LANGUAGES CXX
 )
@@ -73,7 +73,7 @@ endif()
 
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_LIST_DIR}/cmake")
 
-find_package(Boost REQUIRED COMPONENTS unit_test_framework program_options system)
+find_package(Boost 1.70 REQUIRED COMPONENTS unit_test_framework program_options system)
 find_package(ApMon MODULE)
 find_package(CURL MODULE)
 find_package(RdKafka CONFIG)
@@ -128,6 +128,7 @@ add_library(Monitoring SHARED
   src/Transports/TCP.cxx
   src/Transports/Unix.cxx
   src/Transports/StdOut.cxx
+  src/Transports/WebSocket.cxx
   src/Exceptions/MonitoringException.cxx
   $<$<BOOL:${ApMon_FOUND}>:src/Backends/ApMonBackend.cxx>
   $<$<BOOL:${RdKafka_FOUND}>:src/Transports/Kafka.cxx>

README.md

@@ -41,13 +41,14 @@ 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 |
+| Format       | Transport   | URI backend[-protocol] | URI query   | Default verbosity |
 | ------------ |:-----------:|:----------------------:|:-----------:| -----------------:|
-| No-op        | -           | `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`            |
+| InfluxDB     | WebSocket   | `influxdb-ws`          | `token=TOKEN` | `info`          |
 | InfluxDB 2.x | HTTP        | `influxdbv2`           | `org=ORG&bucket=BUCKET&token=TOKEN` | `info` |
 | ApMon        | UDP         | `apmon`                | -           | `info`            |
 | StdOut       | -           | `stdout`, `infologger` | [Prefix]    | `debug`           |

src/MonitoringFactory.cxx

@@ -28,6 +28,7 @@
 #include "Transports/UDP.h"
 #include "Transports/Unix.h"
 #include "Transports/StdOut.h"
+#include "Transports/WebSocket.h"
 
 #ifdef O2_MONITORING_WITH_APPMON
 #include "Backends/ApMonBackend.h"
@@ -116,6 +117,17 @@ std::unique_ptr<Backend> getInfluxDb(http::url uri)
     auto transport = std::make_unique<transports::StdOut>();
     return std::make_unique<backends::InfluxDB>(std::move(transport));
   }
+  if (uri.protocol == "ws") {
+    std::string tokenLabel = "token=";
+    auto tokenSearch = uri.search.find(tokenLabel);
+    uri.path.erase(std::remove(uri.path.begin(), uri.path.end(), '/'), uri.path.end());
+    if (tokenSearch == std::string::npos) {
+      throw MonitoringException("Factory", "Grafana token is required for WebSocket backend");
+    }
+    std::string token = uri.search.substr(tokenSearch + tokenLabel.length());
+    auto transport = std::make_unique<transports::WebSocket>(uri.host, uri.port, token, uri.path);
+    return std::make_unique<backends::InfluxDB>(std::move(transport));
+  }
   if (uri.protocol == "kafka") {
 #ifdef O2_MONITORING_WITH_KAFKA
     auto transport = std::make_unique<transports::Kafka>(uri.host, uri.port, uri.search);
@@ -165,6 +177,7 @@ std::unique_ptr<Backend> MonitoringFactory::GetBackend(std::string& url)
     {"influxdb-unix", getInfluxDb},
     {"influxdb-stdout", getInfluxDb},
     {"influxdb-kafka", getInfluxDb},
+    {"influxdb-ws", getInfluxDb},
     {"influxdbv2", getInfluxDbv2},
     {"apmon", getApMon},
     {"no-op", getNoop}

src/Transports/WebSocket.cxx

@@ -0,0 +1,55 @@
+// 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 WebSocket.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "WebSocket.h"
+#include "Exceptions/MonitoringException.h"
+#include "../MonLogger.h"
+
+namespace o2
+{
+/// ALICE O2 Monitoring system
+namespace monitoring
+{
+/// Monitoring transports
+namespace transports
+{
+
+WebSocket::WebSocket(const std::string& hostname, int port, const std::string& token, const std::string& stream) : mWebSocket(mIoContext)
+{
+  boost::asio::ip::tcp::resolver resolver{mIoContext};
+  auto const results = resolver.resolve(hostname, std::to_string(port));
+  boost::asio::connect(mWebSocket.next_layer(), results.begin(), results.end());
+  mWebSocket.set_option(beast::websocket::stream_base::decorator(
+    [&token](beast::websocket::request_type& req) {
+      req.set(beast::http::field::authorization, "Bearer " + token);
+    }
+  ));
+  mWebSocket.handshake(hostname, "/api/live/push/" + stream);
+}
+
+WebSocket::~WebSocket()
+{
+  mWebSocket.close(beast::websocket::close_code::normal);
+}
+
+void WebSocket::send(std::string&& message)
+{
+  mWebSocket.write(boost::asio::buffer(message));
+}
+
+} // namespace transports
+} // namespace monitoring
+} // namespace o2

src/Transports/WebSocket.h

@@ -0,0 +1,69 @@
+// 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 WebSocket.h
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#ifndef ALICEO2_MONITORING_TRANSPORTS_WEBSOCKET_H
+#define ALICEO2_MONITORING_TRANSPORTS_WEBSOCKET_H
+
+#include "TransportInterface.h"
+
+#include <boost/beast/core.hpp>
+#include <boost/beast/websocket.hpp>
+#include <boost/asio/connect.hpp>
+#include <boost/asio/ip/tcp.hpp>
+#include <string>
+
+namespace beast = boost::beast;
+
+namespace o2
+{
+/// ALICE O2 Monitoring system
+namespace monitoring
+{
+/// Monitoring transports
+namespace transports
+{
+
+/// \brief Transport that sends string formatted metrics via WebSocket
+class WebSocket : public TransportInterface
+{
+ public:
+  /// Constructor
+  /// \param hostname      Grafana host
+  /// \param port          Grafana port
+  /// \param token         Grafana API token
+  /// \param stream        Name of WebSocket stream
+  WebSocket(const std::string& hostname, int port, const std::string& token, const std::string& stream = "alice_o2");
+
+  /// Gracefull disconnect
+  ~WebSocket();
+
+  /// Sends metric via UDP
+  /// \param message   r-value string formated
+  void send(std::string&& message) override;
+
+ private:
+  /// IO context
+  boost::asio::io_context mIoContext;
+
+  /// Websocket stream
+  beast::websocket::stream<boost::asio::ip::tcp::socket> mWebSocket;
+};
+
+} // namespace transports
+} // namespace monitoring
+} // namespace o2
+
+#endif // ALICEO2_MONITORING_TRANSPORTS_WEBSOCKET_H