ipfixprobe - refactor PluginManager

Author: SiskaPavel

src/core/CMakeLists.txt

@@ -3,8 +3,6 @@ add_executable(ipfixprobe
 	ipfixprobe.cpp
 	ipfixprobe.hpp
 	options.cpp
-	pluginmgr.cpp
-	pluginmgr.hpp
 	ring.c
 	stacktrace.cpp
 	stacktrace.hpp
@@ -14,6 +12,8 @@ add_executable(ipfixprobe
 	workers.cpp
 	workers.hpp
 	inputPlugin.cpp
+	pluginManager.cpp
+	pluginManager.hpp
 )
 
 

src/core/ipfixprobe.cpp

@@ -26,9 +26,8 @@
  *
  */
 
-#include "ipfixprobe.hpp"
-
 #include "buildConfig.hpp"
+#include "ipfixprobe.hpp"
 
 #include <fstream>
 #include <future>
@@ -47,6 +46,8 @@
 #endif
 #include "stats.hpp"
 
+#include <ipfixprobe/pluginFactory/pluginFactory.hpp>
+
 namespace ipxp {
 
 volatile sig_atomic_t stop = 0;
@@ -90,56 +91,38 @@ void error(std::string msg)
 	std::cerr << "Error: " << msg << std::endl;
 }
 
-template<typename T>
-static void print_plugins_help(std::vector<Plugin*>& plugins)
+static void printPluginsUsage(const std::vector<PluginManifest>& pluginsManifest)
 {
-	for (auto& it : plugins) {
-		if (dynamic_cast<T*>(it)) {
-			OptionsParser* parser = it->get_parser();
-			parser->usage(std::cout);
-			std::cout << std::endl;
-			delete parser;
+	for (const auto& pluginManifest : pluginsManifest) {
+		if (pluginManifest.usage) {
+			pluginManifest.usage();
 		}
 	}
 }
 
 void print_help(ipxp_conf_t& conf, const std::string& arg)
 {
-	auto deleter = [&](std::vector<Plugin*>* p) {
-		for (auto& it : *p) {
-			delete it;
-		}
-		delete p;
-	};
-	auto plugins = std::unique_ptr<std::vector<Plugin*>, decltype(deleter)>(
-		new std::vector<Plugin*>(conf.mgr.get()),
-		deleter);
-
 	if (arg == "input") {
-		print_plugins_help<InputPlugin>(*plugins);
-	} else if (arg == "storage") {
-		print_plugins_help<StoragePlugin>(*plugins);
-	} else if (arg == "output") {
-		print_plugins_help<OutputPlugin>(*plugins);
-	} else if (arg == "process") {
-		print_plugins_help<ProcessPlugin>(*plugins);
-	} else {
-		Plugin* p;
-		try {
-			p = conf.mgr.get(arg);
-			if (p == nullptr) {
-				std::cout << "No help available for " << arg << std::endl;
-				return;
-			}
-		} catch (PluginManagerError& e) {
-			error(std::string("when loading plugin: ") + e.what());
-			return;
-		}
-		OptionsParser* parser = p->get_parser();
-		parser->usage(std::cout);
-		delete parser;
-		delete p;
+		auto& inputPluginFactory = InputPluginFactory::getInstance();
+		return printPluginsUsage(inputPluginFactory.getRegisteredPlugins());
+	}
+
+	if (arg == "storage") {
+		auto& storagePluginFactory = StoragePluginFactory::getInstance();
+		return printPluginsUsage(storagePluginFactory.getRegisteredPlugins());
 	}
+
+	if (arg == "output") {
+		auto& outputPluginFactory = OutputPluginFactory::getInstance();
+		return printPluginsUsage(outputPluginFactory.getRegisteredPlugins());
+	}
+
+	if (arg == "process") {
+		auto& processPluginFactory = ProcessPluginFactory::getInstance();
+		return printPluginsUsage(processPluginFactory.getRegisteredPlugins());
+	}
+
+	std::cerr << "No help available for " << arg << std::endl;
 }
 
 void process_plugin_argline(
@@ -262,6 +245,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 		if (process_name == BASIC_PLUGIN_NAME) {
 			continue;
 		}
+		/*
 		try {
 			process_plugin = dynamic_cast<ProcessPlugin*>(conf.mgr.get(process_name));
 			if (process_plugin == nullptr) {
@@ -279,6 +263,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 		} catch (PluginManagerError& e) {
 			throw IPXPError(process_name + std::string(": ") + e.what());
 		}
+		*/
 	}
 
 	// telemetry
@@ -297,6 +282,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 	conf.holder.add(statsFile);
 
 	OutputPlugin* output_plugin = nullptr;
+	/*
 	try {
 		output_plugin = dynamic_cast<OutputPlugin*>(conf.mgr.get(output_name));
 		if (output_plugin == nullptr) {
@@ -318,6 +304,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 	} catch (PluginManagerError& e) {
 		throw IPXPError(output_name + std::string(": ") + e.what());
 	}
+	*/
 
 	{
 		std::promise<WorkerResult>* output_res = new std::promise<WorkerResult>();
@@ -360,6 +347,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 		auto pipeline_queue_dir
 			= pipeline_dir->addDir("queues")->addDir(std::to_string(pipeline_idx));
 
+		/*
 		try {
 			input_plugin = dynamic_cast<InputPlugin*>(conf.mgr.get(input_name));
 			if (input_plugin == nullptr) {
@@ -398,6 +386,7 @@ bool process_plugin_args(ipxp_conf_t& conf, IpfixprobeOptParser& parser)
 		} catch (PluginManagerError& e) {
 			throw IPXPError(storage_name + std::string(": ") + e.what());
 		}
+		*/
 
 		std::vector<ProcessPlugin*> storage_process_plugins;
 		for (auto& it : *process_plugins) {
@@ -638,6 +627,7 @@ int run(int argc, char* argv[])
 	IpfixprobeOptParser parser;
 	ipxp_conf_t conf;
 	int status = EXIT_SUCCESS;
+	const bool loadPluginsRecursive = true;
 
 	register_handlers();
 
@@ -649,6 +639,8 @@ int run(int argc, char* argv[])
 		goto EXIT;
 	}
 
+	conf.pluginManager.loadPlugins("/usr/local/lib64/ipfixprobe/", loadPluginsRecursive);
+
 	if (parser.m_help) {
 		if (parser.m_help_str.empty()) {
 			parser.usage(std::cout, 0, IPXP_APP_NAME);

src/core/ipfixprobe.hpp

@@ -29,7 +29,7 @@
 #ifndef IPXP_IPFIXPROBE_HPP
 #define IPXP_IPFIXPROBE_HPP
 
-#include "pluginmgr.hpp"
+#include "pluginManager.hpp"
 #include "workers.hpp"
 
 #include <atomic>
@@ -296,7 +296,7 @@ struct ipxp_conf_t {
 	uint32_t fps;
 	uint32_t max_pkts;
 
-	PluginManager mgr;
+	PluginManager pluginManager;
 	struct Plugins {
 		std::vector<InputPlugin*> input;
 		std::vector<StoragePlugin*> storage;
@@ -333,6 +333,7 @@ struct ipxp_conf_t {
 		, worker_cnt(0)
 		, fps(0)
 		, max_pkts(0)
+		, pluginManager(false)
 		, pkt_bufsize(1600)
 		, blocks_cnt(0)
 		, pkts_cnt(0)

src/core/pluginManager.cpp

@@ -0,0 +1,110 @@
+/**
+ * @file
+ * @brief Implementation of PluginManager class for loading and unloading plugins.
+ * @author Pavel Siska <[email protected]>
+ * @date 2025
+ *
+ * Copyright (c) 2025 CESNET
+ *
+ * SPDX-License-Identifier: BSD-3-Clause
+ */
+
+#include "pluginManager.hpp"
+
+#include <filesystem>
+#include <iostream>
+#include <stdexcept>
+#include <variant>
+
+#include <dlfcn.h>
+
+namespace ipxp {
+
+namespace fs = std::filesystem;
+
+static void clearErrorMessage()
+{
+	dlerror();
+}
+
+static void* openSharedObject(const std::string& path)
+{
+	const int dlFlags = RTLD_LAZY | RTLD_LOCAL;
+	return dlopen(path.c_str(), dlFlags);
+}
+
+template<typename Iterator>
+static bool isValidPlugin(const Iterator& entry)
+{
+	return entry.is_regular_file() && entry.path().extension() == ".so";
+}
+
+static void loadPluginsRecursive(const std::string& dirPath, PluginManager& pluginManager)
+{
+	for (const auto& entry : fs::recursive_directory_iterator(dirPath)) {
+		if (isValidPlugin(entry)) {
+			pluginManager.loadPlugin(entry.path().string());
+		}
+	}
+}
+
+static void loadPluginsNonRecursive(const std::string& dirPath, PluginManager& pluginManager)
+{
+	for (const auto& entry : fs::directory_iterator(dirPath)) {
+		if (isValidPlugin(entry)) {
+			pluginManager.loadPlugin(entry.path().string());
+		}
+	}
+}
+
+PluginManager::PluginManager(bool unloadAtExit)
+	: m_unloadAtExit(unloadAtExit)
+{
+}
+
+void PluginManager::loadPlugins(const std::string& dirPath, bool recursive)
+{
+	const auto loadPluginsFunction = recursive ? loadPluginsRecursive : loadPluginsNonRecursive;
+
+	try {
+		loadPluginsFunction(dirPath, *this);
+	} catch (const fs::filesystem_error& ex) {
+		std::cerr << "Error accessing directory '" << dirPath << "': " << ex.what() << std::endl;
+		// m_logger->error("Error accessing directory '{}': {}", dirPath, ex.what());
+		throw std::runtime_error("PluginManager::loadPlugins() has failed.");
+	}
+}
+
+void PluginManager::loadPlugin(const std::string& pluginPath)
+{
+	clearErrorMessage();
+	void* pluginHandle = openSharedObject(pluginPath);
+
+	if (!pluginHandle) {
+		std::cerr << "Error loading plugin '" << pluginPath << "': " << dlerror() << std::endl;
+		// m_logger->error(std::string(dlerror()));
+		throw std::runtime_error("PluginManager::loadPlugin() has failed.");
+	}
+
+	std::cerr << "Plugin '" << pluginPath << "' loaded." << std::endl;
+	// m_logger->info("Plugin '{}' loaded.", pluginPath);
+
+	m_pluginHandles.emplace_back(pluginHandle);
+}
+
+PluginManager::~PluginManager()
+{
+	if (m_unloadAtExit) {
+		unloadPlugins();
+	}
+}
+
+void PluginManager::unloadPlugins()
+{
+	for (auto handle : m_pluginHandles) {
+		dlclose(handle);
+	}
+	m_pluginHandles.clear();
+}
+
+} // namespace ipxp