WIP - output telemetry

Author: SiskaPavel

include/ipfixprobe/output.hpp

@@ -34,24 +34,91 @@
 #include "process.hpp"
 #include "flowifc.hpp"
 
+#include <telemetry.hpp>
+#include <atomic>
+#include <cstdint>
+
 namespace ipxp {
 
 #define DEFAULT_EXPORTER_ID 1
 
+struct OutputPluginStats {
+   uint64_t flows_seen = 0;
+   uint64_t flows_dropped = 0;
+   uint64_t flows_exported = 0;
+   uint64_t bytes_exported = 0;
+
+   void clear()
+   {
+      flows_seen = 0;
+      flows_dropped = 0;
+      flows_exported = 0;
+      bytes_exported = 0;
+   }
+};
+
+struct OutputPluginAtomicStats {
+   std::atomic<uint64_t> flows_seen = 0;
+   std::atomic<uint64_t> flows_dropped = 0;
+   std::atomic<uint64_t> flows_exported = 0;
+   std::atomic<uint64_t> bytes_exported = 0;
+   std::atomic<uint64_t> fps_exported = 0;
+   std::atomic<uint64_t> mbps_exported = 0;
+
+
+   void updatePktBurstStats(const OutputPluginStats& basicStats, uint64_t timestamp)
+   {
+      flows_seen.fetch_add(basicStats.flows_seen, std::memory_order_relaxed);
+      flows_dropped.fetch_add(basicStats.flows_dropped, std::memory_order_relaxed);
+      flows_exported.fetch_add(basicStats.flows_exported, std::memory_order_relaxed);
+      bytes_exported.fetch_add(basicStats.bytes_exported, std::memory_order_relaxed);
+
+      if (timestamp == 0) {
+         fps_exported.store(0);
+         mbps_exported.store(0);
+         return;
+      }
+
+      fps_exported.store(basicStats.flows_exported * 1000000000 / timestamp);
+      mbps_exported.store(basicStats.bytes_exported * 8 * 1000000000 / timestamp / 1024 / 1024);
+   }
+};
+
+static telemetry::Dict createOutputPluginStatsDict(const OutputPluginAtomicStats& burstStats)
+{
+	telemetry::Dict dict;
+   dict["flows_seen"] = burstStats.flows_seen;
+   dict["flows_dropped"] = burstStats.flows_dropped;
+   dict["flows_exported"] = burstStats.flows_exported;
+   dict["bytes_exported"] = burstStats.bytes_exported;
+   dict["fps_exported"] = burstStats.fps_exported;
+   dict["mbps_exported"] = burstStats.mbps_exported;
+	return dict;
+}
+
 /**
  * \brief Base class for flow exporters.
  */
 class OutputPlugin : public Plugin
 {
 public:
    typedef std::vector<std::pair<std::string, ProcessPlugin *>> Plugins;
-   uint64_t m_flows_seen; /**< Number of flows received to export. */
-   uint64_t m_flows_dropped; /**< Number of flows that could not be exported. */
 
-   OutputPlugin() : m_flows_seen(0), m_flows_dropped(0) {}
+   OutputPlugin() = default;
+
    virtual ~OutputPlugin() {}
 
-   virtual void init(const char *params, Plugins &plugins) = 0;
+   void init(const char *params, Plugins &plugins, const std::shared_ptr<telemetry::Directory>& dir)
+   {
+      const auto statsFile = dir->addFile(
+		"basic_stats",
+		{.read = [this]() { return createOutputPluginStatsDict(m_atomicStats); },
+		 .clear = nullptr});
+
+      m_holder.add(statsFile);
+
+      init_plugin(params, plugins, dir);
+   }
 
    enum class Result {
       EXPORTED = 0,
@@ -70,6 +137,34 @@ class OutputPlugin : public Plugin
    virtual void flush()
    {
    }
+
+   void update_stats(uint64_t timestamp) override final
+   {
+      m_atomicStats.updatePktBurstStats(m_stats, timestamp);
+      m_stats.clear();
+
+      update_plugin_stats(timestamp);
+   }
+
+   const OutputPluginAtomicStats& get_stats() const
+   {
+      return m_atomicStats;
+   }
+
+
+protected:
+   virtual void init_plugin(const char *params, Plugins &plugins, const std::shared_ptr<telemetry::Directory>& dir) = 0;
+
+
+   virtual void update_plugin_stats(uint64_t timestamp)
+   {
+      (void) timestamp;
+   }
+
+   OutputPluginStats m_stats = {};
+   telemetry::Holder m_holder;
+private:
+   OutputPluginAtomicStats m_atomicStats = {};
 };
 
 }

include/ipfixprobe/plugin.hpp

@@ -60,6 +60,11 @@ class Plugin
    virtual void init(const char *params) {}
    virtual void close() {}
 
+   virtual void update_stats(uint64_t timestamp)
+   {
+      (void) timestamp;
+   }
+
    virtual OptionsParser *get_parser() const = 0;
    virtual std::string get_name() const = 0;
 };

ipfixprobe.cpp

@@ -287,7 +287,9 @@ bool process_plugin_args(ipxp_conf_t &conf, IpfixprobeOptParser &parser)
          throw IPXPError("invalid output plugin " + output_name);
       }
 
-      output_plugin->init(output_params.c_str(), *process_plugins);
+      auto output_plugin_dir = output_dir->addDir(output_name);
+
+      output_plugin->init(output_params.c_str(), *process_plugins, output_plugin_dir);
       conf.active.output.push_back(output_plugin);
       conf.active.all.push_back(output_plugin);
    } catch (PluginError &e) {

output/ipfix.cpp

@@ -232,7 +232,7 @@ void IPFIXExporter::init(const char *params)
    signal(SIGPIPE, SIG_IGN);
 }
 
-void IPFIXExporter::init(const char *params, Plugins &plugins)
+void IPFIXExporter::init_plugin(const char *params, Plugins &plugins, const std::shared_ptr<telemetry::Directory>& dir)
 {
    init(params);
 
@@ -258,6 +258,14 @@ void IPFIXExporter::init(const char *params, Plugins &plugins)
       }
       delete ext;
    }
+
+   auto plugin_config_file = dir->addFile(
+      "config",
+      {.read = [this]() { return get_ipfix_config(); },
+       .clear = nullptr});
+
+   m_holder.add(plugin_config_file);
+
 }
 
 void IPFIXExporter::close()
@@ -398,13 +406,13 @@ bool IPFIXExporter::fill_template(const Flow &flow, template_t *tmplt)
 
 int IPFIXExporter::export_flow(const Flow &flow)
 {
-   m_flows_seen++;
+   m_stats.flows_seen++;
    template_t *tmplt = get_template(flow);
    if (!fill_template(flow, tmplt)) {
       flush();
 
       if (!fill_template(flow, tmplt)) {
-         m_flows_dropped++;
+         m_stats.flows_dropped++;
          return 1;
       }
    }
@@ -833,7 +841,7 @@ void IPFIXExporter::send_data()
          ret = send_packet(&pkt);
       }
       if (ret != 0) {
-         m_flows_dropped += pkt.flows;
+         m_stats.flows_dropped += pkt.flows;
       }
    }
 }
@@ -930,10 +938,12 @@ int IPFIXExporter::send_packet(ipfix_packet_t *packet)
 
       /* No error from sendto(), add sent data count to total */
       sent += ret;
+      m_stats.bytes_exported += ret;
    }
 
    /* Update sequence number for next packet */
    sequenceNum += packet->flows;
+   m_stats.flows_exported += packet->flows;
 
    /* Increase packet counter */
    exportedPackets++;
@@ -1154,6 +1164,26 @@ int IPFIXExporter::reconnect()
    return 0;
 }
 
+telemetry::Content IPFIXExporter::get_ipfix_config()
+{
+   telemetry::Dict dict;
+
+   dict["protocol"] = (protocol == IPPROTO_UDP ? "UDP" : "TCP");
+   dict["host"] = host;
+   dict["port"] = static_cast<uint64_t>(port);
+   dict["odid"] = static_cast<uint64_t>(odid);
+   dict["dir_bit_field"] = static_cast<uint64_t>(dir_bit_field);
+   dict["mtu"] = static_cast<uint64_t>(mtu);
+   dict["LZ4_compression"] = (packetDataBuffer.isCompressing() ? "enabled" : "disabled");
+
+   return dict;
+}
+
+void IPFIXExporter::update_plugin_stats(uint64_t timestamp)
+{
+
+}
+
 // compress buffer implementation
 
 CompressBuffer::CompressBuffer() :

output/ipfix.hpp

@@ -40,9 +40,12 @@
 
 #include <vector>
 #include <map>
+#include <atomic>
+#include <cstdint>
 #include <cstdint>
 
 #include <lz4.h>
+#include <telemetry.hpp>
 
 #include <ipfixprobe/output.hpp>
 #include <ipfixprobe/process.hpp>
@@ -414,6 +417,8 @@ class CompressBuffer {
     */
    void close();
 
+   bool isCompressing() const noexcept { return shouldCompress; }
+
    // the maximum aditional size required to send metadata that are needed to
    // to decompress the data, the +4 is there for four 0 bytes that identify
    // ipfix_start_compress_header_t
@@ -452,13 +457,23 @@ class CompressBuffer {
    LZ4_stream_t *lz4Stream;
 };
 
+struct IpfixStats {
+   std::atomic<uint64_t> sent_bytes;
+   std::atomic<uint64_t> sent_flows;
+   std::atomic<uint64_t> dropped_flows;
+   std::atomic<uint64_t> sent_mbps;
+   std::atomic<uint64_t> sent_fps;
+   std::atomic<uint64_t> dropped_fps;
+   std::atomic<uint64_t> timestamp;
+};
+
 class IPFIXExporter : public OutputPlugin
 {
 public:
    IPFIXExporter();
    ~IPFIXExporter();
    void init(const char *params);
-   void init(const char *params, Plugins &plugins);
+   void init_plugin(const char *params, Plugins &plugins, const std::shared_ptr<telemetry::Directory>& dir) override;
    void close();
    OptionsParser *get_parser() const { return new IpfixOptParser(); }
    std::string get_name() const { return "ipfix"; }
@@ -526,6 +541,9 @@ class IPFIXExporter : public OutputPlugin
    bool fill_template(const Flow &flow, template_t *tmplt);
    void flush();
    void shutdown();
+
+   telemetry::Content get_ipfix_config();
+   void update_plugin_stats(uint64_t timestamp) override;
 };
 
 }

output/text.cpp

@@ -77,7 +77,7 @@ void TextExporter::init(const char *params)
    *m_out << "conversation packets bytes tcp-flags time extensions" << std::endl;
 }
 
-void TextExporter::init(const char *params, Plugins &plugins)
+void TextExporter::init_plugin(const char *params, Plugins &plugins, const std::shared_ptr<telemetry::Directory>& dir)
 {
    init(params);
 }
@@ -94,7 +94,7 @@ int TextExporter::export_flow(const Flow &flow)
 {
    RecordExt *ext = flow.m_exts;
 
-   m_flows_seen++;
+   //m_flows_seen++;
    print_basic_flow(flow);
    while (ext != nullptr) {
       *m_out << " " << ext->get_text();

output/text.hpp

@@ -64,7 +64,7 @@ class TextExporter : public OutputPlugin
    TextExporter();
    ~TextExporter();
    void init(const char *params);
-   void init(const char *params, Plugins &plugins);
+   void init_plugin(const char *params, Plugins &plugins, const std::shared_ptr<telemetry::Directory>& dir) override;
    void close();
    OptionsParser *get_parser() const { return new TextOptParser(); }
    std::string get_name() const { return "text"; }

output/unirec.cpp

@@ -346,7 +346,7 @@ int UnirecExporter::export_flow(const Flow &flow)
       trap_send(m_basic_idx, record_ptr, ur_rec_fixlen_size(tmplt_ptr) + ur_rec_varlen_size(tmplt_ptr, record_ptr));
    }
 
-   m_flows_seen++;
+   //m_flows_seen++;
    uint64_t tmplt_dbits = 0; // templates dirty bits
    memset(m_ext_id_flgs, 0, sizeof(int) * m_ext_cnt); // in case one flow has multiple extension of same type
    int ext_processed_cnd = 0;