CTT support

CMakeLists.txt

@@ -19,6 +19,7 @@ option(ENABLE_OUTPUT_UNIREC         "Enable build of output UNIREC plugin"
 option(ENABLE_PROCESS_EXPERIMENTAL  "Enable build of experimental process plugins"            OFF)
 option(ENABLE_MILLISECONDS_TIMESTAMP "Compile ipfixprobe with miliseconds timestamp precesion" OFF)
 option(ENABLE_NEMEA                 "Enable build of NEMEA plugins"                           OFF)
+option(ENABLE_CTT                   "Enable support of conaction tracking table"              OFF)
 
 option(ENABLE_RPMBUILD              "Enable build of RPM package"                             ON)
 option(ENABLE_TESTS                 "Build tests (make test)"                                 OFF)
@@ -49,6 +50,8 @@ set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wall -Wextra")
 set(CMAKE_CXX_FLAGS_RELEASE "${CMAKE_CXX_FLAGS_RELEASE} -O3")
 set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -g -ggdb3")
 
+set(CMAKE_INTERPROCEDURAL_OPTIMIZATION TRUE)
+
 include(cmake/dependencies.cmake)
 
 add_subdirectory(external)

cmake/dependencies.cmake

@@ -15,7 +15,7 @@ if (ENABLE_INPUT_DPDK)
 	pkg_check_modules(DPDK REQUIRED libdpdk)
 endif()
 
-if (ENABLE_INPUT_NFB)
+if (ENABLE_INPUT_NFB OR ENABLE_CTT)
 	find_package(NFB REQUIRED)
 	find_package(NUMA REQUIRED)
 endif()

include/ipfixprobe/cttConfig.hpp

@@ -0,0 +1,12 @@
+#pragma once
+
+#include <string>
+
+namespace ipxp {
+
+struct CttConfig {
+    std::string nfb_device;
+    unsigned dma_channel;
+};
+
+} // namespace ipxp

include/ipfixprobe/cttmeta.hpp

@@ -0,0 +1,126 @@
+#pragma once
+
+#include <cstdint>
+#include <optional>
+#include <bits/types/struct_timeval.h>
+#include <feta.hpp>
+
+namespace ipxp {    
+
+static 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;
+}
+
+enum MessageType : uint8_t
+{
+    FRAME_AND_FULL_METADATA = 0x0, ///< Frame and full metadata
+    FRAME_AND_HALF_METADATA = 0x1, ///< Frame and half metadata
+    FRAME_WITH_TIMESTAMP    = 0x2, ///< Frame with timestamp
+    FRAME_WITH_NO_METADATA  = 0x3, ///< Frame with no metadata
+    ONLY_FULL_METADATA      = 0x4, ///< Only full metadata
+    FLOW_EXPORT             = 0xF  ///< Flow export
+};
+
+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 CttMetadata {
+    constexpr static size_t SIZE = 32;
+
+    static CttMetadata parse(const uint8_t* data, size_t length) noexcept
+    {
+        CttMetadata metadata;
+        if (length != CttMetadata::SIZE) {
+            metadata.flow_hash = 0;
+            return metadata;
+        }
+
+        metadata.vlan_tci = *reinterpret_cast<const uint16_t*>(data + 8);
+        metadata.vlan_vld = *reinterpret_cast<const uint8_t*>(data + 10) & 0x01;       
+        metadata.vlan_stripped = *reinterpret_cast<const uint8_t*>(data + 10) & 0x02;
+        metadata.flow_hash = *reinterpret_cast<const uint64_t*>(data + 16);
+        return metadata;
+    }
+    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
+};
+
+constexpr static timeval CTT_REQUEST_TIMEOUT = {10, 0}; ///< Timeout for CTT request
+
+static constexpr size_t KEY_SIZE = 8;
+static constexpr size_t STATE_SIZE = sizeof(feta::CttRecord);
+static constexpr size_t MASK_SIZE = 21;
+
+
+}

include/ipfixprobe/flowifc.hpp

@@ -37,6 +37,7 @@
 #include <stdint.h>
 #include <stdlib.h>
 #include <sys/time.h>
+#include <bitset>
 
 #ifdef WITH_NEMEA
 #include <unirec/unirec.h>
@@ -233,38 +234,61 @@ struct Record {
 	virtual ~Record() { remove_extensions(); }
 };
 
-#define FLOW_END_INACTIVE 0x01
-#define FLOW_END_ACTIVE 0x02
-#define FLOW_END_EOF 0x03
-#define FLOW_END_FORCED 0x04
-#define FLOW_END_NO_RES 0x05
+enum FlowEndReason : int {
+	FLOW_END_INACTIVE = 0x1, /**< Flow ended due to inactivity timeout. */
+	FLOW_END_ACTIVE = 0x2, /**< Flow ended due to active timeout. */
+	FLOW_END_EOF = 0x3, /**< Flow ended due to end of flow (TCP FIN or RST). */
+	FLOW_END_FORCED = 0x4, /**< Flow ended due to process plugin flushes */
+	FLOW_END_NO_RES = 0x5 /**< Flow ended due to lack of resources (e.g. full cache line). */
+};
 
 /**
  * \brief Flow record struct constaining basic flow record data and extension headers.
  */
 struct Flow : public Record {
-	uint64_t flow_hash;
-
-	struct timeval time_first;
-	struct timeval time_last;
-	uint64_t src_bytes;
-	uint64_t dst_bytes;
-	uint32_t src_packets;
-	uint32_t dst_packets;
-	uint8_t src_tcp_flags;
-	uint8_t dst_tcp_flags;
-
-	uint8_t ip_version;
-
-	uint8_t ip_proto;
-	uint16_t src_port;
-	uint16_t dst_port;
-	ipaddr_t src_ip;
-	ipaddr_t dst_ip;
-
-	uint8_t src_mac[6];
-	uint8_t dst_mac[6];
-	uint8_t end_reason;
+    static inline const int MAXIMAL_PROCESS_PLUGIN_COUNT = 64;
+    /**
+     * \brief Plugins status struct describes flow information required by process plugins.
+     */
+    struct PluginsStatus {
+        // get_no_data[i] == true -> i-th process plugin requires no flow data
+        // get_no_data[i] == false && get_all_data[i] == true -> i-th process plugin requires all
+        // available flow data
+        // get_no_data[i] == false && get_all_data[i] == false -> i-th process plugin requires
+        // only metadata
+        std::bitset<MAXIMAL_PROCESS_PLUGIN_COUNT> get_all_data;
+        std::bitset<MAXIMAL_PROCESS_PLUGIN_COUNT> get_no_data;
+    };
+
+    uint64_t flow_hash;
+    uint64_t flow_hash_ctt;     /**< Flow hash for CTT. */
+
+    PluginsStatus plugins_status; /**< Statuses of the process plugins for this flow, used to check
+                                     if the flow process plugins requires all available data, only
+                                     metadata or nothing of this. */
+
+   struct timeval time_first;
+   struct timeval time_last;
+   uint64_t src_bytes;
+   uint64_t dst_bytes;
+   uint32_t src_packets;
+   uint32_t dst_packets;
+   uint8_t  src_tcp_flags;
+   uint8_t  dst_tcp_flags;
+
+   uint8_t  ip_version;
+   uint16_t vlan_id;
+
+   uint8_t  ip_proto;
+   uint16_t src_port;
+   uint16_t dst_port;
+   ipaddr_t src_ip;
+   ipaddr_t dst_ip;
+
+   uint8_t src_mac[6];
+   uint8_t dst_mac[6];
+   uint8_t end_reason;
+   bool swapped; /**< Flow addresses and ports were swapped on creation. */
 };
 
 } // namespace ipxp

include/ipfixprobe/inputPlugin.hpp

@@ -20,10 +20,12 @@
 #include "parser-stats.hpp"
 #include "plugin.hpp"
 #include "telemetry-utils.hpp"
+#include "cttConfig.hpp"
 
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <optional>
 
 #include <telemetry.hpp>
 
@@ -66,6 +68,11 @@ class IPXP_API InputPlugin
 		std::shared_ptr<telemetry::Directory> plugin_dir,
 		std::shared_ptr<telemetry::Directory> queues_dir);
 
+	virtual std::optional<CttConfig> get_ctt_config() const 
+	{
+		return std::nullopt;
+	}
+
 	/// Number of packets seen by the plugin.
 	uint64_t m_seen = 0;
 	/// Number of packets successfully parsed.

include/ipfixprobe/options.hpp

@@ -49,7 +49,7 @@ class IPXP_API OptionsParser {
 
 	OptionsParser();
 	OptionsParser(const std::string& name, const std::string& info);
-	~OptionsParser();
+	virtual ~OptionsParser();
 	OptionsParser(OptionsParser& p) = delete;
 	OptionsParser(OptionsParser&& p) = delete;
 	void operator=(OptionsParser& p) = delete;

include/ipfixprobe/packet.hpp

@@ -37,6 +37,7 @@
 #include <stdint.h>
 #include <stdlib.h>
 #include <sys/time.h>
+#include <ipfixprobe/cttmeta.hpp>
 
 namespace ipxp {
 
@@ -100,7 +101,10 @@ struct Packet : public Record {
 	uint16_t buffer_size; /**< Size of buffer */
 
 	bool source_pkt; /**< Direction of packet from flow point of view */
-
+
+	std::optional<CttMetadata> cttmeta; /**< Metadata from CTT */
+	bool external_export; /**< True if packet payload is ctt export data */
+
 	/**
 	 * \brief Constructor.
 	 */
@@ -142,6 +146,7 @@ struct Packet : public Record {
 		, buffer(nullptr)
 		, buffer_size(0)
 		, source_pkt(true)
+      	, external_export(false)
 	{
 	}
 };