Add InfluxDB over Unix datagram socket backend (#115)

* Add unix socket transport

* fix verbosities

* Use datagram unix sockets

Author: awegrzyn

CMakeLists.txt

@@ -78,6 +78,7 @@ set(SRCS
   src/Transports/UDP.cxx
   src/Transports/TCP.cxx
   src/Transports/HTTP.cxx
+  src/Transports/Unix.cxx
   src/Exceptions/MonitoringException.cxx
 )
 

README.md

@@ -71,6 +71,7 @@ See table below to find out how to create `URI` for each backend:
 | ------------ |:---------:|:----------------------:|:----------:| -----------------:|
 | InfluxDB     | HTTP      | `influxdb-http`        | `?db=<db>` | `info`            |
 | InfluxDB     | UDP       | `influxdb-udp`         | -          | `info`            |
+| InfluxDB     | Unix datagram | `influxdb-unix`    | -          | `info`            |
 | ApMon        | UDP       | `apmon`                | -          | `info`            |
 | StdOut       | -         | `stdout`, `infologger` | -          | `debug`           |
 | Flume        | UDP       | `flume`                | -          | `info`            |

src/Backends/InfluxDB.cxx

@@ -8,6 +8,7 @@
 #include <string>
 #include "../Transports/UDP.h"
 #include "../Transports/HTTP.h"
+#include "../Transports/Unix.h"
 #include "../Exceptions/MonitoringException.h"
 
 namespace o2
@@ -35,6 +36,12 @@ InfluxDB::InfluxDB(const std::string& host, unsigned int port, const std::string
                    << ":" <<  std::to_string(port) << "/write?" << search << ")" << MonLogger::End();
 }
 
+InfluxDB::InfluxDB(const std::string& socketPath)
+{
+  transport = std::make_unique<transports::Unix>(socketPath);
+  MonLogger::Get() << "InfluxDB/Unix backend initialized (" << socketPath << ")" << MonLogger::End();
+}
+
 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

@@ -38,6 +38,8 @@ class InfluxDB final : public Backend
     /// \param path	 Query search providing database name
     InfluxDB(const std::string& host, unsigned int port, const std::string& search);
 
+    InfluxDB(const std::string& socketPath);
+
     /// Default destructor
     ~InfluxDB() = default;
 

src/MonitoringFactory.cxx

@@ -27,6 +27,12 @@ namespace o2
 namespace monitoring 
 {
 
+static const std::map<std::string, Verbosity> verbosities = {
+  {"/prod", Verbosity::Prod},
+  {"/info", Verbosity::Info},
+  {"/debug", Verbosity::Debug}
+};
+
 std::unique_ptr<Backend> getStdOut(http::url) {
   return std::make_unique<backends::StdOut>();
 }
@@ -42,6 +48,16 @@ std::unique_ptr<Backend> getInfluxDb(http::url uri) {
   if (uri.protocol == "http") {
     return std::make_unique<backends::InfluxDB>(uri.host, uri.port, uri.search);
   }
+  if (uri.protocol == "unix") {
+    std::string path = uri.path;;
+    auto found = std::find_if(begin(verbosities), end(verbosities),
+                       [&](const auto& s)
+                       {return path.find(s.first) != std::string::npos; });
+    if (found != end(verbosities)) {
+      path.erase(path.rfind('/'));
+    }
+    return std::make_unique<backends::InfluxDB>(path);
+  }
   throw std::runtime_error("InfluxDB transport protocol not supported");
 }
 
@@ -64,12 +80,6 @@ std::unique_ptr<Backend> getFlume(http::url uri) {
 }
 
 void MonitoringFactory::SetVerbosity(std::string selected, std::unique_ptr<Backend>& backend) {
-  static const std::map<std::string, Verbosity> verbosities = {
-    {"/prod", Verbosity::Prod},
-    {"/info", Verbosity::Info},
-    {"/debug", Verbosity::Debug}
-  };
-
   auto found = verbosities.find(selected);
   if (found == verbosities.end()) {
     throw std::runtime_error("Unrecognised verbosity");
@@ -86,6 +96,7 @@ std::unique_ptr<Backend> MonitoringFactory::GetBackend(std::string& url) {
     {"stdout", getStdOut},
     {"influxdb-udp", getInfluxDb},
     {"influxdb-http", getInfluxDb},
+    {"influxdb-unix", getInfluxDb},
     {"apmon", getApMon},
     {"flume", getFlume},
     {"no-op", getNoop}
@@ -103,7 +114,7 @@ std::unique_ptr<Backend> MonitoringFactory::GetBackend(std::string& url) {
 
   auto backend = iterator->second(parsedUrl);
   if (!parsedUrl.path.empty() && parsedUrl.path != "/") {
-    SetVerbosity(parsedUrl.path, backend);
+    SetVerbosity(parsedUrl.path.substr(parsedUrl.path.rfind("/")), backend);
   }
   return backend;
 }

src/Transports/Unix.cxx

@@ -0,0 +1,31 @@
+///
+/// \file Unix.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "Unix.h"
+#include <string>
+
+namespace o2
+{
+/// ALICE O2 Monitoring system
+namespace monitoring
+{
+/// Monitoring transports
+namespace transports
+{
+#if defined(BOOST_ASIO_HAS_LOCAL_SOCKETS)
+Unix::Unix(const std::string &socketPath) :
+  mSocket(mIoService), mEndpoint(socketPath)
+{
+  mSocket.open();
+}
+
+void Unix::send(std::string&& message)
+{
+  mSocket.send_to(boost::asio::buffer(message, message.size()), mEndpoint);
+}
+#endif // defined(BOOST_ASIO_HAS_LOCAL_SOCKETS)
+} // namespace transports
+} // namespace monitoring
+} // namespace o2

src/Transports/Unix.h

@@ -0,0 +1,56 @@
+///
+/// \file Unix.h
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#ifndef ALICEO2_MONITORING_TRANSPORTS_UNIX_H
+#define ALICEO2_MONITORING_TRANSPORTS_UNIX_H
+
+#include "TransportInterface.h"
+
+#include <boost/array.hpp>
+#include <boost/asio.hpp>
+#include <boost/algorithm/string.hpp>
+#include <chrono>
+#include <string>
+
+namespace o2
+{
+/// ALICE O2 Monitoring system
+namespace monitoring
+{
+/// Monitoring transports
+namespace transports
+{
+
+/// \brief Transport that sends string formatted metrics via Unix datagram socket
+class Unix : public TransportInterface
+{
+  public:
+    /// \param hostname
+    /// \param port
+    Unix(const std::string &socketPath);
+
+    /// Default destructor
+    ~Unix() = default;
+ 
+    /// \param message   r-value string formated
+    void send(std::string&& message) override;   
+
+  private:
+    /// Boost Asio I/O functionality
+    boost::asio::io_service mIoService;
+#if defined(BOOST_ASIO_HAS_LOCAL_SOCKETS)
+    /// Unix socket
+    boost::asio::local::datagram_protocol::socket mSocket;
+
+    /// Unix endpoint
+    boost::asio::local::datagram_protocol::endpoint mEndpoint;
+#endif // defined(BOOST_ASIO_HAS_LOCAL_SOCKETS)
+};
+
+} // namespace transports
+} // namespace monitoring
+} // namespace o2
+
+#endif // ALICEO2_MONITORING_TRANSPORTS_UNIX_H