Merge pull request #228 from CESNET/ctt-input-parse

CTT Metadata input handling

ChangeLog

@@ -1,3 +1,14 @@
+2024-10-17 ipfixprobe-4.13.0
+	* usability: introduce docker/podman container to convert PCAP files to CSV
+	* IPFIX: fix order of TCP options flags
+	* basicplus: update TCP options mask across flow packets
+	* utils: introduce memcpy_le32toh() for ipfix representation
+	* wg: fix parsing and exporting byte order (IPFIX)
+	* DPDK-ring: optimization: prefetch; read timestamp from HW metadata if available
+	* cache: optimization - prefetch
+	* IPv6: fix header parsing
+	* DPDK&DPDK-ring: fix use of parse_packet(), skip invalid packets causing crash
+
 2024-08-28 ipfixprobe-4.12.0
 	* ipfix plugin: support lz4 compression
 	* ipfixprobe: possibility to set workers affinity

Makefile.am

@@ -203,7 +203,8 @@ ipfixprobe_headers_src=\
 		include/ipfixprobe/ipfix-elements.hpp \
 		include/ipfixprobe/rtp.hpp \
 		include/ipfixprobe/telemetry-utils.hpp \
-		include/ipfixprobe/parser-stats.hpp
+		include/ipfixprobe/parser-stats.hpp \
+		include/ipfixprobe/cttmeta.hpp
 
 ipfixprobe_src=\
 		$(ipfixprobe_input_src) \
@@ -336,3 +337,7 @@ deb:
 else
 endif
 
+.PHONY: doc
+doc:
+	doxygen
+

NEWS

@@ -1,3 +1,23 @@
+2024-10-17 (Tomas Cejka): doc: include generated Doxyfile to create documentation
+2024-10-17 (Tomas Cejka): doc: add doxygen comment for parse_packet()
+2024-10-16 (Jan Sobol): dpdk-ring - fix checking if any packet has actually been parsed
+2024-10-16 (Jan Sobol): dpdk - fix checking if any packet has actually been parsed
+2024-10-02 (Damir Zainullin): Fix IPv6 header parsing
+2024-09-30 (Pavel Šiška): Merge pull request #220 from CESNET/prefetch-optimizations
+2024-09-30 (Pavel Šiška): Merge pull request #219 from CESNET/dpdk-ring-metadata-timestamp
+2024-09-30 (Pavel Šiška): Merge pull request #215 from CESNET/new-version
+2024-09-26 (Jan Sobol): cache - prefetch flow records before checking their expiration
+2024-09-26 (Jan Sobol): dpdk-ring - prefetch dequeued packets before processing
+2024-09-25 (Jan Sobol): dpdk-ring - read timestamp from hw metadata if available
+2024-09-20 (Tomas Cejka): Merge pull request #216 from CESNET/ipfixprobe-docker-wrapper
+2024-09-19 (Tomas Cejka): dist: include docker/ files into distribution archive
+2024-09-19 (Jan Sobol): wg - fix parsing and exporting byte order
+2024-09-19 (Jan Sobol): utils - introduce memcpy_le32toh function
+2024-09-18 (Jan Sobol): basicplus test - fix reference values of tcp options
+2024-09-18 (Jan Sobol): basicplus - update tcp options mask across flow packets
+2024-09-18 (Jan Sobol): parser - fix order of tcp options flags according to ipfix standard https://www.iana.org/assignments/ipfix/ipfix.xhtml, entity 209 - tcpOptions
+2024-09-17 (Jaroslav Pesek): process container - introduce docker/podman container wrapper for processing pcaps to csvs
+
 2024-08-28 (Pavel Siska): ipfixprobed - add new option LZ4_COMPRESSION to init script and config example
 2024-08-28 (Pavel Siska): README.md - add LZ4 compression info
 2024-08-28 (Jakub Antonín Štigler): ipfix plugin: add lz4 compression

configure.ac

@@ -2,7 +2,7 @@
 # Process this file with autoconf to produce a configure script.
 
 AC_PREREQ([2.69])
-AC_INIT([ipfixprobe], [4.12.0], [[email protected]])
+AC_INIT([ipfixprobe], [4.13.0], [[email protected]])
 
 AC_CONFIG_SRCDIR([main.cpp])
 AC_CONFIG_HEADERS([config.h])

include/ipfixprobe/cttmeta.hpp

@@ -0,0 +1,81 @@
+#ifndef IPXP_INPUT_CTT_HPP
+#define IPXP_INPUT_CTT_HPP
+
+#include <bits/types/struct_timeval.h>
+#include <cstdint>
+
+namespace ipxp {
+
+enum CsumStatus : uint8_t {
+    CSUM_UNKNOWN = 0x0, ///< No information about the checksum
+    CSUM_BAD     = 0x1, ///< The checksum in the packet is wrong
+    CSUM_GOOD    = 0x2, ///< The checksum in the packet is valid
+    CSUM_NONE    = 0x3  ///< Checksum not correct but header integrity verified
+};
+
+enum ParserStatus : uint8_t {
+    PA_OK      = 0x0, ///< Parsing completed successfully
+    PA_UNKNOWN = 0x1, ///< Parser stopped at an unknown protocol
+    PA_LIMIT   = 0x2, ///< Parser stopped at its own limit (e.g., VLAN=4)
+    PA_ERROR   = 0x3  ///< Error in protocol header or parsing overflow
+};
+
+enum L2PType : uint8_t {
+    L2_UNKNOWN          = 0x0, ///< Unknown L2 protocol
+    L2_ETHER_IP         = 0x1, ///< Ethernet with IP payload
+    L2_ETHER_TIMESYNC   = 0x2, ///< Ethernet with TimeSync protocol
+    L2_ETHER_ARP        = 0x3, ///< Ethernet with ARP protocol
+    L2_ETHER_LLDP       = 0x4, ///< Ethernet with LLDP protocol
+    L2_ETHER_NSH        = 0x5, ///< Ethernet with NSH protocol
+    L2_ETHER_VLAN       = 0x6, ///< Ethernet with VLAN tagging
+    L2_ETHER_QINQ       = 0x7, ///< Ethernet with QinQ tagging
+    L2_ETHER_PPPOE      = 0x8, ///< Ethernet with PPPoE encapsulation
+    L2_ETHER_FCOE       = 0x9, ///< Ethernet with FCoE protocol
+    L2_ETHER_MPLS       = 0xA  ///< Ethernet with MPLS
+};
+
+enum L3PType : uint8_t {
+    L3_UNKNOWN   = 0x0, ///< Unknown L3 protocol
+    L3_IPV4      = 0x1, ///< IPv4 protocol
+    L3_IPV4_EXT  = 0x3, ///< IPv4 with extensions
+    L3_IPV6      = 0x4, ///< IPv6 protocol
+    L3_IPV6_EXT  = 0xC  ///< IPv6 with extensions
+};
+
+enum L4PType : uint8_t {
+    L4_UNKNOWN = 0x0, ///< Unknown L4 protocol
+    L4_TCP     = 0x1, ///< TCP protocol
+    L4_UDP     = 0x2, ///< UDP protocol
+    L4_FRAG    = 0x3, ///< Fragmented packet
+    L4_SCTP    = 0x4, ///< SCTP protocol
+    L4_ICMP    = 0x5, ///< ICMP protocol
+    L4_NONFRAG = 0x6, ///< Non-fragmented packet
+    L4_IGMP    = 0x7  ///< IGMP protocol
+};
+
+struct Metadata_CTT {
+   struct timeval ts;             ///< Timestamp; invalid if all bits are 1
+   uint16_t vlan_tci;             ///< VLAN Tag Control Information from outer VLAN
+   bool vlan_vld : 1;             ///< VLAN valid flag; indicates if VLAN TCI is valid
+   bool vlan_stripped : 1;        ///< VLAN stripped flag; outer VLAN only
+   CsumStatus ip_csum_status : 2; ///< IP checksum status
+   CsumStatus l4_csum_status : 2; ///< Layer 4 checksum status
+   ParserStatus parser_status : 2;///< Final state of FPGA parser
+   uint8_t ifc;                   ///< Interface (IFC) number
+   uint16_t filter_bitmap;        ///< Filter bitmap; each filter rule can have several mark bits
+   bool ctt_export_trig : 1;      ///< CTT flag; packet triggered export in CTT
+   bool ctt_rec_matched : 1;      ///< CTT flag; packet matched record in CTT
+   bool ctt_rec_created : 1;      ///< CTT flag; packet created record in CTT
+   bool ctt_rec_deleted : 1;      ///< CTT flag; packet deleted record in CTT
+   uint64_t flow_hash;            ///< Flow hash; not the same as RSS hash
+   uint8_t l2_len : 7;            ///< Length of the L2 layer, if known
+   uint16_t l3_len : 9;           ///< Length of the L3 layer, if known
+   uint8_t l4_len : 8;            ///< Length of the L4 layer, if known
+   L2PType l2_ptype : 4;          ///< Type of the L2 layer
+   L3PType l3_ptype : 4;          ///< Type of the L3 layer
+   L4PType l4_ptype : 4;          ///< Type of the L4 layer
+};
+
+}
+
+#endif // IPXP_INPUT_CTT_HPP

include/ipfixprobe/packet.hpp

@@ -38,13 +38,15 @@
 
 #include <ipfixprobe/ipaddr.hpp>
 #include <ipfixprobe/flowifc.hpp>
+#include <ipfixprobe/cttmeta.hpp>
 
 namespace ipxp {
 
 /**
  * \brief Structure for storing parsed packet fields
  */
 struct Packet : public Record {
+   Metadata_CTT cttmeta; /**< Metadata from CTT */
    struct timeval ts;
 
    uint8_t     dst_mac[6];

input/dpdk-ring.cpp

@@ -211,7 +211,7 @@ InputPlugin::Result DpdkRingReader::get(PacketBlock& packets)
     m_stats.receivedPackets += pkts_read_;
     m_stats.receivedBytes += packets.bytes;
 
-    return Result::PARSED;
+    return opt.pblock->cnt ? Result::PARSED : Result::NOT_PARSED;
 }
 
 telemetry::Content DpdkRingReader::get_queue_telemetry()

input/dpdk.cpp

@@ -292,7 +292,7 @@ InputPlugin::Result DpdkReader::get(PacketBlock& packets)
     m_stats.receivedPackets += receivedPackets;
     m_stats.receivedBytes += packets.bytes;
 
-    return Result::PARSED;
+    return packets.cnt ? Result::PARSED : Result::NOT_PARSED;
 }
 
 }

input/ndp.cpp

@@ -27,20 +27,41 @@
  *
  */
 
+#include <cstdint>
 #include <cstdio>
 #include <cstring>
-#include <iostream>
+#include <netinet/in.h>
+#include <sys/types.h>
+#include <cstdint>
+#include <cstddef>
+#include <inttypes.h>
 
 #include "ndp.hpp"
+#include "ipfixprobe/packet.hpp"
+#include "ipfixprobe/plugin.hpp"
 #include "parser.hpp"
 
 namespace ipxp {
 
+uint64_t extract(const uint8_t* bitvec, size_t start_bit, size_t bit_length) {
+   size_t start_byte = start_bit / 8;
+   size_t end_bit = start_bit + bit_length;
+   size_t end_byte = (end_bit + 7) / 8;
+   uint64_t value = 0;
+   for (size_t i = 0; i < end_byte - start_byte; ++i) {
+      value |= static_cast<uint64_t>(bitvec[start_byte + i]) << (8 * i);
+   }
+   value >>= (start_bit % 8);
+   uint64_t mask = (bit_length == 64) ? ~0ULL : ((1ULL << bit_length) - 1);
+   return value & mask;
+}
+
 telemetry::Content NdpPacketReader::get_queue_telemetry()
 {
    telemetry::Dict dict;
    dict["received_packets"] = m_stats.receivedPackets;
    dict["received_bytes"] = m_stats.receivedBytes;
+   dict["bad_metadata"] = m_stats.bad_metadata;
    return dict;
 }
 
@@ -71,6 +92,9 @@ void NdpPacketReader::init(const char *params)
    if (parser.m_dev.empty()) {
       throw PluginError("specify device path");
    }
+   if (parser.m_metadata == "ctt") {
+      m_ctt_metadata = true;
+   }
    init_ifc(parser.m_dev);
 }
 
@@ -86,6 +110,38 @@ void NdpPacketReader::init_ifc(const std::string &dev)
    }
 }
 
+int NdpPacketReader::parse_ctt_metadata(const ndp_packet *ndp_packet, Metadata_CTT &ctt)
+{
+   if (ndp_packet->header_length != 32) {
+      return -1;
+   }
+   const uint8_t *metadata = ndp_packet->header;
+
+   ctt.ts.tv_usec      = extract(metadata, 0,   32);
+   ctt.ts.tv_sec       = extract(metadata, 32,  32);
+   ctt.vlan_tci        = extract(metadata, 64,  16);
+   ctt.vlan_vld        = extract(metadata, 80,  1);
+   ctt.vlan_stripped   = extract(metadata, 81,  1);
+   ctt.ip_csum_status  = static_cast<CsumStatus>(extract(metadata, 82,  2));
+   ctt.l4_csum_status  = static_cast<CsumStatus>(extract(metadata, 84,  2));
+   ctt.parser_status   = static_cast<ParserStatus>(extract(metadata, 86,  2));
+   ctt.ifc             = extract(metadata, 88,  8);
+   ctt.filter_bitmap   = extract(metadata, 96,  16);
+   ctt.ctt_export_trig = extract(metadata, 112, 1);
+   ctt.ctt_rec_matched = extract(metadata, 113, 1);
+   ctt.ctt_rec_created = extract(metadata, 114, 1);
+   ctt.ctt_rec_deleted = extract(metadata, 115, 1);
+   ctt.flow_hash       = extract(metadata, 128, 64);
+   ctt.l2_len          = extract(metadata, 192, 7);
+   ctt.l3_len          = extract(metadata, 199, 9);
+   ctt.l4_len          = extract(metadata, 208, 8);
+   ctt.l2_ptype        = static_cast<L2PType>(extract(metadata, 216, 4));
+   ctt.l3_ptype        = static_cast<L3PType>(extract(metadata, 220, 4));
+   ctt.l4_ptype        = static_cast<L4PType>(extract(metadata, 224, 4));
+
+   return 0;
+}
+
 InputPlugin::Result NdpPacketReader::get(PacketBlock &packets)
 {
    parser_opt_t opt = {&packets, false, false, 0};
@@ -107,7 +163,18 @@ InputPlugin::Result NdpPacketReader::get(PacketBlock &packets)
          throw PluginError(ndpReader.error_msg);
       }
       read_pkts++;
-      parse_packet(&opt, m_parser_stats, timestamp, ndp_packet->data, ndp_packet->data_length, ndp_packet->data_length);
+      if (m_ctt_metadata) {
+         Metadata_CTT ctt;
+         int ret = parse_ctt_metadata(ndp_packet, ctt);
+         if (ret == -1) {
+            m_stats.bad_metadata++;
+            parse_packet(&opt, m_parser_stats, timestamp, ndp_packet->data, ndp_packet->data_length, ndp_packet->data_length);
+         } else {
+            parse_packet_ctt_metadata(&opt, m_parser_stats, ctt, ndp_packet->data, ndp_packet->data_length, ndp_packet->data_length);
+         }
+      } else {
+         parse_packet(&opt, m_parser_stats, timestamp, ndp_packet->data, ndp_packet->data_length, ndp_packet->data_length);
+      }
    }
 
    m_seen += read_pkts;

input/ndp.hpp

@@ -30,12 +30,14 @@
 #ifndef IPXP_INPUT_NDP_HPP
 #define IPXP_INPUT_NDP_HPP
 
+#include <bits/types/struct_timeval.h>
 #include <ndpreader.hpp>
 
 #include <ipfixprobe/input.hpp>
 #include <ipfixprobe/packet.hpp>
 #include <ipfixprobe/options.hpp>
 #include <ipfixprobe/utils.hpp>
+#include <ipfixprobe/cttmeta.hpp>
 
 namespace ipxp {
 
@@ -44,13 +46,14 @@ class NdpOptParser : public OptionsParser
 public:
    std::string m_dev;
    uint64_t m_id;
+   std::string m_metadata;
 
-   NdpOptParser() : OptionsParser("ndp", "Input plugin for reading packets from a ndp device"), m_dev(""), m_id(0)
+   NdpOptParser() : OptionsParser("ndp", "Input plugin for reading packets from a ndp device"), m_dev(""), m_id(0), m_metadata("")
    {
       register_option("d", "dev", "PATH", "Path to a device file", [this](const char *arg){m_dev = arg; return true;}, OptionFlags::RequiredArgument);
       register_option("I", "id", "NUM", "Link identifier number",
-         [this](const char *arg){try {m_id = str2num<decltype(m_id)>(arg);} catch(std::invalid_argument &e) {return false;} return true;},
-         OptionFlags::RequiredArgument);
+         [this](const char *arg){try {m_id = str2num<decltype(m_id)>(arg);} catch(std::invalid_argument &e) {return false;} return true;}, OptionFlags::RequiredArgument);
+      register_option("M", "meta", "Metadata type", "Choose metadata type if any", [this](const char *arg){m_metadata = arg; return true;}, OptionFlags::RequiredArgument);
    }
 };
 
@@ -74,14 +77,18 @@ class NdpPacketReader : public InputPlugin
    struct RxStats {
         uint64_t receivedPackets;
         uint64_t receivedBytes;
+        uint64_t bad_metadata;
    };
 
    telemetry::Content get_queue_telemetry();
 
    NdpReader ndpReader;
    RxStats m_stats = {};
 
+   bool m_ctt_metadata = false;
+
    void init_ifc(const std::string &dev);
+   int parse_ctt_metadata(const ndp_packet *ndp_packet, Metadata_CTT &ctt);
 };
 
 }