parser-stats - add parser stats to input plugins

Author: SiskaPavel

include/ipfixprobe/input.hpp

@@ -73,10 +73,10 @@ class InputPlugin : public TelemetryUtils, public Plugin
       std::shared_ptr<telemetry::Directory> plugin_dir, 
       std::shared_ptr<telemetry::Directory> queues_dir) {};
 
+   ParserStats m_parser_stats;
+
 private:
    void create_parser_stats_telemetry(std::shared_ptr<telemetry::Directory> queues_dir);
-
-   ParserStats m_parser_stats;
 };
 
 }

input/dpdk-ring.cpp

@@ -179,6 +179,7 @@ InputPlugin::Result DpdkRingReader::get(PacketBlock& packets)
     }
     for (auto i = 0; i < pkts_read_; i++) {
         parse_packet(&opt,
+            m_parser_stats,
             getTimestamp(mbufs_[i]),
             rte_pktmbuf_mtod(mbufs_[i], const std::uint8_t*),
             rte_pktmbuf_data_len(mbufs_[i]),

input/dpdk.cpp

@@ -278,6 +278,7 @@ InputPlugin::Result DpdkReader::get(PacketBlock& packets)
         packets.cnt++;
 #else
         parse_packet(&opt,
+            m_parser_stats,
             dpdkDevice.getPacketTimestamp(mBufs[packetID]),
             rte_pktmbuf_mtod(mBufs[packetID], const std::uint8_t*),
             rte_pktmbuf_data_len(mBufs[packetID]),

input/ndp.cpp

@@ -107,7 +107,7 @@ InputPlugin::Result NdpPacketReader::get(PacketBlock &packets)
          throw PluginError(ndpReader.error_msg);
       }
       read_pkts++;
-      parse_packet(&opt, timestamp, ndp_packet->data, ndp_packet->data_length, ndp_packet->data_length);
+      parse_packet(&opt, m_parser_stats, timestamp, ndp_packet->data, ndp_packet->data_length, ndp_packet->data_length);
    }
 
    m_seen += read_pkts;

input/parser.cpp

@@ -640,7 +640,7 @@ inline uint16_t process_pppoe(const u_char *data_ptr, uint16_t data_len, Packet
    return length;
 }
 
-void parse_packet(parser_opt_t *opt, struct timeval ts, const uint8_t *data, uint16_t len, uint16_t caplen)
+void parse_packet(parser_opt_t *opt, ParserStats& stats, struct timeval ts, const uint8_t *data, uint16_t len, uint16_t caplen)
 {
    if (opt->pblock->cnt >= opt->pblock->size) {
       return;
@@ -672,6 +672,8 @@ void parse_packet(parser_opt_t *opt, struct timeval ts, const uint8_t *data, uin
    pkt->tcp_mss = 0;
    pkt->mplsTop = 0;
 
+   stats.seen_packets++;
+
    uint32_t l3_hdr_offset = 0;
    uint32_t l4_hdr_offset = 0;
    try {
@@ -697,6 +699,7 @@ void parse_packet(parser_opt_t *opt, struct timeval ts, const uint8_t *data, uin
 
       if (pkt->ethertype == ETH_P_TRILL) {
          data_offset += parse_trill(data + data_offset, caplen - data_offset, pkt);
+         stats.trill_packets++;
          data_offset += parse_eth_hdr(data + data_offset, caplen - data_offset, pkt);
       }
       l3_hdr_offset = data_offset;
@@ -706,24 +709,39 @@ void parse_packet(parser_opt_t *opt, struct timeval ts, const uint8_t *data, uin
          data_offset += parse_ipv6_hdr(data + data_offset, caplen - data_offset, pkt);
       } else if (pkt->ethertype == ETH_P_MPLS_UC || pkt->ethertype == ETH_P_MPLS_MC) {
          data_offset += process_mpls(data + data_offset, caplen - data_offset, pkt);
+         stats.mpls_packets++;
       } else if (pkt->ethertype == ETH_P_PPP_SES) {
          data_offset += process_pppoe(data + data_offset, caplen - data_offset, pkt);
+         stats.pppoe_packets++;
       } else if (!opt->parse_all) {
+         stats.unknown_packets++;
          DEBUG_MSG("Unknown ethertype %x\n", pkt->ethertype);
          return;
       }
 
       l4_hdr_offset = data_offset;
       if (pkt->ip_proto == IPPROTO_TCP) {
          data_offset += parse_tcp_hdr(data + data_offset, caplen - data_offset, pkt);
+         stats.tcp_packets++;
       } else if (pkt->ip_proto == IPPROTO_UDP) {
          data_offset += parse_udp_hdr(data + data_offset, caplen - data_offset, pkt);
+         stats.udp_packets++;
       }
    } catch (const char *err) {
       DEBUG_MSG("%s\n", err);
       return;
    }
 
+   if (pkt->vlan_id) {
+      stats.vlan_packets++;
+   }
+
+   if (pkt->ethertype == ETH_P_IP) {
+      stats.ipv4_packets++;
+   } else if (pkt->ethertype == ETH_P_IPV6) {
+      stats.ipv6_packets++;
+   }
+
    uint16_t pkt_len = caplen;
    pkt->packet = data;
    pkt->packet_len = caplen;

input/parser.hpp

@@ -30,6 +30,7 @@
 #define IPXP_INPUT_PARSER_HPP
 
 #include <ipfixprobe/packet.hpp>
+#include <ipfixprobe/parser-stats.hpp>
 
 #ifdef WITH_PCAP
 #include <pcap/pcap.h>
@@ -65,7 +66,7 @@ typedef struct parser_opt_s {
    int datalink;
 } parser_opt_t;
 
-void parse_packet(parser_opt_t *opt, struct timeval ts, const uint8_t *data, uint16_t len, uint16_t caplen);
+void parse_packet(parser_opt_t *opt, ParserStats& stats, struct timeval ts, const uint8_t *data, uint16_t len, uint16_t caplen);
 
 }
 #endif /* IPXP_INPUT_PARSER_HPP */

input/pcap.cpp

@@ -41,6 +41,11 @@ namespace ipxp {
 // Read only 1 packet into packet block
 constexpr size_t PCAP_PACKET_BLOCK_SIZE = 1;
 
+struct UserData {
+   parser_opt_t* opt;
+   ParserStats& stats;
+};
+
 __attribute__((constructor)) static void register_this_plugin()
 {
    static PluginRecord rec = PluginRecord("pcap", [](){return new PcapReader();});
@@ -49,12 +54,14 @@ __attribute__((constructor)) static void register_this_plugin()
 
 /**
  * \brief Parsing callback function for pcap_dispatch() call. Parse packets up to transport layer.
- * \param [in,out] arg Serves for passing pointer to Packet structure into callback function.
+ * \param [in,out] arg Serves for passing pointer to Packet user data structure into callback function.
  * \param [in] h Contains timestamp and packet size.
  * \param [in] data Pointer to the captured packet data.
  */
 void packet_handler(u_char *arg, const struct pcap_pkthdr *h, const u_char *data)
 {
+   UserData *user_data = reinterpret_cast<UserData *>(arg);
+
 #ifdef __CYGWIN__
    // WinPcap, uses Microsoft's definition of struct timeval, which has `long` data type
    // used for both tv_sec and tv_usec and has 32 bit even on 64 bit platform.
@@ -64,9 +71,9 @@ void packet_handler(u_char *arg, const struct pcap_pkthdr *h, const u_char *data
    new_h.ts.tv_usec = *(reinterpret_cast<const uint32_t *>(h) + 1);
    new_h.caplen = *(reinterpret_cast<const uint32_t *>(h) + 2);
    new_h.len = *(reinterpret_cast<const uint32_t *>(h) + 3);
-   parse_packet((parser_opt_t *) arg, new_h.ts, data, new_h.len, new_h.caplen);
+   parse_packet(user_data->opt, user_data->stats, new_h.ts, data, new_h.len, new_h.caplen);
 #else
-   parse_packet((parser_opt_t *) arg, h->ts, data, h->len, h->caplen);
+   parse_packet(user_data->opt, user_data->stats, h->ts, data, h->len, h->caplen);
 #endif
 }
 
@@ -249,12 +256,14 @@ InputPlugin::Result PcapReader::get(PacketBlock &packets)
    parser_opt_t opt = {&packets, false, false, m_datalink};
    int ret;
 
+   UserData user_data = {&opt, m_parser_stats};
+
    if (m_handle == nullptr) {
       throw PluginError("no interface capture or file opened");
    }
 
    packets.cnt = 0;
-   ret = pcap_dispatch(m_handle, PCAP_PACKET_BLOCK_SIZE, packet_handler, (u_char *) (&opt));
+   ret = pcap_dispatch(m_handle, PCAP_PACKET_BLOCK_SIZE, packet_handler, (u_char *) (&user_data));
    if (m_live) {
       if (ret == 0) {
          return Result::TIMEOUT;

input/raw.cpp

@@ -309,7 +309,7 @@ int RawReader::process_packets(struct tpacket_block_desc *pbd, PacketBlock &pack
       size_t snaplen = ppd->tp_snaplen;
       struct timeval ts = {ppd->tp_sec, ppd->tp_nsec / 1000};
 
-      parse_packet(&opt, ts, data, len, snaplen);
+      parse_packet(&opt, m_parser_stats, ts, data, len, snaplen);
       ppd = (struct tpacket3_hdr *) ((uint8_t *) ppd + ppd->tp_next_offset);
    }
    m_last_ppd = ppd;