Process plugins - Update ipfixprobe core

Author: Zainullin Damir

src/core/CMakeLists.txt

@@ -13,16 +13,28 @@ add_library(ipfixprobe-core STATIC
 	inputPlugin.cpp
 	pluginManager.cpp
 	pluginManager.hpp
+	fieldManager.cpp
+	xxhash.c
 )
 
-
 target_include_directories(ipfixprobe-core PUBLIC
 	${CMAKE_SOURCE_DIR}/include
+	${CMAKE_SOURCE_DIR}/include/ipfixprobe/processPlugin
+	${CMAKE_SOURCE_DIR}/include/ipfixprobe/outputPlugin
 	${CMAKE_BINARY_DIR}/src
+	${adaptmon_SOURCE_DIR}/lib/include/public/
 )
 
 target_compile_options(ipfixprobe-core PRIVATE -fPIC)
 
+#add_library(ipfixprobe-plugin-factory STATIC
+#	pluginFactory.cpp
+#)
+
+#target_include_directories(ipfixprobe-plugin-factory PUBLIC
+#	${CMAKE_SOURCE_DIR}/include
+#)
+
 set(CORE_LIB -Wl,--whole-archive ipfixprobe-core -Wl,--no-whole-archive)
 
 target_link_libraries(ipfixprobe-core

src/core/fieldManager.cpp

@@ -0,0 +1,141 @@
+/**
+ * @file
+ * @author Pavel Siska <[email protected]>
+ * @brief Implementation of FieldManager methods for registering and accessing fields.
+ *
+ * @copyright Copyright (c) 2025 CESNET, z.s.p.o.
+ */
+
+#include "fieldManager.hpp"
+
+#include "fieldGroup.hpp"
+
+namespace ipxp::process {
+
+void FieldManager::addField(
+	std::vector<FieldDescriptor>& container,
+	std::string_view group,
+	std::string_view name,
+	std::size_t bitIndex,
+	GenericValueGetter getter)
+{
+	container.emplace_back(FieldDescriptor(
+		FieldInfo {
+			.group = std::string(group),
+			.name = std::string(name),
+			.bitIndex = bitIndex,
+			.getter = std::move(getter),
+		}));
+}
+
+[[nodiscard]] FieldGroup FieldManager::createFieldGroup(std::string_view groupName)
+{
+	return FieldGroup(groupName, *this);
+}
+
+const std::vector<FieldDescriptor>& FieldManager::getBiflowFields() const noexcept
+{
+	return m_biflowFields;
+}
+
+const std::vector<FieldDescriptor>& FieldManager::getReverseBiflowFields() const noexcept
+{
+	return m_reverseBiflowFields;
+}
+
+const std::vector<FieldDescriptor>& FieldManager::getUniflowForwardFields() const noexcept
+{
+	return m_uniflowForwardFields;
+}
+
+const std::vector<FieldDescriptor>& FieldManager::getUniflowReverseFields() const noexcept
+{
+	return m_uniflowReverseFields;
+}
+
+[[nodiscard]] FieldHandler FieldManager::registerField(
+	std::string_view groupName,
+	std::string_view fieldName,
+	GenericValueGetter getter)
+{
+	const auto bitIndex = getNextBitIndex();
+	const FieldHandler fieldHandler(bitIndex);
+
+	// biflow
+	addField(m_biflowFields, groupName, fieldName, bitIndex, getter);
+
+	// reverse biflow
+	addField(m_reverseBiflowFields, groupName, fieldName, bitIndex, getter);
+
+	// forward uniflow
+	addField(m_uniflowForwardFields, groupName, fieldName, bitIndex, getter);
+
+	// reverse uniflow
+	addField(m_uniflowReverseFields, groupName, fieldName, bitIndex, getter);
+
+	return fieldHandler;
+}
+
+[[nodiscard]] std::pair<FieldHandler, FieldHandler> FieldManager::registerDirectionalPairFields(
+	std::string_view groupName,
+	std::string_view forwardFieldName,
+	std::string_view reverseFieldName,
+	GenericValueGetter forwardGetter,
+	GenericValueGetter reverseGetter)
+{
+	const auto forwardBitIndex = getNextBitIndex();
+	const auto reverseBitIndex = getNextBitIndex();
+
+	const FieldHandler forwardFieldHandler(forwardBitIndex);
+	const FieldHandler reverseFieldHandler(reverseBitIndex);
+
+	// biflow
+	addField(m_biflowFields, groupName, forwardFieldName, forwardBitIndex, forwardGetter);
+	addField(m_biflowFields, groupName, reverseFieldName, reverseBitIndex, reverseGetter);
+
+	// reverse biflow
+	addField(m_reverseBiflowFields, groupName, forwardFieldName, reverseBitIndex, reverseGetter);
+	addField(m_reverseBiflowFields, groupName, reverseFieldName, forwardBitIndex, forwardGetter);
+
+	// forward uniflow
+	addField(m_uniflowForwardFields, groupName, forwardFieldName, forwardBitIndex, forwardGetter);
+
+	// reverse uniflow
+	addField(m_uniflowReverseFields, groupName, forwardFieldName, reverseBitIndex, reverseGetter);
+
+	return {forwardFieldHandler, reverseFieldHandler};
+}
+
+[[nodiscard]] std::pair<FieldHandler, FieldHandler> FieldManager::registerBiflowPairFields(
+	std::string_view groupName,
+	std::string_view aFieldName,
+	std::string_view bFieldName,
+	GenericValueGetter aGetter,
+	GenericValueGetter bGetter)
+{
+	const std::size_t aBitIndex = getNextBitIndex();
+	const std::size_t bBitIndex = getNextBitIndex();
+
+	const FieldHandler aFieldHandler(aBitIndex);
+	const FieldHandler bFieldHandler(bBitIndex);
+
+	// biflow
+	addField(m_biflowFields, groupName, aFieldName, aBitIndex, aGetter);
+	addField(m_biflowFields, groupName, bFieldName, bBitIndex, bGetter);
+
+	// reverse biflow
+	addField(m_reverseBiflowFields, groupName, aFieldName, bBitIndex, bGetter);
+	addField(m_reverseBiflowFields, groupName, bFieldName, aBitIndex, aGetter);
+
+	// forward uniflow
+	addField(m_uniflowForwardFields, groupName, aFieldName, aBitIndex, aGetter);
+	addField(m_uniflowForwardFields, groupName, bFieldName, bBitIndex, bGetter);
+
+	// reverse uniflow
+	addField(m_uniflowReverseFields, groupName, aFieldName, bBitIndex, bGetter);
+	addField(m_uniflowReverseFields, groupName, bFieldName, aBitIndex, aGetter);
+
+	return {aFieldHandler, bFieldHandler};
+}
+
+} // namespace ipxp::process

src/core/ipfixprobe.cpp

@@ -45,6 +45,8 @@
 #include <poll.h>
 #include <signal.h>
 #include <unistd.h>
+#include <sys/types.h>
+#include <sys/socket.h>
 
 namespace ipxp {
 
@@ -250,7 +252,7 @@ void set_thread_details(pthread_t thread, const std::string& name, const std::ve
 
 bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 {
-	OutputPlugin::ProcessPlugins processPlugins;
+	//OutputPlugin::ProcessPlugins processPlugins;
 	std::string storage_name = "cache";
 	std::string storage_params = "";
 	std::string output_name = "ipfix";
@@ -276,7 +278,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 
 	// Process
 	for (auto& it : parser.m_process) {
-		std::shared_ptr<ProcessPlugin> processPlugin;
+		//std::shared_ptr<ProcessPlugin> processPlugin;
 		std::string process_params;
 		std::string process_name;
 		std::vector<int> affinity;
@@ -287,8 +289,8 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 				"inside "
 				"input threads)");
 		}
-		for (auto& it : processPlugins) {
-			std::string plugin_name = it.first;
+		for (auto& it : conf.manager.getEntries()) {
+			std::string plugin_name = it.name;
 			if (plugin_name == process_name) {
 				throw IPXPError(process_name + " plugin was specified multiple times");
 			}
@@ -298,14 +300,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 		}
 
 		try {
-			auto& processPluginFactory = ProcessPluginFactory::getInstance();
-			const int pluginID = ProcessPluginIDGenerator::instance().generatePluginID();
-			processPlugin
-				= processPluginFactory.createShared(process_name, process_params, pluginID);
-			if (processPlugin == nullptr) {
-				throw IPXPError("invalid process plugin " + process_name);
-			}
-			processPlugins.emplace_back(process_name, processPlugin);
+			conf.manager.addProcessPlugin(process_name, process_params);
 		} catch (PluginError& e) {
 			throw IPXPError(process_name + std::string(": ") + e.what());
 		} catch (PluginExit& e) {
@@ -334,7 +329,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 
 	try {
 		auto& outputPluginFactory = OutputPluginFactory::getInstance();
-		outputPlugin = outputPluginFactory.createShared(output_name, output_params, processPlugins);
+		outputPlugin = outputPluginFactory.createShared(output_name, output_params, conf.manager.getFieldManager(), conf.manager.getEntries());
 		if (outputPlugin == nullptr) {
 			throw IPXPError("invalid output plugin " + output_name);
 		}
@@ -412,7 +407,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 		try {
 			auto& storagePluginFactory = StoragePluginFactory::getInstance();
 			storagePlugin
-				= storagePluginFactory.createShared(storage_name, storage_params, output_queue);
+				= storagePluginFactory.createShared(storage_name, storage_params, output_queue, conf.manager);
 			if (storagePlugin == nullptr) {
 				throw IPXPError("invalid storage plugin " + storage_name);
 			}
@@ -427,11 +422,11 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 		}
 
 		std::vector<ProcessPlugin*> storage_process_plugins;
-		for (auto& it : processPlugins) {
-			ProcessPlugin* tmp = it.second->copy();
-			storagePlugin->add_plugin(tmp);
+		for (auto& it : conf.manager.getEntries()) {
+			ProcessPlugin* tmp = it.plugin.get();
+			//storagePlugin->add_plugin(tmp);
 			conf.active.process.push_back(tmp);
-			conf.active.all.push_back(tmp);
+			//conf.active.all.push_back(tmp);
 			storage_process_plugins.push_back(tmp);
 		}
 
@@ -479,7 +474,7 @@ void finish(ipxp_conf_t& conf)
 	// Terminate all storages
 	for (auto& it : conf.pipelines) {
 		for (auto& itp : it.storage.plugins) {
-			itp->close();
+			//itp->close(); TODO
 		}
 	}
 
@@ -596,7 +591,7 @@ void serve_stat_clients(ipxp_conf_t& conf, struct pollfd pfds[2])
 	}
 
 	if (pfds[0].revents & POLL_IN) {
-		int fd = accept(pfds[0].fd, NULL, NULL);
+		int fd = accept(pfds[-1].fd, NULL, NULL);
 		if (pfds[1].fd == -1) {
 			pfds[1].fd = fd;
 		} else if (fd != -1) {

src/core/ipfixprobe.hpp

@@ -43,10 +43,10 @@
 #include <appFs.hpp>
 #include <ipfixprobe/inputPlugin.hpp>
 #include <ipfixprobe/options.hpp>
-#include <ipfixprobe/outputPlugin.hpp>
+#include <outputPlugin.hpp>
 #include <ipfixprobe/packet.hpp>
 #include <ipfixprobe/plugin.hpp>
-#include <ipfixprobe/processPlugin.hpp>
+#include <processPlugin.hpp>
 #include <ipfixprobe/ring.h>
 #include <ipfixprobe/storagePlugin.hpp>
 #include <ipfixprobe/utils.hpp>
@@ -314,7 +314,8 @@ struct ipxp_conf_t {
 	std::vector<std::shared_ptr<InputPlugin>> inputPlugins;
 	std::vector<std::shared_ptr<StoragePlugin>> storagePlugins;
 	std::shared_ptr<OutputPlugin> outputPlugin;
-
+	FieldManager fieldManager;
+	ProcessPluginManager manager;
 	PluginManager pluginManager;
 	struct Plugins {
 		std::vector<InputPlugin*> input;
@@ -352,6 +353,7 @@ struct ipxp_conf_t {
 		, worker_cnt(0)
 		, fps(0)
 		, max_pkts(0)
+		, manager(fieldManager)
 		, pluginManager(false)
 		, pkt_bufsize(1600)
 		, blocks_cnt(0)

src/core/workers.cpp

@@ -173,25 +173,27 @@ void output_worker(
 	while (1) {
 		gettimeofday(&end, nullptr);
 
-		Flow* flow = static_cast<Flow*>(ipx_ring_pop(queue));
-		if (!flow) {
+		FlowRecordUniquePtr* flowPtr = static_cast<FlowRecordUniquePtr*>(ipx_ring_pop(queue));
+		if (!flowPtr) {
 			if (end.tv_sec - last_flush.tv_sec > 1) {
 				last_flush = end;
-				outputPlugin->flush();
+				//outputPlugin->flush(); TODO flush ? 
 			}
 			if (terminate_export && !ipx_ring_cnt(queue)) {
 				break;
 			}
 			continue;
 		}
 
+		FlowRecordUniquePtr& flow = *flowPtr;
+
 		stats.biflows++;
-		stats.bytes += flow->src_bytes + flow->dst_bytes;
-		stats.packets += flow->src_packets + flow->dst_packets;
-		stats.dropped = outputPlugin->m_flows_dropped;
+		stats.bytes += flow->directionalData[Direction::Forward].bytes + flow->directionalData[Direction::Reverse].bytes;
+		stats.packets += flow->directionalData[Direction::Forward].packets + flow->directionalData[Direction::Reverse].packets;
+		stats.dropped = outputPlugin->getDroppedCount();
 		out_stats->store(stats);
 		try {
-			outputPlugin->export_flow(*flow);
+			outputPlugin->processRecord(flow);
 		} catch (PluginError& e) {
 			res.error = true;
 			res.msg = e.what();
@@ -231,8 +233,8 @@ void output_worker(
 		}
 	}
 
-	outputPlugin->flush();
-	stats.dropped = outputPlugin->m_flows_dropped;
+	//outputPlugin->flush(); TODO flush ?
+	stats.dropped = outputPlugin->getDroppedCount();
 	out_stats->store(stats);
 	out->set_value(res);
 }

src/core/workers.hpp

@@ -35,9 +35,9 @@
 #include <future>
 
 #include <ipfixprobe/inputPlugin.hpp>
-#include <ipfixprobe/outputPlugin.hpp>
+#include <outputPlugin.hpp>
 #include <ipfixprobe/packet.hpp>
-#include <ipfixprobe/processPlugin.hpp>
+#include <processPlugin.hpp>
 #include <ipfixprobe/ring.h>
 #include <ipfixprobe/storagePlugin.hpp>