Merge pull request #276 from CESNET/top-ports

Top 10 ports

Author: SiskaPavel

include/ipfixprobe/inputPlugin.hpp

@@ -96,7 +96,7 @@ class IPXP_API InputPlugin
 	};
 
 	/// Statistics related to packet parsing.
-	ParserStats m_parser_stats = {};
+	ParserStats m_parser_stats {10};
 
 private:
 	void create_parser_stats_telemetry(

include/ipfixprobe/parser-stats.hpp

@@ -25,6 +25,8 @@
 
 #pragma once
 
+#include "../../src/plugins/input/parser/topPorts.hpp"
+
 #include <array>
 #include <cstdint>
 #include <string>
@@ -162,6 +164,23 @@ struct VlanStats {
  * \brief Structure for storing parser statistics.
  */
 struct ParserStats {
+	ParserStats(size_t top_ports_count)
+		: top_ports(top_ports_count)
+		, mpls_packets(0)
+		, vlan_packets(0)
+		, pppoe_packets(0)
+		, trill_packets(0)
+		, ipv4_packets(0)
+		, ipv6_packets(0)
+		, tcp_packets(0)
+		, udp_packets(0)
+		, seen_packets(0)
+		, unknown_packets(0)
+	{
+	}
+
+	TopPorts top_ports;
+
 	uint64_t mpls_packets;
 	uint64_t vlan_packets;
 	uint64_t pppoe_packets;

src/core/CMakeLists.txt

@@ -32,6 +32,7 @@ target_link_libraries(ipfixprobe-core
 	atomic::atomic
 	unwind::unwind
 	${CMAKE_DL_LIBS}
+	top-ports
 )
 
 add_executable(ipfixprobe main.cpp)

src/core/inputPlugin.cpp

@@ -13,6 +13,8 @@
  * SPDX-License-Identifier: BSD-3-Clause
  */
 
+#include <numeric>
+
 #include <ipfixprobe/inputPlugin.hpp>
 
 namespace ipxp {
@@ -35,6 +37,19 @@ static telemetry::Content get_parser_stats_content(const ParserStats& parserStat
 
 	dict["seen_packets"] = parserStats.seen_packets;
 	dict["unknown_packets"] = parserStats.unknown_packets;
+	const std::vector<TopPorts::PortStats>& ports = parserStats.top_ports.get_top_ports();
+	if (ports.empty()) {
+		dict["top_10_ports"] = "";
+	} else {
+		std::string top_ports = ports[0].to_string();
+		dict["top_10_ports"] = std::accumulate(
+			ports.begin() + 1,
+			ports.end(),
+			top_ports,
+			[](std::string acc, const TopPorts::PortStats& port_frequency) {
+				return acc + ", " + port_frequency.to_string();
+			});
+	}
 	return dict;
 }
 

src/plugins/input/CMakeLists.txt

@@ -1,4 +1,5 @@
 add_subdirectory(raw)
+add_subdirectory(parser)
 
 if (ENABLE_INPUT_PCAP)
 	add_subdirectory(pcap)

src/plugins/input/parser/CMakeLists.txt

@@ -0,0 +1,11 @@
+add_library(top-ports STATIC
+	topPorts.cpp
+	topPorts.hpp
+)
+
+target_include_directories(top-ports PUBLIC
+	${CMAKE_SOURCE_DIR}/include
+	${CMAKE_BINARY_DIR}/src
+)
+
+target_compile_options(top-ports PRIVATE -fPIC)

src/plugins/input/parser/parser.cpp

@@ -465,7 +465,8 @@ 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(const u_char* data_ptr, uint16_t data_len, Packet* pkt, ParserStats& stats)
 {
 	struct tcphdr* tcp = (struct tcphdr*) data_ptr;
 	if (sizeof(struct tcphdr) > data_len) {
@@ -479,6 +480,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.increment_tcp_frequency(pkt->src_port);
+	stats.top_ports.increment_tcp_frequency(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));
@@ -544,7 +548,8 @@ 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(const u_char* data_ptr, uint16_t data_len, Packet* pkt, ParserStats& stats)
 {
 	struct udphdr* udp = (struct udphdr*) data_ptr;
 	if (sizeof(struct udphdr) > data_len) {
@@ -554,6 +559,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.increment_udp_frequency(pkt->src_port);
+	stats.top_ports.increment_udp_frequency(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));
@@ -749,10 +757,10 @@ void parse_packet(
 
 		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(data + data_offset, caplen - data_offset, pkt, stats);
 			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(data + data_offset, caplen - data_offset, pkt, stats);
 			stats.udp_packets++;
 		}
 	} catch (const char* err) {

src/plugins/input/parser/topPorts.cpp

@@ -0,0 +1,75 @@
+/**
+ * \file topPorts.cpp
+ * \brief TopPorts class implementation.
+ * \author Damir Zainullin <[email protected]>
+ * \date 2024
+ */
+
+#include "topPorts.hpp"
+
+#include <algorithm>
+#include <array>
+#include <cstdint>
+#include <functional>
+#include <limits>
+#include <string>
+#include <vector>
+
+namespace ipxp {
+
+TopPorts::TopPorts(size_t top_ports_count) noexcept
+	: m_top_ports_count(top_ports_count)
+{
+}
+
+std::string TopPorts::PortStats::to_string() const noexcept
+{
+	return std::to_string(port) + "[" + (protocol == Protocol::TCP ? "TCP" : "UDP") + "] - "
+		+ std::to_string(frequency);
+}
+
+bool update_port_buffer(
+	std::vector<TopPorts::PortStats>& port_buffer,
+	TopPorts::PortStats port_stats) noexcept
+{
+	auto port_pos = std::lower_bound(
+		port_buffer.begin(),
+		port_buffer.end(),
+		port_stats.frequency,
+		[](const TopPorts::PortStats& port_frequency, size_t count) {
+			return port_frequency.frequency >= count;
+		});
+
+	if (port_pos != port_buffer.end()) {
+		std::copy_backward(port_pos, std::prev(port_buffer.end()), port_buffer.end());
+		*port_pos = port_stats;
+		return true;
+	}
+	return false;
+};
+
+std::vector<TopPorts::PortStats> TopPorts::get_top_ports() const noexcept
+{
+	std::vector<PortStats> port_buffer(m_top_ports_count);
+	size_t ports_inserted = 0;
+
+	std::for_each(
+		m_tcp_port_frequencies.begin(),
+		m_tcp_port_frequencies.end(),
+		[&, port = uint16_t {0}](size_t frequency) mutable {
+			ports_inserted
+				+= update_port_buffer(port_buffer, {port++, frequency, PortStats::Protocol::TCP});
+		});
+	std::for_each(
+		m_udp_port_frequencies.begin(),
+		m_udp_port_frequencies.end(),
+		[&, port = uint16_t {0}](size_t frequency) mutable {
+			ports_inserted
+				+= update_port_buffer(port_buffer, {port++, frequency, PortStats::Protocol::UDP});
+		});
+
+	port_buffer.resize(std::min(m_top_ports_count, ports_inserted));
+	return port_buffer;
+}
+
+} // namespace ipxp

src/plugins/input/parser/topPorts.hpp

@@ -0,0 +1,71 @@
+/**
+ * \file topPorts.hpp
+ * \brief TopPorts class declaration implementing the most popular ports.
+ * \author Damir Zainullin <[email protected]>
+ * \date 2024
+ */
+#pragma once
+
+#include <array>
+#include <cstdint>
+#include <limits>
+#include <string>
+#include <vector>
+
+namespace ipxp {
+/**
+ * \brief Top ports counter.
+ */
+class TopPorts {
+public:
+	/**
+	 * \brief Constructor.
+	 * \param top_ports_count Number of the most popular ports to track.
+	 */
+	TopPorts(size_t top_ports_count) noexcept;
+
+	/**
+	 * \brief Increments number of times given tcp port has been seen.
+	 * \param port Port to increment its frequency.
+	 */
+	void increment_tcp_frequency(uint16_t port) noexcept { m_tcp_port_frequencies[port]++; }
+
+	/**
+	 * \brief Increments number of times given udp port has been seen.
+	 * \param port Port to increment its frequency.
+	 */
+	void increment_udp_frequency(uint16_t port) noexcept { m_udp_port_frequencies[port]++; }
+
+	/**
+	 * \brief Port frequency and protocol to which it belongs.
+	 */
+	struct PortStats {
+		/**
+		 * \brief Protocol type.
+		 */
+		enum class Protocol { TCP, UDP };
+
+		uint16_t port; /**< Port number. */
+		size_t frequency; /**< Number of times the port has been seen. */
+		Protocol protocol; /**< Protocol to which the port belongs. */
+
+		/**
+		 * \brief Convert the port stats to string.
+		 * \return String representation of the port stats.
+		 */
+		std::string to_string() const noexcept;
+	};
+
+	/**
+	 * \brief Get the top ports.
+	 * \return Vector of the most popular ports.
+	 */
+	std::vector<TopPorts::PortStats> get_top_ports() const noexcept;
+
+private:
+	std::array<std::size_t, std::numeric_limits<uint16_t>::max() + 1> m_tcp_port_frequencies {};
+	std::array<std::size_t, std::numeric_limits<uint16_t>::max() + 1> m_udp_port_frequencies {};
+	const size_t m_top_ports_count;
+};
+
+} // namespace ipxp