WIP - extends stats

Author: SiskaPavel

output/ipfix.cpp

@@ -134,6 +134,17 @@ const char *basic_tmplt_v6[] = {
    nullptr
 };
 
+static telemetry::Content getExtendedAtomicStats(const IpfixExtendedAtomicStats& stats)
+{
+   telemetry::Dict dict;
+
+   dict["flows_dropped_eagain"] = stats.flows_dropped_eagain;
+   dict["fps_dropped_eagain"] = stats.fps_dropped_eagain;
+   dict["lz4_compression_error"] = stats.lz4_compression_error;
+
+   return dict;
+}
+
 IPFIXExporter::IPFIXExporter() :
    extensions(nullptr), extension_cnt(0),
    templates(nullptr), templatesDataSize(0),
@@ -264,7 +275,13 @@ void IPFIXExporter::init_plugin(const char *params, Plugins &plugins, const std:
       {.read = [this]() { return get_ipfix_config(); },
        .clear = nullptr});
 
+   auto plugin_stats_file = dir->addFile(
+      "plugin_stats",
+      {.read = [this]() { return getExtendedAtomicStats(ext_atomic_stats); },
+       .clear = nullptr});
+
    m_holder.add(plugin_config_file);
+   m_holder.add(plugin_stats_file);
 
 }
 
@@ -866,7 +883,7 @@ void IPFIXExporter::flush()
  * When the collector disconnects, tries to reconnect and resend the data
  *
  * \param packet Packet to send
- * \return 0 on success, -1 on socket error, -2 on compress error,
+ * \return 0 on success, -1 on socket error, -2 on compress error, -3 eagain error
  *         1 when data needs to be resent (after reconnect)
  */
 int IPFIXExporter::send_packet(ipfix_packet_t *packet)
@@ -880,8 +897,16 @@ int IPFIXExporter::send_packet(ipfix_packet_t *packet)
    }
 
    auto dataLen = packetDataBuffer.compress();
+   if (dataLen < 0) {
+      ext_stats.lz4_compression_error++;
+      return -2;
+   }
    auto data = packetDataBuffer.getCompressed();
 
+   static constexpr size_t eagain_limit = 3;
+
+   size_t eagain_count = 0;
+
    /* sendto() does not guarantee that everything will be send in one piece */
    while (sent < dataLen) {
       /* Send data to collector (TCP and SCTP ignores last two arguments) */
@@ -924,6 +949,12 @@ int IPFIXExporter::send_packet(ipfix_packet_t *packet)
             /* Say that we should try to connect and send data again */
             return 1;
          case EAGAIN:
+            eagain_count++;
+            if (eagain_count == eagain_limit) {
+               ext_stats.flows_dropped_eagain += packet->flows;
+               return -1;
+            }
+            usleep(1000);
             // EAGAIN is returned when the socket is non-blocking and the send buffer is full
             // possible wait and stop flag check
             continue;
@@ -1179,9 +1210,21 @@ telemetry::Content IPFIXExporter::get_ipfix_config()
    return dict;
 }
 
+
 void IPFIXExporter::update_plugin_stats(uint64_t timestamp)
 {
+   ext_atomic_stats.flows_dropped_eagain.fetch_add(ext_stats.flows_dropped_eagain, std::memory_order_relaxed);
+   ext_atomic_stats.lz4_compression_error.fetch_add(ext_stats.lz4_compression_error, std::memory_order_relaxed);
+
+   if (timestamp == 0) {
+      ext_atomic_stats.fps_dropped_eagain.store(0);
+      return;
+   }
+
+   ext_atomic_stats.fps_dropped_eagain.store(ext_stats.flows_dropped_eagain * 1000000000 / timestamp);
 
+   ext_stats.flows_dropped_eagain = 0;
+   ext_stats.lz4_compression_error = 0;
 }
 
 // compress buffer implementation

output/ipfix.hpp

@@ -457,14 +457,15 @@ 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;
+struct IpfixExtendedStats {
+  uint64_t flows_dropped_eagain = 0;
+  uint64_t lz4_compression_error = 0;
+};
+
+struct IpfixExtendedAtomicStats {
+  std::atomic<uint64_t> flows_dropped_eagain = 0;
+  std::atomic<uint64_t> fps_dropped_eagain = 0;
+  std::atomic<uint64_t> lz4_compression_error = 0;
 };
 
 class IPFIXExporter : public OutputPlugin
@@ -519,6 +520,9 @@ class IPFIXExporter : public OutputPlugin
    uint16_t mtu; /**< Max size of packet payload sent */
    uint16_t tmpltMaxBufferSize; /**< Size of template buffer, tmpltBufferSize < packetDataBuffer */
 
+   IpfixExtendedStats ext_stats;
+   IpfixExtendedAtomicStats ext_atomic_stats;
+
    void init_template_buffer(template_t *tmpl);
    int fill_template_set_header(uint8_t *ptr, uint16_t size);
    void check_template_lifetime(template_t *tmpl);