From b299d552eb117c21c5ccafea822d83931d5c95bb Mon Sep 17 00:00:00 2001
From: Damir Zainullin <zaidamilda@gmail.com>
Date: Mon, 2 Dec 2024 01:00:25 +0100
Subject: [PATCH] Top 10 ports - Adjust parser

---
 include/ipfixprobe/parser-stats.hpp | 16 ++++++++++++++++
 input/input.cpp                     | 15 +++++++++++++++
 input/parser.cpp                    | 18 ++++++++++++++----
 3 files changed, 45 insertions(+), 4 deletions(-)

diff --git a/include/ipfixprobe/parser-stats.hpp b/include/ipfixprobe/parser-stats.hpp
index 3c5c7777..efb29e64 100644
--- a/include/ipfixprobe/parser-stats.hpp
+++ b/include/ipfixprobe/parser-stats.hpp
@@ -25,7 +25,9 @@
 
 #pragma once
 
+#include "../../input/countminsketch.hpp"
 #include <cstdint>
+#include <array>
 
 namespace ipxp {
 
@@ -46,6 +48,20 @@ struct ParserStats {
 
    uint64_t seen_packets;
    uint64_t unknown_packets;
+
+   CountMinSketch<uint16_t, 2> top_ports{
+        {std::hash<uint16_t>{},
+           [](uint16_t port) {
+            constexpr size_t prime1 = 54059UL;
+            constexpr size_t prime2 = 76963UL;
+            constexpr size_t prime3 = 86969UL;
+            size_t res = 37;
+            res = (res * prime1) ^ ((port >> 8) * prime2);
+            res = (res * prime1) ^ (static_cast<uint8_t>(port) * prime2);
+            return res % prime3;
+        },
+        }
+   };
 };
 
 } // namespace ipxp
diff --git a/input/input.cpp b/input/input.cpp
index 105615bb..2b680786 100644
--- a/input/input.cpp
+++ b/input/input.cpp
@@ -24,6 +24,10 @@
  */
 
 #include <ipfixprobe/input.hpp>
+#include <iterator>
+#include <string>
+#include <sstream>
+#include <numeric>
 
 namespace ipxp {
 
@@ -52,6 +56,17 @@ static telemetry::Content get_parser_stats_content(const ParserStats& parserStat
    dict["seen_packets"] = parserStats.seen_packets;
    dict["unknown_packets"] = parserStats.unknown_packets;
 
+   const auto& [ports, size] = parserStats.top_ports.get_top_events();
+   if (size == 0) {
+      dict["top_10_ports"] = "";
+   } else {
+       std::string top_ports = std::to_string(ports[0].event) + ": " + std::to_string(ports[0].frequency);
+       dict["top_10_ports"] = std::accumulate(ports.begin() + 1, ports.begin() + size, top_ports,
+                       [](std::string& acc, const auto& portFrequency) {
+                           return acc + ", " + std::to_string(portFrequency.event) + ": " + std::to_string(portFrequency.frequency);
+                       });
+   }
+
    return dict;
 }
 
diff --git a/input/parser.cpp b/input/parser.cpp
index 61b696cb..cae4b72e 100644
--- a/input/parser.cpp
+++ b/input/parser.cpp
@@ -37,6 +37,10 @@
 #include "headers.hpp"
 #include <ipfixprobe/packet.hpp>
 
+#include <iterator>
+#include <string>
+#include <sstream>
+
 namespace ipxp {
 
 //#define DEBUG_PARSER
@@ -454,7 +458,7 @@ inline uint16_t parse_ipv6_hdr(const u_char *data_ptr, uint16_t data_len, Packet
  * \param [out] pkt Pointer to Packet structure where parsed fields will be stored.
  * \return Size of header in bytes.
  */
-inline uint16_t parse_tcp_hdr(const u_char *data_ptr, uint16_t data_len, Packet *pkt)
+inline uint16_t parse_tcp_hdr(ParserStats& stats, const u_char *data_ptr, uint16_t data_len, Packet *pkt)
 {
    struct tcphdr *tcp = (struct tcphdr *) data_ptr;
    if (sizeof(struct tcphdr) > data_len) {
@@ -469,6 +473,9 @@ inline uint16_t parse_tcp_hdr(const u_char *data_ptr, uint16_t data_len, Packet
    pkt->tcp_flags = (uint8_t) *(data_ptr + 13) & 0xFF;
    pkt->tcp_window = ntohs(tcp->window);
 
+   stats.top_ports.insert(pkt->src_port);
+   stats.top_ports.insert(pkt->dst_port);
+
    DEBUG_MSG("TCP header:\n");
    DEBUG_MSG("\tSrc port:\t%u\n",   ntohs(tcp->source));
    DEBUG_MSG("\tDest port:\t%u\n",  ntohs(tcp->dest));
@@ -529,7 +536,7 @@ inline uint16_t parse_tcp_hdr(const u_char *data_ptr, uint16_t data_len, Packet
  * \param [out] pkt Pointer to Packet structure where parsed fields will be stored.
  * \return Size of header in bytes.
  */
-inline uint16_t parse_udp_hdr(const u_char *data_ptr, uint16_t data_len, Packet *pkt)
+inline uint16_t parse_udp_hdr(ParserStats& stats, const u_char *data_ptr, uint16_t data_len, Packet *pkt)
 {
    struct udphdr *udp = (struct udphdr *) data_ptr;
    if (sizeof(struct udphdr) > data_len) {
@@ -539,6 +546,9 @@ inline uint16_t parse_udp_hdr(const u_char *data_ptr, uint16_t data_len, Packet
    pkt->src_port = ntohs(udp->source);
    pkt->dst_port = ntohs(udp->dest);
 
+   stats.top_ports.insert(pkt->src_port);
+   stats.top_ports.insert(pkt->dst_port);
+
    DEBUG_MSG("UDP header:\n");
    DEBUG_MSG("\tSrc port:\t%u\n",   ntohs(udp->source));
    DEBUG_MSG("\tDest port:\t%u\n",  ntohs(udp->dest));
@@ -727,10 +737,10 @@ void parse_packet(parser_opt_t *opt, ParserStats& stats, struct timeval ts, cons
 
       l4_hdr_offset = data_offset;
       if (pkt->ip_proto == IPPROTO_TCP) {
-         data_offset += parse_tcp_hdr(data + data_offset, caplen - data_offset, pkt);
+         data_offset += parse_tcp_hdr(stats, 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);
+         data_offset += parse_udp_hdr(stats, data + data_offset, caplen - data_offset, pkt);
          stats.udp_packets++;
       }
    } catch (const char *err) {