feat: Implement logging for ebpfdiscoverysrv (#19)

CMakeLists.txt

@@ -66,6 +66,9 @@ endif()
 if(NOT TARGET fmt)
 	find_package(fmt REQUIRED)
 endif(NOT TARGET fmt)
+if(NOT TARGET spdlog)
+	find_package(spdlog REQUIRED)
+endif(NOT TARGET spdlog)
 
 find_package(Protobuf REQUIRED)
 
@@ -101,11 +104,12 @@ set(LIBBPF_INCLUDE_DIRS ${CMAKE_BINARY_DIR}/thirdparty/libbpf)
 set(LIBBPF_LIBRARIES ${CMAKE_BINARY_DIR}/thirdparty/libbpf/libbpf.a)
 
 add_subdirectory(libbpfload)
-add_subdirectory(libebpfdiscoveryproto)
 add_subdirectory(libebpfdiscovery)
+add_subdirectory(libebpfdiscoveryproto)
 add_subdirectory(libebpfdiscoveryshared)
 add_subdirectory(libebpfdiscoveryskel)
 add_subdirectory(libhttpparser)
+add_subdirectory(liblogging)
 
 include(GNUInstallDirs)
 add_subdirectory(ebpfdiscoverysrv)

ebpfdiscoverysrv/CMakeLists.txt

@@ -5,4 +5,5 @@ add_executable(${TARGET} ${SOURCES})
 
 target_link_libraries(${TARGET} Boost::program_options)
 target_link_libraries(${TARGET} ebpfdiscovery)
+target_link_libraries(${TARGET} logging)
 target_compile_definitions(${TARGET} PUBLIC PROJECT_VERSION="${PROJECT_VERSION}")

ebpfdiscoverysrv/src/main.cpp

@@ -1,62 +1,186 @@
 // SPDX-License-Identifier: GPL-2.0
-
 #include "ebpfdiscovery/Discovery.h"
+#include "ebpfdiscovery/DiscoveryBpf.h"
+#include "ebpfdiscovery/DiscoveryBpfLoader.h"
+#include "logging/Logger.h"
 
-#include <bpf/libbpf.h>
+#include <boost/program_options.hpp>
 
+#include <condition_variable>
+#include <filesystem>
 #include <iostream>
 #include <memory>
 #include <signal.h>
+#include <sstream>
 #include <unistd.h>
 
-static ebpfdiscovery::Discovery discoveryInstance;
+namespace po = boost::program_options;
+using logging::Logger;
+using logging::LogLevel;
 
-std::atomic<bool> isShuttingDown;
+enum class ProgramStatus {
+	Running,
+	UnixShutdownSignalReceived,
+} programStatus;
 
-static void handleUnixExitSignal(int signo) {
-	if (isShuttingDown) {
-		return;
-	}
-	isShuttingDown = true;
+std::condition_variable programStatusCV;
+std::mutex programStatusMutex;
+
+/*
+ * CLI options
+ */
+
+static po::options_description getProgramOptions() {
+	po::options_description desc{"Options"};
+
+	// clang-format off
+	desc.add_options()
+      ("log-level", po::value<logging::LogLevel>()->default_value(logging::LogLevel::Err, "error"), "Set log level {trace,debug,info,warning,error,critical,off}")
+      ("help,h", "Display available options")
+      ("log-dir", po::value<std::filesystem::path>()->default_value(""), "Log files directory")
+      ("log-no-stdout", po::value<bool>()->default_value(false), "Disable logging to stdout")
+      ("version", "Display program version")
+  ;
+	// clang-format on
 
-	discoveryInstance.stopRun();
+	return desc;
 }
 
-void setupUnixSignalHandlers() {
-	struct sigaction action {};
-	action.sa_handler = handleUnixExitSignal;
-	action.sa_flags = 0;
-	sigaction(SIGINT, &action, nullptr);
-	sigaction(SIGTERM, &action, nullptr);
-	sigaction(SIGPIPE, &action, nullptr);
+/*
+ * Logging setup
+ */
+
+static void setupLogging(logging::LogLevel logLevel, bool enableStdout, const std::filesystem::path& logDir) {
+	Logger::getInstance().setup("eBPF-Discovery", enableStdout, logDir);
+	Logger::getInstance().setLevel(logLevel);
+	LOG_TRACE("Logging has been set up. (logDir: {})", logDir.string());
 }
 
+/*
+ * Unix signals setup
+ */
+
+static sigset_t getSigset() {
+	sigset_t sigset;
+	sigfillset(&sigset);
+	return sigset;
+}
+
+static void runUnixSignalHandlerLoop() {
+	while (true) {
+		sigset_t sigset{getSigset()};
+		LOG_TRACE("Waiting for unix signals.");
+		const auto signo{sigwaitinfo(&sigset, nullptr)};
+		if (signo == -1) {
+			LOG_CRITICAL("Failed to wait for unix signals: {}", std::strerror(errno));
+			std::abort();
+		}
+		LOG_DEBUG("Received unix signal. (signo: {})", signo);
+		if (signo == SIGINT || signo == SIGPIPE || signo == SIGTERM) {
+			std::lock_guard<std::mutex> lock(programStatusMutex);
+			programStatus = ProgramStatus::UnixShutdownSignalReceived;
+			LOG_TRACE("Unix signal handler is notifying for shutdown.");
+			programStatusCV.notify_all();
+			break;
+		}
+	}
+}
+
+/*
+ * Libbpf setup
+ */
+
 static int libbpfPrintFn(enum libbpf_print_level level, const char* format, va_list args) {
-#ifdef DEBUG
-	return vfprintf(stderr, format, args);
-#else
+	switch (level) {
+	case LIBBPF_WARN:
+		Logger::getInstance().vlogf(logging::LogLevel::Warn, format, args);
+		return 0;
+	case LIBBPF_INFO:
+		Logger::getInstance().vlogf(logging::LogLevel::Info, format, args);
+		return 0;
+	case LIBBPF_DEBUG:
+		Logger::getInstance().vlogf(logging::LogLevel::Debug, format, args);
+		return 0;
+	}
 	return 0;
-#endif
 }
 
-void setupLibbpf() {
+static void setupLibbpf() {
 	libbpf_set_print(libbpfPrintFn);
 }
 
 int main(int argc, char** argv) {
-	setupLibbpf();
-	setupUnixSignalHandlers();
+	po::options_description desc{getProgramOptions()};
+	po::variables_map vm;
+
+	try {
+		po::store(po::parse_command_line(argc, argv, desc), vm);
+		po::notify(vm);
+	} catch (const po::error& e) {
+		std::cout << e.what() << '\n';
+		return EXIT_FAILURE;
+	}
+
+	if (vm.count("help")) {
+		std::cout << desc;
+		return EXIT_SUCCESS;
+	}
+
+	if (vm.count("version")) {
+		std::cout << "eBPF-Discovery " << PROJECT_VERSION << '\n';
+		return EXIT_SUCCESS;
+	}
+
+	logging::LogLevel logLevel{vm["log-level"].as<logging::LogLevel>()};
+	bool isStdoutLogDisabled{vm["log-no-stdout"].as<bool>()};
+	std::filesystem::path logDir{vm["log-dir"].as<std::filesystem::path>()};
 
 	try {
-		discoveryInstance.load();
+		setupLogging(logLevel, !isStdoutLogDisabled, logDir);
 	} catch (const std::runtime_error& e) {
-		std::cerr << "Couldn't load BPF program: " << e.what() << std::endl;
+		std::cerr << "Couldn't setup logging: " << e.what() << '\n';
 		return EXIT_FAILURE;
 	}
 
-	if (discoveryInstance.run() != 0) {
+	LOG_DEBUG("Starting the program.");
+
+	{
+		LOG_TRACE("Setting up unix signals handling.");
+		sigset_t sigset = getSigset();
+		if (sigprocmask(SIG_BLOCK, &sigset, nullptr) == -1) {
+			LOG_CRITICAL("Failed to block unix signals: {}", std::strerror(errno));
+			return EXIT_FAILURE;
+		}
+	}
+
+	setupLibbpf();
+	ebpfdiscovery::DiscoveryBpfLoader loader;
+	try {
+		loader.load();
+	} catch (const std::runtime_error& e) {
+		LOG_CRITICAL("Couldn't load BPF program. ({})", e.what());
 		return EXIT_FAILURE;
 	}
 
+	ebpfdiscovery::Discovery instance(loader.get());
+	try {
+		instance.start();
+	} catch (const std::runtime_error& e) {
+		LOG_CRITICAL("Couldn't start Discovery: {}", e.what());
+	}
+
+	std::thread unixSignalThread(runUnixSignalHandlerLoop);
+	{
+		std::unique_lock<std::mutex> programStatusLock(programStatusMutex);
+		programStatusCV.wait(programStatusLock, []() { return programStatus != ProgramStatus::Running; });
+	}
+
+	LOG_DEBUG("Exiting the program.");
+	if (unixSignalThread.joinable()) {
+		unixSignalThread.join();
+	}
+	instance.stop();
+	instance.wait();
+
 	return EXIT_SUCCESS;
 }

libebpfdiscovery/CMakeLists.txt

@@ -3,6 +3,8 @@ list(
 	SOURCES
 	src/Config.cpp
 	src/Discovery.cpp
+	src/DiscoveryBpf.cpp
+	src/DiscoveryBpfLoader.cpp
 	src/IpAddressChecker.cpp
 	src/NetlinkCalls.cpp
 	src/Session.cpp
@@ -15,11 +17,13 @@ add_library(${TARGET} STATIC ${SOURCES})
 target_include_directories(${TARGET} PRIVATE src PUBLIC headers)
 
 target_link_libraries(${TARGET} bpfload)
+target_link_libraries(${TARGET} ebpfdiscoveryproto)
 target_link_libraries(${TARGET} ebpfdiscoveryshared)
 target_link_libraries(${TARGET} ebpfdiscoveryskel)
 target_link_libraries(${TARGET} fmt::fmt)
 target_link_libraries(${TARGET} httpparser)
-target_link_libraries(${TARGET} ebpfdiscoveryproto)
+target_link_libraries(${TARGET} logging)
+target_link_libraries(${TARGET} spdlog::spdlog)
 
 if(BUILD_TESTS)
 	list(APPEND TEST_SOURCES test/StringFunctionsTest.cpp test/LRUCacheTest.cpp test/IpAddressCheckerTest.cpp)

libebpfdiscovery/headers/ebpfdiscovery/Discovery.h

@@ -2,17 +2,19 @@
 #pragma once
 
 #include "ebpfdiscovery/Config.h"
+#include "ebpfdiscovery/DiscoveryBpf.h"
 #include "ebpfdiscovery/LRUCache.h"
 #include "ebpfdiscovery/Session.h"
 #include "ebpfdiscoveryshared/Types.h"
 #include "httpparser/HttpRequestParser.h"
 
-#include "discovery.skel.h"
-
 #include <atomic>
 #include <chrono>
+#include <condition_variable>
+#include <mutex>
 #include <queue>
 #include <string>
+#include <thread>
 #include <unordered_map>
 
 namespace ebpfdiscovery {
@@ -21,23 +23,23 @@ using httpparser::HttpRequestParser;
 
 class Discovery {
 public:
-	Discovery();
-	Discovery(const DiscoveryConfig config);
-	~Discovery();
-
-	bool isLoaded() noexcept;
-	void load();
-	void unload() noexcept;
-
-	// Blocks current thread until stopRun() is called
-	int run();
-
-	// Thread safe operation
-	void stopRun();
+	Discovery(DiscoveryBpf discoveryBpf);
+	Discovery(DiscoveryBpf discoveryBpf, const DiscoveryConfig config);
+	Discovery(const Discovery&) = delete;
+	Discovery& operator=(const Discovery&) = delete;
+	Discovery(Discovery&&) = default;
+	Discovery& operator=(Discovery&&) = default;
+	~Discovery() = default;
+
+	void start();
+	void stop();
+	void wait();
 
 private:
 	typedef LRUCache<DiscoverySavedSessionKey, Session, DiscoverySavedSessionKeyHash> SavedSessionsCacheType;
 
+	void run();
+
 	void fetchEvents();
 	void saveSession(const DiscoverySavedSessionKey& session_key, const Session& session);
 
@@ -47,20 +49,27 @@ class Discovery {
 	void handleBufferLookupSuccess(DiscoverySavedBuffer& savedBuffer, DiscoveryEvent& event);
 	void handleExistingSession(SavedSessionsCacheType::iterator it, std::string_view& bufferView, DiscoveryEvent& event);
 	void handleNewSession(std::string_view& bufferView, DiscoveryEvent& event);
+	void handleNewRequest(const Session& session, const DiscoverySessionMeta& meta);
 	void handleCloseEvent(DiscoveryEvent& event);
 	void handleSuccessfulParse(const Session& session, const DiscoverySessionMeta& sessionMeta);
 
 	int bpfDiscoveryResetConfig();
 	int bpfDiscoveryResumeCollecting();
 	int bpfDiscoveryDeleteSession(const DiscoveryTrackedSessionKey& trackedSessionKey);
 
-	DiscoveryConfig config;
+	constexpr auto discoverySkel() const {
+		return discoveryBpf.skel;
+	}
 
-	std::atomic<bool> running;
-	std::atomic<bool> loaded;
-	discovery_bpf* discoverySkel;
-	bpf_object_open_opts discoverySkelOpenOpts;
+	DiscoveryConfig config;
+	DiscoveryBpf discoveryBpf;
 	SavedSessionsCacheType savedSessions;
+
+	std::atomic<bool> running{false};
+	bool stopReceived{false};
+	std::condition_variable stopReceivedCV;
+	std::mutex stopReceivedMutex;
+	std::thread workerThread;
 };
 
 } // namespace ebpfdiscovery

libebpfdiscovery/headers/ebpfdiscovery/DiscoveryBpf.h

@@ -0,0 +1,20 @@
+// SPDX-License-Identifier: Apache-2.0
+#pragma once
+
+#include "discovery.skel.h"
+
+namespace ebpfdiscovery {
+
+class DiscoveryBpf {
+public:
+	DiscoveryBpf(discovery_bpf* skel);
+	DiscoveryBpf(const DiscoveryBpf&) = default;
+	DiscoveryBpf& operator=(const DiscoveryBpf&) = default;
+	DiscoveryBpf(DiscoveryBpf&&) = default;
+	DiscoveryBpf& operator=(DiscoveryBpf&&) = default;
+	~DiscoveryBpf() = default;
+
+	discovery_bpf* skel;
+};
+
+} // namespace ebpfdiscovery