++

Author: Zainullin Damir

include/ipfixprobe/outputPlugin/outputPlugin.hpp

@@ -31,10 +31,13 @@ class IPXP_API OutputPlugin {
 
 	virtual ~OutputPlugin() = default;
 
+	constexpr static std::size_t DEFAULT_EXPORTER_ID = 1;
 protected:
+	
 	const FieldManager& m_fieldManager;
 	const std::vector<ProcessPluginEntry>& m_plugins;
 	std::size_t m_dropped = 0;
+	std::size_t m_seen = 0;
 };
 
 template<typename Base, typename... Args>

include/ipfixprobe/pluginFactory/pluginFactory.hpp

@@ -36,7 +36,7 @@ namespace ipxp {
  * @tparam Args The types of arguments that will be passed to the plugin constructors.
  */
 template<typename Base, typename... Args>
-class IPXP_API PluginFactory {
+class PluginFactory {
 public:
 	/**
 	 * @brief Retrieves the singleton instance of `PluginFactory`.

include/ipfixprobe/processPlugin/flowRecord.hpp

@@ -63,6 +63,7 @@ class FlowRecord {
 	Timestamp timeLastUpdate;
 
 	FlowKey flowKey;
+	DirectionalField<MACAddress> macAddress;
 
 	DirectionalField<DirectionalData> directionalData;
 
@@ -113,19 +114,19 @@ class FlowRecord {
 		m_flow.dst_tcp_flags = 0;*/
 	}
 
-	inline __attribute__((always_inline)) bool is_empty() const
+	constexpr bool isEmpty() const noexcept
 	{
 		return hash == 0;
 	}
 
-	inline __attribute__((always_inline)) bool belongs(uint64_t value) const
+	/*constexpr bool belongs(uint64_t value) const noexcept
 	{
 		return hash == value;
-	}
+	}*/
 
-	void create(const Packet& packet, uint64_t hash)
+	void createFrom(const Packet& packet)
 	{
-		/*m_flow.src_packets = 1;
+		m_flow.directionalData[Direction::Forward].packets = 1;
 
 		m_hash = hash;
 

include/ipfixprobe/processPlugin/macAddress.hpp

@@ -1,10 +1,27 @@
 #pragma once
 
 #include <iostream>
+#include <array>
+#include <span>
+#include <cstddef>
+#include <cstdint>
 
 namespace ipxp {
 
-struct MACAddress {};
+struct MACAddress {
+	std::array<std::byte, 6> address;
+
+	constexpr MACAddress() noexcept
+	: address{}
+	{
+	}
+
+	constexpr MACAddress(std::span<const std::byte, 6> address) noexcept
+	: address{}
+	{
+		std::ranges::copy(address, this->address.begin());
+	}
+};
 
 inline std::ostream& operator<<(std::ostream& os, const MACAddress& mac)
 {

include/ipfixprobe/processPlugin/processPluginManager.hpp

@@ -142,7 +142,7 @@ class ProcessPluginManager {
 		}
 	}
 
-	void exportFlowRecord(FlowRecord & flowRecord)
+	void exportFlowRecord(FlowRecord& flowRecord)
 	{
 		for (std::size_t pluginID = 0; pluginID < m_processPlugins.size(); pluginID++) {
 			const auto& pluginEntry = m_processPlugins[pluginID];

include/ipfixprobe/processPlugin/timestamp.hpp

@@ -32,6 +32,15 @@ struct Timestamp {
 		return ns / 1'000'000'000;
 	}
 
+	constexpr timeval toTimeval() const noexcept
+	{
+		// TODO use chrono
+		timeval tv;
+		tv.tv_sec = static_cast<time_t>(ns / 1'000'000'000);
+		tv.tv_usec = static_cast<suseconds_t>((ns % 1'000'000'000) / 1'000);
+		return tv;
+	}
+
 	constexpr bool operator<(const Timestamp& other) const noexcept
 	{
 		return ns < other.ns;

src/plugins/output/ipfix/src/ipfix.hpp

@@ -44,7 +44,6 @@
 #define RECONNECT_TIMEOUT 60
 #define TEMPLATE_REFRESH_TIME 600
 #define TEMPLATE_REFRESH_PACKETS 0
-#define DEFAULT_EXPORTER_ID 1
 
 namespace ipxp {
 
@@ -69,7 +68,7 @@ class IpfixOptParser : public OptionsParser {
 		, m_mtu(DEFAULT_MTU)
 		, m_udp(false)
 		, m_non_blocking_tcp(false)
-		, m_id(DEFAULT_EXPORTER_ID)
+		, m_id(OutputPlugin::DEFAULT_EXPORTER_ID)
 		, m_dir(0)
 		, m_template_refresh_time(TEMPLATE_REFRESH_TIME)
 		, m_verbose(false)

src/plugins/output/text/src/textOutputPlugin.cpp

@@ -49,35 +49,25 @@ printVectorVariant(const FieldDescriptor& field, const VectorValueGetter& varian
 
 void TextOutputPlugin::processRecord(FlowRecordUniquePtr& flowRecord)
 {
-	auto outputFields = m_fieldManager.getBiflowFields();
-	auto fn = [&](ProcessPlugin* processPlugin) {
-		// const void* pluginExportData = processPlugin->getExportData();
-		const void* pluginExportData = flowRecord->getPluginContext(0);
-
-		for (const auto& outputField : outputFields) {
-			if (!outputField.isInRecord(*flowRecord.get())) {
-				continue;
-			}
-
-			const auto& getter = outputField.getValueGetter();
-
-			std::visit(
-				[&](const auto& variant) {
-					using GetterT = std::decay_t<decltype(variant)>;
-					if constexpr (std::is_same_v<GetterT, ScalarValueGetter>) {
-						printScalarVariant(outputField, variant, pluginExportData);
-					} else if constexpr (std::is_same_v<GetterT, VectorValueGetter>) {
-						printVectorVariant(outputField, variant, pluginExportData);
-					}
-				},
-				getter);
+	std::ranges::for_each(m_fieldManager.getBiflowFields(), [&](const FieldDescriptor& fieldDescriptor) {
+		if (!fieldDescriptor.isInRecord(*flowRecord.get())) {
+			return;
 		}
-	};
 
-	for (const auto& pluginEntry : m_plugins) {
-		std::cout << "Processing " << pluginEntry.name << "\n";
-		fn(reinterpret_cast<ProcessPlugin*>(pluginEntry.plugin.get()));
-	}
+		const void* pluginExportData = flowRecord->getPluginContext(fieldDescriptor.getBitIndex());
+
+		const auto& getter = fieldDescriptor.getValueGetter();
+		std::visit(
+			[&](const auto& variant) {
+				using GetterT = std::decay_t<decltype(variant)>;
+				if constexpr (std::is_same_v<GetterT, ScalarValueGetter>) {
+					printScalarVariant(fieldDescriptor, variant, pluginExportData);
+				} else if constexpr (std::is_same_v<GetterT, VectorValueGetter>) {
+					printVectorVariant(fieldDescriptor, variant, pluginExportData);
+				}
+			},
+			getter);
+	});
 }
 
 } // namespace ipxp

src/plugins/output/text/src/textOutputPlugin.hpp

@@ -10,12 +10,10 @@ class TextOutputPlugin : public OutputPlugin {
 	TextOutputPlugin(const std::string& params, const FieldManager& manager, const std::vector<ProcessPluginEntry>& plugins)
 		: OutputPlugin(manager, plugins)
 	{
-		// TODO parse params
-		(void)params;
 	}
 
 	void processRecord(FlowRecordUniquePtr& flowRecord) override;
-
+	
 };
 
 } // namespace ipxp

src/plugins/output/unirec/CMakeLists.txt

@@ -12,6 +12,8 @@ set_target_properties(ipfixprobe-output-unirec PROPERTIES
 
 target_include_directories(ipfixprobe-output-unirec PRIVATE
 	${CMAKE_SOURCE_DIR}/include/
+	${CMAKE_SOURCE_DIR}/include/ipfixprobe/outputPlugin
+	${CMAKE_SOURCE_DIR}/include/ipfixprobe/processPlugin
 )
 
 target_link_libraries(ipfixprobe-output-unirec PRIVATE