ndp ctt integration - parsing metadata from ndp pkt

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,27 +30,31 @@
 #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 {
 
 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 +78,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);
 };
 
 }