Zabbix backend init (#10)

Author: awegrzyn

CMakeLists.txt

@@ -71,12 +71,14 @@ set(SRCS
         src/Collector.cxx
         src/Metric.cxx
         src/Backends/InfoLoggerBackend.cxx
+	src/Backends/Zabbix.cxx
         src/Backends/Flume.cxx
         src/DerivedMetrics.cxx
         src/ProcessMonitor.cxx
         src/ProcessDetails.cxx
         src/MonitoringFactory.cxx
         src/Transports/UDP.cxx
+	src/Transports/TCP.cxx
         src/Transports/HTTP.cxx
         src/Exceptions/MonitoringException.cxx
         src/Exceptions/MonitoringInternalException.cxx
@@ -143,6 +145,7 @@ set(TEST_SRCS
         test/testDerived.cxx
         test/testFlume.cxx
         test/testMetric.cxx
+	test/testZabbix.cxx
         test/testProcessDetails.cxx
         test/testProcessMonitor.cxx
         )

README.md

@@ -150,6 +150,7 @@ Metrics are pushed to one or multiple backends. The module currently supports th
 | ApMonBackend     | MonALISA Serivce               | UDP                          | ApMon                      | Default tags concatenated with entity; Metric tags concatenated with name |
 | InfoLoggerBackned| O2 Logging module              | -                            | (as log message)           | Added to the end of message |
 | Flume            | Collects, aggragate monitoring data | UDP (JSON)              | boost asio                 | In Flume Event header |
+| Zabbix           | Via Zabbix trapper item        | TCP (Zabbix protocol)        | boost asio                 | Not supported |
 
 Instruction how to install and configure server-sides backends are available in [Server-side backend installation and configuration](#server-side-backend-installation-and-configuration) section.
 
@@ -169,6 +170,10 @@ Instruction how to install and configure server-sides backends are available in
   + enable - enable Flume HTTP backend
   + port 
   + hostname
++ Zabbix
+  + enable - enable Zabbix backend
+  + port
+  + hostname
 + ProcessMonitor
   + enable - enable process monitor
   + interval - updates interval[]

examples/config-default.ini

@@ -17,6 +17,11 @@ enable=0
 hostname=<hostname>
 port=4445
 
+[Zabbix]
+enable=0
+hostname=<hostname>
+port=10051
+
 [ProcessMonitor]
 enable=1
 interval=5

src/Backends/Zabbix.cxx

@@ -0,0 +1,90 @@
+///
+/// \file Zabbix.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "Zabbix.h"
+#include <boost/lexical_cast.hpp>
+#include <boost/property_tree/json_parser.hpp>
+#include <string>
+#include "../Transports/TCP.h"
+#include "../Exceptions/MonitoringInternalException.h"
+
+namespace AliceO2
+{
+/// ALICE O2 Monitoring system
+namespace Monitoring
+{
+/// Monitoring backends
+namespace Backends
+{
+
+Zabbix::Zabbix(const std::string &hostname, int port)
+{ 
+  transport = std::make_unique<Transports::TCP>(hostname, port);
+  MonLogger::Get() << "Zabbix/TCP backend initialized"
+                       << " ("<< hostname << ":" << port << ")" << MonLogger::End();
+}
+
+inline std::string Zabbix::convertTimestamp(const std::chrono::time_point<std::chrono::system_clock>& timestamp)
+{
+  std::string converted = std::to_string(
+    std::chrono::duration_cast <std::chrono::nanoseconds>(
+    timestamp.time_since_epoch()
+  ).count());
+  converted.erase(converted.begin()+10, converted.end());
+  return converted;
+}
+
+std::string Zabbix::metricToZabbix(const Metric& metric)
+{
+  // create JSON payload
+  boost::property_tree::ptree request, dataArray, data;
+  data.put<std::string>("host", hostname);
+  data.put<std::string>("key", metric.getName());
+  data.put<std::string>("value", boost::lexical_cast<std::string>(metric.getValue()));
+  data.put<std::string>("clock", convertTimestamp(metric.getTimestamp()));
+
+  dataArray.push_back(std::make_pair("", data));
+  request.put<std::string>("request", "sender data");
+  request.add_child("data", dataArray);
+
+  std::stringstream ss; 
+  write_json(ss, request);
+  
+  std::string noWhiteSpaces = ss.str();
+  noWhiteSpaces.erase(std::remove_if( noWhiteSpaces.begin(), noWhiteSpaces.end(), 
+      [](char c){ return (c =='\r' || c =='\t' || c == ' ' || c == '\n');}), noWhiteSpaces.end() );
+  noWhiteSpaces.insert(18, " ");
+
+  // prepare Zabbix header
+  uint32_t length = noWhiteSpaces.length();
+  std::vector<unsigned char> header = {
+    'Z', 'B', 'X', 'D', '\x01',
+    static_cast<unsigned char>(length & 0xFF),
+    static_cast<unsigned char>((length >> 8) & 0x00FF),
+    static_cast<unsigned char>((length >> 16) & 0x0000FF),
+    static_cast<unsigned char>((length >> 24) & 0x000000FF),
+    '\x00','\x00','\x00','\x00'
+  };
+
+  return std::string(header.begin(), header.end()) + noWhiteSpaces;
+}
+
+void Zabbix::send(const Metric& metric) {
+  try {
+    transport->send(metricToZabbix(metric));
+  } catch (MonitoringInternalException&) {
+  }
+}
+
+void Zabbix::addGlobalTag(std::string name, std::string value)
+{
+  if (name == "hostname") {
+    hostname = value;
+  }
+}
+
+} // namespace Backends
+} // namespace Monitoring
+} // namespace AliceO2

src/Backends/Zabbix.h

@@ -0,0 +1,67 @@
+///
+/// \file Zabbix.h
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#ifndef ALICEO2_MONITORING_BACKENDS_ZABBIX_H
+#define ALICEO2_MONITORING_BACKENDS_ZABBIX_H
+
+#include "Monitoring/Backend.h"
+#include "../Transports/TCP.h"
+#include "../MonLogger.h"
+#include <chrono>
+#include <string>
+
+namespace AliceO2
+{
+/// ALICE O2 Monitoring system
+namespace Monitoring
+{
+/// Monitoring backends
+namespace Backends
+{
+
+/// Backend that sends metrics to InfluxDB time-series databse
+///
+/// Metrics are converted into Influx Line protocol and then sent via one of available transports
+class Zabbix final : public Backend
+{
+  public:
+    /// Constructor, uses UDP transport
+    /// \param hostname  InfluxDB UDP endpoint hostname
+    /// \param port      InfluxDB UDP endpoint port number
+    Zabbix(const std::string &hostname, int port);
+
+    /// Default destructor
+    ~Zabbix() = default;
+
+    /// Sends metric to InfluxDB instance via one transport
+    /// \param metric    reference to metric object
+    void send(const Metric& metric) override;
+
+    /// Adds tag
+    /// \param name      tag name
+    /// \param value     tag value
+    void addGlobalTag(std::string name, std::string value) override;
+  
+  private:
+    /// TCP transport
+    std::unique_ptr<Transports::TCP> transport;
+
+    /// Hostname as required by Zabbix protocol
+    std::string hostname;
+
+    /// Prepares Zabbix protocol message
+    std::string metricToZabbix(const Metric& metric);
+
+    /// Converts timestamp into unix format
+    /// \param timestamp chrono system_clock timestamp
+    /// \return unix timestamp
+    std::string convertTimestamp(const std::chrono::time_point<std::chrono::system_clock>& timestamp);
+};
+
+} // namespace Backends
+} // namespace Monitoring
+} // namespace AliceO2
+
+#endif // ALICEO2_MONITORING_BACKENDS_ZABBIX_H

src/Collector.cxx

@@ -18,6 +18,7 @@
 #include "Exceptions/MonitoringInternalException.h"
 #include "Backends/InfoLoggerBackend.h"
 #include "Backends/Flume.h"
+#include "Backends/Zabbix.h"
 
 #ifdef _WITH_APPMON
 #include "Backends/ApMonBackend.h"
@@ -89,6 +90,16 @@ Collector::Collector(const std::string& configPath)
     MonLogger::Get() << "Flume backend disabled" << MonLogger::End();
   }
 
+  if (configFile->get<int>("Zabbix/enable").value_or(0) == 1) {
+    mBackends.emplace_back(std::make_unique<Backends::Zabbix>(
+      configFile->get<std::string>("Zabbix/hostname").value(),
+      configFile->get<int>("Zabbix/port").value()
+    )); 
+  }
+  else {
+    MonLogger::Get() << "Zabbix backend disabled" << MonLogger::End();
+  }
+
 #ifdef _OS_LINUX  
   mProcessMonitor = std::make_unique<ProcessMonitor>();
   if (configFile->get<int>("ProcessMonitor/enable").value_or(0) == 1) {

src/Transports/TCP.cxx

@@ -0,0 +1,61 @@
+///
+/// \file TCP.cxx
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#include "TCP.h"
+#include <string>
+#include <iostream>
+#include "Exceptions/MonitoringInternalException.h"
+
+namespace AliceO2
+{
+/// ALICE O2 Monitoring system
+namespace Monitoring
+{
+/// Monitoring transports
+namespace Transports
+{
+
+TCP::TCP(const std::string &hostname, int port) :
+  mSocket(mIoService)
+{
+  boost::asio::ip::tcp::resolver resolver(mIoService);
+  boost::asio::ip::tcp::resolver::query query(hostname, std::to_string(port));
+  boost::asio::ip::tcp::resolver::iterator resolverIterator = resolver.resolve(query);
+
+  boost::asio::ip::tcp::resolver::iterator end;
+  boost::system::error_code error = boost::asio::error::host_not_found;
+  while (error && resolverIterator != end) {
+    mSocket.close();
+    mSocket.connect(*resolverIterator++, error);
+  }
+  if (error) {
+    throw MonitoringInternalException("TCP connection", error.message());
+  } 
+}
+
+void TCP::send(std::string&& message)
+{
+  try {
+    mSocket.send(boost::asio::buffer(message));
+  } catch(const boost::system::system_error& e) {
+    throw MonitoringInternalException("TCP send", e.what());
+  }
+}
+
+void TCP::read() {
+  for (;;) {
+    boost::system::error_code error;
+    boost::array<char, 128> buf;
+    size_t len = mSocket.read_some(boost::asio::buffer(buf), error);
+    if (error == boost::asio::error::eof) {
+      break;
+    }
+    std::cout.write(buf.data(), len);
+  }
+}
+
+} // namespace Transports
+} // namespace Monitoring
+} // namespace AliceO2

src/Transports/TCP.h

@@ -0,0 +1,61 @@
+///
+/// \file TCP.h
+/// \author Adam Wegrzynek <[email protected]>
+///
+
+#ifndef ALICEO2_MONITORING_TRANSPORTS_TCP_H
+#define ALICEO2_MONITORING_TRANSPORTS_TCP_H
+
+#include "TransportInterface.h"
+
+#include <boost/array.hpp>
+#include <boost/asio.hpp>
+#include <boost/algorithm/string.hpp>
+#include <chrono>
+#include <string>
+
+namespace AliceO2
+{
+/// ALICE O2 Monitoring system
+namespace Monitoring
+{
+/// Monitoring transports
+namespace Transports
+{
+
+/// Transport that sends string formatted metrics via UDP
+class TCP : public TransportInterface
+{
+  public:
+    /// Constructor
+    /// \param hostname      InfluxDB instance hostname
+    //  \param port          InfluxDB instance port number
+    TCP(const std::string &hostname, int port);
+
+    /// Default destructor
+    ~TCP() = default;
+ 
+    /// Sends metric via UDP
+    /// \param lineMessage   r-value string formated
+    void send(std::string&& message) override;
+
+    /// Dummy read method - Forwards read out buffer to cout
+    void read();
+
+  private:
+    /// Boost Asio I/O functionality
+    boost::asio::io_service mIoService;
+
+    /// TCP socket
+    boost::asio::ip::tcp::socket mSocket;
+
+    /// TCP endpoint interator
+    boost::asio::ip::tcp::resolver::iterator mEndpoint;
+
+};
+
+} // namespace Transports
+} // namespace Monitoring
+} // namespace AliceO2
+
+#endif // ALICEO2_MONITORING_TRANSPORTS_TCP_H

test/testZabbix.cxx

@@ -0,0 +1,23 @@
+#include "../src/Backends/Zabbix.h"
+#include "../src/Exceptions/MonitoringInternalException.h"
+
+#define BOOST_TEST_MODULE testZabbix
+#include <boost/test/included/unit_test.hpp>
+
+
+namespace AliceO2 {
+namespace Monitoring {
+namespace Test {
+
+BOOST_AUTO_TEST_CASE(crateZabbixInstance)
+{
+  try {
+    AliceO2::Monitoring::Backends::Zabbix zabbixBackend("localhost", 1000);
+  } catch (MonitoringInternalException &e) {
+    BOOST_CHECK_EQUAL(std::string(e.what()), "Connection refused");
+  }
+}
+
+} // namespace Test
+} // namespace Monitoring
+} // namespace AliceO2